From df26609a2ca231888fe33aeb6b037e1e5e3930d2 Mon Sep 17 00:00:00 2001 From: Pascal Sachs Date: Fri, 10 Jul 2026 13:52:22 +0200 Subject: [PATCH] Add totalizer * Add totalizer * Ensure linux file endings are used * Fix several typos and wordings --- .editorconfig | 12 + CHANGELOG.md | 10 +- examples/scc1_slf3x_example/slf3x_usage.py | 60 +-- .../scc1_usb_to_i2c/scc1_usb_2_i2c_usage.py | 2 +- sensirion_uart_scc1/drivers/scc1_slf3x.py | 443 ++++++++++-------- sensirion_uart_scc1/drivers/slf_common.py | 298 ++++++------ .../protocols/i2c_transceiver.py | 126 ++--- sensirion_uart_scc1/scc1_exceptions.py | 30 +- sensirion_uart_scc1/scc1_i2c_transceiver.py | 84 ++-- sensirion_uart_scc1/scc1_shdlc_device.py | 340 ++++++++------ tests/test_scc1_shdlc_device.py | 18 +- tests/test_slf3x.py | 62 +-- 12 files changed, 786 insertions(+), 699 deletions(-) create mode 100644 .editorconfig diff --git a/.editorconfig b/.editorconfig new file mode 100644 index 0000000..9802a24 --- /dev/null +++ b/.editorconfig @@ -0,0 +1,12 @@ +root = true + +[*] +charset = utf-8 +end_of_line = lf +insert_final_newline = true +indent_style = space +trim_trailing_whitespace = true + +[*.bat] +end_of_line = crlf + diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c8ae9d..51605b6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,19 +3,23 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased] + +- Add interface for totalizer +- Ensure linux file endings across the package +- Fix several typos ## [1.1.1] - 2024-4-10 ### Fixed -- Accidential release due to wrong github action definition +- Accidental release due to wrong GitHub action definition ## [1.1.0] - 2024-4-10 ### Added -- Interface to get and set sensor type +- Interface to get and set the sensor type - Interface to scan for i2c addresses - Interface to get/set i2c address diff --git a/examples/scc1_slf3x_example/slf3x_usage.py b/examples/scc1_slf3x_example/slf3x_usage.py index 3bcd70e..adcce3d 100644 --- a/examples/scc1_slf3x_example/slf3x_usage.py +++ b/examples/scc1_slf3x_example/slf3x_usage.py @@ -1,30 +1,30 @@ -import argparse - -from sensirion_shdlc_driver import ShdlcSerialPort, ShdlcConnection - -from sensirion_uart_scc1.drivers.scc1_slf3x import Scc1Slf3x -from sensirion_uart_scc1.drivers.slf_common import get_flow_unit_label -from sensirion_uart_scc1.scc1_shdlc_device import Scc1ShdlcDevice - -parser = argparse.ArgumentParser() -parser.add_argument('--serial-port', '-p', default='COM5') -args = parser.parse_args() - -with ShdlcSerialPort(port=args.serial_port, baudrate=115200) as port: - device = Scc1ShdlcDevice(ShdlcConnection(port), slave_address=0) - device.set_sensor_type(Scc1Slf3x.SENSOR_TYPE) - sensor = Scc1Slf3x(device) - print("serial_number:", sensor.serial_number) - print("product id:", sensor.product_id) - print("Flow;\tTemperature;\t Flag") - flow_scale, unit = sensor.get_flow_unit_and_scale() - sensor.start_continuous_measurement(interval_ms=2) - try: - for _ in range(1000): - remaining, lost, data = sensor.read_extended_buffer() - print(f"Remaining bytes {remaining} and {lost}-sample lost") - print() - for flow, temperature, flag in data: - print(f'{flow / flow_scale} {get_flow_unit_label(unit)};\t{temperature / 200} C;\t {flag}') - finally: - sensor.stop_continuous_measurement() +import argparse + +from sensirion_shdlc_driver import ShdlcSerialPort, ShdlcConnection + +from sensirion_uart_scc1.drivers.scc1_slf3x import Scc1Slf3x +from sensirion_uart_scc1.drivers.slf_common import get_flow_unit_label +from sensirion_uart_scc1.scc1_shdlc_device import Scc1ShdlcDevice + +parser = argparse.ArgumentParser() +parser.add_argument('--serial-port', '-p', default='COM5') +args = parser.parse_args() + +with ShdlcSerialPort(port=args.serial_port, baudrate=115200) as port: + device = Scc1ShdlcDevice(ShdlcConnection(port), target_address=0) + device.set_sensor_type(Scc1Slf3x.SENSOR_TYPE) + sensor = Scc1Slf3x(device) + print("serial_number:", sensor.serial_number) + print("product id:", sensor.product_id) + print("Flow;\tTemperature;\t Flag") + flow_scale, unit = sensor.get_flow_unit_and_scale() + sensor.start_continuous_measurement(interval_ms=2) + try: + for _ in range(1000): + remaining, lost, data = sensor.read_extended_buffer() + print(f"Remaining bytes {remaining} and {lost}-sample lost") + print() + for flow, temperature, flag in data: + print(f'{flow / flow_scale} {get_flow_unit_label(unit)};\t{temperature / 200} C;\t {flag}') + finally: + sensor.stop_continuous_measurement() diff --git a/examples/scc1_usb_to_i2c/scc1_usb_2_i2c_usage.py b/examples/scc1_usb_to_i2c/scc1_usb_2_i2c_usage.py index e3117c6..0b2af90 100644 --- a/examples/scc1_usb_to_i2c/scc1_usb_2_i2c_usage.py +++ b/examples/scc1_usb_to_i2c/scc1_usb_2_i2c_usage.py @@ -11,7 +11,7 @@ from sensirion_uart_scc1.scc1_shdlc_device import Scc1ShdlcDevice with ShdlcSerialPort(port='COM5', baudrate=115200) as port: - bridge = Scc1ShdlcDevice(ShdlcConnection(port), slave_address=0) + bridge = Scc1ShdlcDevice(ShdlcConnection(port), target_address=0) bridge.set_sensor_type(3) # SF06 devices channel = I2cChannel(bridge.get_i2c_transceiver(), diff --git a/sensirion_uart_scc1/drivers/scc1_slf3x.py b/sensirion_uart_scc1/drivers/scc1_slf3x.py index 96309d5..79b5e4c 100644 --- a/sensirion_uart_scc1/drivers/scc1_slf3x.py +++ b/sensirion_uart_scc1/drivers/scc1_slf3x.py @@ -1,206 +1,237 @@ -# -*- 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.scc1_shdlc_device import Scc1ShdlcDevice - - -class Scc1Slf3x: - """ - Scc1 Slf3x Sensor Driver - - 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 - 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.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_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 +# -*- 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.scc1_shdlc_device import Scc1ShdlcDevice + + +class Scc1Slf3x: + """ + Scc1 Slf3x Sensor Driver + + 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 + 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.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 c004a79..6011c1c 100644 --- a/sensirion_uart_scc1/drivers/slf_common.py +++ b/sensirion_uart_scc1/drivers/slf_common.py @@ -1,149 +1,149 @@ -# -*- coding: utf-8 -*- - -""" -This file contains channels, gas modes and product specifications used by multiple SLF products -""" -from __future__ import annotations - -from collections import OrderedDict -from enum import Enum - -from sensirion_uart_scc1.scc1_exceptions import Scc1InvalidProductId - - -class SlfMode(Enum): - LIQUI_0 = 'Liquid 0' - LIQUI_1 = 'Liquid 1' - LIQUI_2 = 'Liquid 2' - LIQUI_3 = 'Liquid 3' - LIQUI_4 = 'Liquid 4' - LIQUI_5 = 'Liquid 5' - LIQUI_6 = 'Liquid 6' - LIQUI_7 = 'Liquid 7' - LIQUI_8 = 'Liquid 8' - - -class SlfMeasurementCommand(object): - MEASUREMENT_COMMANDS = { - SlfMode.LIQUI_0: 0x3603, - SlfMode.LIQUI_1: 0x3608, - SlfMode.LIQUI_2: 0x3615, - SlfMode.LIQUI_3: 0x361E, - SlfMode.LIQUI_4: 0x3624, - SlfMode.LIQUI_5: 0x362F, - SlfMode.LIQUI_6: 0x3632, - SlfMode.LIQUI_7: 0x3639, - SlfMode.LIQUI_8: 0x3646, - } - - @staticmethod - def from_mode(mode): - return SlfMeasurementCommand.MEASUREMENT_COMMANDS[mode] - - -#: Product id to product name mapping for SLF3x family -SLF3x_PRODUCT_NAME = { - 0x070302: 'SLF3S_1300', - 0x070303: 'SLF3S_600', - 0x070304: 'SLF3C_1300F', - 0x070305: 'SLF3S_4000B', -} - -#: Product id to product name mapping for LD20 family -LD20_PRODUCT_NAME = { - 0x070102: 'LD20_2600B', - 0x070103: 'LD20-0600L' -} - -SLF3x_PRODUCT_IDS = list(SLF3x_PRODUCT_NAME.keys()) - -LD20_2600B_PRODUCT_IDS = list(LD20_PRODUCT_NAME.keys()) - - -class SlfProduct(Enum): - SLF3x = 'SLF3x' - LD20 = 'LD20-2600B', - - @staticmethod - def from_product_id(product_id: int) -> SlfProduct: - if product_id in SLF3x_PRODUCT_IDS: - return SlfProduct.SLF3x - elif product_id in LD20_2600B_PRODUCT_IDS: - return SlfProduct.LD20 - raise Scc1InvalidProductId(product_id) - - -class SlfProductName: - @staticmethod - def from_product(product: SlfProduct, product_id: int) -> str: - if product is SlfProduct.SLF3x: - return SLF3x_PRODUCT_NAME.get(product_id, 'SLF3x') - elif product is SlfProduct.LD20: - return LD20_PRODUCT_NAME.get(product_id, 'LD20') - raise Scc1InvalidProductId(product_id) - - -class SlfLiquiConfig(object): - def __init__(self, config): - """ - :param config: A dictionary of form mode: name, where mode is one of - `~sensirion_uart_scc1.drivers.slf_common.SlfMode` - """ - self._liqui_config = config - self._supported_liquids = list(config.keys()) - - @property - def supported_liqui_modes(self): - return self._supported_liquids - - def liqui_mode_name(self, mode): - return self._liqui_config[mode] - - -SLF_PRODUCT_LIQUI_MAP = { - SlfProduct.SLF3x: SlfLiquiConfig(OrderedDict({ - SlfMode.LIQUI_1: 'Water', - SlfMode.LIQUI_2: 'Isopropyl alcohol', - })), - SlfProduct.LD20: SlfLiquiConfig(OrderedDict({ - SlfMode.LIQUI_1: 'Water', - })) -} - -FLOW_UNIT_PREFIX = { - 3: 'n', - 4: 'u', - 5: 'm', - 6: 'c', - 7: 'd', - 8: '', # 1 - 9: '', # 10 - 10: 'h', - 11: 'k', - 12: 'M', - 13: 'G' -} - -FLOW_UNIT_TIME_BASE = { - 0: '', - 1: 'us', - 2: 'ms', - 3: 's', - 4: 'min', - 5: 'h', - 6: 'day' -} - -FLOW_UNIT_VOLUME = { - 0: 'nl', # norm liter - 1: 'sl', # standard liter - 8: 'l', # liter (typ: liquid flow) - 9: 'g', # gram (typ: liquid flow) -} - - -def get_flow_unit_label(flow_unit_raw: int) -> str: - prefix = FLOW_UNIT_PREFIX.get(flow_unit_raw & 0xF, '') - time_base = FLOW_UNIT_TIME_BASE.get((flow_unit_raw >> 4) & 0xF, '') - volume = FLOW_UNIT_VOLUME.get((flow_unit_raw >> 8) & 0xF, '') - return f'{prefix}{volume}/{time_base}' +# -*- coding: utf-8 -*- + +""" +This file contains channels, gas modes, and product specifications used by multiple SLF products +""" +from __future__ import annotations + +from collections import OrderedDict +from enum import Enum + +from sensirion_uart_scc1.scc1_exceptions import Scc1InvalidProductId + + +class SlfMode(Enum): + LIQUI_0 = 'Liquid 0' + LIQUI_1 = 'Liquid 1' + LIQUI_2 = 'Liquid 2' + LIQUI_3 = 'Liquid 3' + LIQUI_4 = 'Liquid 4' + LIQUI_5 = 'Liquid 5' + LIQUI_6 = 'Liquid 6' + LIQUI_7 = 'Liquid 7' + LIQUI_8 = 'Liquid 8' + + +class SlfMeasurementCommand(object): + MEASUREMENT_COMMANDS = { + SlfMode.LIQUI_0: 0x3603, + SlfMode.LIQUI_1: 0x3608, + SlfMode.LIQUI_2: 0x3615, + SlfMode.LIQUI_3: 0x361E, + SlfMode.LIQUI_4: 0x3624, + SlfMode.LIQUI_5: 0x362F, + SlfMode.LIQUI_6: 0x3632, + SlfMode.LIQUI_7: 0x3639, + SlfMode.LIQUI_8: 0x3646, + } + + @staticmethod + def from_mode(mode): + return SlfMeasurementCommand.MEASUREMENT_COMMANDS[mode] + + +#: Product id to product name mapping for SLF3x family +SLF3x_PRODUCT_NAME = { + 0x070302: 'SLF3S_1300', + 0x070303: 'SLF3S_600', + 0x070304: 'SLF3C_1300F', + 0x070305: 'SLF3S_4000B', +} + +#: Product id to product name mapping for LD20 family +LD20_PRODUCT_NAME = { + 0x070102: 'LD20_2600B', + 0x070103: 'LD20-0600L' +} + +SLF3x_PRODUCT_IDS = list(SLF3x_PRODUCT_NAME.keys()) + +LD20_2600B_PRODUCT_IDS = list(LD20_PRODUCT_NAME.keys()) + + +class SlfProduct(Enum): + SLF3x = 'SLF3x' + LD20 = 'LD20-2600B', + + @staticmethod + def from_product_id(product_id: int) -> SlfProduct: + if product_id in SLF3x_PRODUCT_IDS: + return SlfProduct.SLF3x + elif product_id in LD20_2600B_PRODUCT_IDS: + return SlfProduct.LD20 + raise Scc1InvalidProductId(product_id) + + +class SlfProductName: + @staticmethod + def from_product(product: SlfProduct, product_id: int) -> str: + if product is SlfProduct.SLF3x: + return SLF3x_PRODUCT_NAME.get(product_id, 'SLF3x') + elif product is SlfProduct.LD20: + return LD20_PRODUCT_NAME.get(product_id, 'LD20') + raise Scc1InvalidProductId(product_id) + + +class SlfLiquiConfig(object): + def __init__(self, config): + """ + :param config: A dictionary of form mode: name, where mode is one of + `~sensirion_uart_scc1.drivers.slf_common.SlfMode` + """ + self._liqui_config = config + self._supported_liquids = list(config.keys()) + + @property + def supported_liqui_modes(self): + return self._supported_liquids + + def liqui_mode_name(self, mode): + return self._liqui_config[mode] + + +SLF_PRODUCT_LIQUI_MAP = { + SlfProduct.SLF3x: SlfLiquiConfig(OrderedDict({ + SlfMode.LIQUI_1: 'Water', + SlfMode.LIQUI_2: 'Isopropyl alcohol', + })), + SlfProduct.LD20: SlfLiquiConfig(OrderedDict({ + SlfMode.LIQUI_1: 'Water', + })) +} + +FLOW_UNIT_PREFIX = { + 3: 'n', + 4: 'u', + 5: 'm', + 6: 'c', + 7: 'd', + 8: '', # 1 + 9: '', # 10 + 10: 'h', + 11: 'k', + 12: 'M', + 13: 'G' +} + +FLOW_UNIT_TIME_BASE = { + 0: '', + 1: 'us', + 2: 'ms', + 3: 's', + 4: 'min', + 5: 'h', + 6: 'day' +} + +FLOW_UNIT_VOLUME = { + 0: 'nl', # norm liter + 1: 'sl', # standard liter + 8: 'l', # liter (typ: liquid flow) + 9: 'g', # gram (typ: liquid flow) +} + + +def get_flow_unit_label(flow_unit_raw: int) -> str: + prefix = FLOW_UNIT_PREFIX.get(flow_unit_raw & 0xF, '') + time_base = FLOW_UNIT_TIME_BASE.get((flow_unit_raw >> 4) & 0xF, '') + volume = FLOW_UNIT_VOLUME.get((flow_unit_raw >> 8) & 0xF, '') + return f'{prefix}{volume}/{time_base}' diff --git a/sensirion_uart_scc1/protocols/i2c_transceiver.py b/sensirion_uart_scc1/protocols/i2c_transceiver.py index bfa268e..81ba2d4 100644 --- a/sensirion_uart_scc1/protocols/i2c_transceiver.py +++ b/sensirion_uart_scc1/protocols/i2c_transceiver.py @@ -1,63 +1,63 @@ -# -*- coding: utf-8 -*- - -from typing import Protocol, Optional, runtime_checkable, Tuple, Any - - -class RxTx(Protocol): - """ - An object which conforms to this protocol is supplied by the I2cChannel when using the public python drivers. - """ - - @property - def tx_data(self) -> Optional[bytes]: - """ - The byte array with the data to be sent - """ - - @property - def rx_length(self) -> Optional[int]: - """ - Number of bytes to read - """ - - @property - def read_delay(self) -> float: - """ - Time between writing and reading an i2c command. - """ - - def interpret_response(self, data: Optional[bytes]) -> Optional[Tuple[Any, ...]]: - """Split the byte array from the response into the fields of the command. - - :param data: The byte array that needs to be interpreted. - :return: A tuple with the interpreted data. - """ - - -@runtime_checkable -class I2cTransceiver(Protocol): - - def execute(self, slave_addr: int, rx_tx: RxTx) -> Tuple[Any, ...]: - """ - Compatibility method for driver adapters. - - :param slave_addr: i2c slave address - :param rx_tx: Object containing the information to execute the communication with the device - :return: interpreted results - """ - - def transceive(self, slave_address: int, tx_data: Optional[bytes], rx_length: Optional[int], read_delay: float, - timeout: float) -> bytes: - """ - Send data to the sensor and receive back a response. This function can be used for sending or receiving only - as well. - - :param slave_address: I2c address of the sensor. - :param tx_data: The data to be sent. In case no data shall be sent, this parameter is supposed to be None - :param rx_length: Length of the repsonse of the sensor. If the request does not have a response, this parameter - may be 0. - :param read_delay: Defines the time in seconds that needs to be observed after sending the data before - the read is initiated. - :param timeout: Defines the time after receiving the response from the sensor before the next command may be - sent. - """ +# -*- coding: utf-8 -*- + +from typing import Protocol, Optional, runtime_checkable, Tuple, Any + + +class RxTx(Protocol): + """ + An object which conforms to this protocol is supplied by the I2cChannel when using the public python drivers. + """ + + @property + def tx_data(self) -> Optional[bytes]: + """ + The byte array with the data to be sent + """ + + @property + def rx_length(self) -> Optional[int]: + """ + Number of bytes to read + """ + + @property + def read_delay(self) -> float: + """ + Time between writing and reading an i2c command. + """ + + def interpret_response(self, data: Optional[bytes]) -> Optional[Tuple[Any, ...]]: + """Split the byte array from the response into the fields of the command. + + :param data: The byte array that needs to be interpreted. + :return: A tuple with the interpreted data. + """ + + +@runtime_checkable +class I2cTransceiver(Protocol): + + def execute(self, target_address: int, rx_tx: RxTx) -> Optional[Tuple[Any, ...]]: + """ + Compatibility method for driver adapters. + + :param target_address: I2c address or the sensor + :param rx_tx: Object containing the information to execute the communication with the device + :return: interpreted results + """ + + def transceive(self, target_address: int, tx_data: Optional[bytes], rx_length: Optional[int], read_delay: float, + timeout: float) -> bytes: + """ + Send data to the sensor and receive back a response. This function can be used for sending or receiving only + as well. + + :param target_address: I2c address of the sensor. + :param tx_data: The data to be sent. In case no data shall be sent, this parameter is supposed to be None + :param rx_length: Length of the response of the sensor. If the request does not have a response, this parameter + may be 0. + :param read_delay: Defines the time in seconds that needs to be observed after sending the data before + the read is initiated. + :param timeout: Defines the time after receiving the response from the sensor before the next command may be + sent. + """ diff --git a/sensirion_uart_scc1/scc1_exceptions.py b/sensirion_uart_scc1/scc1_exceptions.py index c3457eb..5d63482 100644 --- a/sensirion_uart_scc1/scc1_exceptions.py +++ b/sensirion_uart_scc1/scc1_exceptions.py @@ -1,15 +1,15 @@ -# -*- coding: utf-8 -*- - -class Scc1InvalidProductId(ValueError): - """Indicates a product id that is not supported""" - - def __init__(self, product_id: int) -> None: - super().__init__(f'The {product_id} is not valid!') - - -class Scc1NotSupportedException(ValueError): - """Indicates a feature that is not (yet) supported.""" - - -class Scc1InvalidDataReceived(IOError): - """Indicates the reception of invalid data from the device""" +# -*- coding: utf-8 -*- + +class Scc1InvalidProductId(ValueError): + """Indicates a product id that is not supported""" + + def __init__(self, product_id: int) -> None: + super().__init__(f'The {product_id} is not valid!') + + +class Scc1NotSupportedException(ValueError): + """Indicates a feature that is not (yet) supported.""" + + +class Scc1InvalidDataReceived(IOError): + """Indicates the reception of invalid data from the device""" diff --git a/sensirion_uart_scc1/scc1_i2c_transceiver.py b/sensirion_uart_scc1/scc1_i2c_transceiver.py index 7669f59..1589b2f 100644 --- a/sensirion_uart_scc1/scc1_i2c_transceiver.py +++ b/sensirion_uart_scc1/scc1_i2c_transceiver.py @@ -1,42 +1,42 @@ -# -*- coding: utf-8 -*- - -from struct import pack -from typing import Optional, Any, Tuple - -from sensirion_uart_scc1.protocols.shdlc_transceiver import ShdlcTransceiver -from sensirion_uart_scc1.protocols.i2c_transceiver import RxTx - - -class Scc1I2cTransceiver: - """ - Wrapper that implements the I2cTransceiver protocol. - This wrappers allows to use the public I2c Python drivers drivers wit the SCC1 cable - """ - - def __init__(self, device: ShdlcTransceiver) -> None: - super().__init__() - self._scc1 = device - - def execute(self, slave_address: int, rx_tx: RxTx) -> Tuple[Any, ...]: - """Implements tht I2cTransceiver protocol""" - data = self.transceive(slave_address, rx_tx.tx_data, rx_tx.rx_length, rx_tx.read_delay) - return rx_tx.interpret_response(data) - - def transceive(self, slave_address: int, tx_data: Optional[bytes], rx_length: Optional[int], read_delay: float, - timeout: float = 0.01) -> bytes: - """Implements the I2cTransceiver protocol""" - - if tx_data is None: - tx_data = bytearray() - else: - tx_data = bytearray(tx_data) - if rx_length is None: - rx_length = 0 - tx_size = len(tx_data) - header = pack('>BBBH', slave_address, tx_size, rx_length, int(read_delay * 1000)) - cmd_data = bytearray(header) - cmd_data.extend(tx_data) - result = self._scc1.transceive(0x2A, cmd_data, timeout) - if result is None or rx_length == 0: - return bytearray() - return result +# -*- coding: utf-8 -*- + +from struct import pack +from typing import Optional, Any, Tuple + +from sensirion_uart_scc1.protocols.i2c_transceiver import RxTx, I2cTransceiver +from sensirion_uart_scc1.protocols.shdlc_transceiver import ShdlcTransceiver + + +class Scc1I2cTransceiver(I2cTransceiver): + """ + Wrapper that implements the I2cTransceiver protocol. + This wrapper allows using the public I2c Python drivers with the SCC1 cable + """ + + def __init__(self, device: ShdlcTransceiver) -> None: + super().__init__() + self._scc1 = device + + def execute(self, target_address: int, rx_tx: RxTx) -> Optional[Tuple[Any, ...]]: + """Implements tht I2cTransceiver protocol""" + data = self.transceive(target_address, rx_tx.tx_data, rx_tx.rx_length, rx_tx.read_delay) + return rx_tx.interpret_response(data) + + def transceive(self, target_address: int, tx_data: Optional[bytes], rx_length: Optional[int], read_delay: float, + timeout: float = 0.01) -> bytes: + """Implements the I2cTransceiver protocol""" + + if tx_data is None: + tx_data = bytearray() + else: + tx_data = bytearray(tx_data) + if rx_length is None: + rx_length = 0 + tx_size = len(tx_data) + header = pack('>BBBH', target_address, tx_size, rx_length, int(read_delay * 1000)) + cmd_data = bytearray(header) + cmd_data.extend(tx_data) + result = self._scc1.transceive(0x2A, cmd_data, timeout) + if result is None or rx_length == 0: + return bytearray() + return result diff --git a/sensirion_uart_scc1/scc1_shdlc_device.py b/sensirion_uart_scc1/scc1_shdlc_device.py index 29482b9..109f8fe 100644 --- a/sensirion_uart_scc1/scc1_shdlc_device.py +++ b/sensirion_uart_scc1/scc1_shdlc_device.py @@ -1,150 +1,190 @@ -# -*- coding: utf-8 -*- - -import logging -from struct import unpack -from typing import Optional, Iterable, Union, List - -from packaging.version import Version -from sensirion_shdlc_driver import ShdlcDevice, ShdlcConnection -from sensirion_shdlc_driver.command import ShdlcCommand - -from sensirion_uart_scc1.protocols.i2c_transceiver import I2cTransceiver -from sensirion_uart_scc1.scc1_i2c_transceiver import Scc1I2cTransceiver - -log = logging.getLogger(__name__) - - -class Scc1ShdlcDevice(ShdlcDevice): - """ - The Scc1 SHDLC device is used to communicate with various sensors using the Sensirion SCC1 sensor cable. - """ - - def __init__(self, connection: ShdlcConnection, slave_address: int = 0) -> None: - """Initialize object instance. - - :param connection: The used ShdlcConnection - :param slave_address: The SHDLC slave address that is used by this device. - """ - super().__init__(connection, slave_address) - self._version = self.get_version() - self._serial_number = self.get_serial_number() - self._sensor_type = self.get_sensor_type() - self._i2c_address = self.get_sensor_address() - self._connected_i2c_addresses: List[int] = [] - - def __str__(self) -> str: - return f"SCC1-{self.serial_number}@{self.com_port}" - - @property - def com_port(self) -> str: - return self.connection.port.description.split('@')[0] - - @property - def serial_number(self) -> str: - return self._serial_number - - @property - def firmware_version(self) -> Version: - return Version(str(self._version.firmware)) - - @property - def connected_i2c_addresses(self) -> List[int]: - """Returns the connected I2C addresses. You need to call find_chips to fill this attribute.""" - return self._connected_i2c_addresses - - def perform_i2c_scan(self) -> List[int]: - """ - Looks for i2c devices within a certain range on a certain port - :return: List of i2c addresses that responded to the scan - """ - result = self.transceive(0x29, [0x01], timeout=0.025) - return list(unpack('>{cnt}B'.format(cnt=len(result)), result)) - - def find_chips(self) -> List[int]: - """ - Looking for chips on all ports and sets the _connected_i2c_addresses attribute - :return: List of connected addresses - """ - self._connected_i2c_addresses = self.perform_i2c_scan() - return self._connected_i2c_addresses - - def get_sensor_type(self) -> Optional[int]: - """ - :return: the configured sensor type - """ - result = self.transceive(0x24, [], timeout=0.025) - if result: - return int(unpack('>B', result)[0]) - return None - - def set_sensor_type(self, sensor_type: int): - """ - Set sensor type - 0: Flow Sensor (SF04 based products) - 1: Humidity Sensor (SHTxx products) - 2: Flow Sensor (SF05 based products) - 3: Flow Sensor (SF06 based products) (Firmware ≥1.7) - 4: Reserved - :param sensor_type: One of the supported sensor types 0-4 - - """ - if sensor_type not in range(5): - raise ValueError('Sensor type not supported') - self.transceive(0x24, [sensor_type], timeout=0.01) - self._sensor_type = sensor_type - - def get_sensor_address(self) -> Optional[int]: - """ - :return: the configured i2c address - """ - result = self.transceive(0x25, [], timeout=0.025) - if result: - return int(unpack('>B', result)[0]) - return None - - def set_sensor_address(self, i2c_address: int) -> None: - """ - Configure the sensors i2c address and write it to EEPROM - :param i2c_address: the i2c address - """ - if i2c_address not in range(128): - raise ValueError('I2C address out of range. Address has to be within the range 0...127') - self.transceive(0x25, [i2c_address], timeout=0.01) - self._i2c_address = i2c_address - - def sensor_reset(self) -> None: - """ - Execute a hard reset on the sensor and check for the correct response. - Active continuous/single measurement is stopped, and the sensor is left in idle state. - """ - self.transceive(0x66, [], 0.3) - - def transceive(self, command: int, data: Union[bytes, Iterable], timeout: float = -1.0) -> Optional[bytes]: - """ - Provides a generic way to send shdlc commands. - - :param command: The command to send (one byte). - :param data: Byte array of the data to send as arguments to the command. - :param timeout: Response timeout in seconds (-1 for using default value). - :return: The returned data as bytes. - """ - if timeout <= 0.0: - timeout = 3.0 - result = self.execute(ShdlcCommand( - id=command, - data=data, - max_response_time=float(timeout) - )) - if not result: - return b'' - return result - - def get_i2c_transceiver(self) -> I2cTransceiver: - """ - An I2cTransceiver object is required in or der to use the cable with public python i2c drivers. - - In general, all functionality of the sensors is available in the public python drivers as well. - The throughput of the public python driver will be lower than the throughput that can be achieved with - the sensor-specific api of the SCC1 sensor cable. - """ - return Scc1I2cTransceiver(self) +# -*- coding: utf-8 -*- + +import logging +from struct import unpack +from typing import Optional, Iterable, Union, List + +from packaging.version import Version +from sensirion_shdlc_driver import ShdlcDevice, ShdlcConnection +from sensirion_shdlc_driver.command import ShdlcCommand + +from sensirion_uart_scc1.protocols.i2c_transceiver import I2cTransceiver +from sensirion_uart_scc1.scc1_i2c_transceiver import Scc1I2cTransceiver + +log = logging.getLogger(__name__) + + +class Scc1ShdlcDevice(ShdlcDevice): + """ + The Scc1 SHDLC device is used to communicate with various sensors using the Sensirion SCC1 sensor cable. + """ + + def __init__(self, connection: ShdlcConnection, target_address: int = 0) -> None: + """Initialize SCC1 SHDLC Device. + + :param connection: The used ShdlcConnection + :param target_address: The SHDLC target address used by this device (default: 0). + Usually 0 unless multiple devices are connected to the same USB port. + """ + super().__init__(connection, target_address) + self._version = self.get_version() + self._serial_number = self.get_serial_number() + self._sensor_type = self.get_sensor_type() + self._i2c_address = self.get_sensor_address() + self._connected_i2c_addresses: List[int] = [] + + def __str__(self) -> str: + return f"SCC1-{self.serial_number}@{self.com_port}" + + @property + def com_port(self) -> str: + return self.connection.port.description.split('@')[0] + + @property + def serial_number(self) -> str: + return self._serial_number + + @property + def firmware_version(self) -> Version: + return Version(str(self._version.firmware)) + + @property + def connected_i2c_addresses(self) -> List[int]: + """Returns the connected I2C addresses. You need to call find_chips to fill this attribute.""" + return self._connected_i2c_addresses + + def perform_i2c_scan(self) -> List[int]: + """ + Looks for i2c devices within a certain range on a certain port + :return: List of i2c addresses that responded to the scan + """ + result = self.transceive(0x29, [0x01], timeout=0.025) + if not result: + return [] + return list(unpack('>{cnt}B'.format(cnt=len(result)), result)) + + def find_chips(self) -> List[int]: + """ + Looking for chips on all ports and sets the _connected_i2c_addresses attribute + :return: List of connected addresses + """ + self._connected_i2c_addresses = self.perform_i2c_scan() + return self._connected_i2c_addresses + + def get_sensor_type(self) -> Optional[int]: + """ + :return: the configured sensor type + """ + result = self.transceive(0x24, [], timeout=0.025) + if result: + return int(unpack('>B', result)[0]) + return None + + def set_sensor_type(self, sensor_type: int): + """ + Set sensor type + 0: Flow Sensor (SF04 based products) + 1: Humidity Sensor (SHTxx products) + 2: Flow Sensor (SF05 based products) + 3: Flow Sensor (SF06 based products) (Firmware ≥1.7) + 4: Reserved + :param sensor_type: One of the supported sensor types 0-4 + + """ + if sensor_type not in range(5): + raise ValueError('Sensor type not supported') + self.transceive(0x24, [sensor_type], timeout=0.01) + self._sensor_type = sensor_type + + def get_sensor_address(self) -> Optional[int]: + """ + :return: the configured i2c address + """ + result = self.transceive(0x25, [], timeout=0.025) + if result: + return int(unpack('>B', result)[0]) + return None + + def set_sensor_address(self, i2c_address: int) -> None: + """ + Configure the sensors i2c address and write it to EEPROM + :param i2c_address: the i2c address + """ + if i2c_address not in range(128): + raise ValueError('I2C address out of range. Address has to be within the range 0...127') + self.transceive(0x25, [i2c_address], timeout=0.01) + self._i2c_address = i2c_address + + 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 + """ + result = self.transceive(0x37, [], timeout=0.01) + if result is None: + return None + return bool(result[0]) + + 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.transceive(0x37, [1 if enabled else 0], timeout=0.01) + + def get_totalizator_value(self) -> Optional[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 + """ + result = self.transceive(0x38, [], timeout=0.01) + if result is None: + return None + return int(unpack('>q', result)[0]) + + 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.transceive(0x39, [], timeout=0.01) + + def sensor_reset(self) -> None: + """ + Execute a hard reset on the sensor and check for the correct response. + Active continuous/single measurement is stopped, and the sensor is left in the idle state. + """ + self.transceive(0x66, [], 0.3) + + def transceive(self, command: int, data: Union[bytes, Iterable], timeout: float = -1.0) -> Optional[bytes]: + """ + Provides a generic way to send SHDLC commands. + + :param command: The command to send (one byte). + :param data: Byte array of the data to send as arguments to the command. + :param timeout: Response timeout in seconds (-1 for using the default value). + :return: The returned data as bytes. + """ + if timeout <= 0.0: + timeout = 3.0 + result = self.execute(ShdlcCommand( + id=command, + data=data, + max_response_time=float(timeout) + )) + if not result: + return b'' + return result + + def get_i2c_transceiver(self) -> I2cTransceiver: + """ + An I2cTransceiver object is required in or der to use the cable with public python i2c drivers. + + In general, all functionality of the sensors is available in the public python drivers as well. + The throughput of the public python driver will be lower than the throughput that can be achieved with + the sensor-specific api of the SCC1 sensor cable. + """ + return Scc1I2cTransceiver(self) diff --git a/tests/test_scc1_shdlc_device.py b/tests/test_scc1_shdlc_device.py index 07c54e4..6d3a5ab 100644 --- a/tests/test_scc1_shdlc_device.py +++ b/tests/test_scc1_shdlc_device.py @@ -1,9 +1,9 @@ -# -*- coding: utf-8 -*- -import pytest -import re - - -@pytest.mark.needs_hardware -def test_scc1_device(scc1_device): - assert scc1_device is not None - assert re.match("SCC1-([^@])*@COM.", str(scc1_device)) +# -*- coding: utf-8 -*- +import pytest +import re + + +@pytest.mark.needs_hardware +def test_scc1_device(scc1_device): + assert scc1_device is not None + assert re.match("SCC1-([^@])*@COM.", str(scc1_device)) diff --git a/tests/test_slf3x.py b/tests/test_slf3x.py index f22727b..abdb64c 100644 --- a/tests/test_slf3x.py +++ b/tests/test_slf3x.py @@ -1,31 +1,31 @@ -# -*- coding: utf-8 -*- -import pytest - -from sensirion_uart_scc1.drivers import scc1_slf3x - - -@pytest.fixture -def slf3x(scc1_device): - yield scc1_slf3x.Scc1Slf3x(scc1_device) - - -@pytest.mark.needs_hardware -def test_slf3x_serial_number_and_product_id(slf3x): - assert slf3x is not None - assert isinstance(slf3x.serial_number, int) - assert isinstance(slf3x.product_id, int) - - -@pytest.mark.needs_hardware -def test_scc1_sf06_start_measurements(slf3x): - assert slf3x is not None - slf3x.start_continuous_measurement(100) - for i in range(3): - remaining, lost, data = slf3x.read_extended_buffer() - assert isinstance(remaining, int) - assert isinstance(lost, int) - assert isinstance(data, list) - for record in data: - assert isinstance(record, tuple) - assert len(record) == 3 - slf3x.stop_continuous_measurement() +# -*- coding: utf-8 -*- +import pytest + +from sensirion_uart_scc1.drivers import scc1_slf3x + + +@pytest.fixture +def slf3x(scc1_device): + yield scc1_slf3x.Scc1Slf3x(scc1_device) + + +@pytest.mark.needs_hardware +def test_slf3x_serial_number_and_product_id(slf3x): + assert slf3x is not None + assert isinstance(slf3x.serial_number, int) + assert isinstance(slf3x.product_id, int) + + +@pytest.mark.needs_hardware +def test_scc1_sf06_start_measurements(slf3x): + assert slf3x is not None + slf3x.start_continuous_measurement(100) + for i in range(3): + remaining, lost, data = slf3x.read_extended_buffer() + assert isinstance(remaining, int) + assert isinstance(lost, int) + assert isinstance(data, list) + for record in data: + assert isinstance(record, tuple) + assert len(record) == 3 + slf3x.stop_continuous_measurement()