From ceffafff21d698111e45c054b83c026269d5df54 Mon Sep 17 00:00:00 2001 From: Franck Nijhof Date: Tue, 7 Jul 2026 19:32:53 +0000 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Replace=20Mashumaro=20with?= =?UTF-8?q?=20Probatio?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Parse the Fumis WiRCU API into the model dataclasses with Probatio instead of Mashumaro. Models become plain frozen dataclasses that validate and construct through Probatio's SchemaMixin, dropping unmodeled keys, with field aliases on Key() and coercions on inline Coerce/FromEpoch/AsTimedelta/Maybe hints. Serialization for the `info --json` command moves to orjson, since Probatio validates but does not serialize. Also resolves error and alert codes to non-optional enums with explicit NO_ERROR/NO_ALERT members instead of None. Probatio requires Python 3.12, so 3.11 is dropped. --- .github/workflows/linting.yaml | 2 +- .github/workflows/release.yaml | 2 +- .github/workflows/tests.yaml | 4 +- .github/workflows/typing.yaml | 2 +- poetry.lock | 45 ++-- pyproject.toml | 7 +- src/fumis/cli/__init__.py | 34 +-- src/fumis/cli/tui.py | 4 +- src/fumis/const.py | 26 +-- src/fumis/models.py | 315 +++++++------------------- tests/cli/__snapshots__/test_cli.ambr | 40 +++- tests/test_fumis.py | 8 +- 12 files changed, 179 insertions(+), 310 deletions(-) diff --git a/.github/workflows/linting.yaml b/.github/workflows/linting.yaml index 552c803..09c30f9 100644 --- a/.github/workflows/linting.yaml +++ b/.github/workflows/linting.yaml @@ -8,7 +8,7 @@ on: workflow_dispatch: env: - DEFAULT_PYTHON: "3.11" + DEFAULT_PYTHON: "3.12" permissions: contents: read diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml index b570b49..290bea6 100644 --- a/.github/workflows/release.yaml +++ b/.github/workflows/release.yaml @@ -8,7 +8,7 @@ on: - published env: - DEFAULT_PYTHON: "3.11" + DEFAULT_PYTHON: "3.12" permissions: contents: read diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index d60adb1..eb22960 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -8,7 +8,7 @@ on: workflow_dispatch: env: - DEFAULT_PYTHON: "3.11" + DEFAULT_PYTHON: "3.12" permissions: contents: read @@ -19,7 +19,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python: ["3.11", "3.12", "3.13", "3.14"] + python: ["3.12", "3.13", "3.14"] steps: - name: ⤵️ Check out code from GitHub uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 diff --git a/.github/workflows/typing.yaml b/.github/workflows/typing.yaml index 9e19fc1..e3470d9 100644 --- a/.github/workflows/typing.yaml +++ b/.github/workflows/typing.yaml @@ -8,7 +8,7 @@ on: workflow_dispatch: env: - DEFAULT_PYTHON: "3.11" + DEFAULT_PYTHON: "3.12" permissions: contents: read diff --git a/poetry.lock b/poetry.lock index aef5de7..c298576 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 2.3.2 and should not be changed by hand. +# This file is automatically @generated by Poetry 2.4.1 and should not be changed by hand. [[package]] name = "aiohappyeyeballs" @@ -1120,27 +1120,6 @@ files = [ {file = "marshmallow-4.3.0.tar.gz", hash = "sha256:fb43c53b3fe240b8f6af37223d6ef1636f927ad9bea8ab323afad95dff090880"}, ] -[[package]] -name = "mashumaro" -version = "3.22" -description = "Fast and well tested serialization library" -optional = false -python-versions = ">=3.10" -groups = ["main"] -files = [ - {file = "mashumaro-3.22-py3-none-any.whl", hash = "sha256:17dc4d7294c33ef380a8b929dda0608577aa2141988c00a0c4932310108fe71d"}, - {file = "mashumaro-3.22.tar.gz", hash = "sha256:64538cc365204402a060ebde683a86505b5a4344acf6870d79021e9fbfe57360"}, -] - -[package.dependencies] -typing_extensions = ">=4.14.0" - -[package.extras] -msgpack = ["msgpack (>=0.5.6)"] -orjson = ["orjson"] -toml = ["tomli (>=1.1.0) ; python_version < \"3.11\"", "tomli-w (>=1.0)"] -yaml = ["pyyaml (>=3.13)"] - [[package]] name = "mccabe" version = "0.7.0" @@ -1571,6 +1550,18 @@ files = [ {file = "prek-0.4.8.tar.gz", hash = "sha256:d15d8bef72ab7b02c7dc01458ac9e05b3131534492b5ce9bb11c4f6f636fa868"}, ] +[[package]] +name = "probatio" +version = "0.9.0" +description = "A modern Python data validation library" +optional = false +python-versions = ">=3.12" +groups = ["main"] +files = [ + {file = "probatio-0.9.0-py3-none-any.whl", hash = "sha256:82944240c4dc4c68f870a3fcb87b664b30c091eac31611f84d2536ddee2c214f"}, + {file = "probatio-0.9.0.tar.gz", hash = "sha256:3dd9ff27d8152a3e97e76ab019b7b0234b83001d0ddeef54aaae69609be331cf"}, +] + [[package]] name = "propcache" version = "0.5.2" @@ -1901,10 +1892,7 @@ files = [ [package.dependencies] astroid = ">=4.0.2,<=4.1.dev0" colorama = {version = ">=0.4.5", markers = "sys_platform == \"win32\""} -dill = [ - {version = ">=0.3.7", markers = "python_version >= \"3.12\""}, - {version = ">=0.3.6", markers = "python_version == \"3.11\""}, -] +dill = {version = ">=0.3.7", markers = "python_version >= \"3.12\""} isort = ">=5,<5.13 || >5.13,<9" mccabe = ">=0.6,<0.8" platformdirs = ">=2.2" @@ -2493,6 +2481,7 @@ files = [ {file = "typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8"}, {file = "typing_extensions-4.16.0.tar.gz", hash = "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5"}, ] +markers = {main = "python_version < \"4.0\" and extra == \"cli\" or python_version == \"3.12\""} [[package]] name = "typing-inspection" @@ -2689,5 +2678,5 @@ cli = ["rich", "textual", "textual-plotext", "typer"] [metadata] lock-version = "2.1" -python-versions = ">=3.11" -content-hash = "a4c1ee6f85401ba6f7f7d2dcc7a9abb52b39b08f6d99a0f2a9e6901699bfce9d" +python-versions = ">=3.12" +content-hash = "c51d7752b59a9c7ee1b105bc9f104d003515b2ec2a0847eb5adbee641b9cd161" diff --git a/pyproject.toml b/pyproject.toml index 9236c7c..259f392 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ description = "Asynchronous Python client for the Fumis WiRCU API." authors = [{ name = "Franck Nijhof", email = "opensource@frenck.dev" }] maintainers = [{ name = "Franck Nijhof", email = "opensource@frenck.dev" }] license = "MIT" -requires-python = ">=3.11" +requires-python = ">=3.12" readme = "README.md" keywords = [ "api", @@ -29,7 +29,6 @@ classifiers = [ "Framework :: AsyncIO", "Intended Audience :: Developers", "Natural Language :: English", - "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", @@ -38,7 +37,7 @@ classifiers = [ ] dynamic = ["dependencies"] packages = [{ include = "fumis", from = "src" }] -dependencies = ['aiohttp (>=3.0.0)', 'awesomeversion (>=22.0.0)', 'mashumaro (>=3.13,<4.0.0)', 'orjson (>=3.9.8)', 'yarl (>=1.6.0)'] +dependencies = ['aiohttp (>=3.0.0)', 'awesomeversion (>=22.0.0)', 'orjson (>=3.9.8)', 'probatio (>=0.9.0)', 'yarl (>=1.6.0)'] [project.optional-dependencies] cli = ["rich>=14.0.0", "textual>=8.0.0,<9.0.0; python_version<'4.0'", "textual-plotext>=1.0.0,<2.0.0; python_version<'4.0'", "typer>=0.15.1"] @@ -89,7 +88,7 @@ omit = [ ] [tool.ty.environment] -python-version = "3.11" +python-version = "3.12" [tool.pylint.MASTER] ignore = ["tests"] diff --git a/src/fumis/cli/__init__.py b/src/fumis/cli/__init__.py index 4e50d0d..238a4a2 100644 --- a/src/fumis/cli/__init__.py +++ b/src/fumis/cli/__init__.py @@ -4,8 +4,10 @@ import json import sys +from dataclasses import replace from typing import TYPE_CHECKING, Annotated +import orjson import typer from rich.console import Console from rich.panel import Panel @@ -153,11 +155,6 @@ async def _fetch_info(mac: str, password: str) -> FumisInfo: return await fumis.update_info() -def _emit_json(payload: object) -> None: - """Emit a payload as indented JSON on stdout.""" - typer.echo(json.dumps(payload, indent=2, default=str)) - - STATUS_ICONS: dict[StoveStatus, str] = { StoveStatus.OFF: "[dim]\u2b58[/dim]", StoveStatus.PRE_HEATING: "[yellow]\u25b2[/yellow]", @@ -215,13 +212,13 @@ def _render_info( # noqa: PLR0912, PLR0915 # pylint: disable=too-many-branches status_table.add_row("\U0001f3e0 Status", _status_display(c)) - if error := c.stove_error: + if (error := c.stove_error) != StoveError.NO_ERROR: error_label = f"E{c.error:03d}" if error == StoveError.UNKNOWN else str(error) status_table.add_row( "\u274c Error", f"[red bold]{error_label}[/red bold] [dim]{error.description}[/dim]", ) - if alert := c.stove_alert: + if (alert := c.stove_alert) != StoveAlert.NO_ALERT: alert_label = f"A{c.alert:03d}" if alert == StoveAlert.UNKNOWN else str(alert) status_table.add_row( "\u26a0\ufe0f Alert", @@ -349,9 +346,10 @@ async def info_command( info = await _fetch_info(mac, password) if output_json: - data = info.to_dict() - data["unit"]["id"] = "**REDACTED**" - _emit_json(data) + redacted = replace(info, unit=replace(info.unit, id="**REDACTED**")) + # pylint: disable-next=no-member + payload = orjson.dumps(redacted, default=str, option=orjson.OPT_INDENT_2) + typer.echo(payload.decode()) return _render_info(info) @@ -532,21 +530,23 @@ async def errors_command( c = info.controller # Current error - if error := c.stove_error: + error = c.stove_error + if error == StoveError.NO_ERROR: + console.print("\u2705 [green bold]No active error[/green bold]") + else: label = f"E{c.error:03d}" if error == StoveError.UNKNOWN else str(error) console.print(f"\u274c [red bold]Error {label}:[/red bold] {error.description}") - else: - console.print("\u2705 [green bold]No active error[/green bold]") # Current alert - if alert := c.stove_alert: + alert = c.stove_alert + if alert == StoveAlert.NO_ALERT: + console.print("\u2705 [green bold]No active alert[/green bold]") + else: label = f"A{c.alert:03d}" if alert == StoveAlert.UNKNOWN else str(alert) console.print( f"\u26a0\ufe0f [yellow bold]Alert {label}:[/yellow bold]" f" {alert.description}" ) - else: - console.print("\u2705 [green bold]No active alert[/green bold]") # Error history from diagnostic variables # var[36..95] in groups of 4: [sequence, error_code, date(YYYYMMDD), time] @@ -573,7 +573,7 @@ async def errors_command( table.add_column("Date") for i, (code, date_val, time_val) in enumerate(history, 1): err = StoveError.from_code(code) - if err is None or err == StoveError.UNKNOWN: + if err == StoveError.UNKNOWN: label = f"E{code:03d}" desc = f"Unknown ({code})" else: diff --git a/src/fumis/cli/tui.py b/src/fumis/cli/tui.py index 05136ea..cd48c68 100644 --- a/src/fumis/cli/tui.py +++ b/src/fumis/cli/tui.py @@ -73,10 +73,10 @@ def update_info(self, info: FumisInfo) -> None: eco_str = "\U0001f331 On" if eco.enabled else "Off" error_str = "" - if error := c.stove_error: + if (error := c.stove_error) != StoveError.NO_ERROR: label = f"E{c.error:03d}" if error == StoveError.UNKNOWN else str(error) error_str = f"\u274c {label}: {error.description}" - if alert := c.stove_alert: + if (alert := c.stove_alert) != StoveAlert.NO_ALERT: label = f"A{c.alert:03d}" if alert == StoveAlert.UNKNOWN else str(alert) error_str += f" \u26a0\ufe0f {label}: {alert.description}" diff --git a/src/fumis/const.py b/src/fumis/const.py index 9c8dc40..736ba79 100644 --- a/src/fumis/const.py +++ b/src/fumis/const.py @@ -121,6 +121,9 @@ class StoveError(StrEnum): Use `StoveError.from_code()` to convert the raw integer from the API. """ + NO_ERROR = "E000" + """No active error.""" + IGNITION_FAILED = "E101" """Ignition failed, water overtemperature, or backfire protection.""" @@ -197,13 +200,8 @@ class StoveError(StrEnum): """Unknown or unmapped error code from the API.""" @classmethod - def from_code(cls, code: int) -> StoveError | None: - """Convert a raw error code integer to a StoveError enum. - - Returns None when there is no active error (code 0). - """ - if code == 0: - return None + def from_code(cls, code: int) -> StoveError: + """Convert a raw error code integer to a StoveError enum.""" formatted = f"E{code:03d}" try: return cls(formatted) @@ -225,6 +223,9 @@ class StoveAlert(StrEnum): Use `StoveAlert.from_code()` to convert the raw integer from the API. """ + NO_ALERT = "A000" + """No active alert.""" + LOW_FUEL = "A001" """Low fuel level - refuel the tank.""" @@ -250,13 +251,8 @@ class StoveAlert(StrEnum): """Unknown or unmapped alert code from the API.""" @classmethod - def from_code(cls, code: int) -> StoveAlert | None: - """Convert a raw alert code integer to a StoveAlert enum. - - Returns None when there is no active alert (code 0). - """ - if code == 0: - return None + def from_code(cls, code: int) -> StoveAlert: + """Convert a raw alert code integer to a StoveAlert enum.""" formatted = f"A{code:03d}" try: return cls(formatted) @@ -272,6 +268,7 @@ def description(self) -> str: _ERROR_DESCRIPTIONS: dict[StoveError, str] = { + StoveError.NO_ERROR: "No error", StoveError.IGNITION_FAILED: ( "Ignition failed / water overtemperature / backfire protection" ), @@ -301,6 +298,7 @@ def description(self) -> str: } _ALERT_DESCRIPTIONS: dict[StoveAlert, str] = { + StoveAlert.NO_ALERT: "No alert", StoveAlert.LOW_FUEL: "Low fuel level", StoveAlert.SERVICE_DUE: "Service due", StoveAlert.FLUE_GAS_WARNING: "Flue gas temperature warning", diff --git a/src/fumis/models.py b/src/fumis/models.py index e36e663..0a7357b 100644 --- a/src/fumis/models.py +++ b/src/fumis/models.py @@ -4,12 +4,18 @@ from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta +from typing import Annotated from awesomeversion import AwesomeVersion -from mashumaro import field_options -from mashumaro.config import BaseConfig -from mashumaro.mixins.orjson import DataClassORJSONMixin -from mashumaro.types import SerializationStrategy +from probatio import ( + REMOVE_EXTRA, + AsTimedelta, + Coerce, + FromEpoch, + Key, + Maybe, + SchemaMixin, +) from .const import ( STOVE_MODELS, @@ -21,106 +27,20 @@ ) -class _AwesomeVersionStrategy(SerializationStrategy): - """Serialize AwesomeVersion to/from string for mashumaro.""" - - def serialize(self, value: AwesomeVersion) -> str: - """Serialize to string.""" - return str(value) - - def deserialize(self, value: str) -> AwesomeVersion: - """Deserialize from string.""" - return AwesomeVersion(value) - - -class _StringToIntStrategy(SerializationStrategy): - """Deserialize a string-encoded integer (e.g., RSSI "-48").""" - - def serialize(self, value: int) -> str: - """Serialize to string.""" - return str(value) - - def deserialize(self, value: str | int) -> int: - """Deserialize from string or int.""" - return int(value) - - -class _OptionalIntStrategy(SerializationStrategy): - """Deserialize a string-or-int to int|None, with -1 as None.""" - - def serialize(self, value: int | None) -> int: - """Serialize None back to -1.""" - return -1 if value is None else value - - def deserialize(self, value: str | int) -> int | None: - """Deserialize from string or int, -1 becomes None.""" - result = int(value) - return None if result == -1 else result - - -class _StringToFloatStrategy(SerializationStrategy): - """Deserialize a string-encoded float (e.g., heatingSlope "0.0").""" - - def serialize(self, value: float) -> float: - """Serialize to float.""" - return value - - def deserialize(self, value: str | float) -> float: - """Deserialize from string, float, or int.""" - return float(value) - - -class _TimestampStrategy(SerializationStrategy): - """Convert a unix timestamp to a UTC datetime.""" - - def serialize(self, value: datetime) -> int: - """Serialize to unix timestamp.""" - return int(value.timestamp()) - - def deserialize(self, value: int) -> datetime: - """Deserialize from unix timestamp.""" - return datetime.fromtimestamp(value, tz=UTC) - - -class _OptionalTimestampStrategy(SerializationStrategy): - """Convert a unix timestamp to a UTC datetime, with -1 as None.""" - - def serialize(self, value: datetime | None) -> int: - """Serialize to unix timestamp, None becomes -1.""" - return -1 if value is None else int(value.timestamp()) - - def deserialize(self, value: int) -> datetime | None: - """Deserialize from unix timestamp, -1 becomes None.""" - return None if value == -1 else datetime.fromtimestamp(value, tz=UTC) - - -class _TimedeltaSecondsStrategy(SerializationStrategy): - """Convert seconds to a timedelta.""" +def _sentinel_int(value: str | int) -> int | None: + """Map the -1 sentinel to None, otherwise coerce to int.""" + result = int(value) + return None if result == -1 else result - def serialize(self, value: timedelta) -> int: - """Serialize to seconds.""" - return int(value.total_seconds()) - def deserialize(self, value: int | str) -> timedelta: - """Deserialize from seconds (may be string-encoded).""" - return timedelta(seconds=int(value)) - - -class _BaseModel(DataClassORJSONMixin): - """Base model for all Fumis models.""" - - # pylint: disable-next=too-few-public-methods - class Config(BaseConfig): - """Mashumaro configuration.""" - - omit_none = True - serialization_strategy = { # noqa: RUF012 - AwesomeVersion: _AwesomeVersionStrategy(), - } +def _sentinel_epoch(value: int) -> datetime | None: + """Map the -1 sentinel to None, otherwise a unix timestamp to a UTC datetime.""" + seconds = int(value) + return None if seconds == -1 else datetime.fromtimestamp(seconds, tz=UTC) @dataclass(frozen=True) -class FumisDiagnosticItem(_BaseModel): +class FumisDiagnosticItem: """A single diagnostic variable or parameter entry.""" id: int @@ -128,7 +48,7 @@ class FumisDiagnosticItem(_BaseModel): @dataclass(frozen=True) -class FumisDiagnostic(_BaseModel): +class FumisDiagnostic: """FumisDiagnostic data from the Fumis controller.""" variables: list[FumisDiagnosticItem] = field(default_factory=list) @@ -167,65 +87,52 @@ def timer(self, timer_id: int) -> int: @dataclass(frozen=True) -class FumisTemperature(_BaseModel): +class FumisTemperature: """A temperature channel from the Fumis controller.""" id: int actual: float = 0 - setpoint: float = field(default=0, metadata=field_options(alias="set")) - on_main_screen: bool = field( - default=False, metadata=field_options(alias="onMainScreen") - ) - actual_type: int = field(default=0, metadata=field_options(alias="actualType")) - set_type: int = field(default=0, metadata=field_options(alias="setType")) + setpoint: Annotated[float, Key(alias="set")] = 0 + on_main_screen: Annotated[bool, Key(alias="onMainScreen")] = False + actual_type: Annotated[int, Key(alias="actualType")] = 0 + set_type: Annotated[int, Key(alias="setType")] = 0 name: str | None = None weight: int = 0 @dataclass(frozen=True) -class FumisPower(_BaseModel): +class FumisPower: """FumisPower state of the Fumis controller.""" - kw: float = field( - default=0, - metadata=field_options(serialization_strategy=_StringToFloatStrategy()), - ) - actual_power: int = field(default=0, metadata=field_options(alias="actualPower")) - set_power: int = field(default=0, metadata=field_options(alias="setPower")) - set_type: int = field(default=0, metadata=field_options(alias="setType")) - actual_type: int = field(default=0, metadata=field_options(alias="actualType")) + kw: Annotated[float, Coerce(float)] = 0 + actual_power: Annotated[int, Key(alias="actualPower")] = 0 + set_power: Annotated[int, Key(alias="setPower")] = 0 + set_type: Annotated[int, Key(alias="setType")] = 0 + actual_type: Annotated[int, Key(alias="actualType")] = 0 @dataclass(frozen=True) -class FumisFan(_BaseModel): +class FumisFan: """A fan entry from the Fumis controller.""" id: int speed: int = 0 - speed_type: int = field(default=0, metadata=field_options(alias="speedType")) + speed_type: Annotated[int, Key(alias="speedType")] = 0 weight: int = 0 @dataclass(frozen=True) -class FumisFuel(_BaseModel): +class FumisFuel: """A fuel entry from the Fumis controller.""" id: int quality: int = 0 - quality_type: int = field(default=0, metadata=field_options(alias="qualityType")) - quality_actual: int | None = field( - default=None, metadata=field_options(alias="qualityActual") - ) + quality_type: Annotated[int, Key(alias="qualityType")] = 0 + quality_actual: Annotated[int | None, Key(alias="qualityActual")] = None quantity: float | None = None - quantity_display: int | None = field( - default=None, metadata=field_options(alias="quantityDisplay") - ) - quantity_set_type: int = field( - default=0, metadata=field_options(alias="quantitySetType") - ) - quantity_actual_type: int = field( - default=0, metadata=field_options(alias="quantityActualType") - ) + quantity_display: Annotated[int | None, Key(alias="quantityDisplay")] = None + quantity_set_type: Annotated[int, Key(alias="quantitySetType")] = 0 + quantity_actual_type: Annotated[int, Key(alias="quantityActualType")] = 0 name: str | None = None @property @@ -240,15 +147,11 @@ def quantity_percentage(self) -> float | None: @dataclass(frozen=True) -class FumisEcoMode(_BaseModel): +class FumisEcoMode: """Eco mode state of the Fumis controller.""" - eco_mode_enable: int | None = field( - default=None, metadata=field_options(alias="ecoModeEnable") - ) - eco_mode_set_type: int | None = field( - default=None, metadata=field_options(alias="ecoModeSetType") - ) + eco_mode_enable: Annotated[int | None, Key(alias="ecoModeEnable")] = None + eco_mode_set_type: Annotated[int | None, Key(alias="ecoModeSetType")] = None @property def enabled(self) -> bool: @@ -257,16 +160,16 @@ def enabled(self) -> bool: @dataclass(frozen=True) -class FumisHybrid(_BaseModel): +class FumisHybrid: """FumisHybrid mode state (wood + pellet stoves).""" - actual_type: int = field(default=0, metadata=field_options(alias="actualType")) + actual_type: Annotated[int, Key(alias="actualType")] = 0 operation: int = 0 state: int = 0 @dataclass(frozen=True) -class FumisAntifreeze(_BaseModel): +class FumisAntifreeze: """FumisAntifreeze protection settings.""" temperature: float | None = None @@ -274,53 +177,35 @@ class FumisAntifreeze(_BaseModel): @dataclass(frozen=True) -class FumisStatistic(_BaseModel): +class FumisStatistic: """Statistics counters from the Fumis controller.""" - igniter_starts: int = field( - default=0, metadata=field_options(alias="igniterStarts") + igniter_starts: Annotated[int, Key(alias="igniterStarts")] = 0 + uptime: Annotated[timedelta, AsTimedelta()] = field( + default_factory=lambda: timedelta(0) ) - uptime: timedelta = field( - default_factory=lambda: timedelta(0), - metadata=field_options( - serialization_strategy=_TimedeltaSecondsStrategy(), - ), + heating_time: Annotated[timedelta, Key(alias="heatingTime"), AsTimedelta()] = field( + default_factory=lambda: timedelta(0) ) - heating_time: timedelta = field( - default_factory=lambda: timedelta(0), - metadata=field_options( - alias="heatingTime", - serialization_strategy=_TimedeltaSecondsStrategy(), - ), - ) - service_time: timedelta = field( - default_factory=lambda: timedelta(0), - metadata=field_options( - alias="serviceTime", - serialization_strategy=_TimedeltaSecondsStrategy(), - ), + service_time: Annotated[timedelta, Key(alias="serviceTime"), AsTimedelta()] = field( + default_factory=lambda: timedelta(0) ) overheatings: int = 0 misfires: int = 0 - fuel_quantity_used: int = field( - default=0, metadata=field_options(alias="fuelQuantityUsed") - ) + fuel_quantity_used: Annotated[int, Key(alias="fuelQuantityUsed")] = 0 @dataclass(frozen=True) -class FumisUnit(_BaseModel): +class FumisUnit: """WiRCU box information.""" id: str = "Unknown" type: int = 0 - version: AwesomeVersion = field(default_factory=lambda: AwesomeVersion("0")) - command: int | None = None - rssi: int = field( - default=-100, - metadata=field_options( - serialization_strategy=_StringToIntStrategy(), - ), + version: Annotated[AwesomeVersion, Coerce(AwesomeVersion)] = field( + default_factory=lambda: AwesomeVersion("0") ) + command: int | None = None + rssi: Annotated[int, Coerce(int)] = -100 ip: str = "Unknown" timezone: str | None = None temperature: float | None = None @@ -388,61 +273,37 @@ def active_days(self) -> list[str]: @dataclass(frozen=True) # pylint: disable-next=too-many-instance-attributes,too-many-public-methods -class FumisController(_BaseModel): +class FumisController: """Fumis stove controller state.""" type: int = 0 - version: AwesomeVersion = field(default_factory=lambda: AwesomeVersion("0")) + version: Annotated[AwesomeVersion, Coerce(AwesomeVersion)] = field( + default_factory=lambda: AwesomeVersion("0") + ) command: int = -1 status: int = -1 error: int = 0 alert: int = 0 - heating_slope: float = field( - default=0, - metadata=field_options( - alias="heatingSlope", - serialization_strategy=_StringToFloatStrategy(), - ), - ) - current_time: datetime = field( - default_factory=lambda: datetime.fromtimestamp(0, tz=UTC), - metadata=field_options( - alias="currentTime", - serialization_strategy=_TimestampStrategy(), - ), - ) - timer_enable: bool = field( - default=False, metadata=field_options(alias="timerEnable") - ) - fuel_type: int = field(default=0, metadata=field_options(alias="fuelType")) - time_to_service: int | None = field( - default=None, - metadata=field_options( - alias="timeToService", - serialization_strategy=_OptionalIntStrategy(), - ), - ) - delayed_start_at: datetime | None = field( - default=None, - metadata=field_options( - alias="delayedStartAt", - serialization_strategy=_OptionalTimestampStrategy(), - ), - ) - delayed_stop_at: datetime | None = field( - default=None, - metadata=field_options( - alias="delayedStopAt", - serialization_strategy=_OptionalTimestampStrategy(), - ), + heating_slope: Annotated[float, Key(alias="heatingSlope"), Coerce(float)] = 0 + current_time: Annotated[datetime, Key(alias="currentTime"), FromEpoch()] = field( + default_factory=lambda: datetime.fromtimestamp(0, tz=UTC) ) + timer_enable: Annotated[bool, Key(alias="timerEnable")] = False + fuel_type: Annotated[int, Key(alias="fuelType")] = 0 + time_to_service: Annotated[ + int | None, Key(alias="timeToService"), Maybe(Coerce(_sentinel_int)) + ] = None + delayed_start_at: Annotated[ + datetime | None, Key(alias="delayedStartAt"), Maybe(Coerce(_sentinel_epoch)) + ] = None + delayed_stop_at: Annotated[ + datetime | None, Key(alias="delayedStopAt"), Maybe(Coerce(_sentinel_epoch)) + ] = None power: FumisPower = field(default_factory=FumisPower) statistic: FumisStatistic = field(default_factory=FumisStatistic) diagnostic: FumisDiagnostic = field(default_factory=FumisDiagnostic) - eco_mode: FumisEcoMode | None = field( - default=None, metadata=field_options(alias="ecoMode") - ) + eco_mode: Annotated[FumisEcoMode | None, Key(alias="ecoMode")] = None hybrid: FumisHybrid | None = None antifreeze: FumisAntifreeze | None = None fans: list[FumisFan] = field(default_factory=list) @@ -457,8 +318,8 @@ def stove_status(self) -> StoveStatus: return StoveStatus(self.status) @property - def stove_error(self) -> StoveError | None: - """Return the stove error as an enum, or None if no error. + def stove_error(self) -> StoveError: + """Return the stove error as an enum. Converts the raw integer error code to a StoveError enum whose value matches the device display (e.g., E102). @@ -466,8 +327,8 @@ def stove_error(self) -> StoveError | None: return StoveError.from_code(self.error) @property - def stove_alert(self) -> StoveAlert | None: - """Return the stove alert as an enum, or None if no alert. + def stove_alert(self) -> StoveAlert: + """Return the stove alert as an enum. Converts the raw integer alert code to a StoveAlert enum whose value matches the device display (e.g., A004). @@ -662,17 +523,17 @@ def temperature_channel(self, channel_id: int) -> FumisTemperature | None: @dataclass(frozen=True) -class FumisInfo(_BaseModel): +class FumisInfo(SchemaMixin, extra=REMOVE_EXTRA): """Top-level Fumis WiRCU API response. This is the complete device status returned by GET /v1/status. All structured data is accessible via the nested `unit` and - `controller` objects. + `controller` objects. Parse a raw API payload with `FumisInfo.from_dict`, + which validates and constructs the whole tree, dropping unmodeled keys. """ unit: FumisUnit = field(default_factory=FumisUnit) controller: FumisController = field(default_factory=FumisController) - api_version: AwesomeVersion = field( - default_factory=lambda: AwesomeVersion("0"), - metadata=field_options(alias="apiVersion"), - ) + api_version: Annotated[ + AwesomeVersion, Key(alias="apiVersion"), Coerce(AwesomeVersion) + ] = field(default_factory=lambda: AwesomeVersion("0")) diff --git a/tests/cli/__snapshots__/test_cli.ambr b/tests/cli/__snapshots__/test_cli.ambr index 5f0fcf5..2eb8987 100644 --- a/tests/cli/__snapshots__/test_cli.ambr +++ b/tests/cli/__snapshots__/test_cli.ambr @@ -305,8 +305,11 @@ "id": "**REDACTED**", "type": 0, "version": "2.0.0", - "rssi": "-48", - "ip": "192.168.1.2" + "command": null, + "rssi": -48, + "ip": "192.168.1.2", + "timezone": null, + "temperature": null }, "controller": { "type": 0, @@ -316,9 +319,12 @@ "error": 0, "alert": 0, "heating_slope": 0.0, - "current_time": 1574360658, + "current_time": "2019-11-21T18:24:18+00:00", "timer_enable": false, "fuel_type": 1, + "time_to_service": null, + "delayed_start_at": null, + "delayed_stop_at": null, "power": { "kw": 0.0, "actual_power": 0, @@ -328,9 +334,9 @@ }, "statistic": { "igniter_starts": 392, - "uptime": 58184580, - "heating_time": 1823340, - "service_time": 1823340, + "uptime": "673 days, 10:23:00", + "heating_time": "21 days, 2:29:00", + "service_time": "21 days, 2:29:00", "overheatings": 0, "misfires": 0, "fuel_quantity_used": 0 @@ -1055,8 +1061,15 @@ } ] }, - "eco_mode": {}, - "antifreeze": {}, + "eco_mode": { + "eco_mode_enable": null, + "eco_mode_set_type": null + }, + "hybrid": null, + "antifreeze": { + "temperature": null, + "enable": null + }, "fans": [ { "id": 1, @@ -1073,6 +1086,7 @@ "on_main_screen": true, "actual_type": 1, "set_type": 2, + "name": null, "weight": 0 }, { @@ -1082,6 +1096,7 @@ "on_main_screen": false, "actual_type": 0, "set_type": 0, + "name": null, "weight": 0 }, { @@ -1091,6 +1106,7 @@ "on_main_screen": false, "actual_type": 0, "set_type": 0, + "name": null, "weight": 0 }, { @@ -1100,6 +1116,7 @@ "on_main_screen": false, "actual_type": 0, "set_type": 0, + "name": null, "weight": 0 }, { @@ -1109,6 +1126,7 @@ "on_main_screen": false, "actual_type": 0, "set_type": 0, + "name": null, "weight": 0 }, { @@ -1118,6 +1136,7 @@ "on_main_screen": false, "actual_type": 0, "set_type": 0, + "name": null, "weight": 0 } ], @@ -1126,9 +1145,12 @@ "id": 1, "quality": 1, "quality_type": 3, + "quality_actual": null, "quantity": 0.0, + "quantity_display": null, "quantity_set_type": 0, - "quantity_actual_type": 0 + "quantity_actual_type": 0, + "name": null } ] }, diff --git a/tests/test_fumis.py b/tests/test_fumis.py index 748eac7..e18c8d4 100644 --- a/tests/test_fumis.py +++ b/tests/test_fumis.py @@ -696,7 +696,7 @@ def test_stove_state_is_str_enum() -> None: def test_stove_error_from_code() -> None: """Test StoveError.from_code conversion.""" - assert StoveError.from_code(0) is None + assert StoveError.from_code(0) == StoveError.NO_ERROR assert StoveError.from_code(102) == StoveError.CHIMNEY_DIRTY assert StoveError.from_code(241) == StoveError.CHIMNEY_ALARM assert StoveError.from_code(999) == StoveError.UNKNOWN @@ -719,7 +719,7 @@ def test_stove_error_description() -> None: def test_stove_alert_from_code() -> None: """Test StoveAlert.from_code conversion.""" - assert StoveAlert.from_code(0) is None + assert StoveAlert.from_code(0) == StoveAlert.NO_ALERT assert StoveAlert.from_code(4) == StoveAlert.LOW_BATTERY assert StoveAlert.from_code(1) == StoveAlert.LOW_FUEL assert StoveAlert.from_code(99) == StoveAlert.UNKNOWN @@ -744,7 +744,7 @@ def test_controller_stove_error_property() -> None: assert info.controller.stove_error == StoveError.CHIMNEY_DIRTY info_none = FumisInfo.from_dict({"controller": {"error": 0}}) - assert info_none.controller.stove_error is None + assert info_none.controller.stove_error == StoveError.NO_ERROR def test_controller_stove_alert_property() -> None: @@ -753,7 +753,7 @@ def test_controller_stove_alert_property() -> None: assert info.controller.stove_alert == StoveAlert.LOW_BATTERY info_none = FumisInfo.from_dict({"controller": {"alert": 0}}) - assert info_none.controller.stove_alert is None + assert info_none.controller.stove_alert == StoveAlert.NO_ALERT def test_eco_mode_enabled() -> None: