From 8e17aae3550028c30d77a9c69a62d2854d8cc7d0 Mon Sep 17 00:00:00 2001 From: Alexander Bessman Date: Tue, 4 Mar 2025 12:04:54 +0100 Subject: [PATCH 1/3] Begin update to new protocol --- pslab/connection/connection.py | 48 ++++++++++++++++++++++++---------- pslab/protocol.py | 2 ++ 2 files changed, 36 insertions(+), 14 deletions(-) diff --git a/pslab/connection/connection.py b/pslab/connection/connection.py index d87fd18..35860e0 100644 --- a/pslab/connection/connection.py +++ b/pslab/connection/connection.py @@ -70,6 +70,36 @@ def write(self, data: bytes) -> int: """ ... + def exchange(self, cmd: bytes, data: bytes = b"") -> bytes: + """Send command and input data to device, and return output data. + + Parameters + ---------- + cmd : int + Command code. + data : bytes, default b'' + Input data for command, if any. + + Returns + ------- + bytes + Output data from command, if any. + """ + (cmd_int,) = CP.ShortInt.unpack(cmd) + header = CP.Header.pack(cmd_int, len(data)) + self.write(header + data) + status, response_size = CP.Header.unpack(self.read(CP.Header.size)) + + if status: + raise Exception(status) + + response = self.read(response_size) + + if len(response) < response_size: + raise TimeoutError + + return response + def get_byte(self) -> int: """Read a single one-byte of integer value. @@ -164,10 +194,7 @@ def get_version(self) -> str: str Version string. """ - self.send_byte(CP.COMMON) - self.send_byte(CP.GET_VERSION) - version_length = 9 - version = self.read(version_length) + version = self.exchange(CP.COMMON + CP.GET_VERSION) try: if b"PSLab" not in version: @@ -177,23 +204,16 @@ def get_version(self) -> str: msg = "device not found" raise ConnectionError(msg) from exc - return version.decode("utf-8") + return version.rstrip(b"\x00").decode("utf-8") def get_firmware_version(self) -> FirmwareVersion: """Get firmware version. Returns ------- - tuple[int, int, int] + FirmwareVersion major, minor, patch. """ - self.send_byte(CP.COMMON) - self.send_byte(CP.GET_FW_VERSION) - - # Firmware version query was added in firmware version 3.0.0. - major = self.get_byte() - minor = self.get_byte() - patch = self.get_byte() - + major, minor, patch = self.exchange(CP.COMMON + CP.GET_FW_VERSION) return FirmwareVersion(major, minor, patch) diff --git a/pslab/protocol.py b/pslab/protocol.py index de7f258..b86406b 100644 --- a/pslab/protocol.py +++ b/pslab/protocol.py @@ -9,6 +9,8 @@ ShortInt = struct.Struct("H") # size 2 Integer = struct.Struct("I") # size 4 +Header = struct.Struct(" Date: Fri, 14 Mar 2025 23:31:17 +0700 Subject: [PATCH 2/3] Update several functions to match changes in firmware --- pslab/instrument/oscilloscope.py | 13 ++----------- pslab/protocol.py | 29 +---------------------------- pslab/sciencelab.py | 30 +++++++++++------------------- 3 files changed, 14 insertions(+), 58 deletions(-) diff --git a/pslab/instrument/oscilloscope.py b/pslab/instrument/oscilloscope.py index b8d3e30..9372d73 100644 --- a/pslab/instrument/oscilloscope.py +++ b/pslab/instrument/oscilloscope.py @@ -38,8 +38,8 @@ def __init__(self, device: ConnectionHandler | None = None): self._trigger_voltage = None self._trigger_enabled = False self._trigger_channel = "CH1" - self._set_gain("CH1", 1) - self._set_gain("CH2", 1) + # self._set_gain("CH1", 1) + # self._set_gain("CH2", 1) def capture( self, @@ -375,12 +375,6 @@ def select_range(self, channel: str, voltage_range: Union[int, float]): self._set_gain(channel, gain) def _set_gain(self, channel: str, gain: int): - spi_config_supported = self._check_spi_config() - - if not spi_config_supported: - spi_parameters = SPIMaster.get_parameters() - spi = SPIMaster(self._device) # Initializing SPIMaster will reset config. - self._channels[channel].gain = gain pga = self._channels[channel].programmable_gain_amplifier gain_idx = GAIN_VALUES.index(gain) @@ -390,9 +384,6 @@ def _set_gain(self, channel: str, gain: int): self._device.send_byte(gain_idx) self._device.get_ack() - if not spi_config_supported: - spi.set_parameters(*spi_parameters) - @staticmethod def _check_spi_config() -> bool: """Check whether current SPI config is supported by PGA. diff --git a/pslab/protocol.py b/pslab/protocol.py index b86406b..8f5dfa7 100644 --- a/pslab/protocol.py +++ b/pslab/protocol.py @@ -193,36 +193,9 @@ BAUD230400_LEGACY = Byte.pack(8) BAUD1000000_LEGACY = Byte.pack(9) -# /*-----------NRFL01 radio module----------*/ -NRFL01 = Byte.pack(13) -NRF_SETUP = Byte.pack(1) -NRF_RXMODE = Byte.pack(2) -NRF_TXMODE = Byte.pack(3) -NRF_POWER_DOWN = Byte.pack(4) -NRF_RXCHAR = Byte.pack(5) -NRF_TXCHAR = Byte.pack(6) -NRF_HASDATA = Byte.pack(7) -NRF_FLUSH = Byte.pack(8) -NRF_WRITEREG = Byte.pack(9) -NRF_READREG = Byte.pack(10) -NRF_GETSTATUS = Byte.pack(11) -NRF_WRITECOMMAND = Byte.pack(12) -NRF_WRITEPAYLOAD = Byte.pack(13) -NRF_READPAYLOAD = Byte.pack(14) -NRF_WRITEADDRESS = Byte.pack(15) -NRF_TRANSACTION = Byte.pack(16) -NRF_START_TOKEN_MANAGER = Byte.pack(17) -NRF_STOP_TOKEN_MANAGER = Byte.pack(18) -NRF_TOTAL_TOKENS = Byte.pack(19) -NRF_REPORTS = Byte.pack(20) -NRF_WRITE_REPORT = Byte.pack(21) -NRF_DELETE_REPORT_ROW = Byte.pack(22) - -NRF_WRITEADDRESSES = Byte.pack(23) - # ---------Non standard IO protocols-------- -NONSTANDARD_IO = Byte.pack(14) +NONSTANDARD_IO = Byte.pack(13) # HX711_HEADER = Byte.pack(1) HCSR04_HEADER = Byte.pack(2) # AM2302_HEADER = Byte.pack(3) diff --git a/pslab/sciencelab.py b/pslab/sciencelab.py index bf4b16f..959bcff 100644 --- a/pslab/sciencelab.py +++ b/pslab/sciencelab.py @@ -36,6 +36,7 @@ class ScienceLab: def __init__(self, device: ConnectionHandler | None = None): self.device = device if device is not None else autoconnect() + self.version = self.device.get_version() self.logic_analyzer = LogicAnalyzer(device=self.device) self.oscilloscope = Oscilloscope(device=self.device) self.waveform_generator = WaveformGenerator(device=self.device) @@ -163,8 +164,10 @@ def rgb_led(self, colors: List, output: str = "RGB", order: str = "GRB"): >>> psl.rgb_led([[10,0,0],[0,10,10],[10,0,10]], output="SQ1", order="RGB") """ - if "6" in self.device.version: + if "6" in self.version: pins = {"ONBOARD": 0, "SQ1": 1, "SQ2": 2, "SQ3": 3, "SQ4": 4} + if output == "RGB": + output = "ONBOARD" else: pins = {"RGB": CP.SET_RGB1, "PGC": CP.SET_RGB2, "SQ1": CP.SET_RGB3} @@ -189,24 +192,13 @@ def rgb_led(self, colors: List, output: str = "RGB", order: str = "GRB"): f"Invalid order: {order}. order must contain 'R', 'G', and 'B'." ) - self.device.send_byte(CP.COMMON) - - if "6" in self.device.version: - self.device.send_byte(CP.SET_RGB_COMMON) - else: - self.device.send_byte(pin) - - self.device.send_byte(len(colors) * 3) - - for color in colors: - self.device.send_byte(color[order.index("R")]) - self.device.send_byte(color[order.index("G")]) - self.device.send_byte(color[order.index("B")]) - - if "6" in self.device.version: - self.device.send_byte(pin) - - self.device.get_ack() + cmd = CP.COMMON + CP.SET_RGB_COMMON + args = CP.Byte.pack(pin) + args += CP.Byte.pack(len(colors) * 3) + args += bytes( + color[order.index(channel)] for channel in "RGB" for color in colors + ) + self.device.exchange(cmd, args) def _read_program_address(self, address: int): """Return the value stored at the specified address in program memory. From ea167801e77f689d192284bd7fc9eec3d97ed02c Mon Sep 17 00:00:00 2001 From: Alexander Bessman Date: Fri, 14 Mar 2025 23:33:37 +0700 Subject: [PATCH 3/3] Add solution to blink exercise --- pslab/blink.py | 56 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) create mode 100644 pslab/blink.py diff --git a/pslab/blink.py b/pslab/blink.py new file mode 100644 index 0000000..8b3e9a0 --- /dev/null +++ b/pslab/blink.py @@ -0,0 +1,56 @@ +"""FOSSASIA Summit 2025 - PSLab Development 101 exercises.""" + +import time + +import pslab +import pslab.protocol as CP + + +def blink(psl: pslab.ScienceLab, color: tuple[int, int, int], period: int) -> None: + """Blink the onbard RGB LED. + + Parameters + ---------- + psl : pslab.ScienceLab + color : tuple[int, int, int] + Green, red, blue, each in range 0-255. + period : int + Blink period in milliseconds. + """ + toggle = time.time() + state = 0 + + while True: + if period / 2 < (time.time() - toggle) * 1000: + if state: + # Turn off LED. + psl.rgb_led((0, 0, 0)) + else: + # Turn on LED. + psl.rgb_led(color) + + state = not state + toggle = time.time() + + +def blink_c(psl: pslab.ScienceLab, color: tuple[int, int, int], period: int) -> None: + """Blink the RGB LED using firmware implementation. + + Parameters + ---------- + psl : pslab.ScienceLab + color : tuple[int, int, int] + Green, red, blue, each in range 0-255. + period : int + Blink period in milliseconds. + """ + if not period: + cmd = CP.NONSTANDARD_IO + CP.Byte.pack(11) + psl.device.exchange(cmd) + psl.rgb_led(color) + return + + cmd = CP.NONSTANDARD_IO + CP.Byte.pack(10) + args = CP.ShortInt.pack(period) + args += bytes(color) + psl.device.exchange(cmd, args)