diff --git a/pslab/instrument/logic_analyzer.py b/pslab/instrument/logic_analyzer.py index 8888034..5e59ed4 100644 --- a/pslab/instrument/logic_analyzer.py +++ b/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) diff --git a/tests/test_logic_analyzer.py b/tests/test_logic_analyzer.py index 550a929..a6b7895 100644 --- a/tests/test_logic_analyzer.py +++ b/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