feat: add Pico SCPI transport - #286
Conversation
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Sorry @IM-TechieScientist, you have reached your weekly rate limit of 500000 diff characters.
Please try again later or upgrade to continue using Sourcery
Reviewer's GuideAdds initial PSLab Pico SCPI support, including USB and Wi-Fi transports, a shared SCPI client, a high-level PicoDevice wrapper, top-level exports, and comprehensive unit tests around SCPI command/query behavior and transport helpers. Sequence diagram for PicoDevice USB identification flowsequenceDiagram
actor User
participant PicoDevice
participant ScpiClient
participant PicoUsbTransport
participant Serial
User->>PicoDevice: usb(port, baudrate, timeout)
activate PicoDevice
PicoDevice->>PicoUsbTransport: __init__(port, baudrate, timeout)
PicoDevice->>ScpiClient: __init__(transport)
deactivate PicoDevice
User->>PicoDevice: connect()
activate PicoDevice
PicoDevice->>ScpiClient: connect()
activate ScpiClient
ScpiClient->>PicoUsbTransport: connect()
activate PicoUsbTransport
PicoUsbTransport->>Serial: Serial(port, baudrate, timeout)
Serial-->>PicoUsbTransport: serial_instance
deactivate PicoUsbTransport
deactivate ScpiClient
deactivate PicoDevice
User->>PicoDevice: identify()
activate PicoDevice
PicoDevice->>ScpiClient: identify()
activate ScpiClient
ScpiClient->>ScpiClient: query(*IDN?)
ScpiClient->>PicoUsbTransport: write(b"*IDN?\n")
ScpiClient->>PicoUsbTransport: readline()
PicoUsbTransport-->>ScpiClient: b"PSLab Pico...\n"
ScpiClient-->>PicoDevice: id_string
deactivate ScpiClient
PicoDevice-->>User: id_string
deactivate PicoDevice
File-Level Changes
Possibly linked issues
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (6)
pslab/pico/transport.py:190
- Similar to
read(),PicoWifiTransport.readline()will currently raise a rawsocket.timeout/TimeoutErroron timeouts, which makes error handling inconsistent withScpiClient.query()/_read_exact()(which raiseScpiTimeoutError). Catch and rethrow asScpiTimeoutError.
def readline(self) -> bytes:
self.connect()
while b"\n" not in self._rx_buffer:
chunk = self._socket.recv(4096)
if not chunk:
raise ScpiTimeoutError("SCPI TCP connection closed.")
self._rx_buffer.extend(chunk)
end = self._rx_buffer.index(b"\n") + 1
line = bytes(self._rx_buffer[:end])
del self._rx_buffer[:end]
return line
pslab/pico/transport.py:253
read_block()fails for SCPI arbitrary blocks that use the standard#0(indefinite-length) header:digit_countbecomes 0, thenint(self._read_exact(0).decode(...))raises aValueError. This makes the block reader incompatible with instruments/firmware that emit#0...\nblocks.
digit_count_text = self._read_exact(1)
try:
digit_count = int(digit_count_text.decode("ascii"))
byte_count = int(self._read_exact(digit_count).decode("ascii"))
except ValueError as exc:
raise ScpiError("Invalid SCPI arbitrary block header.") from exc
pslab/pico/transport.py:336
_drain_block_terminator()consumes one byte even when it is not a newline/CR. That byte is then lost, so the “next read” cannot reliably detect the protocol mismatch (it will already be desynchronized). It’s safer to fail fast if the terminator isn’t present.
if char in (b"\n", b"\r"):
return
# The firmware emits a newline after blocks. If a different byte appears,
# leave higher-level code to catch the protocol mismatch on the next read.
pslab/pico/transport.py:298
set_transport()accepts "WIRELESS" but then sends it verbatim (COMM:TRAN WIRELESS). If the firmware command set only supportsUSB,WIFI, andAUTO(as the docstring suggests), this will produce an invalid SCPI command. Consider accepting "WIRELESS" as an alias but mapping it toWIFIon the wire.
def set_transport(self, mode: str) -> None:
"""Select firmware capture transport: ``USB``, ``WIFI``, or ``AUTO``."""
normalized = mode.strip().upper()
if normalized not in ("USB", "WIFI", "WIRELESS", "AUTO"):
raise ValueError("mode must be USB, WIFI, WIRELESS, or AUTO.")
self.command(f"COMM:TRAN {normalized}")
pslab/pico/transport.py:178
PicoWifiTransport.read()treats a remote close asScpiTimeoutError, but a real socket timeout currently bubbles up assocket.timeout/TimeoutError(different exception type than the rest of the SCPI stack). Also,read(0)will read 1 byte because ofmax(1, ...). Handlesize <= 0and normalize timeouts toScpiTimeoutErrorfor consistent behavior.
This issue also appears on line 180 of the same file.
def read(self, size: int) -> bytes:
self.connect()
while len(self._rx_buffer) < size:
chunk = self._socket.recv(max(1, size - len(self._rx_buffer)))
if not chunk:
raise ScpiTimeoutError("SCPI TCP connection closed.")
self._rx_buffer.extend(chunk)
data = bytes(self._rx_buffer[:size])
del self._rx_buffer[:size]
return data
tests/test_pico_scpi_transport.py:91
- Block parsing tests cover only definite-length blocks (
#14...). Since the client is intended to support “arbitrary block parsing”, add a unit test for the standard#0...\n(indefinite-length) form so regressions are caught.
def test_query_block_reads_definite_length_payload():
client = ScpiClient(FakeTransport())
payload = client.query_block("LA:READ?")
assert payload == b"abcd"
| def flush(self) -> None: | ||
| """Flush pending output bytes if the backend supports it.""" | ||
|
|
||
| def reset_input_buffer(self) -> None: |
There was a problem hiding this comment.
Is this function used anywhere?
There was a problem hiding this comment.
not at the moment but I felt it might be useful for when i implement waveform streaming to drop stale bytes before a new command group
CloudyPadmal
left a comment
CloudyPadmal
left a comment
There was a problem hiding this comment.
What prevents us from reusing the existing ConnectionHandler? From what I can see, most of the functions here in PicoTransport are abstracted in the ConnectionHandler.
|
I made a seperate |
There was a problem hiding this comment.
Copilot flagged the following:
_drain_block_terminator()catchesTimeoutError, but wifi transport raisesScpiTimeoutErrorand pyserial raisesSerialTimeoutException— a missing terminator can leave the client out of sync.command()/query()never checkSYST:ERR/drain_errors()after operations.close()could null outself._serialto avoid reuse of a closed handle.
|
@CloudyPadmal please let me know if there are any changes required in this PR. |
|
How would you plan to wire |
|
This from copilot seems like a valid concern PicoWifiTransport.read() treats a remote close as ScpiTimeoutError, but a real socket timeout currently bubbles up as socket.timeout/TimeoutError (different exception type than the rest of the SCPI stack). Also, read(0) will read 1 byte because of max(1, ...). Handle size <= 0 and normalize timeouts to ScpiTimeoutError for consistent behavior. |
|
I plan to wire the existing pico instruments through The transport layer is only the byte/connection layer aka USB CDC or TCP to the ESP bridge.
Planned structure: PicoDevice For example |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.
Suppressed comments (1)
pslab/pico/transport.py:95
PicoUsbTransport.is_opendefaults toTruewhen the injected serial instance lacks anis_openattribute, which can causeconnect()to be skipped even if the transport is not actually open (e.g., when using a stub/mock serial backend). Defaulting toFalseis safer and aligns with the intent of the guard.
def is_open(self) -> bool:
return bool(self._serial is not None and getattr(self._serial, "is_open", True))
| def _drain_block_terminator(self) -> None: | ||
| try: | ||
| char = self.transport.read(1) | ||
| except TimeoutError: | ||
| return | ||
| if char in (b"\n", b"\r"): | ||
| return | ||
| # The firmware emits a newline after blocks. If a different byte appears, | ||
| # leave higher-level code to catch the protocol mismatch on the next read. |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.
Suppressed comments (4)
tests/test_pico_scpi_transport.py:129
- There is no regression test ensuring a CRLF block terminator (
\r\n) is fully consumed. Without it, a leftover\ncan cause the nextquery()to read an empty line. Adding a targeted test will prevent stream-desync regressions.
def test_query_block_reads_definite_length_payload():
client = ScpiClient(FakeTransport())
payload = client.query_block("LA:READ?")
pslab/pico/transport.py:304
set_transport()acceptsWIRELESSas a valid mode, but the docstring only lists USB/WIFI/AUTO. Update the docstring so it matches the supported inputs.
def set_transport(self, mode: str) -> None:
"""Select firmware capture transport: ``USB``, ``WIFI``, or ``AUTO``."""
pslab/pico/device.py:86
PicoDevice.wifi_status()is missing a return type annotation even though the underlyingScpiClient.wifi_status()returns a fixed 4-tuple. Adding the type helps keep the public API self-describing and consistent with the rest of this typed module.
def wifi_status(self):
"""Return Wi-Fi transport status counters."""
return self.scpi.wifi_status()
pslab/pico/device.py:92
- Most context-manager
__exit__methods in this codebase use the conventional parameter names(exc_type, exc_val, exc_tb)(e.g.pslab/bus/busio.py). Using the same names here improves consistency and readability.
def __exit__(self, exc_type, exc, traceback) -> None:
| def _drain_block_terminator(self) -> None: | ||
| try: | ||
| char = self.transport.read(1) | ||
| except TimeoutError: | ||
| return | ||
| if not char: | ||
| return | ||
| if char in (b"\n", b"\r"): | ||
| return | ||
| raise ScpiError(f"Unexpected SCPI block terminator byte: {char!r}") |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.
Suppressed comments (3)
pslab/pico/transport.py:111
close()leaves_serialset after closing. If a custom/injected backend doesn't updateis_open, subsequentconnect()calls may not reopen the port becauseis_openstill evaluates truthy. Clearing_serialon close avoids this stale-state issue.
def close(self) -> None:
if self._serial is not None:
self._serial.close()
pslab/pico/transport.py:337
_read_exact()assumes the underlying transport returns exactlysizebytes in a singleread()call.pyserial.Serial.read()(and many stream transports) can legally return short reads, which would make SCPI block parsing fail spuriously withScpiTimeoutErroreven though more bytes are available.
def _read_exact(self, size: int) -> bytes:
data = self.transport.read(size)
if len(data) != size:
raise ScpiTimeoutError(f"Expected {size} bytes, received {len(data)}.")
return data
pslab/pico/transport.py:95
is_opendefaults toTruewhen the injected serial backend does not expose anis_openattribute. That can incorrectly treat a closed backend as open and causeconnect()to no-op.
This issue also appears on line 109 of the same file.
def is_open(self) -> bool:
return bool(self._serial is not None and getattr(self._serial, "is_open", True))
First PR adding support for PSLab Pico in pslab-python. Working on #285.
This PR introduces the initial pslab.pico package with:
from pslab import PicoDeviceexportSummary by Sourcery
Add initial PSLab Pico SCPI support, including transports, client helper, and high-level device wrapper.
New Features:
Tests: