diff --git a/pyvisa_sim/devices.py b/pyvisa_sim/devices.py index 3e0102f..04cb2b9 100644 --- a/pyvisa_sim/devices.py +++ b/pyvisa_sim/devices.py @@ -6,8 +6,12 @@ """ +import itertools +import re from typing import Deque, Dict, List, Optional, Tuple, Union +import stringparser + from pyvisa import constants, rname from .channels import Channels @@ -227,6 +231,74 @@ def add_eom( to_bytes(response_termination), ) + def _process_single_query(self, the_bytes: bytes) -> int: + """Process one valid query in the input buffer. + + Searches for a valid query (according to the yaml file) from the + beginning of the given bytes. If a valid query is found, the matching + dialogue response is added to the device output buffer, and the matched + query is removed from the input buffer. Everything before the matched + query is also removed from the input buffer. + + Returns + ------- + + int: The number of bytes removed from the input buffer. It may + be 0, in which case no query matching a dialogue was found. + + """ + # the stringparser Parsers for each setter query + setter_query_parsers = filter( + lambda i: isinstance(i, stringparser.Parser), + [next(iter(t[1:])) for t in self._setters], + ) + # the setter query regular expressions + setter_query_regex_patterns = [parser._regex for parser in setter_query_parsers] # type: ignore[union-attr] # parsers guaranteed to be parser objects by filter iterator + for query in itertools.chain( + iter(self._dialogues | self._getters), setter_query_regex_patterns + ): + # query is a setter query regular expression + if isinstance(query, re.Pattern): + try: + # remove the ^ and $ added by stringparser around the regex + # that stringparser adds, because the buffer can contain + # other messages + new_pattern = re.compile( + # TODO are there downsides to not using UTF-8? latin-1 + # preserves raw bytes above 127, which is useful for + # instruments that accept non-ascii bytes. + query.pattern.removeprefix("^") + .removesuffix("$") + .encode("latin-1") + ) + match = new_pattern.search(the_bytes) + query_index = match.start() # type: ignore[union-attr] # None will raise AttributeError handled by try block + query_len = len(match[0]) # type: ignore[index] # only executed if match is not None + # search returned None + except AttributeError: + continue + + # query is a dialogue or getter + elif isinstance(query, bytes): + # try the next key if this one not found + if (query_index := the_bytes.find(query)) == -1: + continue + query_len = len(query) + + if ( + response := self._match( + the_bytes[query_index : query_index + query_len] + ) + ) is not None and response is not NoResponse: + self._output_buffers.append(bytearray(response)) + + # to keep the garbage bytes: + # del self._input_buffer[query_index:query_index+query_len] + del self._input_buffer[0 : query_index + query_len] + return query_len + + return 0 + def write(self, data: bytes) -> None: """Write data into the device input buffer.""" logger.debug("Writing into device input buffer: %r" % data) @@ -239,22 +311,47 @@ def write(self, data: bytes) -> None: if not self._input_buffer.endswith(self._query_eom): return - try: - message = bytes(self._input_buffer[:-le]) - queries = message.split(self.delimiter) if self.delimiter else [message] - for query in queries: - response = self._match(query) - eom = self._response_eom + # TODO: I feel like a simplifying refactor can be done + + # handle the no write termination case separately. importantly, the + # input buffer is not cleared if no valid message is found. once a + # valid message is found (starting from the end of the buffer), it and + # only it is removed from the input buffer + if self._query_eom == b"": + message = bytes(self._input_buffer) + response = self._match(message) + + # initial dialogue lookup failed + if response is None: + # incrementally scan for a match starting from the beginning of + # the buffer + while self._process_single_query(message) != 0: + message = bytes(self._input_buffer) + + # initial dialogue lookup succeeded + else: + if response is not NoResponse: + self._output_buffers.append(bytearray(response)) + del self._input_buffer[-len(message) :] - if response is None: - response = self.error_response("command_error") - assert response is not None + else: + try: + message = bytes(self._input_buffer[:-le]) - if response is not NoResponse: - self._output_buffers.append(bytearray(response) + eom) + queries = message.split(self.delimiter) if self.delimiter else [message] + for query in queries: + response = self._match(query) + eom = self._response_eom + + if response is None: + response = self.error_response("command_error") + assert response is not None + + if response is not NoResponse: + self._output_buffers.append(bytearray(response) + eom) - finally: - self._input_buffer = bytearray() + finally: + self._input_buffer = bytearray() def read(self) -> Tuple[bytes, bool]: """ diff --git a/pyvisa_sim/testsuite/conftest.py b/pyvisa_sim/testsuite/conftest.py index 2ac2d85..0cc2021 100644 --- a/pyvisa_sim/testsuite/conftest.py +++ b/pyvisa_sim/testsuite/conftest.py @@ -18,3 +18,13 @@ def channels(): rm = pyvisa.ResourceManager(path + "@sim") yield rm rm.close() + + +@pytest.fixture +def no_termination_chars_resource_manager(): + path = os.path.join( + os.path.dirname(__file__), "fixtures", "no_termination_chars.yaml" + ) + rm = pyvisa.ResourceManager(path + "@sim") + yield rm + rm.close() diff --git a/pyvisa_sim/testsuite/fixtures/no_termination_chars.yaml b/pyvisa_sim/testsuite/fixtures/no_termination_chars.yaml new file mode 100644 index 0000000..b6fd49a --- /dev/null +++ b/pyvisa_sim/testsuite/fixtures/no_termination_chars.yaml @@ -0,0 +1,33 @@ +# TODO: write dialogues and properties of a real non-IEEE 488.2 or SCPI device, +# as those ones could have no write termination. IEEE 488.2 over serial (not +# GPIB) is required to have linefeed as write termination. + +spec: "1.1" +devices: + device 1: + eom: + ASRL INSTR: + q: "" + r: "" + dialogues: + - q: "*IDN?" + r: "SCPI,MOCK,VERSION_1.0" + - q: "*ESR?" + r: "32" + - q: "*STB?" + r: "255" + properties: + current: + default: 1.0 + getter: + q: ":CURR:IMM:AMPL?" + r: "{:+.8E}" + setter: + q: ":CURR:IMM:AMPL {:.3f}" + specs: + min: 1 + max: 6 + type: float +resources: + ASRL1::INSTR: + device: device 1 diff --git a/pyvisa_sim/testsuite/test_serial.py b/pyvisa_sim/testsuite/test_serial.py index 984fd8b..7184b85 100644 --- a/pyvisa_sim/testsuite/test_serial.py +++ b/pyvisa_sim/testsuite/test_serial.py @@ -1,4 +1,6 @@ # -*- coding: utf-8 -*- +import pytest + import pyvisa from pyvisa_sim.sessions import serial @@ -25,3 +27,81 @@ def test_serial_write_with_termination_last_bit(resource_manager): instr.write("*IDN?") assert instr.read() == "SCPI,MOCK,VERSION_1.0" + + +# an IEEE 488.2 (and therefore SCPI as well) over RS-232 device must always use +# a newline to terminate program messages (commands). these test cases do not +# correspond to real-world scenarios, but are used as a dummy to test non-IEEE +# 488.2 device behavior. +@pytest.mark.parametrize( + "writes, wanted_response, wanted_input_buffer", + [ + # fragmented write + (("*ID", "N?"), b"SCPI,MOCK,VERSION_1.0", b""), + # sequential write + (("*IDN?", "*ESR?", "*STB?"), b"SCPI,MOCK,VERSION_1.0" + b"32" + b"255", b""), + # burst write + # (("*IDN?*ESR?*STB?"), b"SCPI,MOCK,VERSION_1.0" + b"32" + b"255", b""), + # burst write with garbage + ( + ("ab*IDN?cd*ESR?*STB?qwerty"), + b"SCPI,MOCK,VERSION_1.0" + b"32" + b"255", + b"qwerty", + ), + # fragmented write (properties) + ((":CURR:", "IMM:AMP", "L?"), b"+1.00000000E+00", b""), + ((":CURR:", "IMM:AMP", "L 2.0", ":CURR:IMM:AMPL?"), b"+2.00000000E+00", b""), + # sequential write (properties) + ( + (":CURR:IMM:AMPL?", ":CURR:IMM:AMPL 1.2345", ":CURR:IMM:AMPL?"), + b"+1.00000000E+00" + b"+1.23450000E+00", + b"", + ), + # sequential write (properties) + ( + (":CURR:IMM:AMPL?", ":CURR:IMM:AMPL?", ":CURR:IMM:AMPL?"), + b"+1.00000000E+00" * 3, + b"", + ), + ( + ( + ":CURR:IMM:AMPL 1.0", + ":CURR:IMM:AMPL 2.0", + ":CURR:IMM:AMPL 3.0", + ":CURR:IMM:AMPL?", + ":CURR:IMM:AMPL 4.0", + ), + b"+3.00000000E+00", + b"", + ), + # this test fails; however, no SCPI serial devices without termination + # character exist. + # ( + # ( + # "ab :CURR:IMM:AMPL? cd :CURR:IMM:AMPL 5.4321 ef:CURR:IMM:AMPL? gh" + # ), + # b"+1.00000000E+00" + b"+5.43210000E+00", + # b" gh", + # ), + ], +) +def test_serial_write_no_termination( + no_termination_chars_resource_manager, writes, wanted_response, wanted_input_buffer +): + instr = no_termination_chars_resource_manager.open_resource( + "ASRL1::INSTR", + write_termination=None, + read_termination=None, + end_input=pyvisa.constants.SerialTermination.none, + end_output=pyvisa.constants.SerialTermination.none, + send_end=False, + ) + visa_library = instr.visalib + session_handle = instr.session + session = visa_library.sessions[session_handle] + device = session.device + for write in writes: + instr.write(write) + assert instr.read_bytes(len(wanted_response)) == wanted_response + assert device._input_buffer == wanted_input_buffer + device._input_buffer = bytearray()