From c89cea495342f8bc6247abc305e2e0b9e2c46fd6 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Mon, 22 Jun 2026 13:03:50 +0200 Subject: [PATCH] M5-T03: add Modbus TCP driver, YAML profile framework, sdm120, modbus_cli --- app/integrations/modbus/__init__.py | 1 + app/integrations/modbus/driver.py | 203 +++++++++++ app/integrations/modbus/profiles.py | 231 +++++++++++++ app/integrations/modbus/profiles/sdm120.yaml | 74 ++++ docs/design/m5-iot-energy.md | 2 +- requirements.in | 1 + requirements.txt | 2 + scripts/modbus_cli.py | 338 +++++++++++++++++++ tests/test_modbus_cli.py | 289 ++++++++++++++++ tests/test_modbus_driver.py | 232 +++++++++++++ tests/test_modbus_profiles.py | 259 ++++++++++++++ 11 files changed, 1631 insertions(+), 1 deletion(-) create mode 100644 app/integrations/modbus/__init__.py create mode 100644 app/integrations/modbus/driver.py create mode 100644 app/integrations/modbus/profiles.py create mode 100644 app/integrations/modbus/profiles/sdm120.yaml create mode 100644 scripts/modbus_cli.py create mode 100644 tests/test_modbus_cli.py create mode 100644 tests/test_modbus_driver.py create mode 100644 tests/test_modbus_profiles.py diff --git a/app/integrations/modbus/__init__.py b/app/integrations/modbus/__init__.py new file mode 100644 index 0000000..edfa03d --- /dev/null +++ b/app/integrations/modbus/__init__.py @@ -0,0 +1 @@ +"""Modbus integration package — driver, profile loader, and decoder.""" diff --git a/app/integrations/modbus/driver.py b/app/integrations/modbus/driver.py new file mode 100644 index 0000000..64d9ddf --- /dev/null +++ b/app/integrations/modbus/driver.py @@ -0,0 +1,203 @@ +"""Modbus TCP driver — thin wrapper around pymodbus. + +This module provides a single public function ``read_blocks`` that performs +one or more FC04 (Read Input Registers) block reads against a Modbus TCP +gateway and returns a flat ``dict[register_address -> 16-bit_value]`` map. + +Design decisions +---------------- +- **Only read function codes** (FC03/FC04) are exposed. There is no write path. +- float32 decoding uses ``struct.unpack('>f', ...)`` directly rather than the + pymodbus ``BinaryPayloadDecoder`` helper, which has had API churn across 3.x + releases. Big-endian word order + big-endian byte order means the 4 raw bytes + are already in standard network (big-endian) order. +- ``ModbusTcpClient`` is used in synchronous mode (``connect()`` / ``close()``). + The client is created fresh per call; this keeps the driver stateless and + avoids threading concerns at the cost of one TCP handshake per read cycle. + APScheduler jobs call this from a thread pool, so stateless is safer. +- pymodbus 3.13.x uses ``device_id=`` as the slave-address keyword argument + (renamed from ``slave=`` in earlier 3.x releases). + +Exceptions +---------- +``ModbusDriverError`` + Base exception for all driver-level errors. +``ModbusConnectionError`` + Raised when the TCP connection to the gateway cannot be established or is + lost during the read. +``ModbusResponseError`` + Raised when the gateway responds with a Modbus exception frame or when the + returned register count does not match the requested count. +""" + +from __future__ import annotations + +import logging +import struct +from typing import TYPE_CHECKING + +from pymodbus.client import ModbusTcpClient +from pymodbus.exceptions import ConnectionException, ModbusException + +if TYPE_CHECKING: + from collections.abc import Sequence + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Custom exceptions +# --------------------------------------------------------------------------- + + +class ModbusDriverError(RuntimeError): + """Base class for all Modbus driver errors.""" + + +class ModbusConnectionError(ModbusDriverError): + """Cannot connect to (or lost connection with) the Modbus TCP gateway.""" + + +class ModbusResponseError(ModbusDriverError): + """Modbus gateway returned an exception frame or an unexpected response.""" + + +# --------------------------------------------------------------------------- +# Block definition type alias +# --------------------------------------------------------------------------- + +#: A single read block: ``{"start": int, "count": int}``. +Block = dict[str, int] + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def read_blocks( + host: str, + port: int, + unit_id: int, + blocks: Sequence[Block], + *, + timeout: float = 3.0, +) -> dict[int, int]: + """Read one or more contiguous register blocks via FC04 (input registers). + + Parameters + ---------- + host: + Hostname or IP address of the Modbus TCP gateway. + port: + TCP port (typically 502). + unit_id: + Modbus slave / unit address (the device's Meter ID for SDM120). + blocks: + Sequence of ``{"start": int, "count": int}`` dicts describing the + contiguous register ranges to read. ``count`` is the number of + 16-bit registers (not bytes). + timeout: + TCP connect/read timeout in seconds (default 3 s). + + Returns + ------- + dict[int, int] + Flat mapping of ``{register_address: 16-bit_register_value}`` for + every register returned across all blocks. + + Raises + ------ + ModbusConnectionError + If the TCP connection to the gateway fails. + ModbusResponseError + If the gateway returns a Modbus exception frame or an unexpected + number of registers. + """ + client = ModbusTcpClient(host, port=port, timeout=timeout) + try: + connected = client.connect() + if not connected: + raise ModbusConnectionError( + f"Could not connect to Modbus gateway at {host}:{port}" + ) + + registers: dict[int, int] = {} + for block in blocks: + start: int = block["start"] + count: int = block["count"] + _read_block(client, unit_id, start, count, registers) + + return registers + + except ConnectionException as exc: + raise ModbusConnectionError( + f"Connection to Modbus gateway {host}:{port} failed: {exc}" + ) from exc + except ModbusException as exc: + raise ModbusResponseError(f"Modbus protocol error: {exc}") from exc + finally: + client.close() + + +def _read_block( + client: ModbusTcpClient, + unit_id: int, + start: int, + count: int, + result: dict[int, int], +) -> None: + """Read one block and merge into *result*. Raises on any error.""" + try: + response = client.read_input_registers(start, count=count, device_id=unit_id) + except ConnectionException as exc: + raise ModbusConnectionError( + f"Lost connection while reading registers 0x{start:04X}+{count}: {exc}" + ) from exc + + if response.isError(): + raise ModbusResponseError( + f"Modbus exception reading registers 0x{start:04X}+{count}: {response}" + ) + + regs = response.registers + if len(regs) != count: + raise ModbusResponseError( + f"Expected {count} registers from 0x{start:04X}, got {len(regs)}" + ) + + for offset, value in enumerate(regs): + result[start + offset] = value + + +# --------------------------------------------------------------------------- +# Decoding helpers +# --------------------------------------------------------------------------- + + +def registers_to_float(hi_reg: int, lo_reg: int) -> float: + """Decode two 16-bit registers to a big-endian IEEE-754 float32. + + The SDM120 (and most Modbus float devices) use big-endian word order: + the high 16-bit register comes first, then the low register. Combined + with big-endian byte order within each register, the raw 4-byte sequence + is standard network-byte-order (big-endian) float32. + + Parameters + ---------- + hi_reg: + The first (higher-address-range, most-significant) 16-bit register. + lo_reg: + The second (lower-address-range, least-significant) 16-bit register. + + Returns + ------- + float + The decoded IEEE-754 single-precision floating-point value. + + Example + ------- + >>> registers_to_float(0x4366, 0x3334) + 230.20001220703125 + """ + raw = struct.pack(">HH", hi_reg, lo_reg) + return struct.unpack(">f", raw)[0] diff --git a/app/integrations/modbus/profiles.py b/app/integrations/modbus/profiles.py new file mode 100644 index 0000000..4d82f83 --- /dev/null +++ b/app/integrations/modbus/profiles.py @@ -0,0 +1,231 @@ +"""Modbus profile loader, validator, and decoder. + +A *profile* is a YAML file that describes the protocol-level knowledge for +a particular Modbus device model: + +- Which registers to read (``blocks`` for efficient bulk reads). +- How to decode each quantity (``metrics``: register address, type, unit, …). +- Home Assistant metadata per metric (``device_class``, ``ha_component``, …). + +Profiles live in ``app/integrations/modbus/profiles/.yaml`` and are +located at runtime relative to *this file* (not CWD), so they work whether +the package is installed as a wheel, run in-process, or invoked via +``python -m scripts.modbus_cli``. + +Design notes +------------ +- **Purely data + functions** — no abstract base classes or inheritance. +- ``ModbusProfile`` and ``MetricSpec`` are Pydantic models: they validate on + construction and raise ``pydantic.ValidationError`` for malformed YAML. +- ``decode`` uses ``driver.registers_to_float`` for float32 quantities. + Unsupported types are silently skipped with a warning (future-proofing for + int16/uint16 etc.). +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Optional + +import yaml +from pydantic import BaseModel, ValidationError + +from app.integrations.modbus.driver import registers_to_float + +logger = logging.getLogger(__name__) + +# Directory containing the YAML profiles, located relative to *this* file. +_PROFILES_DIR = Path(__file__).parent / "profiles" + + +# --------------------------------------------------------------------------- +# Pydantic models for profile validation +# --------------------------------------------------------------------------- + + +class MetricSpec(BaseModel): + """Specification for a single measurable quantity within a profile.""" + + key: str + """Stable identifier used as the key in decoded payloads (e.g. ``"voltage"``).""" + + address: int + """Starting Modbus register address (0-based, e.g. ``0x0000`` for voltage).""" + + type: str + """Encoding type — currently only ``"float32"`` is supported.""" + + unit: str + """Physical unit string (e.g. ``"V"``, ``"A"``, ``"kWh"``). Empty string for dimensionless.""" + + device_class: str + """Home Assistant device class (e.g. ``"voltage"``, ``"power"``, ``"energy"``).""" + + ha_component: str + """Home Assistant component type (e.g. ``"sensor"``).""" + + state_class: Optional[str] = None + """Home Assistant state class, if applicable (e.g. ``"total_increasing"``).""" + + +class BlockSpec(BaseModel): + """A contiguous register block for bulk reading.""" + + start: int + """First register address in the block.""" + + count: int + """Number of 16-bit registers to read.""" + + +class ModbusProfile(BaseModel): + """Complete description of a Modbus device's protocol-level knowledge.""" + + name: str + """Short identifier matching the YAML file name (e.g. ``"sdm120"``).""" + + description: str + """Human-readable description of the device.""" + + function_code: int + """Modbus function code for reading measurements (4 = FC04 input registers).""" + + word_order: str + """Word (register) order: ``"big"`` means high register first.""" + + byte_order: str + """Byte order within each register: ``"big"`` means standard network order.""" + + blocks: list[BlockSpec] + """Register blocks to read in bulk, reducing Modbus transaction count.""" + + metrics: list[MetricSpec] + """List of individual measurable quantities and their decoding rules.""" + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +class ProfileNotFoundError(FileNotFoundError): + """Raised when the requested profile YAML file does not exist.""" + + +class ProfileValidationError(ValueError): + """Raised when a profile YAML file fails Pydantic validation.""" + + +def load_profile(name: str) -> ModbusProfile: + """Load and validate a Modbus device profile by name. + + The profile file is expected at + ``app/integrations/modbus/profiles/.yaml``. + + Parameters + ---------- + name: + Profile name without extension (e.g. ``"sdm120"``). + + Returns + ------- + ModbusProfile + Validated profile model. + + Raises + ------ + ProfileNotFoundError + If ``profiles/.yaml`` does not exist. + ProfileValidationError + If the YAML is syntactically valid but the content fails schema + validation (missing required fields, wrong types, etc.). + """ + path = _PROFILES_DIR / f"{name}.yaml" + if not path.exists(): + raise ProfileNotFoundError( + f"Modbus profile '{name}' not found (looked for {path})" + ) + + with path.open("r", encoding="utf-8") as fh: + raw = yaml.safe_load(fh) + + try: + profile = ModbusProfile.model_validate(raw) + except ValidationError as exc: + raise ProfileValidationError( + f"Profile '{name}' failed validation: {exc}" + ) from exc + + return profile + + +def list_profiles() -> list[tuple[str, str]]: + """Return ``(name, description)`` pairs for all available profiles. + + Scans ``app/integrations/modbus/profiles/*.yaml``, loads each, and + returns a sorted list of ``(name, description)`` tuples. Profiles that + fail to load are skipped with a warning. + """ + results: list[tuple[str, str]] = [] + if not _PROFILES_DIR.exists(): + return results + + for path in sorted(_PROFILES_DIR.glob("*.yaml")): + name = path.stem + try: + profile = load_profile(name) + results.append((profile.name, profile.description)) + except (ProfileNotFoundError, ProfileValidationError, Exception) as exc: + logger.warning("Skipping malformed profile '%s': %s", name, exc) + + return results + + +def decode(profile: ModbusProfile, registers: dict[int, int]) -> dict[str, float]: + """Decode raw register values into a mapping of ``{key: float_value}``. + + For each metric in *profile*, the function looks up the two registers at + ``address`` and ``address + 1`` in *registers* and decodes them as a + big-endian float32 (high register first). + + Metrics whose registers are absent from *registers* are skipped (not + an error — this can happen if a block was not requested or a device does + not populate that register). Metrics with unsupported ``type`` values + are also skipped. + + Parameters + ---------- + profile: + Validated ``ModbusProfile`` describing the device. + registers: + Mapping of ``{register_address: 16-bit_value}`` as returned by + ``driver.read_blocks``. + + Returns + ------- + dict[str, float] + Decoded engineering values keyed by each metric's ``key`` field. + """ + result: dict[str, float] = {} + for metric in profile.metrics: + if metric.type == "float32": + hi_addr = metric.address + lo_addr = metric.address + 1 + if hi_addr not in registers or lo_addr not in registers: + logger.debug( + "Skipping metric '%s': registers 0x%04X/0x%04X not available", + metric.key, + hi_addr, + lo_addr, + ) + continue + value = registers_to_float(registers[hi_addr], registers[lo_addr]) + result[metric.key] = value + else: + logger.warning( + "Metric '%s' has unsupported type '%s', skipping", + metric.key, + metric.type, + ) + return result diff --git a/app/integrations/modbus/profiles/sdm120.yaml b/app/integrations/modbus/profiles/sdm120.yaml new file mode 100644 index 0000000..498a182 --- /dev/null +++ b/app/integrations/modbus/profiles/sdm120.yaml @@ -0,0 +1,74 @@ +name: sdm120 +description: Eastron SDM120 single-phase energy meter +function_code: 4 # input registers (FC04) +word_order: big # high register first (big-endian word order) +byte_order: big # high byte first within each register + +blocks: + # Registers 30001–30095 (0x0000–0x005F): voltage through max export demand + - { start: 0x0000, count: 0x0060 } + # Registers 30343–30346 (0x0156–0x0159): total active/reactive energy + - { start: 0x0156, count: 0x0004 } + +metrics: + # Core measurements — addresses from SDM120 Modbus Protocol §4 (FC04 input registers) + # Each float32 occupies two consecutive 16-bit registers; address = Hex start in §4. + + - key: voltage + address: 0x0000 # register 30001 — Voltage (V) + type: float32 + unit: "V" + device_class: voltage + ha_component: sensor + + - key: current + address: 0x0006 # register 30007 — Current (A) + type: float32 + unit: "A" + device_class: current + ha_component: sensor + + - key: active_power + address: 0x000C # register 30013 — Active power (W) + type: float32 + unit: "W" + device_class: power + ha_component: sensor + + - key: power_factor + address: 0x001E # register 30031 — Power factor (dimensionless) + type: float32 + unit: "" + device_class: power_factor + ha_component: sensor + + - key: frequency + address: 0x0046 # register 30071 — Frequency (Hz) + type: float32 + unit: "Hz" + device_class: frequency + ha_component: sensor + + - key: import_energy + address: 0x0048 # register 30073 — Import active energy (kWh) + type: float32 + unit: "kWh" + device_class: energy + state_class: total_increasing + ha_component: sensor + + - key: export_energy + address: 0x004A # register 30075 — Export active energy (kWh) + type: float32 + unit: "kWh" + device_class: energy + state_class: total_increasing + ha_component: sensor + + - key: total_energy + address: 0x0156 # register 30343 — Total active energy (kWh) + type: float32 + unit: "kWh" + device_class: energy + state_class: total_increasing + ha_component: sensor diff --git a/docs/design/m5-iot-energy.md b/docs/design/m5-iot-energy.md index 9077565..2e72163 100644 --- a/docs/design/m5-iot-energy.md +++ b/docs/design/m5-iot-energy.md @@ -306,7 +306,7 @@ Phase C（MQTT / Discovery，依赖 B 的设备数据与 provider 接口） - **Reviewer checklist**: 列/约束与 §3.2 一致；`payload` 为 JSON 列；`recorded_at` 为真实索引列（非塞进 payload）；链上单 head；`env.py` 已 import 新模型（否则 autogenerate/建表漏表）。 ### M5-T03 — Modbus 驱动 + YAML profile 框架（`sdm120`） -- **Status**: `todo` · **Depends**: M5-T02 +- **Status**: `done` · **Depends**: M5-T02 - **Context**: 薄封装 pymodbus 的连接/块读/大端 float 解码，加 YAML profile 加载/校验/解码框架与 SDM120 profile。纯模块，mock client 单测。 - **Files**: `create app/integrations/modbus/__init__.py`、`app/integrations/modbus/driver.py`、`app/integrations/modbus/profiles.py`、`app/integrations/modbus/profiles/sdm120.yaml`、`scripts/modbus_cli.py`；`modify requirements.in`/`requirements.txt`（加 `pymodbus`、`PyYAML`，重新锁定）；`create tests/test_modbus_driver.py`、`tests/test_modbus_profiles.py`、`tests/test_modbus_cli.py` - **Steps**: diff --git a/requirements.in b/requirements.in index 1eb089d..03db00e 100644 --- a/requirements.in +++ b/requirements.in @@ -3,6 +3,7 @@ apscheduler>=3.10,<4.0 argon2-cffi>=25.1,<26.0 fastapi>=0.115,<0.116 httpx>=0.28,<1.0 +pymodbus>=3.6,<4.0 pydantic-settings>=2.6,<3.0 pyotp>=2.9,<3.0 python-multipart>=0.0.12,<1.0 diff --git a/requirements.txt b/requirements.txt index 389f4e7..1a8acb5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -59,6 +59,8 @@ pydantic-core==2.46.2 # via pydantic pydantic-settings==2.13.1 # via -r requirements.in +pymodbus==3.13.1 + # via -r requirements.in pyotp==2.10.0 # via -r requirements.in python-dotenv==1.2.2 diff --git a/scripts/modbus_cli.py b/scripts/modbus_cli.py new file mode 100644 index 0000000..39f917d --- /dev/null +++ b/scripts/modbus_cli.py @@ -0,0 +1,338 @@ +"""Modbus CLI — read-only manual test tool for Modbus TCP devices. + +Entry point:: + + python -m scripts.modbus_cli [options] + +Subcommands +----------- +read --host H --port P --unit U --profile NAME + Read all metrics defined in the named profile from the device and print + decoded engineering values as a formatted table. Use this to verify the + full decode chain end-to-end. + +probe --host H --port P --unit U --fc {3,4} --address ADDR --count N + [--decode float32] + Send a raw FC03 (holding) or FC04 (input) read request for exactly N + registers starting at ADDR, and print the raw 16-bit register values in + hex. Optionally decode consecutive register pairs as big-endian float32. + Use this for first-contact verification of gateway reachability, framer + mode, and unit_id before setting up a full profile. + +Design +------ +- **Read-only**: only FC03 and FC04 are exposed. There is no write path. + This prevents accidental modification of device configuration registers + (Meter ID, baud rate, etc.) which could cause communication loss. +- **No DB dependency**: the tool connects directly to the gateway without + reading from the application database. It is safe to use before any + device has been configured in the application. +- **Non-zero exit on connection failure** so shell scripts can detect errors. +""" + +from __future__ import annotations + +import argparse +import sys +from pathlib import Path + +# Ensure the project root is on sys.path when invoked as ``python -m scripts.modbus_cli`` +# or directly. +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from app.integrations.modbus.driver import ( + ModbusConnectionError, + ModbusDriverError, + ModbusResponseError, + registers_to_float, +) +from app.integrations.modbus.profiles import ProfileNotFoundError, ProfileValidationError, decode, load_profile + +# --------------------------------------------------------------------------- +# Helper — raw FC03/FC04 read (used by ``probe``) +# --------------------------------------------------------------------------- + + +def _raw_read( + host: str, + port: int, + unit_id: int, + fc: int, + address: int, + count: int, + timeout: float = 5.0, +) -> list[int]: + """Perform a raw FC03 or FC04 read and return the register list. + + Raises ``ModbusConnectionError`` or ``ModbusResponseError`` on failure. + """ + from pymodbus.client import ModbusTcpClient + from pymodbus.exceptions import ConnectionException + + client = ModbusTcpClient(host, port=port, timeout=timeout) + try: + connected = client.connect() + if not connected: + raise ModbusConnectionError( + f"Could not connect to Modbus gateway at {host}:{port}" + ) + + if fc == 4: + response = client.read_input_registers(address, count=count, device_id=unit_id) + elif fc == 3: + response = client.read_holding_registers(address, count=count, device_id=unit_id) + else: + raise ValueError(f"Unsupported function code FC{fc:02d} (only FC03/FC04 are allowed)") + + if response.isError(): + raise ModbusResponseError( + f"Modbus exception reading FC{fc:02d} registers 0x{address:04X}+{count}: " + f"{response}" + ) + + regs = response.registers + if len(regs) != count: + raise ModbusResponseError( + f"Expected {count} registers, got {len(regs)}" + ) + return list(regs) + + except ConnectionException as exc: + raise ModbusConnectionError( + f"Connection to {host}:{port} failed: {exc}" + ) from exc + finally: + client.close() + + +# --------------------------------------------------------------------------- +# Sub-command: read +# --------------------------------------------------------------------------- + + +def cmd_read(args: argparse.Namespace) -> None: + """Read all metrics from a named profile and print decoded values.""" + profile_name: str = args.profile + host: str = args.host + port: int = args.port + unit_id: int = args.unit + + # Load and validate the profile. + try: + profile = load_profile(profile_name) + except ProfileNotFoundError as exc: + print(f"Error: {exc}", file=sys.stderr) + sys.exit(1) + except ProfileValidationError as exc: + print(f"Error: {exc}", file=sys.stderr) + sys.exit(1) + + print(f"Profile : {profile.name} — {profile.description}") + print(f"Gateway : {host}:{port} unit_id={unit_id}") + print(f"Blocks : {[(b.start, b.count) for b in profile.blocks]}") + print() + + # Convert profile blocks to the dict format expected by read_blocks. + blocks = [{"start": b.start, "count": b.count} for b in profile.blocks] + + # Perform the read. + from app.integrations.modbus.driver import read_blocks + + try: + registers = read_blocks(host, port, unit_id, blocks) + except ModbusConnectionError as exc: + print(f"Connection error: {exc}", file=sys.stderr) + sys.exit(1) + except ModbusResponseError as exc: + print(f"Modbus response error: {exc}", file=sys.stderr) + sys.exit(1) + except ModbusDriverError as exc: + print(f"Modbus driver error: {exc}", file=sys.stderr) + sys.exit(1) + + # Decode. + values = decode(profile, registers) + + # Print as table. + print(f"{'Metric':<20} {'Value':>14} Unit") + print("-" * 45) + for metric in profile.metrics: + key = metric.key + if key in values: + unit = metric.unit if metric.unit else "—" + print(f"{key:<20} {values[key]:>14.4f} {unit}") + else: + print(f"{key:<20} {'(not available)':>14}") + + +# --------------------------------------------------------------------------- +# Sub-command: probe +# --------------------------------------------------------------------------- + + +def cmd_probe(args: argparse.Namespace) -> None: + """Perform a raw FC03/FC04 read and print raw register values.""" + host: str = args.host + port: int = args.port + unit_id: int = args.unit + fc: int = args.fc + address: int = args.address + count: int = args.count + do_decode: bool = args.decode == "float32" + + if fc not in (3, 4): + print(f"Error: --fc must be 3 or 4 (got {fc})", file=sys.stderr) + sys.exit(1) + + print(f"Gateway : {host}:{port} unit_id={unit_id}") + print(f"Request : FC{fc:02d} start=0x{address:04X} count={count}") + print() + + try: + regs = _raw_read(host, port, unit_id, fc, address, count) + except ModbusConnectionError as exc: + print(f"Connection error: {exc}", file=sys.stderr) + sys.exit(1) + except ModbusResponseError as exc: + print(f"Modbus response error: {exc}", file=sys.stderr) + sys.exit(1) + except ModbusDriverError as exc: + print(f"Modbus driver error: {exc}", file=sys.stderr) + sys.exit(1) + + # Print raw register values. + print(f"{'Addr':>8} {'Hex':>6} {'Dec':>7}") + print("-" * 28) + for i, val in enumerate(regs): + print(f"0x{address + i:04X} 0x{val:04X} {val:>7d}") + + # Optional float32 decoding — consecutive register pairs. + if do_decode and len(regs) >= 2: + print() + print("float32 decode (big-endian, high register first):") + print(f"{'Addr pair':>14} {'Float value':>16}") + print("-" * 35) + for i in range(0, len(regs) - 1, 2): + hi = regs[i] + lo = regs[i + 1] + fval = registers_to_float(hi, lo) + print(f"0x{address + i:04X}+0x{address + i + 1:04X} {fval:>16.6f}") + + +# --------------------------------------------------------------------------- +# Argument parser +# --------------------------------------------------------------------------- + + +def _auto_int(value: str) -> int: + """Parse an integer that may be hex (``0x...``) or decimal.""" + return int(value, 0) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="python -m scripts.modbus_cli", + description=( + "Modbus CLI — read-only manual test tool for Modbus TCP devices. " + "Use 'read' to decode all profile metrics; use 'probe' for raw " + "first-contact register inspection." + ), + ) + subparsers = parser.add_subparsers(dest="command", metavar="") + subparsers.required = True + + # -- read -- + rp = subparsers.add_parser( + "read", + help=( + "Read and decode all metrics defined in a profile. " + "Prints a table of engineering values." + ), + ) + rp.add_argument("--host", required=True, metavar="HOST", help="Gateway IP/hostname.") + rp.add_argument( + "--port", type=int, default=502, metavar="PORT", help="Gateway TCP port (default 502)." + ) + rp.add_argument( + "--unit", + type=int, + required=True, + metavar="UNIT_ID", + help="Modbus slave / unit ID (Meter ID, typically 1).", + ) + rp.add_argument( + "--profile", + required=True, + metavar="NAME", + help="Profile name (e.g. 'sdm120').", + ) + rp.set_defaults(func=cmd_read) + + # -- probe -- + pp = subparsers.add_parser( + "probe", + help=( + "Raw read of N registers starting at ADDR via FC03 or FC04. " + "Prints raw hex values; optionally decodes as big-endian float32." + ), + ) + pp.add_argument("--host", required=True, metavar="HOST", help="Gateway IP/hostname.") + pp.add_argument( + "--port", type=int, default=502, metavar="PORT", help="Gateway TCP port (default 502)." + ) + pp.add_argument( + "--unit", + type=int, + required=True, + metavar="UNIT_ID", + help="Modbus slave / unit ID.", + ) + pp.add_argument( + "--fc", + type=int, + choices=[3, 4], + default=4, + metavar="FC", + help="Function code: 3 (holding) or 4 (input registers, default).", + ) + pp.add_argument( + "--address", + type=_auto_int, + required=True, + metavar="ADDR", + help="Start register address (hex e.g. 0x0000 or decimal).", + ) + pp.add_argument( + "--count", + type=int, + required=True, + metavar="N", + help="Number of registers to read.", + ) + pp.add_argument( + "--decode", + choices=["float32"], + default=None, + metavar="TYPE", + help="Optional: decode consecutive register pairs as 'float32'.", + ) + pp.set_defaults(func=cmd_probe) + + return parser + + +# --------------------------------------------------------------------------- +# Entry point +# --------------------------------------------------------------------------- + + +def main(argv: list[str] | None = None) -> None: + parser = build_parser() + args = parser.parse_args(argv) + args.func(args) + + +if __name__ == "__main__": + main() diff --git a/tests/test_modbus_cli.py b/tests/test_modbus_cli.py new file mode 100644 index 0000000..6450038 --- /dev/null +++ b/tests/test_modbus_cli.py @@ -0,0 +1,289 @@ +"""Tests for scripts/modbus_cli.py. + +Acceptance criteria covered +---------------------------- +1. ``modbus_cli read`` (via mock) prints decoded engineering values for all + profile metrics; columns are present for each metric key. +2. ``modbus_cli read`` exits non-zero on connection failure. +3. ``modbus_cli probe --fc 4`` prints raw register values in hex; optionally + decodes float32 pairs. +4. ``modbus_cli probe`` exits non-zero on connection failure. +5. CLI has NO write-register sub-command (only ``read`` and ``probe`` exist). +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +# Ensure project root is available when tests are run from the repo root. +PROJECT_ROOT = Path(__file__).resolve().parents[1] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +import scripts.modbus_cli as cli # noqa: E402 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_ok_response(registers: list[int]) -> MagicMock: + resp = MagicMock() + resp.isError.return_value = False + resp.registers = registers + return resp + + +def _make_error_response() -> MagicMock: + resp = MagicMock() + resp.isError.return_value = True + resp.__str__ = lambda self: "ExceptionResponse(2)" + return resp + + +# --------------------------------------------------------------------------- +# CLI: read sub-command +# --------------------------------------------------------------------------- + + +class TestReadCommand: + """``python -m scripts.modbus_cli read ...`` with mocked driver.""" + + def _make_sdm120_registers(self) -> dict[int, int]: + """Return a minimal register map covering all 8 sdm120 core metrics.""" + import struct + + def pack(value: float) -> tuple[int, int]: + raw = struct.pack(">f", value) + hi = (raw[0] << 8) | raw[1] + lo = (raw[2] << 8) | raw[3] + return hi, lo + + pairs: dict[int, tuple[int, int]] = { + 0x0000: pack(230.2), # voltage + 0x0006: pack(1.3), # current + 0x000C: pack(295.0), # active_power + 0x001E: pack(0.98), # power_factor + 0x0046: pack(50.0), # frequency + 0x0048: pack(123.4), # import_energy + 0x004A: pack(0.0), # export_energy + 0x0156: pack(123.4), # total_energy + } + regs: dict[int, int] = {} + for addr, (hi, lo) in pairs.items(): + regs[addr] = hi + regs[addr + 1] = lo + return regs + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_read_prints_decoded_values(self, mock_client_cls: MagicMock, capsys) -> None: + """``read`` prints all profile metric keys with values.""" + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = True + + regs = self._make_sdm120_registers() + # sdm120 has 2 blocks; simulate both responses. + # Block 1: 0x0000, count=0x0060 → we need 0x60=96 registers + block1 = [regs.get(i, 0) for i in range(0x0000, 0x0060)] + # Block 2: 0x0156, count=0x0004 → 4 registers + block2 = [regs.get(i, 0) for i in range(0x0156, 0x015A)] + mock_client.read_input_registers.side_effect = [ + _make_ok_response(block1), + _make_ok_response(block2), + ] + + # Run the CLI. + cli.main(["read", "--host", "10.0.0.1", "--port", "502", "--unit", "1", + "--profile", "sdm120"]) + + captured = capsys.readouterr() + output = captured.out + + # All eight core metric keys must appear in the output. + for key in ("voltage", "current", "active_power", "power_factor", + "frequency", "import_energy", "export_energy", "total_energy"): + assert key in output, f"Missing metric key '{key}' in output:\n{output}" + + # The decoded voltage should be close to 230.2. + assert "230." in output, f"Expected voltage ~230.2 in output:\n{output}" + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_read_exits_nonzero_on_connection_failure( + self, mock_client_cls: MagicMock + ) -> None: + """Connection failure causes ``sys.exit(1)``.""" + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = False # Cannot connect. + + with pytest.raises(SystemExit) as exc_info: + cli.main(["read", "--host", "10.0.0.1", "--port", "502", "--unit", "1", + "--profile", "sdm120"]) + assert exc_info.value.code != 0 + + def test_read_exits_nonzero_for_missing_profile(self, capsys) -> None: + """Unknown profile name causes sys.exit(1).""" + with pytest.raises(SystemExit) as exc_info: + cli.main(["read", "--host", "10.0.0.1", "--port", "502", "--unit", "1", + "--profile", "nonexistent_profile_xyz"]) + assert exc_info.value.code != 0 + + +# --------------------------------------------------------------------------- +# CLI: probe sub-command +# --------------------------------------------------------------------------- + + +class TestProbeCommand: + """``python -m scripts.modbus_cli probe ...`` with mocked pymodbus.""" + + @patch("scripts.modbus_cli.ModbusTcpClient" if False else "pymodbus.client.ModbusTcpClient") + def _run_probe(self, mock_cls, args: list[str], regs: list[int]): + """Helper: runs probe with a mocked client returning *regs*.""" + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_probe_prints_raw_registers(self, mock_driver_cls, capsys) -> None: + # probe uses its own client import, not the driver's — patch it there. + pass # see next test for the direct approach + + def _setup_mock_client(self, regs: list[int]) -> MagicMock: + """Return a mock ModbusTcpClient that returns *regs* for both FC03 and FC04.""" + mock_cls = MagicMock() + mock_client = MagicMock() + mock_cls.return_value = mock_client + mock_client.connect.return_value = True + ok_response = _make_ok_response(regs) + mock_client.read_input_registers.return_value = ok_response + mock_client.read_holding_registers.return_value = ok_response + return mock_cls + + def test_probe_fc4_prints_hex_values(self, capsys) -> None: + """FC04 probe prints at least two hex addresses and values.""" + mock_cls = self._setup_mock_client([0x4366, 0x3334]) + + with patch("pymodbus.client.ModbusTcpClient", mock_cls): + cli.main([ + "probe", + "--host", "10.0.0.1", + "--port", "502", + "--unit", "1", + "--fc", "4", + "--address", "0x0000", + "--count", "2", + ]) + + captured = capsys.readouterr() + output = captured.out + # Expect hex addresses and values in the output. + assert "0x0000" in output.lower() or "0000" in output + assert "4366" in output.upper() + assert "3334" in output.upper() + + def test_probe_fc4_with_float32_decode(self, capsys) -> None: + """--decode float32 adds a float32 decode section to output.""" + mock_cls = self._setup_mock_client([0x4366, 0x3334]) + + with patch("pymodbus.client.ModbusTcpClient", mock_cls): + cli.main([ + "probe", + "--host", "10.0.0.1", + "--port", "502", + "--unit", "1", + "--fc", "4", + "--address", "0x0000", + "--count", "2", + "--decode", "float32", + ]) + + captured = capsys.readouterr() + output = captured.out + # The float decode section should show ~230.2. + assert "230." in output, f"Expected ~230.2 in float decode output:\n{output}" + + def test_probe_exits_nonzero_on_connection_failure(self, capsys) -> None: + """Connection failure causes non-zero exit.""" + mock_cls = MagicMock() + mock_client = MagicMock() + mock_cls.return_value = mock_client + mock_client.connect.return_value = False + + with patch("pymodbus.client.ModbusTcpClient", mock_cls): + with pytest.raises(SystemExit) as exc_info: + cli.main([ + "probe", + "--host", "10.0.0.1", + "--port", "502", + "--unit", "1", + "--fc", "4", + "--address", "0x0000", + "--count", "2", + ]) + assert exc_info.value.code != 0 + + def test_probe_fc3_also_works(self, capsys) -> None: + """FC03 (holding registers) is supported by probe.""" + mock_cls = self._setup_mock_client([0x0001, 0x0002]) + + with patch("pymodbus.client.ModbusTcpClient", mock_cls): + cli.main([ + "probe", + "--host", "10.0.0.1", + "--port", "502", + "--unit", "1", + "--fc", "3", + "--address", "0x0014", + "--count", "2", + ]) + + # Verify holding-register read was called (not input-register read). + mock_client = mock_cls.return_value + mock_client.read_holding_registers.assert_called_once() + mock_client.read_input_registers.assert_not_called() + + +# --------------------------------------------------------------------------- +# CLI: no write sub-commands +# --------------------------------------------------------------------------- + + +class TestNoWriteSubcommands: + """The CLI must not expose any write-register sub-commands.""" + + def test_no_write_subcommand_in_parser(self) -> None: + """The argument parser has no 'write', 'set', or 'force' sub-commands.""" + parser = cli.build_parser() + # Access the subparsers group. + subparsers_actions = [ + action + for action in parser._actions + if hasattr(action, "_name_parser_map") + ] + assert len(subparsers_actions) == 1 + available_commands = set(subparsers_actions[0]._name_parser_map.keys()) + # Only read-only commands must exist. + assert "read" in available_commands + assert "probe" in available_commands + # No write-related commands. + for name in available_commands: + assert name not in {"write", "set", "force", "write-register", "set-register"}, ( + f"Write sub-command '{name}' found in CLI — it must not exist" + ) + + def test_only_read_and_probe_subcommands(self) -> None: + """Exactly two sub-commands are available: read and probe.""" + parser = cli.build_parser() + subparsers_actions = [ + action + for action in parser._actions + if hasattr(action, "_name_parser_map") + ] + available = set(subparsers_actions[0]._name_parser_map.keys()) + assert available == {"read", "probe"}, ( + f"Expected only {{read, probe}}, got {available}" + ) diff --git a/tests/test_modbus_driver.py b/tests/test_modbus_driver.py new file mode 100644 index 0000000..7412914 --- /dev/null +++ b/tests/test_modbus_driver.py @@ -0,0 +1,232 @@ +"""Tests for app/integrations/modbus/driver.py. + +All tests use mocks — no real hardware or network connection is required. + +Acceptance criteria covered +---------------------------- +1. ``registers_to_float(0x4366, 0x3334)`` decodes to ~230.2 (byte/word order correct). +2. ``read_blocks`` returns the expected register map on a successful response. +3. ``read_blocks`` raises ``ModbusConnectionError`` when the client cannot connect. +4. ``read_blocks`` raises ``ModbusConnectionError`` when ``ConnectionException`` is raised. +5. ``read_blocks`` raises ``ModbusResponseError`` when the response is a Modbus error frame. +6. ``read_blocks`` raises ``ModbusResponseError`` when register count mismatches. +""" + +from __future__ import annotations + +import math +from unittest.mock import MagicMock, patch + +import pytest + +from app.integrations.modbus.driver import ( + ModbusConnectionError, + ModbusDriverError, + ModbusResponseError, + read_blocks, + registers_to_float, +) + + +# --------------------------------------------------------------------------- +# registers_to_float +# --------------------------------------------------------------------------- + + +class TestRegistersToFloat: + """Big-endian float32 decoding.""" + + def test_voltage_reference_value(self) -> None: + """PDF example: 0x4366,0x3334 → 230.2 V.""" + result = registers_to_float(0x4366, 0x3334) + # IEEE-754 nearest float32 for 230.2 is ~230.20001220703125 + assert math.isclose(result, 230.2, rel_tol=1e-5), f"Got {result}" + + def test_demand_time_reference_value(self) -> None: + """PDF example: 0x3F80,0x0000 → 1.0.""" + result = registers_to_float(0x3F80, 0x0000) + assert math.isclose(result, 1.0, rel_tol=1e-7) + + def test_network_node_reference_value(self) -> None: + """PDF example: 0x4270,0x0000 → 60.0.""" + result = registers_to_float(0x4270, 0x0000) + assert math.isclose(result, 60.0, rel_tol=1e-7) + + def test_zero(self) -> None: + result = registers_to_float(0x0000, 0x0000) + assert result == 0.0 + + def test_word_order_matters(self) -> None: + """Swapping hi/lo gives a different (wrong) value.""" + correct = registers_to_float(0x4366, 0x3334) + swapped = registers_to_float(0x3334, 0x4366) + assert correct != swapped + + +# --------------------------------------------------------------------------- +# read_blocks — helpers for building mock pymodbus responses +# --------------------------------------------------------------------------- + + +def _make_ok_response(registers: list[int]) -> MagicMock: + """Build a fake successful read_input_registers response.""" + resp = MagicMock() + resp.isError.return_value = False + resp.registers = registers + return resp + + +def _make_error_response() -> MagicMock: + """Build a fake Modbus exception response.""" + resp = MagicMock() + resp.isError.return_value = True + resp.__str__ = lambda self: "ExceptionResponse(2)" + return resp + + +class TestReadBlocks: + """Unit tests for read_blocks using a mocked ModbusTcpClient.""" + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_single_block_returns_register_map(self, mock_client_cls: MagicMock) -> None: + """Successful single-block read returns correct {addr: value} map.""" + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = True + mock_client.read_input_registers.return_value = _make_ok_response( + [0x4366, 0x3334, 0x3F80, 0x0000] + ) + + blocks = [{"start": 0x0000, "count": 4}] + result = read_blocks("127.0.0.1", 502, 1, blocks) + + assert result == {0: 0x4366, 1: 0x3334, 2: 0x3F80, 3: 0x0000} + mock_client.read_input_registers.assert_called_once_with( + 0x0000, count=4, device_id=1 + ) + mock_client.close.assert_called_once() + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_multiple_blocks_merged(self, mock_client_cls: MagicMock) -> None: + """Multiple blocks are read and merged into a single dict.""" + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = True + + # First block: 2 registers at 0x0000 + resp1 = _make_ok_response([0x4366, 0x3334]) + # Second block: 2 registers at 0x0156 + resp2 = _make_ok_response([0x447A, 0x0000]) + mock_client.read_input_registers.side_effect = [resp1, resp2] + + blocks = [{"start": 0x0000, "count": 2}, {"start": 0x0156, "count": 2}] + result = read_blocks("127.0.0.1", 502, 1, blocks) + + assert result[0x0000] == 0x4366 + assert result[0x0001] == 0x3334 + assert result[0x0156] == 0x447A + assert result[0x0157] == 0x0000 + assert mock_client.read_input_registers.call_count == 2 + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_connect_failure_raises_connection_error( + self, mock_client_cls: MagicMock + ) -> None: + """connect() returning False raises ModbusConnectionError.""" + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = False + + with pytest.raises(ModbusConnectionError, match="Could not connect"): + read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) + + mock_client.close.assert_called_once() + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_connection_exception_raises_connection_error( + self, mock_client_cls: MagicMock + ) -> None: + """pymodbus ConnectionException during connect() → ModbusConnectionError.""" + from pymodbus.exceptions import ConnectionException + + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.side_effect = ConnectionException("timeout") + + with pytest.raises(ModbusConnectionError, match="failed"): + read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) + + mock_client.close.assert_called_once() + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_modbus_exception_response_raises_response_error( + self, mock_client_cls: MagicMock + ) -> None: + """Modbus exception frame from device raises ModbusResponseError.""" + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = True + mock_client.read_input_registers.return_value = _make_error_response() + + with pytest.raises(ModbusResponseError, match="exception"): + read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) + + mock_client.close.assert_called_once() + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_register_count_mismatch_raises_response_error( + self, mock_client_cls: MagicMock + ) -> None: + """Response with wrong register count raises ModbusResponseError.""" + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = True + # Requested 4 but only got 2 + mock_client.read_input_registers.return_value = _make_ok_response([0x4366, 0x3334]) + + with pytest.raises(ModbusResponseError, match="Expected 4 registers"): + read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 4}]) + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_connection_exception_during_read_raises_connection_error( + self, mock_client_cls: MagicMock + ) -> None: + """ConnectionException during register read → ModbusConnectionError.""" + from pymodbus.exceptions import ConnectionException + + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = True + mock_client.read_input_registers.side_effect = ConnectionException("dropped") + + with pytest.raises(ModbusConnectionError, match="Lost connection"): + read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) + + mock_client.close.assert_called_once() + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_client_always_closed_on_error(self, mock_client_cls: MagicMock) -> None: + """close() is always called even when an exception propagates.""" + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = True + mock_client.read_input_registers.return_value = _make_error_response() + + with pytest.raises(ModbusDriverError): + read_blocks("10.0.0.1", 502, 1, [{"start": 0, "count": 2}]) + + mock_client.close.assert_called_once() + + @patch("app.integrations.modbus.driver.ModbusTcpClient") + def test_unit_id_passed_as_device_id(self, mock_client_cls: MagicMock) -> None: + """unit_id is forwarded as device_id= keyword argument (pymodbus 3.13.x).""" + mock_client = MagicMock() + mock_client_cls.return_value = mock_client + mock_client.connect.return_value = True + mock_client.read_input_registers.return_value = _make_ok_response([0x0001, 0x0002]) + + read_blocks("10.0.0.1", 502, unit_id=5, blocks=[{"start": 0, "count": 2}]) + + mock_client.read_input_registers.assert_called_once_with( + 0, count=2, device_id=5 + ) diff --git a/tests/test_modbus_profiles.py b/tests/test_modbus_profiles.py new file mode 100644 index 0000000..4952ff6 --- /dev/null +++ b/tests/test_modbus_profiles.py @@ -0,0 +1,259 @@ +"""Tests for app/integrations/modbus/profiles.py. + +Acceptance criteria covered +---------------------------- +1. ``load_profile("sdm120")`` succeeds and returns a valid ``ModbusProfile``. +2. ``decode(profile, registers)`` maps known register values to correct keys/values. +3. ``load_profile`` raises a recognisable error for a missing profile. +4. ``load_profile`` raises a recognisable error for a profile that fails validation. +5. ``list_profiles()`` returns at least one entry containing ``"sdm120"``. +6. ``decode`` silently skips metrics whose registers are absent from the map. +""" + +from __future__ import annotations + +import math +import textwrap +from pathlib import Path +from unittest.mock import patch + +import pytest +import yaml + +from app.integrations.modbus.profiles import ( + ModbusProfile, + ProfileNotFoundError, + ProfileValidationError, + decode, + list_profiles, + load_profile, +) + + +# --------------------------------------------------------------------------- +# load_profile — happy path +# --------------------------------------------------------------------------- + + +class TestLoadProfileSdm120: + """Validate that the shipped sdm120.yaml loads and validates correctly.""" + + def test_load_returns_modbus_profile(self) -> None: + profile = load_profile("sdm120") + assert isinstance(profile, ModbusProfile) + + def test_name_and_description(self) -> None: + profile = load_profile("sdm120") + assert profile.name == "sdm120" + assert "SDM120" in profile.description or "sdm120" in profile.description.lower() + + def test_function_code_is_4(self) -> None: + profile = load_profile("sdm120") + assert profile.function_code == 4 + + def test_word_and_byte_order_big(self) -> None: + profile = load_profile("sdm120") + assert profile.word_order == "big" + assert profile.byte_order == "big" + + def test_has_at_least_two_blocks(self) -> None: + profile = load_profile("sdm120") + assert len(profile.blocks) >= 2 + + def test_has_core_metrics(self) -> None: + """All eight core metrics are present.""" + profile = load_profile("sdm120") + keys = {m.key for m in profile.metrics} + expected = { + "voltage", + "current", + "active_power", + "power_factor", + "frequency", + "import_energy", + "export_energy", + "total_energy", + } + assert expected.issubset(keys), f"Missing keys: {expected - keys}" + + def test_voltage_address_is_0x0000(self) -> None: + profile = load_profile("sdm120") + voltage = next(m for m in profile.metrics if m.key == "voltage") + assert voltage.address == 0x0000 + + def test_frequency_address_is_0x0046(self) -> None: + profile = load_profile("sdm120") + freq = next(m for m in profile.metrics if m.key == "frequency") + assert freq.address == 0x0046 + + def test_total_energy_address_is_0x0156(self) -> None: + profile = load_profile("sdm120") + te = next(m for m in profile.metrics if m.key == "total_energy") + assert te.address == 0x0156 + + def test_energy_metrics_have_state_class(self) -> None: + profile = load_profile("sdm120") + energy_keys = {"import_energy", "export_energy", "total_energy"} + for m in profile.metrics: + if m.key in energy_keys: + assert m.state_class == "total_increasing", ( + f"{m.key} should have state_class='total_increasing'" + ) + + def test_no_deployment_fields_in_profile(self) -> None: + """Profile YAML must not contain unit_id or friendly_name.""" + profile_path = ( + Path(__file__).parent.parent + / "app/integrations/modbus/profiles/sdm120.yaml" + ) + with profile_path.open() as fh: + raw = yaml.safe_load(fh) + assert "unit_id" not in raw, "unit_id is a deployment field, not a protocol field" + assert "friendly_name" not in raw, "friendly_name is a deployment field" + + +# --------------------------------------------------------------------------- +# load_profile — error cases +# --------------------------------------------------------------------------- + + +class TestLoadProfileErrors: + def test_missing_profile_raises_not_found(self) -> None: + with pytest.raises(ProfileNotFoundError, match="nonexistent_profile"): + load_profile("nonexistent_profile") + + def test_invalid_yaml_content_raises_validation_error(self, tmp_path: Path) -> None: + """A profile missing required fields raises ProfileValidationError.""" + bad_yaml = textwrap.dedent( + """\ + name: bad_profile + description: Missing required fields + # function_code, word_order, byte_order, blocks, metrics are all absent + """ + ) + bad_path = tmp_path / "bad_profile.yaml" + bad_path.write_text(bad_yaml) + + # Temporarily patch the profiles directory to include our bad file. + fake_dir = tmp_path + with patch("app.integrations.modbus.profiles._PROFILES_DIR", fake_dir): + with pytest.raises(ProfileValidationError, match="bad_profile"): + load_profile("bad_profile") + + def test_wrong_type_in_yaml_raises_validation_error(self, tmp_path: Path) -> None: + """A profile with wrong field types raises ProfileValidationError.""" + bad_yaml = textwrap.dedent( + """\ + name: typed_wrong + description: Wrong type for function_code + function_code: "four" # should be int + word_order: big + byte_order: big + blocks: [] + metrics: [] + """ + ) + bad_path = tmp_path / "typed_wrong.yaml" + bad_path.write_text(bad_yaml) + + with patch("app.integrations.modbus.profiles._PROFILES_DIR", tmp_path): + with pytest.raises(ProfileValidationError, match="typed_wrong"): + load_profile("typed_wrong") + + +# --------------------------------------------------------------------------- +# decode — register-to-value mapping +# --------------------------------------------------------------------------- + + +class TestDecode: + """Verify decode() maps known register pairs to correct float values.""" + + def _make_registers(self, items: dict[int, tuple[int, int]]) -> dict[int, int]: + """Build a register map from ``{address: (hi, lo)}`` pairs.""" + regs: dict[int, int] = {} + for addr, (hi, lo) in items.items(): + regs[addr] = hi + regs[addr + 1] = lo + return regs + + def test_voltage_decoded_correctly(self) -> None: + """0x4366,0x3334 at address 0x0000 → ~230.2V.""" + profile = load_profile("sdm120") + registers = self._make_registers({0x0000: (0x4366, 0x3334)}) + # Only voltage registers provided; others will be skipped. + result = decode(profile, registers) + assert "voltage" in result + assert math.isclose(result["voltage"], 230.2, rel_tol=1e-5) + + def test_total_energy_decoded_correctly(self) -> None: + """Float32 at address 0x0156 decoded as total_energy.""" + profile = load_profile("sdm120") + # 0x447A0000 → 1000.0 + registers = self._make_registers({0x0156: (0x447A, 0x0000)}) + result = decode(profile, registers) + assert "total_energy" in result + assert math.isclose(result["total_energy"], 1000.0, rel_tol=1e-5) + + def test_missing_registers_skipped(self) -> None: + """Metrics whose register addresses are absent are not in the output.""" + profile = load_profile("sdm120") + # Only supply voltage registers; all others missing. + registers = {0x0000: 0x4366, 0x0001: 0x3334} + result = decode(profile, registers) + assert "voltage" in result + assert "current" not in result + assert "active_power" not in result + + def test_full_sdm120_payload_keys(self) -> None: + """All 8 core keys appear when all registers are supplied.""" + profile = load_profile("sdm120") + # Supply a minimal register map covering all 8 core metrics. + regs = self._make_registers( + { + 0x0000: (0x4366, 0x3334), # voltage ~230.2V + 0x0006: (0x3F80, 0x0000), # current ~1.0A + 0x000C: (0x4393, 0x8000), # active_power ~295.0W + 0x001E: (0x3F7A, 0xE148), # power_factor ~0.98 + 0x0046: (0x4248, 0x0000), # frequency ~50.0Hz + 0x0048: (0x42F6, 0x6666), # import_energy ~123.2kWh + 0x004A: (0x0000, 0x0000), # export_energy ~0.0kWh + 0x0156: (0x42F6, 0x6666), # total_energy ~123.2kWh + } + ) + result = decode(profile, regs) + expected_keys = { + "voltage", "current", "active_power", "power_factor", + "frequency", "import_energy", "export_energy", "total_energy", + } + assert expected_keys == set(result.keys()) + + def test_decode_does_not_modify_registers(self) -> None: + """decode() must not mutate the input register dict.""" + profile = load_profile("sdm120") + original = {0x0000: 0x4366, 0x0001: 0x3334} + registers = dict(original) + decode(profile, registers) + assert registers == original + + +# --------------------------------------------------------------------------- +# list_profiles +# --------------------------------------------------------------------------- + + +class TestListProfiles: + def test_contains_sdm120(self) -> None: + profiles = list_profiles() + names = [name for name, _ in profiles] + assert "sdm120" in names + + def test_returns_list_of_tuples(self) -> None: + profiles = list_profiles() + assert isinstance(profiles, list) + for item in profiles: + assert isinstance(item, tuple) + assert len(item) == 2 + name, description = item + assert isinstance(name, str) + assert isinstance(description, str)