Feat: dynamic frequency

scanner.py

@@ -30,6 +30,7 @@
 import msgproto
 import numpy as np
 import pins
+from collections import deque
 from clocksync import SecondarySync
 from configfile import ConfigWrapper
 from gcode import GCodeCommand
@@ -60,6 +61,10 @@
 THRESHOLD_STEP_MULTIPLIER = 10
 # Require a qualified threshold to pass at 0.66 of the QUALIFY_SAMPLES
 THRESHOLD_ACCEPTANCE_FACTOR = 0.66
+SLIDING_WINDOW_SIZE = 50
+SIGMA_MULTIPLIER = 3
+THRESHOLD_MULTIPLIER = 1.2
+SHORTED_COIL_VALUE = 0xFFFFFFF
 
 
 class TriggerMethod(IntEnum):
@@ -170,7 +175,7 @@
                 )
 
         atypes = {"median": "median", "average": "average"}
         self.samples_config = {
             "samples": config.getfloat("samples", 5, above=0.0),
             "retract_dist": config.getfloat("samples_retract_dist", 5, above=0.0),
             "tolerance": config.getfloat("samples_tolerance", 0.2, minval=0.0),
@@ -215,7 +220,7 @@
             "scanner_touch_max_speed", 10, above=0, maxval=30
         )
         ## NEW VARIABLES HERE
         self.scanner_touch_config = {
             "accel": config.getfloat("scanner_touch_accel", 100, above=0, minval=100),
             "max_speed": max_speed_value,
             "speed": config.getfloat("scanner_touch_speed", 3, maxval=max_speed_value),
@@ -241,7 +246,7 @@
             raise self.printer.command_error(
                 "Please change detect_threshold_z to scanner_touch_threshold in printer.cfg"
             )
         self.detect_threshold_z = self.scanner_touch_config["threshold"]
         self.previous_probe_success = None
 
         self.cal_config = {
@@ -278,7 +283,7 @@
         self._stream_buffer = []
         self._stream_buffer_limit = STREAM_BUFFER_LIMIT_DEFAULT
         self._stream_buffer_limit_new = self._stream_buffer_limit
         self._stream_samples_queue = queue.Queue()
         self._stream_flush_event = threading.Event()
         self._log_stream = None
         self._data_filter = AlphaBetaFilter(
@@ -314,7 +319,7 @@
         )
         self._mcu.register_config_callback(self._build_config)
         self._mcu.register_response(
             self._handle_scanner_data, self.sensor.lower() + "_data"
         )
         # Register webhooks
         webhooks = self.printer.lookup_object("webhooks")
@@ -400,11 +405,11 @@
             self.calibration_method = "scan"
             self._start_calibration(gcmd)
 
     def _get_common_variables(self, gcmd: GCodeCommand):
         return {
             "speed": gcmd.get_float(
                 "SPEED",
                 self.scanner_touch_config["speed"],
                 above=0,
                 maxval=self.scanner_touch_config["max_speed"],
             ),
@@ -1714,23 +1719,74 @@
         # Streaming mode
 
     def _check_hardware(self, sample):
+        # Validate sample input
+        if "data" not in sample or "freq" not in sample:
+            raise self._mcu.error("Sample must contain 'data' and 'freq' keys.")
+
+        # Initialize variables on the first call
+        if not hasattr(self, "freq_window"):
+            self.freq_window = deque(maxlen=SLIDING_WINDOW_SIZE)  # Sliding window
+            self.min_threshold = None  # Minimum frequency threshold
+
+        # Add the current frequency to the sliding window
+        freq = sample["freq"]
+        self.freq_window.append(freq)
+
+        # Calculate statistics from the sliding window
+        if len(self.freq_window) > 1:
+            freq_window_array = np.array(self.freq_window)  # Convert deque to numpy array
+            f_avg = np.mean(freq_window_array)
+            f_std = np.std(freq_window_array)
+            dynamic_threshold = f_avg + SIGMA_MULTIPLIER * f_std  # Dynamic threshold
+        else:
+            # Fallback during initialization
+            f_avg = freq
+            f_std = 0
+            dynamic_threshold = freq * THRESHOLD_MULTIPLIER  # Fallback threshold
+
+
+        # Ensure a minimum threshold is set
+        if self.min_threshold is None:
+            self.min_threshold = freq * THRESHOLD_MULTIPLIER  # Initial minimum threshold
+
+        # Final threshold (whichever is greater: dynamic or minimum)
+        final_threshold = max(dynamic_threshold, self.min_threshold)
+
+        # Debug log for threshold values
+        logging.debug(
+            f"Sliding Window Threshold Debug: freq={freq}, f_avg={f_avg}, "
+            f"f_std={f_std}, dynamic_threshold={dynamic_threshold}, "
+            f"min_threshold={self.min_threshold}, final_threshold={final_threshold}"
+        )
+
+        # Check for hardware issues
         if not self.hardware_failure:
             msg = None
-            if sample["data"] == 0xFFFFFFF:
-                msg = "coil is shorted or not connected"
-            elif self.fmin is not None and sample["freq"] > 1.35 * self.fmin:
-                msg = "coil expected max frequency exceeded"
+
+            if sample["data"] == SHORTED_COIL_VALUE:
+                msg = "Coil is shorted or not connected."
+                logging.debug(f"Debug: data={sample['data']} indicates connection issue.")
+            elif freq > final_threshold:
+                msg = "Coil expected max frequency exceeded (sliding window)."
+                logging.debug(
+                    f"Frequency {freq} exceeded final threshold {final_threshold}."
+                )
+
             if msg:
-                msg = "Scanner hardware issue: " + msg
-                self.hardware_failure = msg
-                logging.error(msg)
+                # Log and handle hardware failure
+                full_msg = f"Scanner hardware issue: {msg}"
+                self.hardware_failure = full_msg
+                logging.error(full_msg)
+
                 if self._stream_en:
-                    self.printer.invoke_shutdown(msg)
+                    self.printer.invoke_shutdown(full_msg)
                 else:
-                    self.gcode.respond_raw("!! " + msg + "\n")
+                    self.gcode.respond_raw(f"!! {full_msg}\n")
         elif self._stream_en:
+            # Handle already detected hardware failure
             self.printer.invoke_shutdown(self.hardware_failure)
 
+
     def _enrich_sample_time(self, sample):
         clock = sample["clock"] = self._mcu.clock32_to_clock64(sample["clock"])
         sample["time"] = self._mcu.clock_to_print_time(clock)
@@ -3252,7 +3308,7 @@
             pa = (begin_a, pos_p) if even else (end_a, pos_p)
             pb = (end_a, pos_p) if even else (begin_a, pos_p)
 
             l = (pa, pb)
 
             if len(points) > 0 and corner_radius > 0:
                 # We need to insert an overscan corner. Basically we insert
@@ -3846,4 +3902,4 @@
         scanner.register_model(name, model)
         return model
     else:
-        raise config.error("Unknown scanner config directive '%s'" % (name[7:],))
+        raise config.error("Unknown scanner config directive '%s'" % (name[7:],))

typings/mcu.pyi

@@ -16,6 +16,9 @@ class MCUStatus(TypedDict):
 class _CommandQueue:
     pass
 
+class error(Exception):
+    pass
+
 class MCU:
     _mcu_freq: float
     _clocksync: ClockSync
@@ -59,6 +62,8 @@ class MCU:
         pass
     def is_fileoutput(self) -> bool:
         pass
+    def error(self, message: str) -> string:
+        pass
 
     @overload
     def get_constant(self, name: str, default: type[sentinel] | str = sentinel) -> str: