Summit/blink/solution

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)

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)

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.

pslab/protocol.py

@@ -9,6 +9,8 @@
 ShortInt = struct.Struct("H")  # size 2
 Integer = struct.Struct("I")  # size 4
 
+Header = struct.Struct("<HHxxxx")
+
 ACKNOWLEDGE = Byte.pack(254)
 MAX_SAMPLES = 10000
 DATA_SPLITTING = 200
@@ -191,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)

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.