From 5efabeb70d96ae5f6b0dfe1301d7aa9895c3f478 Mon Sep 17 00:00:00 2001 From: Pascal Sachs Date: Fri, 10 Jul 2026 15:38:46 +0200 Subject: [PATCH] Add missing commands SCC1 cable Add missing commands for SCC1 device. Add support for all SF06 sensors and add missing commands for those sensors. Add tests for SLF3x, LD20 and generic SF06 sensors. Add more tests for SCC1 device. Ensure tests run stable and faster for tests with hardware. --- CHANGELOG.md | 1 + sensirion_uart_scc1/drivers/scc1_ld20.py | 25 +++ sensirion_uart_scc1/drivers/scc1_sf06.py | 257 ++++++++++++++++++++++ sensirion_uart_scc1/drivers/scc1_slf3x.py | 225 +------------------ sensirion_uart_scc1/drivers/slf_common.py | 24 +- sensirion_uart_scc1/scc1_shdlc_device.py | 139 +++++++++++- tests/conftest.py | 2 + tests/test_ld20.py | 39 ++++ tests/test_scc1_shdlc_device.py | 93 +++++++- tests/test_sf06.py | 40 ++++ tests/test_slf3x.py | 2 +- 11 files changed, 623 insertions(+), 224 deletions(-) create mode 100644 sensirion_uart_scc1/drivers/scc1_ld20.py create mode 100644 sensirion_uart_scc1/drivers/scc1_sf06.py create mode 100644 tests/test_ld20.py create mode 100644 tests/test_sf06.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b096261..bef5502 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +- Add SHDLC commands: User Data (0x21), Device Selftest (0x22), Sensor Voltage (0x23), Measure Sensor Voltage (0x26) - Add interface for totalizer - Ensure linux file endings across the package - Fix several typos diff --git a/sensirion_uart_scc1/drivers/scc1_ld20.py b/sensirion_uart_scc1/drivers/scc1_ld20.py new file mode 100644 index 0000000..7625f65 --- /dev/null +++ b/sensirion_uart_scc1/drivers/scc1_ld20.py @@ -0,0 +1,25 @@ +# -*- coding: utf-8 -*- + +from sensirion_uart_scc1.drivers.slf_common import SlfMode, SLF_PRODUCT_LIQUI_MAP, SlfProduct +from sensirion_uart_scc1.scc1_shdlc_device import Scc1ShdlcDevice +from sensirion_uart_scc1.drivers.scc1_sf06 import Scc1Sf06 + + +class Scc1Ld20(Scc1Sf06): + """ + Scc1 LD20 Sensor Driver + + The Scc1 provides features to support the LD20 liquid flow sensors. This driver accesses the sensor through the + API specified by scc1. + """ + SENSOR_TYPE = 3 #: Sensor type for LD20 (same as SF06/SLF3x) + + def __init__(self, device: Scc1ShdlcDevice, liquid_mode: SlfMode = SlfMode.LIQUI_1) -> None: + """ + Initialize object instance. + + :param device: The Scc1 device that provides the access to the sensor. + :param liquid_mode: The liquid that is measured. + """ + super().__init__(device, liquid_mode) + self._liquid_config = SLF_PRODUCT_LIQUI_MAP[SlfProduct.LD20] diff --git a/sensirion_uart_scc1/drivers/scc1_sf06.py b/sensirion_uart_scc1/drivers/scc1_sf06.py new file mode 100644 index 0000000..86bf74b --- /dev/null +++ b/sensirion_uart_scc1/drivers/scc1_sf06.py @@ -0,0 +1,257 @@ +# -*- coding: utf-8 -*- + +import struct +import time +from typing import List, Tuple, Optional, Any + +from sensirion_uart_scc1.drivers.slf_common import SlfMeasurementCommand, SlfMode, SLF_PRODUCT_LIQUI_MAP, SlfProduct +from sensirion_uart_scc1.scc1_exceptions import Scc1InvalidDataReceived +from sensirion_uart_scc1.scc1_shdlc_device import Scc1ShdlcDevice + + +class Scc1Sf06: + """ + Driver for the SF06 sensor family connected via SCC1 cable. + SF06 sensors are sensor type 3. + """ + SENSOR_TYPE = 3 + START_MEASUREMENT_DELAY_S = 0.015 + + def __init__(self, device: Scc1ShdlcDevice, liquid_mode: SlfMode = SlfMode.LIQUI_1) -> None: + """ + Initialize object instance. + + :param device: The Scc1 device that provides the access to the sensor. + :param liquid_mode: The liquid that is measured. + """ + self._scc1 = device + self._liquid_config = SLF_PRODUCT_LIQUI_MAP[SlfProduct.SF06] + self._serial_number, self._product_id = self._get_serial_number_and_product_id() + self._is_measuring = False + self._sampling_interval_ms = 100 # Default 10Hz + self._liquid_mode = liquid_mode + self._measurement_command = SlfMeasurementCommand.from_mode(self._liquid_mode) + + @property + def serial_number(self) -> Optional[int]: + """ + Get the serial number + + :return: The serial number as integer + """ + return self._serial_number + + @property + def product_id(self) -> Optional[int]: + """ + Get the product id + + :return: The product identifier as integer + """ + return self._product_id + + @property + def liquid_mode(self) -> SlfMode: + """ + Liquid measurement mode + """ + return self._liquid_mode + + @liquid_mode.setter + def liquid_mode(self, mode: SlfMode) -> None: + """ + Set liquid measurement mode. + + :param mode: One of the liquid measurement modes + """ + from sensirion_uart_scc1.scc1_exceptions import Scc1NotSupportedException + if not isinstance(mode, SlfMode): + raise Scc1NotSupportedException(f"Invalid liquid mode: {mode}") + + if self._is_measuring: + raise Scc1NotSupportedException("Set liquid mode not allowed while measurement is running") + + self._liquid_mode = mode + self._measurement_command = SlfMeasurementCommand.from_mode(self._liquid_mode) + + @property + def liquid_mode_name(self) -> str: + """ + Get the name of the liquid. + + :return: Name of current liquid measurement mode + """ + return self.get_liquid_mode_name(self._liquid_mode) + + def get_liquid_mode_name(self, mode: SlfMode) -> str: + """ + Get the name of the liquid + + :param mode: A liquid mode + + :return: The name of a specific liquid measurement mode + """ + return self._liquid_config.liqui_mode_name(mode) + + @property + def sampling_interval_ms(self) -> int: + """ + Sampling interval for synchronous measurement + + :return: Current internal sampling interval + """ + return self._sampling_interval_ms + + @sampling_interval_ms.setter + def sampling_interval_ms(self, interval_ms: int): + """ + Set sampling interval for continuous measurement + This will not be applied while the measurement is running + + :param interval_ms: The requested measurement interval in milliseconds + """ + self._sampling_interval_ms = interval_ms + + def get_serial_number(self) -> Optional[int]: + """ + Get the serial number of the device. + + :return: The sensor serial number + """ + return self._serial_number + + def get_flow_unit_and_scale(self, command: Optional[int] = None) -> Optional[Tuple[int, int]]: + """ + Get the scale factor, unit, and sensor sanity check result of the sensor for the given argument. + (available on SF06 sensor products) + + :param command: The 16-bit command to read flow unit and scale factor for. If no value is supplied, + the actual measurement command is used + :return: A tuple with (scale_factor, flow_unit), None if the command is not supported + """ + if command is None: + command = self._measurement_command + args = list(struct.pack('>h', command)) + data = self._scc1.transceive(0x53, args, 0.01) + if len(data) != 6: + return None + scale, unit, _ = struct.unpack('>HHH', data) + return scale, unit + + def get_last_measurement(self) -> Optional[Tuple[int, int, int]]: + """ + Read current measurement and starts internal continuous measurement with the configured interval, if not + already started. + + :return: A tuple with flow, temperature, and flag + """ + data = self._scc1.transceive(0x35, [self.SENSOR_TYPE], 0.01) + if not data: + # Measurement is not ready + return None + return struct.unpack('>hhH', data) + + def start_continuous_measurement(self, interval_ms=0) -> None: + """ + Start a continuous measurement with a given interval. + + :param interval_ms: Measurement interval in milliseconds. + """ + if self._is_measuring: + return + data = bytearray() + data.extend(bytearray(struct.pack('>H', int(interval_ms)))) + data.extend(bytearray(struct.pack('>H', int(self._measurement_command)))) + self._scc1.transceive(0x33, data, 0.01) + time.sleep(self.START_MEASUREMENT_DELAY_S) + self._is_measuring = True + + def stop_continuous_measurement(self) -> None: + """Stop continuous measurement""" + if not self._is_measuring: + return + self._scc1.transceive(0x34, [], 0.01) + self._is_measuring = False + + def read_extended_buffer(self) -> Tuple[int, int, List[Tuple[Any, ...]]]: + """ + Read out measurement buffer for SF06. + SF06 might return multiple signals. + + :return: A tuple with (bytes_remaining, bytes_lost, data) + """ + data = self._scc1.transceive(0x36, [self.SENSOR_TYPE], 0.01) + if not data: + return 0, 0, [] + bytes_lost = int(struct.unpack('>I', data[:4])[0]) + bytes_remaining = int(struct.unpack('>H', data[4:6])[0]) + num_signals = struct.unpack('>H', data[6:8])[0] + # For SF06, signals are typically i16 + num_packets = int(len(data[8:]) / 2 / num_signals) + buffer = list(struct.unpack(">" + "h" * (num_packets * num_signals), data[8:])) + + if len(buffer) % num_signals != 0: + raise Scc1InvalidDataReceived("Received unexpected amount of data") + + num_samples = len(buffer) // num_signals + out = [tuple(buffer[i * num_signals:(i + 1) * num_signals]) + for i in range(num_samples)] + return bytes_remaining, bytes_lost, out + + def get_totalizator_status(self) -> Optional[bool]: + """ + Get the Status (enabled / disabled) of the Totalizator. + :return: True if the totalizator is enabled, False if disabled + """ + return self._scc1.get_totalizator_status() + + def set_totalizator_status(self, enabled: bool) -> None: + """ + Enable or disable the Totalizator. The value of the Totalizator is not changed with this command. + :param enabled: True to enable the totalizator, false to disable it + """ + self._scc1.set_totalizator_status(enabled) + + def get_totalizator_value(self) -> Optional[int]: + """ + Get the value of the Totalizator for SF06. + For sensor type 3 only: Only the flow values (signal 1) are totalized, and + the values are interpreted as i16 signed integers. + :return: Totalizator value + """ + return self._scc1.get_totalizator_value() + + def reset_totalizator(self) -> None: + """ + Set the Totalizator value to zero, the Totalizator Status (enabled/disabled) is + not changed. The Totalizator can be reset anytime. + """ + self._scc1.reset_totalizator() + + def get_sensor_status(self) -> Optional[int]: + """ + Get the status of the sensor and the continuous measurement. + + :return: Sensor status as integer, None if not available. + """ + return self._scc1.get_sensor_status() + + def get_continuous_measurement_status(self) -> Optional[int]: + """ + Get the interval or status of the continuous Measurement. + + :return: Measurement interval in ms if started, None if not started. + """ + return self._scc1.get_continuous_measurement_status() + + def _get_serial_number_and_product_id(self) -> Tuple[Optional[int], Optional[int]]: + """ + :return: The sensor serial number and product id as tuple + """ + data = self._scc1.transceive(0x50, [], 0.01) + if not data: + return None, None + data = data.rstrip(b'\x00').decode('utf-8') + product_id = int(data[:8], 16) + serial_number = int(data[8:], 16) + return serial_number, product_id diff --git a/sensirion_uart_scc1/drivers/scc1_slf3x.py b/sensirion_uart_scc1/drivers/scc1_slf3x.py index 79b5e4c..7da31b3 100644 --- a/sensirion_uart_scc1/drivers/scc1_slf3x.py +++ b/sensirion_uart_scc1/drivers/scc1_slf3x.py @@ -1,23 +1,18 @@ # -*- coding: utf-8 -*- -import math -import struct -import time -from typing import List, Tuple, Optional, Any, cast - -from sensirion_uart_scc1.drivers.slf_common import SlfMeasurementCommand, SlfMode, SLF_PRODUCT_LIQUI_MAP, SlfProduct -from sensirion_uart_scc1.scc1_exceptions import Scc1NotSupportedException, Scc1InvalidDataReceived +from sensirion_uart_scc1.drivers.scc1_sf06 import Scc1Sf06 +from sensirion_uart_scc1.drivers.slf_common import SlfMode, SLF_PRODUCT_LIQUI_MAP, SlfProduct from sensirion_uart_scc1.scc1_shdlc_device import Scc1ShdlcDevice -class Scc1Slf3x: +class Scc1Slf3x(Scc1Sf06): """ - Scc1 Slf3x Sensor Driver + Scc1 SLF3x Sensor Driver - The Scc1 provides features to support the Slf3x liquid flow sensors. This driver accesses the sensor through the + The Scc1 provides features to support the SLF3x liquid flow sensors. This driver accesses the sensor through the API specified by scc1. """ - SENSOR_TYPE = 3 #: Sensor type for Slf3x + SENSOR_TYPE = 3 #: Sensor type for SLF3x START_MEASUREMENT_DELAY_S = 0.015 def __init__(self, device: Scc1ShdlcDevice, liquid_mode: SlfMode = SlfMode.LIQUI_1) -> None: @@ -27,211 +22,5 @@ def __init__(self, device: Scc1ShdlcDevice, liquid_mode: SlfMode = SlfMode.LIQUI :param device: The Scc1 device that provides the access to the sensor. :param liquid_mode: The liquid that is measured. """ - self._scc1 = device + super().__init__(device, liquid_mode) self._liquid_config = SLF_PRODUCT_LIQUI_MAP[SlfProduct.SLF3x] - self._serial_number, self._product_id = self._get_serial_number_and_product_id() - self._is_measuring = False - self._sampling_interval_ms = 100 # Default 10Hz - self._liquid_mode = liquid_mode - self._measurement_command = SlfMeasurementCommand.from_mode(self._liquid_mode) - - @property - def serial_number(self) -> int: - """ - Get the serial number - - :return: The serial number as integer - """ - return self._serial_number - - @property - def product_id(self) -> int: - """ - Get the product Id - - :return: The product identifier as integer - """ - return self._product_id - - @property - def liquid_mode(self) -> SlfMode: - """ - Liquid measurement mode - """ - return self._liquid_mode - - @liquid_mode.setter - def liquid_mode(self, mode: SlfMode) -> None: - """ - Set liquid measurement mode. - - :param mode: One of the liquid measurement modes - """ - if not isinstance(mode, SlfMode): - raise Scc1NotSupportedException(f"Invalid liquid mode: {mode}") - - if self._is_measuring: - raise Scc1NotSupportedException("Set liquid mode not allowed while measurement is running") - - self._liquid_mode = mode - self._measurement_command = SlfMeasurementCommand.from_mode(self._liquid_mode) - - @property - def liquid_mode_name(self) -> str: - """ - Get the name of the liquid. - - :return: Name of current liquid measurement mode - """ - return self.get_liquid_mode_name(self._liquid_mode) - - def get_liquid_mode_name(self, mode: SlfMode) -> str: - """ - Get the name of the liquid - - :param mode: A liquid mode - :return: Get name of a specific liquid measurement mode - """ - return self._liquid_config.liqui_mode_name(mode) - - @property - def sampling_interval_ms(self) -> int: - """ - Sampling interval for synchronous measurement - - :return: Current internal sampling interval - """ - return self._sampling_interval_ms - - @sampling_interval_ms.setter - def sampling_interval_ms(self, interval_ms: int): - """ - Set sampling interval for continuous measurement - This will not be applied while measurement is running - - :param interval_ms: The requested measurement interval in milliseconds - """ - self._sampling_interval_ms = interval_ms - - def get_serial_number(self) -> int: - """ - Get the serial number of the device. - - :return: The sensor serial number - """ - return self._serial_number - - def get_flow_unit_and_scale(self, command: Optional[int] = None) -> Optional[Tuple[int, int]]: - """ - Get the scale factor, unit and sensor sanity check result of the sensor for the given argument. - (only available on some SLF3x sensor products) - - :param command: The 16-bit command to read flow unit and scale factor for. If no value is supplied - the actual measurement command is used - :return: A tuple with (scale_factor, flow_unit), None if command is not supported - """ - if command is None: - command = self._measurement_command - args = list(struct.pack('>h', command)) - data = self._scc1.transceive(0x53, args, 0.01) - if len(data) != 6: - return None - scale, unit, _ = struct.unpack('>HHH', data) - return scale, unit - - def get_last_measurement(self) -> Optional[Tuple[int, int, int]]: - """ - Read current measurement and starts internal continuous measurement with configured interval, if not - already started. - - :return: A tuple with flow, temperature and flag - """ - data = self._scc1.transceive(0x35, [self.SENSOR_TYPE], 0.01) - if not data: - # Measurement not ready - return None - return cast(Tuple[int, int, int], struct.unpack('>hhH', data)) - - def start_continuous_measurement(self, interval_ms=0) -> None: - """ - Start a continuous measurement with a give interval. - - :param interval_ms: Measurement interval in milliseconds. - """ - if self._is_measuring: - return - data = bytearray() - data.extend(bytearray(struct.pack('>H', int(interval_ms)))) - data.extend(bytearray(struct.pack('>H', int(self._measurement_command)))) - self._scc1.transceive(0x33, data, 0.01) - time.sleep(self.START_MEASUREMENT_DELAY_S) - self._is_measuring = True - - def stop_continuous_measurement(self) -> None: - """Stop continuous measurement""" - if not self._is_measuring: - return - self._scc1.transceive(0x34, [], 0.01) - self._is_measuring = False - - def read_extended_buffer(self) -> Tuple[int, int, List[Tuple[Any, ...]]]: - """ - Read out measurement buffer - - :return: A tuple with (bytes_remaining, bytes_lost, data) - """ - data = self._scc1.transceive(0x36, [self.SENSOR_TYPE], 0.01) - bytes_lost = int(struct.unpack('>I', data[:4])[0]) - bytes_remaining = int(struct.unpack('>H', data[4:6])[0]) - num_signals = struct.unpack('>H', data[6:8])[0] - num_packets = int(len(data[8:]) / 2 / num_signals) - buffer = list(struct.unpack(">" + "hhh" * num_packets, data[8:])) - fractional, integer = math.modf(len(buffer) / num_signals) - if fractional != 0: - raise Scc1InvalidDataReceived("Received unexpected amount of data") - num_samples = int(integer) - # Output buffer a list of tuples with (flow, temp, flags). - out = [tuple(buffer[i * num_signals:i * num_signals + num_signals]) - for i in range(num_samples)] - return bytes_remaining, bytes_lost, out - - def get_totalizator_status(self) -> bool: - """ - Get the Status (enabled / disabled) of the Totalizator. - :return: True if the totalizator is enabled, False if disabled - """ - return self._scc1.get_totalizator_status() - - def set_totalizator_status(self, enabled: bool) -> None: - """ - Enable or disable the Totalizator. The value of the Totalizator is not changed with this command. - :param enabled: true to enable the totalizator, false to disable it - """ - self._scc1.set_totalizator_status(enabled) - - def get_totalizator_value(self) -> int: - """ - Get the value of the Totalizator. This value is the sum of all unscaled - measurements while in continuous measurement. - Note for sensor type 3 only: Only the flow values (signal 1) are totalized, and - the values are interpreted as i16 signed integers. - :return: totalizator value - """ - return self._scc1.get_totalizator_value() - - def reset_totalizator(self) -> None: - """ - Set the Totalizator value to zero, the Totalizator Status (enabled/disabled) is - not changed. The Totalizator can be reset anytime. - """ - self._scc1.reset_totalizator() - - def _get_serial_number_and_product_id(self) -> Tuple[int, int]: - """ - :return: The sensor serial number and product id as tuple - """ - data = self._scc1.transceive(0x50, [], 0.01) - data = data.rstrip(b'\x00').decode('utf-8') - product_id = int(data[:8], 16) - serial_number = int(data[8:], 16) - return serial_number, product_id diff --git a/sensirion_uart_scc1/drivers/slf_common.py b/sensirion_uart_scc1/drivers/slf_common.py index 6011c1c..4b2e7aa 100644 --- a/sensirion_uart_scc1/drivers/slf_common.py +++ b/sensirion_uart_scc1/drivers/slf_common.py @@ -57,12 +57,26 @@ def from_mode(mode): SLF3x_PRODUCT_IDS = list(SLF3x_PRODUCT_NAME.keys()) +#: Product id to product name mapping for SF06 family +SF06_PRODUCT_NAME = { + 0x070105: 'SLQ-QT105', + 0x070101: 'SLQ-QT500', + 0x070201: 'SLI-0430', + 0x070202: 'SLI-1000', + 0x070203: 'SLI-2000', + 0x070204: 'SLI-0430_P', + 0x070205: 'SLG-0430', +} + +SF06_PRODUCT_IDS = list(SF06_PRODUCT_NAME.keys()) + LD20_2600B_PRODUCT_IDS = list(LD20_PRODUCT_NAME.keys()) class SlfProduct(Enum): SLF3x = 'SLF3x' - LD20 = 'LD20-2600B', + LD20 = 'LD20-2600B' + SF06 = 'SF06' @staticmethod def from_product_id(product_id: int) -> SlfProduct: @@ -70,6 +84,8 @@ def from_product_id(product_id: int) -> SlfProduct: return SlfProduct.SLF3x elif product_id in LD20_2600B_PRODUCT_IDS: return SlfProduct.LD20 + elif product_id in SF06_PRODUCT_IDS: + return SlfProduct.SF06 raise Scc1InvalidProductId(product_id) @@ -80,6 +96,8 @@ def from_product(product: SlfProduct, product_id: int) -> str: return SLF3x_PRODUCT_NAME.get(product_id, 'SLF3x') elif product is SlfProduct.LD20: return LD20_PRODUCT_NAME.get(product_id, 'LD20') + elif product is SlfProduct.SF06: + return SF06_PRODUCT_NAME.get(product_id, 'SF06') raise Scc1InvalidProductId(product_id) @@ -107,6 +125,10 @@ def liqui_mode_name(self, mode): })), SlfProduct.LD20: SlfLiquiConfig(OrderedDict({ SlfMode.LIQUI_1: 'Water', + })), + SlfProduct.SF06: SlfLiquiConfig(OrderedDict({ + SlfMode.LIQUI_1: 'Water', + SlfMode.LIQUI_2: 'Isopropyl alcohol', })) } diff --git a/sensirion_uart_scc1/scc1_shdlc_device.py b/sensirion_uart_scc1/scc1_shdlc_device.py index 109f8fe..06aea23 100644 --- a/sensirion_uart_scc1/scc1_shdlc_device.py +++ b/sensirion_uart_scc1/scc1_shdlc_device.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- import logging +import struct from struct import unpack from typing import Optional, Iterable, Union, List @@ -71,6 +72,61 @@ def find_chips(self) -> List[int]: self._connected_i2c_addresses = self.perform_i2c_scan() return self._connected_i2c_addresses + def get_user_data(self, block_number: int = 0) -> bytes: + """ + Get 20 bytes of User Data from EEPROM. + :param block_number: The block number to read (0..4). + :return: 20 bytes of user data. + """ + if block_number not in range(5): + raise ValueError("Block number must be between 0 and 4.") + result = self.transceive(0x21, [block_number], timeout=0.01) + if len(result) != 21: + raise ValueError(f"Unexpected response length for User Data: {len(result)} bytes (expected 21)") + received_block = result[0] + if received_block != block_number: + raise ValueError(f"Received block number {received_block} does not match requested block {block_number}") + return result[1:] + + def set_user_data(self, block_number: int, data: bytes) -> None: + """ + Save 20 bytes of User Data in EEPROM. + :param block_number: The block number to write (0..4). + :param data: 20 bytes of user data. + """ + if block_number not in range(5): + raise ValueError("Block number must be between 0 and 4.") + if len(data) != 20: + raise ValueError("User data must be 20 bytes long.") + payload = bytearray([block_number]) + payload.extend(data) + self.transceive(0x21, payload, timeout=0.01) + + def device_selftest(self) -> int: + """ + Execute a device selftest. + :return: Selftest result (0: success, other: error code). + """ + result = self.transceive(0x22, [], timeout=0.5) + return int(unpack('>H', result)[0]) + + def get_sensor_voltage(self) -> int: + """ + Get the configured sensor supply voltage. + :return: Sensor voltage (0: 3.3V, 1: 5.0V). + """ + result = self.transceive(0x23, [], timeout=0.01) + return int(result[0]) + + def set_sensor_voltage(self, voltage: int) -> None: + """ + Set the sensor supply voltage. + :param voltage: Sensor voltage (0: 3.3V, 1: 5.0V). + """ + if voltage not in [0, 1]: + raise ValueError("Voltage must be 0 (3.3V) or 1 (5.0V).") + self.transceive(0x23, [voltage], timeout=0.01) + def get_sensor_type(self) -> Optional[int]: """ :return: the configured sensor type @@ -115,6 +171,61 @@ def set_sensor_address(self, i2c_address: int) -> None: self.transceive(0x25, [i2c_address], timeout=0.01) self._i2c_address = i2c_address + def measure_sensor_voltage(self) -> int: + """ + Measure the sensor supply voltage. + :return: Output voltage in mV. + """ + result = self.transceive(0x26, [], timeout=0.01) + return int(unpack('>H', result)[0]) + + def get_i2c_delay(self) -> int: + """ + Get the I2C delay. + :return: The I2C delay in microseconds + """ + result = self.transceive(0x28, [], timeout=0.01) + return int(unpack('>H', result)[0]) + + def set_i2c_delay(self, delay_us: int) -> None: + """ + Set the I2C delay. + :param delay_us: The I2C delay in microseconds + """ + self.transceive(0x28, list(struct.pack('>H', delay_us)), timeout=0.01) + + def get_sensor_serial_number(self, sensor_type: int) -> str: + """ + Get the serial number of the connected sensor. + :param sensor_type: The sensors type + :return: the sensor serial number as string + """ + result = self.transceive(0x54, [sensor_type], timeout=0.01) + return result.rstrip(b'\x00').decode('utf-8') + + def get_sensor_part_name(self, sensor_type: int) -> str: + """ + Get the part name of the connected sensor. + :param sensor_type: The sensors' type + :return: the sensors' part name as string + """ + result = self.transceive(0x50, [sensor_type], timeout=0.01) + return result.rstrip(b'\x00').decode('utf-8') + + def i2c_transceive(self, i2c_address: int, tx_data: bytes, rx_length: int, timeout_ms: int) -> bytes: + """ + Perform an I2C transceive operation. + :param i2c_address: The I2C address + :param tx_data: data to transmit + :param rx_length: the number of bytes to receive + :param timeout_ms: timeout in milliseconds + :return: received data + """ + data = bytearray([i2c_address, rx_length]) + data.extend(struct.pack('>H', timeout_ms)) + data.extend(tx_data) + return self.transceive(0x2a, data, timeout=timeout_ms / 1000.0 + 0.05) + def get_totalizator_status(self) -> Optional[bool]: """ Get the Status (enabled / disabled) of the Totalizator. @@ -152,14 +263,36 @@ def reset_totalizator(self) -> None: """ self.transceive(0x39, [], timeout=0.01) + def get_sensor_status(self) -> Optional[int]: + """ + Get the status of sensor and continuous measurement. + + :return: Sensor status as integer, None if not available. + """ + result = self.transceive(0x30, [], timeout=0.01) + if not result: + return None + return int(result[0]) + + def get_continuous_measurement_status(self) -> Optional[int]: + """ + Get the interval or status of the Continuous Measurement. + + :return: Measurement interval in ms if started, None if not started. + """ + result = self.transceive(0x33, [], timeout=0.01) + if not result: + return None + return int(unpack('>H', result)[0]) + def sensor_reset(self) -> None: """ - Execute a hard reset on the sensor and check for the correct response. + Execute a hard reset on the sensor. Sensor must be idle for execution of this command. Active continuous/single measurement is stopped, and the sensor is left in the idle state. """ - self.transceive(0x66, [], 0.3) + self.transceive(0x65, [], 0.3) - def transceive(self, command: int, data: Union[bytes, Iterable], timeout: float = -1.0) -> Optional[bytes]: + def transceive(self, command: int, data: Union[bytes, Iterable], timeout: float = -1.0) -> bytes: """ Provides a generic way to send SHDLC commands. diff --git a/tests/conftest.py b/tests/conftest.py index edb11e2..0f4b09f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -9,6 +9,8 @@ @pytest.fixture def scc1_device(): for port_info in comports(): + if port_info.vid != 0x0403: + continue try: shdlc_port = ShdlcSerialPort(port=port_info.device, baudrate=115200) with shdlc_port: diff --git a/tests/test_ld20.py b/tests/test_ld20.py new file mode 100644 index 0000000..2778efd --- /dev/null +++ b/tests/test_ld20.py @@ -0,0 +1,39 @@ +# -*- coding: utf-8 -*- +from unittest.mock import MagicMock + +import pytest + +from sensirion_uart_scc1.drivers import scc1_ld20 +from sensirion_uart_scc1.drivers.slf_common import SlfMode + + +def test_ld20_initialization(): + mock_device = MagicMock() + # Mock transceive to return something valid for _get_serial_number_and_product_id + # It expects 8 hex chars for product id and some hex chars for serial number + mock_device.transceive.return_value = b"007010201234567\x00" + driver = scc1_ld20.Scc1Ld20(mock_device) + assert driver.SENSOR_TYPE == 3 + assert driver.liquid_mode == SlfMode.LIQUI_1 + + +def test_ld20_liquid_mode_change(): + mock_device = MagicMock() + mock_device.transceive.return_value = b"007010201234567\x00" + driver = scc1_ld20.Scc1Ld20(mock_device) + driver.liquid_mode = SlfMode.LIQUI_1 + assert driver.liquid_mode == SlfMode.LIQUI_1 + # Check if it uses the correct liquid config + assert driver.get_liquid_mode_name(SlfMode.LIQUI_1) == 'Water' + + +@pytest.fixture +def ld20(scc1_device): + yield scc1_ld20.Scc1Ld20(scc1_device) + + +@pytest.mark.needs_hardware +def test_ld20_serial_number_and_product_id(ld20): + assert ld20 is not None + assert isinstance(ld20.serial_number, int) + assert isinstance(ld20.product_id, int) diff --git a/tests/test_scc1_shdlc_device.py b/tests/test_scc1_shdlc_device.py index 9e96896..d5bd1c0 100644 --- a/tests/test_scc1_shdlc_device.py +++ b/tests/test_scc1_shdlc_device.py @@ -1,9 +1,100 @@ # -*- coding: utf-8 -*- -import pytest import re +import pytest + @pytest.mark.needs_hardware def test_scc1_device(scc1_device): assert scc1_device is not None assert re.match("SCC1-([^@])*@*", str(scc1_device)) + + +@pytest.mark.needs_hardware +def test_scc1_baudrate(scc1_device): + baudrate = scc1_device.get_baudrate() + assert isinstance(baudrate, int) + # Test setting it back to the same value + scc1_device.set_baudrate(baudrate) + + +@pytest.mark.needs_hardware +def test_scc1_i2c_delay(scc1_device): + delay = scc1_device.get_i2c_delay() + assert isinstance(delay, int) + scc1_device.set_i2c_delay(delay) + + +@pytest.mark.needs_hardware +def test_scc1_user_data(scc1_device): + data = scc1_device.get_user_data(0) + assert isinstance(data, bytes) + assert len(data) == 20 + scc1_device.set_user_data(0, data) + + +def test_scc1_user_data_invalid_block(): + from sensirion_uart_scc1.scc1_shdlc_device import Scc1ShdlcDevice + from unittest.mock import MagicMock + # Mock connection to return something for execute to avoid initialization errors + connection = MagicMock() + connection.execute.return_value = (b'', 0) + device = Scc1ShdlcDevice(connection) + with pytest.raises(ValueError, match="Block number must be between 0 and 4."): + device.get_user_data(5) + with pytest.raises(ValueError, match="Block number must be between 0 and 4."): + device.set_user_data(5, b'0' * 20) + with pytest.raises(ValueError, match="User data must be 20 bytes long."): + device.set_user_data(0, b'0' * 19) + + +def test_scc1_get_user_data_parsing(): + from sensirion_uart_scc1.scc1_shdlc_device import Scc1ShdlcDevice + from unittest.mock import MagicMock + connection = MagicMock() + # Correct response: block number 2 followed by 20 bytes of 'A' + mock_data = b'\x02' + b'A' * 20 + # Initialization calls: get_version, get_serial_number, get_sensor_type, get_sensor_address + # Then get_user_data + connection.execute.side_effect = [ + (b'\x01\x02\x00\x01\x02\x03', 0), # get_version (6 bytes) + (b'123456', 0), # get_serial_number + (b'\x03', 0), # get_sensor_type (1 byte) + (b'\x10', 0), # get_sensor_address (1 byte) + (mock_data, 0), # get_user_data (first call) + (mock_data, 0), # get_user_data (second call) + (b'\x02' + b'A' * 19, 0), # get_user_data (third call, wrong length) + ] + device = Scc1ShdlcDevice(connection) + + # Test successful parsing + data = device.get_user_data(2) + assert data == b'A' * 20 + + # Test block number mismatch + with pytest.raises(ValueError, match="Received block number 2 does not match requested block 1"): + device.get_user_data(1) + + # Test invalid response length + with pytest.raises(ValueError, match="Unexpected response length for User Data: 20 bytes"): + device.get_user_data(2) + + +@pytest.mark.needs_hardware +def test_scc1_device_selftest(scc1_device): + result = scc1_device.device_selftest() + assert isinstance(result, int) + + +@pytest.mark.needs_hardware +def test_scc1_sensor_voltage(scc1_device): + voltage = scc1_device.get_sensor_voltage() + assert voltage in [0, 1] + scc1_device.set_sensor_voltage(voltage) + + +@pytest.mark.needs_hardware +def test_scc1_measure_sensor_voltage(scc1_device): + voltage_mv = scc1_device.measure_sensor_voltage() + assert isinstance(voltage_mv, int) + assert 3000 <= voltage_mv <= 6000 diff --git a/tests/test_sf06.py b/tests/test_sf06.py new file mode 100644 index 0000000..5a4ba6e --- /dev/null +++ b/tests/test_sf06.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +import pytest + +from sensirion_uart_scc1.drivers import scc1_sf06 + + +@pytest.fixture +def sf06(scc1_device): + yield scc1_sf06.Scc1Sf06(scc1_device) + + +@pytest.mark.needs_hardware +def test_sf06_serial_number_and_product_id(sf06): + assert sf06 is not None + assert isinstance(sf06.serial_number, int) + assert isinstance(sf06.product_id, int) + + +@pytest.mark.needs_hardware +def test_sf06_start_measurements(sf06): + assert sf06 is not None + sf06.start_continuous_measurement(100) + # Test get_continuous_measurement_status + status = sf06.get_continuous_measurement_status() + assert status == 100 + + # Test get_sensor_status + sensor_status = sf06.get_sensor_status() + assert isinstance(sensor_status, int) + + for i in range(3): + remaining, lost, data = sf06.read_extended_buffer() + assert isinstance(remaining, int) + assert isinstance(lost, int) + assert isinstance(data, list) + for record in data: + assert isinstance(record, tuple) + # SF06 might have more than 3 signals, but at least flow/temp/flags + assert len(record) >= 3 + sf06.stop_continuous_measurement() diff --git a/tests/test_slf3x.py b/tests/test_slf3x.py index abdb64c..3295afb 100644 --- a/tests/test_slf3x.py +++ b/tests/test_slf3x.py @@ -17,7 +17,7 @@ def test_slf3x_serial_number_and_product_id(slf3x): @pytest.mark.needs_hardware -def test_scc1_sf06_start_measurements(slf3x): +def test_slf3x_start_measurements(slf3x): assert slf3x is not None slf3x.start_continuous_measurement(100) for i in range(3):