Analog input can now accept multiple triggers

Author: alustig3

devices/rotary_encoder.py

@@ -9,9 +9,7 @@ def __init__(
         name,
         sampling_rate,
         output="velocity",
-        threshold=None,
-        rising_event=None,
-        falling_event=None,
+        triggers=None,
         bytes_per_sample=2,
         reverse=False,
     ):
@@ -28,14 +26,13 @@ def __init__(
         self.position = 0
         self.velocity = 0
         self.sampling_rate = sampling_rate
+
         Analog_input.__init__(
             self,
             None,
             name,
             int(sampling_rate),
-            threshold,
-            rising_event,
-            falling_event,
+            triggers,
             data_type={2: "h", 4: "i"}[bytes_per_sample],
         )
 

source/pyControl/hardware.py

@@ -235,14 +235,8 @@ class Analog_input(IO_object):
     # streams data to computer. Optionally can generate framework events when voltage
     #  goes above / below specified value theshold.
 
-    def __init__(self, pin, name, sampling_rate, threshold=None, rising_event=None, falling_event=None, data_type="H"):
-        if rising_event or falling_event:
-            if threshold is None:
-                raise ValueError("A threshold must be specified if rising or falling events are defined.")
-            self.threshold_watcher = Analog_threshold_watcher(threshold, rising_event, falling_event)
-        else:
-            self.threshold_watcher = False
-
+    def __init__(self, pin, name, sampling_rate, triggers=None, data_type="H"):
+        self.triggers = triggers
         self.timer = pyb.Timer(available_timers.pop())
         if pin:  # pin argument can be None when Analog_input subclassed.
             self.ADC = pyb.ADC(pin)
@@ -255,8 +249,6 @@ def _run_start(self):
         # Start sampling timer, initialise threshold, aquire first sample.
         self.timer.init(freq=self.channel.sampling_rate)
         self.timer.callback(self._timer_ISR)
-        if self.threshold_watcher:
-            self.threshold_watcher.run_start(self.read_sample())
         self._timer_ISR(0)
 
     def _run_stop(self):
@@ -267,11 +259,9 @@ def _timer_ISR(self, t):
         # Read a sample to the buffer, update write index.
         sample = self.read_sample()
         self.channel.put(sample)
-        if self.threshold_watcher:
-            self.threshold_watcher.check(sample)
-
-    def change_threshold(self, new_threshold):
-        self.threshold_watcher.set_threshold(new_threshold)
+        if self.triggers:
+            for trigger in self.triggers:
+                trigger.check(sample)
 
     def record(self):  # For backward compatibility.
         pass
@@ -356,33 +346,89 @@ def send_buffer(self, run_stop=False):
             fw.usb_serial.send(self.buffers[buffer_n])
 
 
+class ValueTrigger(IO_object):
+    # Generates framework events when an analog signal goes above or below specified threshold value.
+
+    def __init__(self, threshold, rising_event=None, falling_event=None):
+        if rising_event is None and falling_event is None:
+            raise ValueError("Either rising_event or falling_event or both must be specified.")
+        self.rising_event = rising_event
+        self.falling_event = falling_event
+        self.set_threshold(threshold)
+        self.timestamp = 0
+        self.crossing_direction = False
+        assign_ID(self)
+
+    def _initialise(self):
+        # Set event codes for rising and falling events.
+        self.rising_event_ID = sm.events[self.rising_event] if self.rising_event in sm.events else False
+        self.falling_event_ID = sm.events[self.falling_event] if self.falling_event in sm.events else False
+        self.threshold_active = self.rising_event_ID or self.falling_event_ID
+
+    def _process_interrupt(self):
+        # Put event generated by threshold crossing in event queue.
+        if self.crossing_direction:
+            fw.event_queue.put(fw.Datatuple(self.timestamp, fw.EVENT_TYP, "i", self.rising_event_ID))
+        else:
+            fw.event_queue.put(fw.Datatuple(self.timestamp, fw.EVENT_TYP, "i", self.falling_event_ID))
+
+    @micropython.native
+    def check(self, sample):
+        if self.reset_above_threshold:
+            # this gets run when the first sample is taken and whenever the threshold is changed
+            self.reset_above_threshold = False
+            self.above_threshold = sample > self.threshold
+            return
+        new_above_threshold = sample > self.threshold
+        if new_above_threshold != self.above_threshold:  # Threshold crossing.
+            self.above_threshold = new_above_threshold
+            if (self.above_threshold and self.rising_event_ID) or (not self.above_threshold and self.falling_event_ID):
+                self.timestamp = fw.current_time
+                self.crossing_direction = self.above_threshold
+
+                interrupt_queue.put(self.ID)
+
+    def set_threshold(self, threshold):
+        if not isinstance(threshold, int):
+            raise ValueError(f"Threshold must be an integer, got {type(threshold).__name__}.")
+        self.threshold = threshold
+        self.reset_above_threshold = True
+
+
 class Crossing:
     above = "above"
     below = "below"
     none = "none"
 
 
-class Analog_threshold_watcher(IO_object):
+class SchmittTrigger(IO_object):
     """
-    Generates framework events when an analog signal goes above or below specified threshold.
-    If given single threshold value, rising event is triggered when signal > threshold, falling event is triggered when signal <= threshold.
-    If given tuple of two threshold values, rising event is triggered when signal > upper bound, falling event is triggered when signal < lower bound.
+    Generates framework events when an analog signal goes above an upper threshold and/or below a lower threshold.
+    The rising event is triggered when signal > upper bound, falling event is triggered when signal < lower bound.
+
+    This trigger implements hysteresis, which is a technique to prevent rapid oscillations or "bouncing" of events:
+    - Hysteresis creates a "dead zone" between the upper and lower thresholds
+    - Once a rising event is triggered (when signal crosses above the upper bound),
+      it cannot be triggered again until the signal has fallen below the lower bound
+    - Similarly, once a falling event is triggered (when signal crosses below the lower bound),
+      it cannot be triggered again until the signal has risen above the upper bound
+
+    This behavior is particularly useful for noisy signals that might otherwise rapidly cross a single threshold
+    multiple times, generating unwanted repeated events.
     """
 
-    def __init__(self, threshold, rising_event=None, falling_event=None):
-        self.set_threshold(threshold)
+    def __init__(self, bounds, rising_event=None, falling_event=None):
+        if rising_event is None and falling_event is None:
+            raise ValueError("Either rising_event or falling_event or both must be specified.")
+        self.set_bounds(bounds)
         self.rising_event = rising_event
         self.falling_event = falling_event
         self.timestamp = 0
-        self.last_crossing = Crossing.none
         assign_ID(self)
 
-    def set_threshold(self, threshold):
-        if isinstance(threshold, int):  # single threshold value
-            self.upper_threshold = threshold
-            self.lower_threshold = threshold + 1  # +1 so falling event is triggered when crossing into <= threshold
-        elif isinstance(threshold, tuple):
-            threshold_requirements_str = "The threshold must be a single integer or a tuple of two integers (lower_bound, upper_bound) where lower_bound <= upper_bound."
+    def set_bounds(self, threshold):
+        if isinstance(threshold, tuple):
+            threshold_requirements_str = "The threshold must be a tuple of two integers (lower_bound, upper_bound) where lower_bound <= upper_bound."
             if len(threshold) != 2:
                 raise ValueError("{} is not a valid threshold. {}".format(threshold, threshold_requirements_str))
             lower, upper = threshold
@@ -396,17 +442,14 @@ def set_threshold(self, threshold):
             self.lower_threshold = lower
         else:
             raise ValueError("{} is not a valid threshold. {}".format(threshold, threshold_requirements_str))
+        self.reset_crossing = True
 
     def _initialise(self):
         # Set event codes for rising and falling events.
         self.rising_event_ID = sm.events[self.rising_event] if self.rising_event in sm.events else False
         self.falling_event_ID = sm.events[self.falling_event] if self.falling_event in sm.events else False
         self.threshold_active = self.rising_event_ID or self.falling_event_ID
 
-    def run_start(self, sample):
-        self.was_above = sample > self.upper_threshold
-        self.was_below = sample < self.lower_threshold
-
     def _process_interrupt(self):
         # Put event generated by threshold crossing in event queue.
         if self.was_above:
@@ -416,6 +459,13 @@ def _process_interrupt(self):
 
     @micropython.native
     def check(self, sample):
+        if self.reset_crossing:
+            # this gets run when the first sample is taken and whenever the threshold is changed
+            self.reset_crossing = False
+            self.was_above = sample > self.upper_threshold
+            self.was_below = sample < self.lower_threshold
+            self.last_crossing = Crossing.none
+            return
         is_above_threshold = sample > self.upper_threshold
         is_below_threshold = sample < self.lower_threshold