Analog threshold improvements (#141)

* analog threshold improvements

- can now provide a upper and lower threshold
- raise an error if a rising or falling event is provided without a threshold
- raise errors if threshold(s) not provided correctly
- add ability to change threshold(s) after task has started

* Analog input can now accept multiple triggers

* automatically log trigger thresholds/bounds

* move schmitt trigger to separate device file

* change name back to Analog_threshold and make sure Analog_input() is backwards compatible

* make rotary_encoder backwards compatible

* add threshold:set type:subtype

* updated running_wheel.py example

- uses new syntax for adding trigger
- demonstrates ability to have multiple triggers

* task and run_start subtypes for threshold

* renamed Analog_threshold class to AnalogTrigger

* add information about what analog input the trigger is attached to

* updated running wheel example

Author: alustig3

devices/rotary_encoder.py

@@ -14,6 +14,7 @@ def __init__(
         falling_event=None,
         bytes_per_sample=2,
         reverse=False,
+        triggers=None,
     ):
         assert output in ("velocity", "position"), "ouput argument must be 'velocity' or 'position'."
         assert bytes_per_sample in (2, 4), "bytes_per_sample must be 2 or 4"
@@ -28,6 +29,7 @@ def __init__(
         self.position = 0
         self.velocity = 0
         self.sampling_rate = sampling_rate
+
         Analog_input.__init__(
             self,
             None,
@@ -37,6 +39,7 @@ def __init__(
             rising_event,
             falling_event,
             data_type={2: "h", 4: "i"}[bytes_per_sample],
+            triggers=triggers,
         )
 
     def read_sample(self):

devices/schmitt_trigger.py

@@ -0,0 +1,109 @@
+from pyControl.hardware import IO_object, assign_ID, interrupt_queue
+import pyControl.framework as fw
+import pyControl.state_machine as sm
+
+
+class Crossing:
+    above = "above"
+    below = "below"
+    none = "none"
+
+
+class SchmittTrigger(IO_object):
+    """
+    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, 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.rising_event = rising_event
+        self.falling_event = falling_event
+        self.bounds = bounds
+        self.timestamp = 0
+        assign_ID(self)
+
+    def run_start(self, attached_to):
+        self.attached_to = attached_to
+        self.set_bounds(self.bounds, run_start=True)
+
+    def set_bounds(self, threshold, run_start=False):
+        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
+            if not upper >= lower:
+                raise ValueError(
+                    "{} is not a valid threshold because the lower bound {} is greater than the upper bound {}. {}".format(
+                        threshold, lower, upper, threshold_requirements_str
+                    )
+                )
+            self.upper_threshold = upper
+            self.lower_threshold = lower
+        else:
+            raise ValueError("{} is not a valid threshold. {}".format(threshold, threshold_requirements_str))
+        self.reset_crossing = True
+
+        content = {"bounds": (self.lower_threshold, self.upper_threshold), "attached_to": self.attached_to}
+        if self.rising_event is not None:
+            content["rising_event"] = self.rising_event
+        if self.falling_event is not None:
+            content["falling_event"] = self.falling_event
+        fw.data_output_queue.put(
+            fw.Datatuple(
+                fw.current_time,
+                fw.THRSH_TYP,
+                "s" if run_start else "t",
+                str(content),
+            )
+        )
+
+    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.was_above:
+            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_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
+
+        if is_above_threshold and not self.was_above and self.last_crossing != Crossing.above:
+            self.timestamp = fw.current_time
+            self.last_crossing = Crossing.above
+            if self.rising_event_ID:
+                interrupt_queue.put(self.ID)
+        elif is_below_threshold and not self.was_below and self.last_crossing != Crossing.below:
+            self.timestamp = fw.current_time
+            self.last_crossing = Crossing.below
+            if self.falling_event_ID:
+                interrupt_queue.put(self.ID)
+
+        self.was_above, self.was_below = is_above_threshold, is_below_threshold

source/communication/data_logger.py

@@ -73,7 +73,7 @@ def write_info_line(self, subtype, content, time=0):
         self.data_file.write(self.tsv_row_str("info", time, subtype, content))
 
     def tsv_row_str(self, rtype, time, subtype="", content=""):
-        time_str = f"{time/1000:.3f}" if isinstance(time, int) else time
+        time_str = f"{time / 1000:.3f}" if isinstance(time, int) else time
         return f"{time_str}\t{rtype}\t{subtype}\t{content}\n"
 
     def copy_task_file(self, data_dir, tasks_dir, dir_name="task_files"):
@@ -140,6 +140,8 @@ def data_to_string(self, new_data, prettify=False, max_len=60):
                             var_str += f'\t\t\t"{var_name}": {var_value}\n'
                         var_str += "\t\t\t}"
                 data_string += self.tsv_row_str("variable", time, nd.subtype, content=var_str)
+            elif nd.type == MsgType.THRSH:  # Threshold
+                data_string += self.tsv_row_str("threshold", time, nd.subtype, content=nd.content)
             elif nd.type == MsgType.WARNG:  # Warning
                 data_string += self.tsv_row_str("warning", time, content=nd.content)
             elif nd.type in (MsgType.ERROR, MsgType.STOPF):  # Error or stop framework.

source/communication/message.py

@@ -16,6 +16,7 @@ class MsgType(Enum):
     ERROR = b"!!"  # Error
     STOPF = b"X"  # Stop framework
     ANLOG = b"A"  # Analog
+    THRSH = b"T"  # Threshold
 
     @classmethod
     def from_byte(cls, byte_value):
@@ -51,5 +52,10 @@ def get_subtype(self, subtype_char):
                     "t": "task",
                     "a": "api",
                     "u": "user",
+                    "s": "trigger",
+                },
+                MsgType.THRSH: {
+                    "s": "run_start",
+                    "t": "task",
                 },
             }[self][subtype_char]

source/communication/pycboard.py

@@ -492,7 +492,7 @@ def process_data(self):
                         self.timestamp = msg_timestamp
                     if msg_type in (MsgType.EVENT, MsgType.STATE):
                         content = int(content_bytes.decode())  # Event/state ID.
-                    elif msg_type in (MsgType.PRINT, MsgType.WARNG):
+                    elif msg_type in (MsgType.PRINT, MsgType.WARNG, MsgType.THRSH):
                         content = content_bytes.decode()  # Print or error string.
                     elif msg_type == MsgType.VARBL:
                         content = content_bytes.decode()  # JSON string

source/pyControl/framework.py

@@ -24,6 +24,7 @@ class pyControlError(BaseException):  # Exception for pyControl errors.
 VARBL_TYP = b"V"  # Variable change  : (time, VARBL_TYP, [g]et/user_[s]et/[a]pi_set/[p]rint/s[t]art/[e]nd, json_str)
 WARNG_TYP = b"!"  # Warning          : (time, WARNG_TYP, "", print_string)
 STOPF_TYP = b"X"  # Stop framework   : (time, STOPF_TYP, "", "")
+THRSH_TYP = b"T"  # Threshold        : (time, THRSH_TYP, [s]et)
 
 # Event_queue -----------------------------------------------------------------
 

source/pyControl/hardware.py

@@ -235,25 +235,36 @@ 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:
-            self.threshold = Analog_threshold(threshold, rising_event, falling_event)
-        else:
-            self.threshold = False
+    def __init__(
+        self,
+        pin,
+        name,
+        sampling_rate,
+        threshold=None,
+        rising_event=None,
+        falling_event=None,
+        data_type="H",
+        triggers=None,
+    ):
+        self.triggers = triggers if triggers is not None else []
+
+        if threshold is not None:  # For backward compatibility
+            self.triggers.append(AnalogTrigger(threshold, rising_event, falling_event))
+
         self.timer = pyb.Timer(available_timers.pop())
         if pin:  # pin argument can be None when Analog_input subclassed.
             self.ADC = pyb.ADC(pin)
             self.read_sample = self.ADC.read
         self.name = name
-        self.Analog_channel = Analog_channel(name, sampling_rate, data_type)
+        self.channel = Analog_channel(name, sampling_rate, data_type)
         assign_ID(self)
 
     def _run_start(self):
         # Start sampling timer, initialise threshold, aquire first sample.
-        self.timer.init(freq=self.Analog_channel.sampling_rate)
+        self.timer.init(freq=self.channel.sampling_rate)
         self.timer.callback(self._timer_ISR)
-        if self.threshold:
-            self.threshold.run_start(self.read_sample())
+        for trigger in self.triggers:
+            trigger.run_start(self.name)
         self._timer_ISR(0)
 
     def _run_stop(self):
@@ -263,9 +274,10 @@ def _run_stop(self):
     def _timer_ISR(self, t):
         # Read a sample to the buffer, update write index.
         sample = self.read_sample()
-        self.Analog_channel.put(sample)
-        if self.threshold:
-            self.threshold.check(sample)
+        self.channel.put(sample)
+        if self.triggers:
+            for trigger in self.triggers:
+                trigger.check(sample)
 
     def record(self):  # For backward compatibility.
         pass
@@ -286,15 +298,21 @@ class Analog_channel(IO_object):
     #     data array bytes (variable)
 
     def __init__(self, name, sampling_rate, data_type, plot=True):
-        assert data_type in ("b", "B", "h", "H", "i", "I"), "Invalid data_type."
-        assert not any([name == io.name for io in IO_dict.values() if isinstance(io, Analog_channel)]), (
-            "Analog signals must have unique names."
-        )
+        if data_type not in ("b", "B", "h", "H", "i", "I"):
+            raise ValueError("Invalid data_type.")
+        if any([name == io.name for io in IO_dict.values() if isinstance(io, Analog_channel)]):
+            raise ValueError(
+                "Analog signals must have unique names.{} {}".format(
+                    name, [io.name for io in IO_dict.values() if isinstance(io, Analog_channel)]
+                )
+            )
+
         self.name = name
         assign_ID(self)
         self.sampling_rate = sampling_rate
         self.data_type = data_type
         self.plot = plot
+
         self.bytes_per_sample = {"b": 1, "B": 1, "h": 2, "H": 2, "i": 4, "I": 4}[data_type]
         self.buffer_size = max(4, min(256 // self.bytes_per_sample, sampling_rate // 10))
         self.buffers = (array(data_type, [0] * self.buffer_size), array(data_type, [0] * self.buffer_size))
@@ -344,16 +362,15 @@ def send_buffer(self, run_stop=False):
             fw.usb_serial.send(self.buffers[buffer_n])
 
 
-class Analog_threshold(IO_object):
-    # Generates framework events when an analog signal goes above or below specified threshold.
+class AnalogTrigger(IO_object):
+    # Generates framework events when an analog signal goes above or below specified threshold value.
 
-    def __init__(self, threshold=None, rising_event=None, falling_event=None):
-        assert isinstance(threshold, int), (
-            "Integer threshold must be specified if rising or falling events are defined."
-        )
-        self.threshold = threshold
+    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.threshold = threshold
         self.timestamp = 0
         self.crossing_direction = False
         assign_ID(self)
@@ -364,8 +381,9 @@ def _initialise(self):
         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.above_threshold = sample > self.threshold
+    def run_start(self, attached_to):
+        self.attached_to = attached_to
+        self.set_threshold(self.threshold, run_start=True)
 
     def _process_interrupt(self):
         # Put event generated by threshold crossing in event queue.
@@ -376,14 +394,40 @@ def _process_interrupt(self):
 
     @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, run_start=False):
+        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
+
+        content = {"value": self.threshold, "attached_to": self.attached_to}
+        if self.rising_event is not None:
+            content["rising_event"] = self.rising_event
+        if self.falling_event is not None:
+            content["falling_event"] = self.falling_event
+        fw.data_output_queue.put(
+            fw.Datatuple(
+                fw.current_time,
+                fw.THRSH_TYP,
+                "s" if run_start else "t",
+                str(content),
+            )
+        )
+
 
 # Digital Output --------------------------------------------------------------
 
@@ -509,4 +553,3 @@ def _timer_callback(self):
             fw.data_output_queue.put(fw.Datatuple(fw.current_time, fw.EVENT_TYP, "s", self.event_ID))
         self.state = not self.state
         self.sync_pin.value(self.state)
-