diff --git a/src/opendisplay/__init__.py b/src/opendisplay/__init__.py index 0f9268e..c7c8a11 100644 --- a/src/opendisplay/__init__.py +++ b/src/opendisplay/__init__.py @@ -36,10 +36,12 @@ AdvertisementTracker, ButtonChangeEvent, ButtonEventData, + Sht40Reading, TouchChangeEvent, TouchEventData, TouchTracker, decode_button_event, + decode_sht40, parse_advertisement, ) from .models.buzzer_activate import BuzzerActivateConfig, BuzzerPattern, BuzzerStep, note_to_index @@ -79,6 +81,7 @@ RefreshMode, Rotation, SeeedBoardType, + SensorType, SolumBoardType, TouchIcType, WaveshareBoardType, @@ -90,6 +93,7 @@ from .ota import find_nrf_dfu_device, perform_nrf_dfu, perform_silabs_ota from .partial import PartialState from .protocol import MANUFACTURER_ID, SERVICE_UUID +from .sensors import SensorReading, read_sensor_values from .transport import BleTransport, TcpTransport, Transport __version__ = "0.1.0" @@ -146,6 +150,9 @@ "firmware_ota_asset", "firmware_release_repo", "SensorData", + "SensorReading", + "SensorType", + "Sht40Reading", "DataBus", "BinaryInputs", "PassiveBuzzer", @@ -189,7 +196,9 @@ "get_board_type_name", "get_manufacturer_name", # Utilities + "decode_sht40", "parse_advertisement", + "read_sensor_values", "decode_button_event", "voltage_to_percent", "build_landing_url", diff --git a/src/opendisplay/cli.py b/src/opendisplay/cli.py index 60d1948..73cc742 100644 --- a/src/opendisplay/cli.py +++ b/src/opendisplay/cli.py @@ -9,7 +9,7 @@ import os import sys from collections.abc import Callable, Coroutine -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Any, NoReturn, TypeVar @@ -18,7 +18,7 @@ from rich.console import Console from rich.live import Live from rich.logging import RichHandler -from rich.progress import BarColumn, Progress, SpinnerColumn, TaskProgressColumn, TextColumn +from rich.progress import BarColumn, Progress, SpinnerColumn, TaskID, TaskProgressColumn, TextColumn from rich.table import Table from rich.tree import Tree @@ -32,7 +32,7 @@ BLETimeoutError, OpenDisplayError, ) -from .models.config import GlobalConfig +from .models.config import GlobalConfig, SensorData from .models.enums import ( PANEL_IC_NAMES, BinaryInputType, @@ -54,6 +54,9 @@ ) from .models.firmware import FirmwareVersion from .partial import PartialState +from .sensors import SensorReading + +_LOGGER = logging.getLogger(__name__) _T = TypeVar("_T") @@ -337,6 +340,51 @@ def _sensor_name(sensor_type: int) -> str: return _enum_name(SensorType, sensor_type, digits=4) or "" +@dataclass(frozen=True) +class _SensorRow: + """A configured sensor paired with its current reading, if it has one.""" + + sensor: SensorData + reading: SensorReading | None + + +def _sensor_rows(ctx: _InfoContext) -> list[_SensorRow]: + """Pair each configured sensor with its live reading.""" + if not ctx.config: + return [] + return [ + _SensorRow(sensor=sensor, reading=ctx.sensor_readings.get(sensor.instance_number)) + for sensor in ctx.config.sensors + ] + + +def _sensor_line(row: _SensorRow) -> str: + """Tree line for one sensor, with live values appended when available.""" + sensor = row.sensor + line = f"Sensor {sensor.instance_number} {_sensor_name(sensor.sensor_type)} (bus {sensor.bus_id})" + values = [ + f"{value:.1f} {unit}" + for value, unit in ( + (row.reading.temperature_c if row.reading else None, "°C"), + (row.reading.humidity_percent if row.reading else None, "%RH"), + ) + if value is not None + ] + return f"{line} {' '.join(values)}" if values else line + + +def _sensor_entry(row: _SensorRow) -> dict[str, Any]: + """JSON object for one sensor; live values are null when unavailable.""" + sensor = row.sensor + return { + "instance": sensor.instance_number, + "type": _sensor_name(sensor.sensor_type), + "bus": sensor.bus_id, + "temperature_c": row.reading.temperature_c if row.reading else None, + "humidity_percent": row.reading.humidity_percent if row.reading else None, + } + + @dataclass(frozen=True) class _InfoContext: """Everything the info report renders, gathered once from the device.""" @@ -350,6 +398,8 @@ class _InfoContext: color_scheme_name: str rotation: Any board_type_name: str | None + # Live sensor values by instance number; empty when none could be read. + sensor_readings: dict[int, SensorReading] = field(default_factory=dict) @property def display(self) -> Any: @@ -752,9 +802,9 @@ def _bus_line(bus: Any) -> str: title="Sensors", key="sensors", json_parent="hardware", - items=lambda c: c.config.sensors if c.config else [], - line=lambda s: f"Sensor {s.instance_number} {_sensor_name(s.sensor_type)} (bus {s.bus_id})", - entry=lambda s: {"instance": s.instance_number, "type": _sensor_name(s.sensor_type), "bus": s.bus_id}, + items=_sensor_rows, + line=_sensor_line, + entry=_sensor_entry, ), _ListSection( title="Buttons", @@ -910,6 +960,25 @@ def _cmd_info(args: argparse.Namespace) -> None: _run(_info(_device_kwargs(args.device, key, args.timeout, args.host, args.port, args.tls), args.output_json)) +async def _read_sensors(device: OpenDisplayDevice, progress: Progress, task: TaskID) -> dict[int, SensorReading]: + """Read live sensor values, keyed by instance number. + + Best-effort: firmware too old for READ_MSD (0x0044) still reports its sensor + hardware, just without values, so a failure here degrades the report rather + than failing the command. + """ + if not device.config or not device.config.sensors: + return {} + + progress.update(task, description="Reading sensors...") + try: + readings = await device.read_sensors() + except OpenDisplayError as exc: + _LOGGER.debug("Could not read sensor values: %s", exc) + return {} + return {reading.instance_number: reading for reading in readings} + + async def _info(device_kwargs: dict[str, Any], output_json: bool) -> None: try: with _spinner() as progress: @@ -929,6 +998,7 @@ async def _info(device_kwargs: dict[str, Any], output_json: bool) -> None: color_scheme_name=device.color_scheme.name, rotation=display.rotation_enum if display else device.rotation, board_type_name=device.get_board_type_name() if config else None, + sensor_readings=await _read_sensors(device, progress, task), ) except OpenDisplayError as exc: _handle_ble_error(exc) diff --git a/src/opendisplay/device.py b/src/opendisplay/device.py index d940145..4bb05f2 100644 --- a/src/opendisplay/device.py +++ b/src/opendisplay/device.py @@ -52,6 +52,7 @@ TruncatedConfigError, ) from .landing import build_landing_url +from .models.advertisement import AdvertisementData, parse_advertisement from .models.buzzer_activate import BuzzerActivateConfig from .models.capabilities import DeviceCapabilities from .models.config import GlobalConfig @@ -108,6 +109,7 @@ build_pipe_write_start_command, build_read_config_command, build_read_fw_version_command, + build_read_msd_command, build_reboot_command, build_write_config_command, classify_pipe_frame, @@ -135,10 +137,12 @@ is_compressed_failure_frame, parse_authenticate_challenge, parse_authenticate_success, + parse_read_msd, strip_command_echo, unpack_command_code, validate_nfc_response, ) +from .sensors import SensorReading, read_sensor_values from .transport import BLEConnection, TcpTransport, Transport if TYPE_CHECKING: @@ -1140,6 +1144,42 @@ async def read_firmware_version(self) -> FirmwareVersion: return self._fw_version + @_serialized + async def read_msd(self) -> AdvertisementData: + """Read the device's manufacturer-specific data record. + + Returns the same 16 bytes the device broadcasts in its advertisement, + but over the open connection -- so it works on transports where no BLE + advertisement is observable, and needs no scan. + + Returns: + The parsed record, including the dynamic block that carries live + sensor readings. Use :meth:`read_sensors` to decode those. + + Raises: + InvalidResponseError: If the device returns a malformed record + """ + _LOGGER.debug("Reading MSD from device %s", self.mac_address) + + await self._write(build_read_msd_command()) + response = await self._read(self.TIMEOUT_ACK) + + return parse_advertisement(parse_read_msd(response)) + + async def read_sensors(self) -> list[SensorReading]: + """Read live values from the device's configured sensors. + + Combines the device config (which sensors exist, and where each one's + bytes sit in the MSD) with a fresh :meth:`read_msd`. + + Returns: + One entry per sensor with a usable reading. Empty when the device + has no sensors configured, or none currently has a valid reading. + """ + if not self.config or not self.config.sensors: + return [] + return read_sensor_values(self.config, await self.read_msd()) + @_serialized async def reboot(self) -> None: """Reboot the device. diff --git a/src/opendisplay/models/__init__.py b/src/opendisplay/models/__init__.py index 9244e81..9c70cd4 100644 --- a/src/opendisplay/models/__init__.py +++ b/src/opendisplay/models/__init__.py @@ -5,7 +5,9 @@ AdvertisementTracker, ButtonChangeEvent, ButtonEventData, + Sht40Reading, decode_button_event, + decode_sht40, parse_advertisement, ) from .buzzer_activate import BuzzerActivateConfig, BuzzerPattern, BuzzerStep, note_to_index @@ -59,6 +61,8 @@ "AdvertisementTracker", "ButtonChangeEvent", "ButtonEventData", + "Sht40Reading", + "decode_sht40", "parse_advertisement", "decode_button_event", "ActiveLevel", diff --git a/src/opendisplay/models/advertisement.py b/src/opendisplay/models/advertisement.py index 5ee62a6..5adab62 100644 --- a/src/opendisplay/models/advertisement.py +++ b/src/opendisplay/models/advertisement.py @@ -7,6 +7,22 @@ from collections.abc import Iterable from dataclasses import dataclass, field +# SHT40 temperature/humidity readings, bit-packed into 3 bytes of the v1 dynamic +# block. Firmware writes them as a 24-bit little-endian word (21 bits used): +# v = (rh_deci & 0x3FF) | ((t_deci + 400) << 10) +# See sensor_sht40.cpp write_sht40_msd() / opendisplay_sensor_sht40.c. +SHT40_MSD_LENGTH = 3 +SHT40_DEFAULT_MSD_START = 7 # firmware default when msd_data_start_byte is 0 or 0xFF +SHT40_MAX_MSD_START = 8 # a 3-byte block must fit the 11-byte dynamic area +_SHT40_RH_MASK = 0x3FF +_SHT40_TEMP_MASK = 0x7FF +_SHT40_TEMP_SHIFT = 10 +_SHT40_TEMP_BIAS = 400 # temperature is stored as (deci-degrees + 400) to avoid a sign bit +_SHT40_MAX_RH_DECI = 1000 # 100.0 %RH +_SHT40_MAX_TEMP_UNITS = 1650 # (125.0 C * 10) + bias, the SHT40's upper limit +_SHT40_UNWRITTEN = 0x000000 # slot the firmware has never written +_SHT40_INVALID = 0xFFFFFF # firmware's explicit read-failure sentinel + @dataclass class AdvertisementData: @@ -35,7 +51,9 @@ class AdvertisementData: Attributes: battery_mv: Battery voltage in millivolts - temperature_c: Chip temperature in Celsius + temperature_c: MCU chip temperature in Celsius. This is the microcontroller's + own temperature, *not* an attached sensor -- for a board with an SHT40, + read ambient temperature and humidity via :meth:`sht40_reading` instead. loop_counter: Incrementing counter for each advertisement format_version: Parsed advertisement format ("legacy" or "v1") reboot_flag: Reboot flag from status byte (v1 only) @@ -118,6 +136,27 @@ def touch_event(self, start_byte: int) -> TouchEventData | None: y=y, ) + def sht40_reading(self, start_byte: int = SHT40_DEFAULT_MSD_START) -> Sht40Reading | None: + """Decode a 3-byte SHT40 block from dynamic_data at the given offset (v1 only). + + Args: + start_byte: Offset within the 11-byte dynamic return block (0-8). Pass + ``SensorData.sht40_msd_start_byte`` from the device config; the + default matches the firmware's own default slot. + + Returns: + Parsed temperature and humidity, or None if this is not a v1 + advertisement, the block does not fit, or the sensor has no valid + reading (not yet read, or the firmware's read failed). + """ + if self.format_version != "v1": + return None + if not 0 <= start_byte <= SHT40_MAX_MSD_START: + return None + if start_byte + SHT40_MSD_LENGTH > len(self.dynamic_data): + return None + return decode_sht40(self.dynamic_data[start_byte : start_byte + SHT40_MSD_LENGTH], start_byte) + @dataclass(frozen=True) class ButtonEventData: @@ -185,6 +224,16 @@ class TouchChangeEvent: timestamp: float +@dataclass(frozen=True) +class Sht40Reading: + """Decoded SHT40 measurement from a 3-byte block in v1 dynamic return data.""" + + start_byte: int + temperature_c: float + humidity_percent: float + raw: bytes + + def decode_button_event(raw: int, byte_index: int) -> ButtonEventData: """Decode one dynamic return byte into button fields.""" return ButtonEventData( @@ -196,6 +245,50 @@ def decode_button_event(raw: int, byte_index: int) -> ButtonEventData: ) +def decode_sht40(raw: bytes, start_byte: int = SHT40_DEFAULT_MSD_START) -> Sht40Reading | None: + """Decode a 3-byte SHT40 block into temperature and humidity. + + Two byte patterns are not measurements and decode to None: + + - ``FF FF FF`` -- the firmware's explicit read-failure sentinel. It decodes to + 164.7 C / 102.3 %RH, which the range checks below reject. + - ``00 00 00`` -- a slot the firmware has never written. It decodes to exactly + -40.0 C / 0.0 %RH, the simultaneous floor of *both* ranges, so the range + checks alone would let it through as a plausible-looking reading. Reporting + it would write a hard -40 C into consumers' long-term statistics, so it is + rejected outright; a real reading sitting on both floors at once is not + physically meaningful. + + Args: + raw: Exactly 3 bytes, as written by the firmware into the dynamic block. + start_byte: Offset the block was read from, recorded on the result. + + Returns: + Parsed reading, or None when the block holds no valid measurement. + + Raises: + ValueError: If raw is not exactly 3 bytes. + """ + if len(raw) != SHT40_MSD_LENGTH: + raise ValueError(f"SHT40 block must be {SHT40_MSD_LENGTH} bytes, got {len(raw)}") + + packed = int.from_bytes(raw, "little") + if packed in (_SHT40_UNWRITTEN, _SHT40_INVALID): + return None + + rh_deci = packed & _SHT40_RH_MASK + temp_units = (packed >> _SHT40_TEMP_SHIFT) & _SHT40_TEMP_MASK + if rh_deci > _SHT40_MAX_RH_DECI or temp_units > _SHT40_MAX_TEMP_UNITS: + return None + + return Sht40Reading( + start_byte=start_byte, + temperature_c=(temp_units - _SHT40_TEMP_BIAS) / 10.0, + humidity_percent=rh_deci / 10.0, + raw=bytes(raw), + ) + + class AdvertisementTracker: """Track per-device v1 advertisements and emit button transitions. diff --git a/src/opendisplay/models/config.py b/src/opendisplay/models/config.py index 25c2d4d..73c3891 100644 --- a/src/opendisplay/models/config.py +++ b/src/opendisplay/models/config.py @@ -13,6 +13,7 @@ from epaper_dithering import ColorScheme +from .advertisement import SHT40_DEFAULT_MSD_START from .enums import ( ActiveLevel, BinaryInputType, @@ -502,6 +503,21 @@ def sensor_type_enum(self) -> SensorType | int: except ValueError: return self.sensor_type + @property + def sht40_msd_start_byte(self) -> int: + """Offset of this SHT40's readings in the advertisement's dynamic block. + + Mirrors the firmware's sht40_msd_start(): 0 and 0xFF both mean "use the + default slot", so callers never have to re-derive that rule. + + This rule is specific to the SHT40. The fuel gauges (BQ27220, NPM1300) + read msd_data_start_byte literally -- 0 means byte 0, and 0xFF means + "do not publish" -- so they must not use this property. + """ + if self.msd_data_start_byte in (0, 0xFF): + return SHT40_DEFAULT_MSD_START + return self.msd_data_start_byte + SIZE: ClassVar[int] = 30 @classmethod diff --git a/src/opendisplay/models/enums.py b/src/opendisplay/models/enums.py index 93ce63a..6e6b13b 100644 --- a/src/opendisplay/models/enums.py +++ b/src/opendisplay/models/enums.py @@ -190,6 +190,7 @@ class SensorType(IntEnum): AXP2101_PMIC = 3 SHT40 = 4 BQ27220 = 5 + NPM1300 = 6 # Nordic nPM1300 PMIC (default I2C 0x6B) class WifiEncryption(IntEnum): diff --git a/src/opendisplay/protocol/__init__.py b/src/opendisplay/protocol/__init__.py index ba9aed8..687adbc 100644 --- a/src/opendisplay/protocol/__init__.py +++ b/src/opendisplay/protocol/__init__.py @@ -53,6 +53,7 @@ build_pipe_write_start_command, build_read_config_command, build_read_fw_version_command, + build_read_msd_command, build_reboot_command, build_write_config_command, ) @@ -68,6 +69,7 @@ parse_pipe_data_ack, parse_pipe_data_nack, parse_pipe_start_response, + parse_read_msd, unpack_ack_ranges, validate_ack_response, ) @@ -94,6 +96,7 @@ "TIMEOUT_PIPE_START", "build_read_config_command", "build_read_fw_version_command", + "build_read_msd_command", "build_enter_dfu_command", "build_deep_sleep_command", "build_reboot_command", @@ -132,6 +135,7 @@ "calculate_config_crc", "validate_ack_response", "parse_firmware_version", + "parse_read_msd", "PipeParams", "classify_pipe_frame", "parse_pipe_start_response", diff --git a/src/opendisplay/protocol/commands.py b/src/opendisplay/protocol/commands.py index 572b9dc..2e39704 100644 --- a/src/opendisplay/protocol/commands.py +++ b/src/opendisplay/protocol/commands.py @@ -20,6 +20,7 @@ class CommandCode(IntEnum): # Firmware commands READ_FW_VERSION = 0x0043 # Read firmware version + READ_MSD = 0x0044 # Read the 16-byte manufacturer-specific data record REBOOT = 0x000F # Reboot device # Authentication command (firmware with encryption support) @@ -127,6 +128,19 @@ def build_read_fw_version_command() -> bytes: return CommandCode.READ_FW_VERSION.to_bytes(2, byteorder="big") +def build_read_msd_command() -> bytes: + """Build command to read the manufacturer-specific data record. + + The MSD holds the same 16 bytes the device broadcasts in its advertisement, + including the dynamic block that carries live sensor readings -- so this is + the connected equivalent of listening for an advertisement. + + Returns: + Command bytes: 0x0044 (2 bytes, big-endian) + """ + return CommandCode.READ_MSD.to_bytes(2, byteorder="big") + + def build_reboot_command() -> bytes: """Build command to reboot device. diff --git a/src/opendisplay/protocol/responses.py b/src/opendisplay/protocol/responses.py index e8f24fd..25470b5 100644 --- a/src/opendisplay/protocol/responses.py +++ b/src/opendisplay/protocol/responses.py @@ -280,6 +280,35 @@ def parse_firmware_version(data: bytes) -> FirmwareVersion: } +MSD_LENGTH = 16 # company_id:2 + dynamic:11 + chip_temp:1 + battery_low:1 + status:1 + + +def parse_read_msd(data: bytes) -> bytes: + """Parse a READ_MSD response into the raw 16-byte MSD record. + + Format: [echo:2][msd:16] + + Args: + data: Raw READ_MSD response + + Returns: + The 16-byte manufacturer-specific data record, still carrying its + leading company ID -- ``parse_advertisement`` accepts it as-is. + + Raises: + InvalidResponseError: If the echo or length is wrong + """ + echo = unpack_command_code(data) if len(data) >= 2 else None + if echo not in (0x0044, 0x0044 | RESPONSE_HIGH_BIT_FLAG): + raise InvalidResponseError(f"READ_MSD echo mismatch: expected 0x0044, got {echo and f'0x{echo:04x}'}") + + msd = data[2:] + if len(msd) != MSD_LENGTH: + raise InvalidResponseError(f"READ_MSD payload must be {MSD_LENGTH} bytes, got {len(msd)}") + + return msd + + # ─── PIPE_WRITE (0x0080-0x0082) sliding-window responses ────────────────────── # 0x0080 START NACK error codes (Part 1 §1.1) diff --git a/src/opendisplay/sensors.py b/src/opendisplay/sensors.py new file mode 100644 index 0000000..88ca9f0 --- /dev/null +++ b/src/opendisplay/sensors.py @@ -0,0 +1,75 @@ +"""Read attached-sensor values out of a device's BLE advertisement. + +Sensor readings are not a separate protocol path: the firmware bit-packs them into +the 11-byte dynamic block that is broadcast in every advertisement, at the offset +recorded in the device's own config (TLV packet 0x23, ``SensorData``). Decoding one +therefore needs both halves -- the config for the offset, the advertisement for the +bytes -- which is what this module joins together. + +This is a convenience layer for scripts and the CLI. Consumers that render one +entity per measurement (Home Assistant, for example) should skip it and call +:meth:`~opendisplay.models.advertisement.AdvertisementData.sht40_reading` directly +with ``SensorData.sht40_msd_start_byte``, rather than rebuilding and searching a +list on every read. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .models.advertisement import AdvertisementData +from .models.config import GlobalConfig, SensorData +from .models.enums import SensorType + +__all__ = ["SensorReading", "read_sensor_values"] + + +@dataclass(frozen=True) +class SensorReading: + """One configured sensor's current measurements. + + Fields a given sensor type does not report stay None -- a temperature-only + sensor leaves ``humidity_percent`` unset. + """ + + instance_number: int + sensor_type: int + sensor_type_enum: SensorType | int + temperature_c: float | None = None + humidity_percent: float | None = None + + +def _read_one(sensor: SensorData, advertisement: AdvertisementData) -> SensorReading | None: + """Decode one configured sensor, or None if it has no readable value.""" + if sensor.sensor_type_enum is not SensorType.SHT40: + return None + + reading = advertisement.sht40_reading(sensor.sht40_msd_start_byte) + if reading is None: + return None + + return SensorReading( + instance_number=sensor.instance_number, + sensor_type=sensor.sensor_type, + sensor_type_enum=sensor.sensor_type_enum, + temperature_c=reading.temperature_c, + humidity_percent=reading.humidity_percent, + ) + + +def read_sensor_values(config: GlobalConfig, advertisement: AdvertisementData) -> list[SensorReading]: + """Decode every configured sensor that currently has a valid reading. + + Args: + config: Device config, read via ``OpenDisplayDevice.config``. Supplies both + which sensors exist and where each one's bytes live. + advertisement: A v1 advertisement from the same device. + + Returns: + One entry per sensor with a usable reading, in config order. Sensors of an + unsupported type, and sensors whose slot holds no valid measurement, are + omitted -- so an empty list means "nothing readable right now", not + "no sensors fitted". + """ + readings = (_read_one(sensor, advertisement) for sensor in config.sensors) + return [reading for reading in readings if reading is not None] diff --git a/tests/unit/test_cli_info_report.py b/tests/unit/test_cli_info_report.py index d8d571c..7496390 100644 --- a/tests/unit/test_cli_info_report.py +++ b/tests/unit/test_cli_info_report.py @@ -9,6 +9,7 @@ from __future__ import annotations +from dataclasses import replace from typing import Any import pytest @@ -31,7 +32,8 @@ TouchController, WifiConfig, ) -from opendisplay.models.enums import LedType +from opendisplay.models.enums import LedType, SensorType +from opendisplay.sensors import SensorReading _FW = {"major": 2, "minor": 26, "patch": 0, "sha": "987c6d9"} @@ -513,3 +515,90 @@ def test_report_survives_missing_config() -> None: out = _text(None) assert "OD405BD8" in out assert "2.26" in out + + +# ── sensor readings ────────────────────────────────────────────────────────── + + +def _sht40(instance: int = 0, start_byte: int = 7) -> SensorData: + return SensorData( + instance_number=instance, + sensor_type=SensorType.SHT40, + bus_id=2, + i2c_addr_7bit=0x44, + msd_data_start_byte=start_byte, + reserved=b"\x00" * 24, + ) + + +def _ctx_with_readings(config: GlobalConfig, readings: dict[int, SensorReading]) -> _InfoContext: + return replace(_ctx(config), sensor_readings=readings) + + +def _reading(instance: int = 0, temperature_c: float = 22.4, humidity_percent: float = 47.1) -> SensorReading: + return SensorReading( + instance_number=instance, + sensor_type=int(SensorType.SHT40), + sensor_type_enum=SensorType.SHT40, + temperature_c=temperature_c, + humidity_percent=humidity_percent, + ) + + +def _tree_text(ctx: _InfoContext) -> str: + console = Console(width=120, record=True, force_terminal=False) + console.print(_build_info_tree(ctx)) + return console.export_text() + + +def test_sensor_line_shows_live_values() -> None: + ctx = _ctx_with_readings(_config(sensors=[_sht40()]), {0: _reading()}) + + out = _tree_text(ctx) + + assert "SHT40" in out + assert "22.4 °C" in out + assert "47.1 %RH" in out + + +def test_sensor_line_without_reading_is_unchanged() -> None: + """A sensor we could not read still reports its hardware.""" + ctx = _ctx_with_readings(_config(sensors=[_sht40()]), {}) + + out = _tree_text(ctx) + + assert "Sensor 0" in out + assert "bus 2" in out + assert "°C" not in out.split("Sensors")[1] + + +def test_sensor_json_carries_readings() -> None: + ctx = _ctx_with_readings(_config(sensors=[_sht40()]), {0: _reading()}) + + entry = _info_to_json(ctx)["hardware"]["sensors"][0] + + assert entry["instance"] == 0 + assert entry["type"] == "SHT40" + assert entry["bus"] == 2 + assert entry["temperature_c"] == 22.4 + assert entry["humidity_percent"] == 47.1 + + +def test_sensor_json_nulls_when_unread() -> None: + ctx = _ctx_with_readings(_config(sensors=[_sht40()]), {}) + + entry = _info_to_json(ctx)["hardware"]["sensors"][0] + + assert entry["temperature_c"] is None + assert entry["humidity_percent"] is None + + +def test_readings_match_their_own_sensor_instance() -> None: + """Two sensors, one readable: the value must not leak onto the other row.""" + config = _config(sensors=[_sht40(instance=0), _sht40(instance=1, start_byte=1)]) + ctx = _ctx_with_readings(config, {1: _reading(instance=1, temperature_c=18.2)}) + + entries = _info_to_json(ctx)["hardware"]["sensors"] + + assert entries[0]["temperature_c"] is None + assert entries[1]["temperature_c"] == 18.2 diff --git a/tests/unit/test_models_advertisement.py b/tests/unit/test_models_advertisement.py index 6347603..98b7140 100644 --- a/tests/unit/test_models_advertisement.py +++ b/tests/unit/test_models_advertisement.py @@ -7,6 +7,7 @@ AdvertisementTracker, TouchTracker, decode_button_event, + decode_sht40, parse_advertisement, ) @@ -502,3 +503,127 @@ def test_tracker_with_no_button_bytes_emits_nothing() -> None: tracker.update("AA", parse_advertisement(_v1_payload(_dynamic(b0=0x28)))) assert tracker.update("AA", parse_advertisement(_v1_payload(_dynamic(b0=0x4C)))) == [] + + +# ── SHT40 sensor readings ───────────────────────────────────────────────────── + + +def _encode_sht40(temp_deci: int, rh_centi: int) -> bytes: + """Encode a reading exactly as the firmware does (sensor_sht40.cpp).""" + temp_units = temp_deci + 400 + rh_deci = (rh_centi + 5) // 10 + packed = (rh_deci & 0x3FF) | (temp_units << 10) + return bytes([packed & 0xFF, (packed >> 8) & 0xFF, (packed >> 16) & 0xFF]) + + +def _dynamic_with_sht40(block: bytes, start_byte: int = 7) -> bytes: + """An 11-byte dynamic block carrying a 3-byte SHT40 reading.""" + dynamic = bytearray(11) + dynamic[start_byte : start_byte + 3] = block + return bytes(dynamic) + + +def test_decode_sht40_known_vector() -> None: + """Decodes the firmware's own encoding of 22.4 C / 47.1 %RH.""" + reading = decode_sht40(bytes.fromhex("d7c109")) + + assert reading is not None + assert reading.temperature_c == pytest.approx(22.4) + assert reading.humidity_percent == pytest.approx(47.1) + assert reading.start_byte == 7 + assert reading.raw == bytes.fromhex("d7c109") + + +@pytest.mark.parametrize( + ("temp_deci", "rh_centi"), + [ + (-400, 100), # temperature floor + (1250, 10000), # both range ceilings + (0, 0), # humidity floor + (0, 5000), + (224, 4712), + (-123, 8888), + ], +) +def test_decode_sht40_round_trips_firmware_encoding(temp_deci: int, rh_centi: int) -> None: + """Every value the firmware can encode decodes back to itself.""" + reading = decode_sht40(_encode_sht40(temp_deci, rh_centi)) + + assert reading is not None + assert reading.temperature_c == pytest.approx(temp_deci / 10.0) + assert reading.humidity_percent == pytest.approx(round(rh_centi / 10) / 10.0) + + +def test_decode_sht40_read_failure_sentinel() -> None: + """FF FF FF is the firmware's explicit read-failure marker, not a reading.""" + assert decode_sht40(b"\xff\xff\xff") is None + + +def test_decode_sht40_unwritten_slot() -> None: + """00 00 00 decodes to a plausible -40 C / 0 %RH but means 'never written'. + + This is the one lossy case: a real -40.0 C at exactly 0 %RH encodes to the + same three zero bytes as a slot the firmware never touched. Reporting it is + the worse failure -- it would write a hard -40 C into consumers' long-term + statistics -- so the ambiguous pattern is treated as "no reading". + """ + assert decode_sht40(b"\x00\x00\x00") is None + assert decode_sht40(_encode_sht40(-400, 0)) is None + + +def test_decode_sht40_rejects_wrong_length() -> None: + with pytest.raises(ValueError, match="must be 3 bytes"): + decode_sht40(b"\x01\x02") + + +def test_sht40_reading_from_advertisement() -> None: + """A v1 advertisement decodes the block at the configured offset.""" + payload = _v1_payload(_dynamic_with_sht40(bytes.fromhex("d7c109"), start_byte=7)) + + reading = parse_advertisement(payload).sht40_reading(7) + + assert reading is not None + assert reading.temperature_c == pytest.approx(22.4) + assert reading.humidity_percent == pytest.approx(47.1) + + +def test_sht40_reading_defaults_to_firmware_slot() -> None: + """The default start byte matches the firmware's own default of 7.""" + payload = _v1_payload(_dynamic_with_sht40(bytes.fromhex("d7c109"), start_byte=7)) + + assert parse_advertisement(payload).sht40_reading() is not None + + +def test_sht40_reading_at_relocated_offset() -> None: + """Config can move the block anywhere that fits in the dynamic area.""" + payload = _v1_payload(_dynamic_with_sht40(bytes.fromhex("d7c109"), start_byte=0)) + adv = parse_advertisement(payload) + + assert adv.sht40_reading(0) is not None + assert adv.sht40_reading(7) is None # zeroed elsewhere → unwritten + + +@pytest.mark.parametrize("start_byte", [-1, 9, 11, 99]) +def test_sht40_reading_rejects_out_of_range_offset(start_byte: int) -> None: + """A 3-byte block must fit the 11-byte dynamic area (firmware caps start at 8).""" + payload = _v1_payload(_dynamic_with_sht40(bytes.fromhex("d7c109"))) + + assert parse_advertisement(payload).sht40_reading(start_byte) is None + + +def test_sht40_reading_is_none_for_legacy_advertisement() -> None: + """Legacy advertisements carry no dynamic block at all.""" + legacy = bytes.fromhex("0236006c00c301") + b"\x6e\x0f" + b"\x16" + b"\x01" + + assert parse_advertisement(legacy).sht40_reading() is None + + +def test_chip_temperature_is_not_the_sensor_reading() -> None: + """The advertisement's temperature_c is the MCU's, distinct from the SHT40's.""" + payload = _v1_payload(_dynamic_with_sht40(bytes.fromhex("d7c109")), temperature_c=31.0) + adv = parse_advertisement(payload) + + reading = adv.sht40_reading() + assert reading is not None + assert adv.temperature_c == pytest.approx(31.0) + assert reading.temperature_c == pytest.approx(22.4) diff --git a/tests/unit/test_models_config.py b/tests/unit/test_models_config.py index 9587d81..40516c2 100644 --- a/tests/unit/test_models_config.py +++ b/tests/unit/test_models_config.py @@ -2,7 +2,7 @@ import pytest -from opendisplay.models.config import BinaryInputs, DisplayConfig, ManufacturerData, PowerOption +from opendisplay.models.config import BinaryInputs, DisplayConfig, ManufacturerData, PowerOption, SensorData from opendisplay.models.enums import ( BoardManufacturer, DIYBoardType, @@ -197,3 +197,22 @@ def test_returns_none_for_not_published(self) -> None: def test_returns_none_for_index_past_block(self) -> None: assert self._inputs(11).published_button_byte_index is None + + +class TestSensorDataMsdStartByte: + """The firmware treats 0 and 0xFF as 'use the default slot' (sht40_msd_start).""" + + def test_zero_means_default_slot(self) -> None: + sensor = SensorData(instance_number=0, sensor_type=4, bus_id=1, msd_data_start_byte=0) + + assert sensor.sht40_msd_start_byte == 7 + + def test_ff_means_default_slot(self) -> None: + sensor = SensorData(instance_number=0, sensor_type=4, bus_id=1, msd_data_start_byte=0xFF) + + assert sensor.sht40_msd_start_byte == 7 + + def test_explicit_offset_is_kept(self) -> None: + sensor = SensorData(instance_number=0, sensor_type=4, bus_id=1, msd_data_start_byte=3) + + assert sensor.sht40_msd_start_byte == 3 diff --git a/tests/unit/test_protocol_responses.py b/tests/unit/test_protocol_responses.py index f83e2ad..250efef 100644 --- a/tests/unit/test_protocol_responses.py +++ b/tests/unit/test_protocol_responses.py @@ -5,6 +5,7 @@ import pytest from opendisplay.exceptions import InvalidResponseError, NfcNotSupportedError, NfcWriteError, ProtocolError +from opendisplay.models.advertisement import parse_advertisement from opendisplay.protocol.commands import CommandCode from opendisplay.protocol.responses import ( NFC_ERROR_MESSAGES, @@ -12,6 +13,7 @@ NFC_STATUS_WRITE_OK, check_response_type, parse_firmware_version, + parse_read_msd, strip_command_echo, unpack_command_code, validate_ack_response, @@ -341,3 +343,38 @@ def test_nfc_not_supported_error_hedges_in_message(self): """ exc = NfcNotSupportedError() assert "may not support" in str(exc).lower() + + +class TestParseReadMsd: + """READ_MSD (0x0044) returns the same 16 bytes the device advertises.""" + + MSD = bytes.fromhex("4624") + bytes(11) + bytes([125, 200, 0x10]) + + def test_parses_payload(self) -> None: + assert parse_read_msd(bytes.fromhex("0044") + self.MSD) == self.MSD + + def test_accepts_ack_high_bit_echo(self) -> None: + assert parse_read_msd(bytes.fromhex("8044") + self.MSD) == self.MSD + + def test_rejects_wrong_echo(self) -> None: + with pytest.raises(InvalidResponseError, match="echo mismatch"): + parse_read_msd(bytes.fromhex("0043") + self.MSD) + + def test_rejects_short_response(self) -> None: + with pytest.raises(InvalidResponseError, match="echo mismatch"): + parse_read_msd(b"\x00") + + def test_rejects_wrong_payload_length(self) -> None: + with pytest.raises(InvalidResponseError, match="must be 16 bytes"): + parse_read_msd(bytes.fromhex("0044") + self.MSD[:-1]) + + def test_output_feeds_advertisement_parser(self) -> None: + """The record keeps its company ID, which parse_advertisement strips.""" + msd = bytes.fromhex("4624") + bytes(7) + bytes.fromhex("d7c109") + bytes(1) + bytes([125, 200, 0x10]) + + adv = parse_advertisement(parse_read_msd(bytes.fromhex("0044") + msd)) + + assert adv.format_version == "v1" + reading = adv.sht40_reading() + assert reading is not None + assert reading.temperature_c == 22.4 diff --git a/tests/unit/test_sensors.py b/tests/unit/test_sensors.py new file mode 100644 index 0000000..8e309fc --- /dev/null +++ b/tests/unit/test_sensors.py @@ -0,0 +1,110 @@ +"""Test joining device config with advertisement data to read sensor values.""" + +from opendisplay.models.advertisement import AdvertisementData, parse_advertisement +from opendisplay.models.config import ( + GlobalConfig, + ManufacturerData, + PowerOption, + SensorData, + SystemConfig, +) +from opendisplay.models.enums import SensorType +from opendisplay.sensors import read_sensor_values + +SHT40_BLOCK = bytes.fromhex("d7c109") # 22.4 C / 47.1 %RH + + +def _advertisement(block: bytes = SHT40_BLOCK, start_byte: int = 7) -> AdvertisementData: + """A v1 advertisement carrying one SHT40 reading at ``start_byte``.""" + dynamic = bytearray(11) + dynamic[start_byte : start_byte + 3] = block + return parse_advertisement(bytes(dynamic) + bytes([124, 139, 0x00])) + + +def _config(*sensors: SensorData) -> GlobalConfig: + """A config carrying only what read_sensor_values looks at.""" + return GlobalConfig( + system=SystemConfig(ic_type=2, communication_modes=0x05, device_flags=0, pwr_pin=0xFF, reserved=b""), + manufacturer=ManufacturerData(manufacturer_id=1, board_type=1, board_revision=1, reserved=b""), + power=PowerOption( + power_mode=1, + battery_capacity_mah=(3000).to_bytes(3, "little"), + sleep_timeout_ms=40000, + tx_power=8, + sleep_flags=0, + battery_sense_pin=1, + battery_sense_enable_pin=0x28, + battery_sense_flags=0, + capacity_estimator=5, + voltage_scaling_factor=0xA1, + deep_sleep_current_ua=0, + deep_sleep_time_seconds=0, + charge_enable_pin=0, + charge_state_pin=0, + charger_flags=0, + min_wake_time_seconds=0, + screen_timeout_seconds=0, + reserved=b"", + ), + sensors=list(sensors), + ) + + +def test_reads_sht40_at_default_slot() -> None: + config = _config(SensorData(instance_number=0, sensor_type=SensorType.SHT40, bus_id=1)) + + readings = read_sensor_values(config, _advertisement()) + + assert len(readings) == 1 + assert readings[0].instance_number == 0 + assert readings[0].sensor_type_enum is SensorType.SHT40 + assert readings[0].temperature_c == 22.4 + assert readings[0].humidity_percent == 47.1 + + +def test_reads_sht40_at_relocated_slot() -> None: + """The offset comes from config, not from the default.""" + config = _config( + SensorData(instance_number=0, sensor_type=SensorType.SHT40, bus_id=1, msd_data_start_byte=2), + ) + + readings = read_sensor_values(config, _advertisement(start_byte=2)) + + assert len(readings) == 1 + assert readings[0].temperature_c == 22.4 + + +def test_ignores_unsupported_sensor_type() -> None: + """A sensor we cannot decode is omitted rather than guessed at.""" + config = _config(SensorData(instance_number=0, sensor_type=SensorType.AXP2101_PMIC, bus_id=1)) + + assert read_sensor_values(config, _advertisement()) == [] + + +def test_omits_sensor_without_valid_reading() -> None: + """A configured-but-failing sensor reports nothing, not a bogus value.""" + config = _config(SensorData(instance_number=0, sensor_type=SensorType.SHT40, bus_id=1)) + + assert read_sensor_values(config, _advertisement(block=b"\xff\xff\xff")) == [] + + +def test_no_sensors_configured() -> None: + assert read_sensor_values(_config(), _advertisement()) == [] + + +def test_multiple_sensors_keep_config_order() -> None: + """Each sensor is decoded from its own slot, in the order config lists them.""" + dynamic = bytearray(11) + dynamic[7:10] = SHT40_BLOCK # instance 0, at the default slot + dynamic[4:7] = bytes.fromhex("f44106") # instance 1, 0.0 C / 50.0 %RH + adv = parse_advertisement(bytes(dynamic) + bytes([124, 139, 0x00])) + config = _config( + SensorData(instance_number=1, sensor_type=SensorType.SHT40, bus_id=1, msd_data_start_byte=4), + SensorData(instance_number=0, sensor_type=SensorType.SHT40, bus_id=1, msd_data_start_byte=0), + ) + + readings = read_sensor_values(config, adv) + + assert [r.instance_number for r in readings] == [1, 0] + assert readings[0].temperature_c == 0.0 + assert readings[1].temperature_c == 22.4