fix(logic_analyzer): count pulses from the measured signal period

pslab/instrument/logic_analyzer.py

@@ -713,12 +713,15 @@ def get_states(self) -> Dict[str, bool]:
         }
 
     def count_pulses(
-        self, channel: str = "FRQ", interval: float = 1, block: bool = True
-    ) -> Union[int, None]:
+        self, channel: str = "FRQ", interval: float = 1, timeout: float = 1
+    ) -> int:
         """Count pulses on a digital input.
 
-        The counter is 16 bits, so it will roll over after 65535 pulses. This
-        method can be used at the same time as the oscilloscope.
+        The number of pulses is calculated from the period of the signal, which
+        is measured by capturing logic events. This method cannot be used at the
+        same time as the oscilloscope. The signal must be slow enough for the
+        capture to resolve the time between two events, and fast enough for two
+        events to be captured within the timeout.
 
         Parameters
         ----------
@@ -727,46 +730,43 @@ def count_pulses(
         interval : float, optional
             Time in seconds during which to count pulses. The default value is
             1 second.
-        block : bool, optional
-            Whether to block while counting pulses or not. If False, this
-            method will return None, and the pulses must be manually fetched
-            using :meth:`fetch_pulse_count`. Additionally, the interval
-            argument has no meaning if block is False; the counter will keep
-            counting even after the interval time has expired. The default
-            value is True.
+        timeout : float, optional
+            Timeout in seconds before cancelling the capture. The default value
+            is 1 second.
 
         Returns
         -------
-        Union[int, None]
-            Number of pulses counted during the interval, or None if block is
-            False.
+        int
+            Number of pulses counted during the interval, or 0 if the signal is
+            too slow, or too fast, to be measured.
+
+        Raises
+        ------
+        ValueError if the interval is negative.
         """
-        self._reset_prescaler()
-        self._device.send_byte(CP.COMMON)
-        self._device.send_byte(CP.START_COUNTING)
-        self._device.send_byte(self._channels[channel].number)
-        self._device.get_ack()
+        if interval < 0:
+            raise ValueError("Interval cannot be negative.")
 
-        if block:
-            time.sleep(interval)
-        else:
-            return
+        tmp = self._channel_one_map
+        self._channel_one_map = channel
+        try:
+            timestamps = self.capture(
+                channel, 2, modes=["sixteen rising"], timeout=timeout
+            )[0]
+        finally:
+            self._channel_one_map = tmp
 
-        return self.fetch_pulse_count()
+        try:
+            # Sixteen rising edges are captured, so the time between two
+            # captured events covers 16 periods of the signal.
+            period = (timestamps[1] - timestamps[0]) * 1e-6 / 16
+        except IndexError:
+            return 0
 
-    def fetch_pulse_count(self) -> int:
-        """Return the number of pulses counted since calling :meth:`count_pulses`.
+        if period <= 0:
+            return 0
 
-        Returns
-        -------
-        int
-            Number of pulses counted since calling :meth:`count_pulses`.
-        """
-        self._device.send_byte(CP.COMMON)
-        self._device.send_byte(CP.FETCH_COUNT)
-        count = self._device.get_int()
-        self._device.get_ack()
-        return count
+        return int(round(interval / period))
 
     def _reset_prescaler(self):
         self._device.send_byte(CP.TIMING)

tests/test_logic_analyzer.py

@@ -280,3 +280,93 @@ def test_count_pulses(la):
     pulses = la.count_pulses("LA2", interval)
     expected_pulses = FREQUENCY * interval
     assert expected_pulses == pytest.approx(pulses, rel=0.1)  # Pretty bad accuracy.
+
+
+# The tests below do not need a connected PSLab, since the capture is mocked.
+
+
+class FakeDevice:
+    """Stand-in for a SerialHandler, recording the commands it receives."""
+
+    def __init__(self):
+        self.commands = []
+
+    def send_byte(self, value):
+        self.commands.append(value)
+
+    def send_int(self, value):
+        self.commands.append(value)
+
+    def get_ack(self):
+        return True
+
+    def get_byte(self):
+        return 0
+
+    def get_int(self):
+        return 0
+
+
+def test_count_pulses_from_the_measured_period(mocker):
+    la = LogicAnalyzer(FakeDevice())
+    # A 100 kHz signal, sampled every 16th rising edge.
+    mocker.patch.object(la, "capture", return_value=[np.array([0, 160])])
+    assert la.count_pulses("LA2", 0.2) == FREQUENCY * 0.2
+
+
+@pytest.mark.parametrize("timestamps", [np.array([]), np.array([500, 500])])
+def test_count_pulses_without_a_measurable_period(mocker, timestamps):
+    la = LogicAnalyzer(FakeDevice())
+    mocker.patch.object(la, "capture", return_value=[timestamps])
+    assert la.count_pulses("LA2", 0.2) == 0
+
+
+def test_count_pulses_does_not_use_the_firmware_counter(mocker):
+    device = FakeDevice()
+    la = LogicAnalyzer(device)
+    mocker.patch.object(la, "capture", return_value=[np.array([0, 160])])
+    la.count_pulses("LA2", 0.2)
+    assert CP.START_COUNTING not in device.commands
+    assert CP.FETCH_COUNT not in device.commands
+
+
+def capture_updating_the_channel_map(la, timestamps):
+    """Stand-in for capture(), which maps a channel name to channel one."""
+
+    def capture(channels, events, **kwargs):
+        la._check_arguments(channels, events)
+        return [timestamps]
+
+    return capture
+
+
+def test_count_pulses_restores_the_channel_map(mocker):
+    la = LogicAnalyzer(FakeDevice())
+    mocker.patch.object(
+        la,
+        "capture",
+        side_effect=capture_updating_the_channel_map(la, np.array([0, 160])),
+    )
+    assert la.count_pulses("LA2", 0.2) == FREQUENCY * 0.2
+    assert la._channel_one_map == "LA1"
+
+
+def test_count_pulses_restores_the_channel_map_when_capture_fails(mocker):
+    la = LogicAnalyzer(FakeDevice())
+
+    def capture(channels, events, **kwargs):
+        la._check_arguments(channels, events)
+        raise ValueError("capture failed")
+
+    mocker.patch.object(la, "capture", side_effect=capture)
+    with pytest.raises(ValueError):
+        la.count_pulses("LA2", 0.2)
+    assert la._channel_one_map == "LA1"
+
+
+def test_count_pulses_rejects_a_negative_interval(mocker):
+    la = LogicAnalyzer(FakeDevice())
+    capture = mocker.patch.object(la, "capture", return_value=[np.array([0, 160])])
+    with pytest.raises(ValueError):
+        la.count_pulses("LA2", -0.2)
+    assert not capture.called