Feat/improved autofocuser

docs/source/user_guide/operation.md

@@ -66,6 +66,14 @@ Run the [autofocus action](scheduling.md#autofocus-action). Ensure the `filter`
 
 Upon successful completion, the optimized focus position is updated in the configuration. Autofocus images and V-curve plot are saved to the `images/autofocus` directory.
 
+```{admonition} Focusing Tip
+:class: tip
+
+**Coarse Search (e.g., `fft`):** Best for broad ranges where stars appear as blurry "donuts." These non-parametric operators measure overall frame sharpness without needing to identify individual stars, making them nearly impossible to "confuse" with distorted optics.
+
+**Fine Tuning (e.g., `HFR`):** Best for precision near the focus peak. These algorithms fit a "V-curve" to the diameters of detected stars. They provide great accuracy once stars are point-like, but will fail during wide searches if the stars are too bloated to be recognized by the star-finder.
+```
+
 <!-- **3. Optional: Pointing model**
 Once focused, build a pointing model. Each pointing is plate solved and sends SyncToCoordinates commands to the mount. The receipt of these commands can be used to build a pointing model in the mount control software. _Astra_ itself does not build or maintain a pointing model.
 

pyproject.toml

@@ -39,7 +39,7 @@ classifiers = [
   "Programming Language :: Python :: 3"
 ]
 dependencies = [
-  "astrafocus>=0.1.1",
+  "astrafocus>=0.1.3",
   "astropy",
   "photutils",
   "scipy",

src/astra/action_configs.py

@@ -1041,6 +1041,15 @@ class AutofocusConfig(BaseActionConfig):
         4. Measure star sharpness in each image
         5. Fit a curve to determine optimal focus
         6. Save plots/results and save the best focus position in the observatory configuration
+
+    Note:
+        Coarse searches (for example `fft`, `normalized_variance`) use non-parametric
+        focus measures that characterise overall frame sharpness and are well suited
+        for very broad search ranges where stars appear as large, defocused "donuts".
+        Analytic response-function autofocusers (for example `HFR`/StarSize), which fit
+        a V-curve to measured star sizes, are better for fine-tuning near the focus
+        peak but can give incorrect results if applied over an excessively large range
+        because the assumed response model may not fit across the whole span.
     """
 
     exptime: float | int = field(default=3.0)
@@ -1095,6 +1104,10 @@ class AutofocusConfig(BaseActionConfig):
         "fwhm": "DAOStarFinder FWHM of the Gaussian kernel in pixels.",
         "percent_to_cut": "Percentage of worst-performing focus samples to drop when shrinking the range.",
         "focus_measure_operator": "Focus metric to optimize (e.g., hfr, gauss, tenengrad, fft, normalized_variance).",
+        "focus_measure_operator_note": (
+            "Prefer non-parametric metrics (e.g., 'fft', 'normalized_variance') for coarse/broad searches; "
+            "use analytic measures (e.g., 'HFR') for fine tuning near the focus peak."
+        ),
         "reduce_exposure_time": "Automatically shorten exposures to prevent saturation.",
         "save": "Persist the optimal focus position back into observatory configuration.",
         "extremum_estimator": "Curve-fitting method used to determine the minimum (LOWESS, medianfilter, spline, rbf).",
@@ -1113,6 +1126,7 @@ class AutofocusConfig(BaseActionConfig):
         "action_value": {
             "exptime": 1.0,
             "filter": "V",
+            "focus_measure_operator": "HFR",
             "search_range_is_relative": True,
             "search_range": 1000,
             "n_steps": [30, 20],

src/astra/autofocus.py

@@ -25,7 +25,6 @@
 import matplotlib
 import matplotlib.pyplot as plt
 import numpy as np
-import pandas as pd
 from astrafocus import ExtremumEstimatorRegistry, FocusMeasureOperatorRegistry
 from astrafocus.autofocuser import (
     AnalyticResponseAutofocuser,
@@ -563,8 +562,10 @@ def __init__(
         self.paired_devices = paired_devices
         self.action_value = action.action_value
         self.autofocuser = autofocuser
+        self.error_message: str | None = None
         self.success = success
         self.config: AutofocusConfig = action.action_value  # type: ignore
+        self._run_timestamp: str | None = None
 
         if (
             self.config.calibration_field.fov_width == 0
@@ -689,7 +690,6 @@ def _setup(self) -> None:
             n_exposures=self.config.n_exposures,
             decrease_search_range=self.config.decrease_search_range,
             exposure_time=self.config.exptime,
-            # save_path=self.config.save_path,
             secondary_focus_measure_operators=self.config._secondary_focus_measure_operators,
             focus_measure_operator_kwargs=self.config.focus_measure_operator_kwargs,
             search_range_is_relative=self.config.search_range_is_relative,
@@ -959,6 +959,17 @@ def reduce_exposure_time(
 
         return new_exposure_time
 
+    @property
+    def run_timestamp(self) -> str:
+        """Timestamp shared across all output files for a single run."""
+        if self._run_timestamp is None:
+            self._run_timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        return self._run_timestamp
+
+    def _output_path(self, save_dir: Path, suffix: str, ext: str) -> Path:
+        save_dir.mkdir(parents=True, exist_ok=True)
+        return save_dir / f"autofocus_{self.run_timestamp}_{suffix}.{ext}"
+
     def make_summary_plot(self) -> None:
         """Create visualization plot of autofocus results.
 
@@ -985,41 +996,39 @@ def make_summary_plot(self) -> None:
                     )
                     last_image_path = getattr(image_handler, "last_image_path", None)
                 except Exception:
+                    self.observatory.logger.warning(
+                        "Unable to determine last image path from image handler. "
+                        "No summary plot will be saved."
+                    )
                     last_image_path = None
 
                 if last_image_path is None:
                     self.observatory.logger.warning(
-                        "Skipping creation of summary plot: no save_path configured and no last image available."
+                        "Skipping creation of autofocus summary plot: "
+                        "unable to determine save directory."
                     )
                     return
 
                 save_dir = last_image_path.parent
 
-            # Obtain focus record dataframe. Prefer in-memory record from the
-            # astrafocus autofocuser instance; if not available, try reading CSVs
-            # from the chosen directory.
-            df: pd.DataFrame | None = None
             if (
                 hasattr(self, "autofocuser")
                 and getattr(self.autofocuser, "focus_record", None) is not None
             ):
                 try:
                     df = self.autofocuser.focus_record
                 except Exception:
-                    df = None
-
-            if df is None:
-                assert save_dir is not None
-                csv_files = sorted(
-                    [p for p in Path(save_dir).iterdir() if p.suffix == ".csv"],
-                    key=lambda p: p.stat().st_mtime,
-                )
-                if not csv_files:
-                    self.observatory.logger.error(
-                        f"No focus record CSV found in {save_dir}. Skipping summary plot."
+                    self.observatory.logger.warning(
+                        "Skipping creation of autofocus summary plot: "
+                        "unable to retrieve focus record from autofocuser instance."
                     )
                     return
-                df = pd.read_csv(csv_files[-1])
+            else:
+                self.observatory.logger.warning(
+                    "Skipping creation of autofocus summary plot: "
+                    "autofocuser instance does not have a focus_record attribute."
+                )
+                return
 
             df = df.sort_values("focus_pos")
 
@@ -1040,18 +1049,12 @@ def make_summary_plot(self) -> None:
             )
             ax.legend()
 
-            # Build output filename: if we read a CSV file use its stem, otherwise timestamp it.
-            if "csv_files" in locals() and csv_files:
-                out_name = f"{csv_files[-1].stem}.png"
-            else:
-                out_name = (
-                    f"autofocus_summary_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
-                )
-
             assert save_dir is not None
-            out_path = Path(save_dir) / out_name
-            plt.savefig(out_path)
+            plt.savefig(self._output_path(save_dir, "summary", "png"))
             plt.close()
+
+            df.to_csv(self._output_path(save_dir, "record", "csv"), index=False)
+
         except Exception as e:
             self.observatory.logger.exception(f"Error creating summary plot: {str(e)}")
 
@@ -1073,31 +1076,23 @@ def create_result_file(self) -> None:
                     self.action.device_name
                 )
                 last_image_path = getattr(image_handler, "last_image_path", None)
-            except Exception:
+            except Exception as e:
+                self.observatory.logger.error(
+                    f"Error occurred while fetching last image path: {str(e)}"
+                )
                 last_image_path = None
 
             if last_image_path is None:
                 self.observatory.logger.error(
-                    "Skipping creation of log file: no save_path configured and no last image available."
+                    "Skipping creation of result file: "
+                    "no save_path configured and no last image available."
                 )
                 return
 
             save_dir = last_image_path.parent
 
-        # derive a timestring for filename; prefer an adjacent CSV if present
-        timestr = None
         assert save_dir is not None
-        csv_files = sorted(
-            [p for p in Path(save_dir).iterdir() if p.suffix == ".csv"],
-            key=lambda p: p.stat().st_mtime,
-        )
-        if csv_files:
-            timestr = csv_files[-1].stem.split("_")[0]
-
-        if not timestr:
-            timestr = datetime.now().strftime("%Y%m%d_%H%M%S")
-
-        result_file_path = Path(save_dir) / f"{timestr}_result.txt"
+        result_file_path = self._output_path(save_dir, "result", "txt")
         try:
             with open(result_file_path, "w") as result_file:
                 result_file.write(f"Best focus position: {self.best_focus_position}\n")
@@ -1106,7 +1101,7 @@ def create_result_file(self) -> None:
                 )
                 result_file.write(f"Autofocuser: {self.autofocuser}\n")
         except Exception as e:
-            self.observatory.logger.exception(f"Error creating log file: {str(e)}")
+            self.observatory.logger.exception(f"Error creating result file: {str(e)}")
 
     def _initialise_logging(self) -> None:
         """Set up logging integration with astrafocus library.
@@ -1217,18 +1212,23 @@ def calculate_field_of_view(self, paired_devices):
 
         except Exception as e:
             field_of_view = np.array([np.nan, np.nan])
-            self.observatory.error(
+            self.observatory.logger.error(
                 f"Error calculating field of view from paired devices. Exception: {e}"
             )
 
         return field_of_view
 
     def determine_default_field_of_view(self, paired_devices):
-        field_of_view = self.calculate_field_of_view(paired_devices)
-        fov_width = float(field_of_view[0])
-        fov_height = float(field_of_view[1])
+        try:
+            field_of_view = self.calculate_field_of_view(paired_devices)
+            fov_width = float(field_of_view[0])
+            fov_height = float(field_of_view[1])
 
-        self.observatory.logger.info(
-            f"Determined field of view width={fov_width}, height={fov_height}."
-        )
-        return fov_width, fov_height
+            self.observatory.logger.info(
+                f"Determined field of view width={fov_width}, height={fov_height}."
+            )
+            return fov_width, fov_height
+        except Exception as e:
+            raise ValueError(
+                f"Error determining default field of view from paired devices: {str(e)}"
+            )

src/astra/observatory.py

@@ -183,6 +183,8 @@ def __init__(
         # log start up
         self.logger.debug("Database and DatabaseLoggingHandler initialized")
         self.logger.info(f"Starting observatory {self.name}")
+        if type(self) is not Observatory:
+            self.logger.info(f"Using observatory subclass: {type(self).__name__}")
 
         # warn if debug mode
         if self.logger.getEffectiveLevel() == logging.DEBUG:
@@ -2859,11 +2861,17 @@ def autofocus_sequence(self, action: Action, paired_devices: PairedDevices) -> b
             if not self.check_conditions(action=action):
                 return False
 
-            autofocuser = Autofocuser(
-                observatory=self,
-                action=action,
-                paired_devices=paired_devices,
-            )
+            try:
+                autofocuser = Autofocuser(
+                    observatory=self,
+                    action=action,
+                    paired_devices=paired_devices,
+                )
+            except Exception as e:
+                self.logger.warning(
+                    f"Autofocuser initialization failed for {action.device_name}: {e}"
+                )
+                return False
             autofocuser.determine_autofocus_calibration_field()
             autofocuser.slew_to_calibration_field()
             autofocuser.setup()