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: