webserver: Upload and choose log configuration (#404)

* webserver: Upload and choose log configuration
* Silence mypy

Author: jscheidtmann

.gitignore

@@ -1,3 +1,7 @@
+# Active log configuration symlink (auto-created at startup)
+python/pifinder_logconf.json
+python/logconf_*.json
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

python/PiFinder/catalog_base.py

@@ -171,7 +171,7 @@ def assign_virtual_object_ids(catalog, low_id: int) -> int:
 class TimerMixin:
     """Provides timer functionality via composition"""
 
-    def __init__(self):
+    def __init__(self) -> None:
         self.timer: Optional[threading.Timer] = None
         self.is_running: bool = False
         self.time_delay_seconds: Union[int, Callable[[], int]] = (

python/PiFinder/catalog_imports/database.py

@@ -4,11 +4,16 @@
 This module provides centralized access to database objects for all catalog loaders.
 """
 
+from typing import Optional
+
+from PiFinder.db.objects_db import ObjectsDatabase
+from PiFinder.db.observations_db import ObservationsDatabase
+
 from .catalog_import_utils import init_databases
 
 # Global database objects shared across all catalog loaders
-objects_db = None
-observations_db = None
+objects_db: Optional[ObjectsDatabase] = None
+observations_db: Optional[ObservationsDatabase] = None
 
 
 def init_shared_database():

python/PiFinder/catalog_imports/harris_loader.py

@@ -381,10 +381,13 @@ def create_cluster_object(entry: npt.NDArray, seq: int) -> Dict[str, Any]:
     return result
 
 
-def load_harris():
+def load_harris() -> None:
     logging.info("Loading Harris Globular Cluster catalog")
     catalog: str = "Har"
     obj_type: str = "Gb"  # Globular Cluster
+    
+    if objects_db is None:
+        raise RuntimeError("Database not initialized. Call init_shared_database() first.")
     conn, _ = objects_db.get_conn_cursor()
 
     # Enable bulk mode to prevent commits during insert operations

python/PiFinder/main.py

@@ -870,6 +870,11 @@ def main(
 if __name__ == "__main__":
     import sys
 
+    # Ensure the active log config symlink exists, defaulting to logconf_default.json
+    _logconf_link = Path("pifinder_logconf.json")
+    if not _logconf_link.exists():
+        _logconf_link.symlink_to("logconf_default.json")
+
     debug_no_file_logs = "--debug-no-file-logs" in sys.argv
     if debug_no_file_logs:
         os.environ["PIFINDER_DEBUG_NO_FILE_LOGS"] = "1"

python/PiFinder/server.py

@@ -844,6 +844,83 @@ def get_component_levels():
                 logging.error(f"Error reading log configuration: {e}")
                 return {"status": "error", "message": str(e)}
 
+        @app.route("/logs/configs")
+        @auth_required
+        def list_log_configs():
+            """Return all available logconf_*.json files with display names."""
+            import glob
+
+            configs = []
+            active = (
+                os.path.realpath("pifinder_logconf.json")
+                if os.path.exists("pifinder_logconf.json")
+                else None
+            )
+            for path in sorted(glob.glob("logconf_*.json")):
+                stem = path[len("logconf_") : -len(".json")]
+                display = stem.replace("_", " ").title()
+                configs.append(
+                    {
+                        "file": path,
+                        "name": display,
+                        "active": os.path.realpath(path) == active,
+                    }
+                )
+            return {"configs": configs}
+
+        @app.route("/logs/switch_config", method="post")
+        @auth_required
+        def switch_log_config():
+            """Atomically repoint pifinder_logconf.json to the chosen config, then restart."""
+            logconf_file = request.forms.get("logconf_file", "").strip()
+            if (
+                not logconf_file
+                or not logconf_file.startswith("logconf_")
+                or not logconf_file.endswith(".json")
+            ):
+                return {"status": "error", "message": "Invalid log config file name"}
+            if not os.path.exists(logconf_file):
+                return {
+                    "status": "error",
+                    "message": f"Log config file not found: {logconf_file}",
+                }
+            try:
+                link = "pifinder_logconf.json"
+                tmp = link + ".tmp"
+                os.symlink(logconf_file, tmp)
+                os.replace(tmp, link)
+                logger.info("Switched log config to %s", logconf_file)
+            except Exception as e:
+                logger.error("Failed to switch log config: %s", e)
+                return {"status": "error", "message": str(e)}
+            return template("restart_pifinder")
+
+        @app.route("/logs/upload_config", method="post")
+        @auth_required
+        def upload_log_config():
+            """Upload a new logconf_*.json file."""
+            upload = request.files.get("config_file")
+            if not upload:
+                return {"status": "error", "message": "No file provided"}
+            filename = upload.filename
+            if not filename.startswith("logconf_") or not filename.endswith(".json"):
+                return {
+                    "status": "error",
+                    "message": "File must be named logconf_<name>.json",
+                }
+            if os.path.exists(filename):
+                return {
+                    "status": "error",
+                    "message": f"File already exists: {filename}",
+                }
+            try:
+                upload.save(filename, overwrite=False)
+                logger.info("Uploaded log config: %s", filename)
+                return {"status": "ok", "file": filename}
+            except Exception as e:
+                logger.error("Failed to save uploaded log config: %s", e)
+                return {"status": "error", "message": str(e)}
+
         @app.route("/logs/download")
         @auth_required
         def download_logs():

python/PiFinder/sqm/save_sweep_metadata.py

@@ -80,7 +80,8 @@ def save_sweep_metadata(
     # Noise floor estimation details (from NoiseFloorEstimator)
     if noise_floor_details is not None:
         metadata["noise_floor_estimator"] = {
-            k: v for k, v in noise_floor_details.items()
+            k: v
+            for k, v in noise_floor_details.items()
             if k != "request_zero_sec_sample"  # Exclude internal flags
         }
         if camera_type is not None:

python/PiFinder/sqm/sqm.py

@@ -65,7 +65,9 @@ def _load_calibration(self) -> bool:
         )
 
         if not calibration_file.exists():
-            logger.debug(f"No calibration file found at {calibration_file}, using defaults")
+            logger.debug(
+                f"No calibration file found at {calibration_file}, using defaults"
+            )
             return False
 
         try:
@@ -193,7 +195,9 @@ def _measure_star_flux_with_local_background(
 
             # Check for saturation in aperture
             aperture_pixels = image_patch[aperture_mask]
-            max_aperture_pixel = np.max(aperture_pixels) if len(aperture_pixels) > 0 else 0
+            max_aperture_pixel = (
+                np.max(aperture_pixels) if len(aperture_pixels) > 0 else 0
+            )
 
             if max_aperture_pixel >= saturation_threshold:
                 # Mark saturated star with flux=-1 to be excluded from mzero calculation
@@ -202,7 +206,6 @@ def _measure_star_flux_with_local_background(
                 n_saturated += 1
                 continue
 
-
             # Total flux in aperture (includes background)
             total_flux = np.sum(aperture_pixels)
 
@@ -263,9 +266,7 @@ def _calculate_mzero(
         # Flux-weighted mean: brighter stars contribute more
         valid_mzeros_arr = np.array(valid_mzeros)
         valid_fluxes_arr = np.array(valid_fluxes)
-        weighted_mzero = float(
-            np.average(valid_mzeros_arr, weights=valid_fluxes_arr)
-        )
+        weighted_mzero = float(np.average(valid_mzeros_arr, weights=valid_fluxes_arr))
 
         return weighted_mzero, mzeros
 
@@ -540,7 +541,9 @@ def calculate(
         # Following ASTAP: zenith is reference point where extinction = 0
         # Only ADDITIONAL extinction below zenith is added: 0.28 * (airmass - 1)
         # This allows comparing measurements at different altitudes
-        extinction_for_altitude = self._atmospheric_extinction(altitude_deg)  # 0.28*(airmass-1)
+        extinction_for_altitude = self._atmospheric_extinction(
+            altitude_deg
+        )  # 0.28*(airmass-1)
 
         # Main SQM value: no extinction correction (raw measurement)
         sqm_final = sqm_uncorrected

python/PiFinder/state.py

@@ -238,7 +238,7 @@ def from_json(cls, json_str):
 
 
 class SharedStateObj:
-    def __init__(self):
+    def __init__(self) -> None:
         self.__power_state = 1  # 0 = sleep state, 1 = awake state
         # self.__solve_state
         # None = No solve attempted yet

python/PiFinder/ui/radec_entry.py

@@ -586,7 +586,7 @@ class LayoutConfig:
 class UIRADecEntry(UIModule):
     __title__ = _("RA/DEC Entry")
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
 
         self.callback = self.item_definition.get("callback")