feat: Set NotImplemented from device schema.

.github/workflows/pytestPR.yaml

@@ -0,0 +1,30 @@
+name: "PyTest PR"
+
+on:
+  pull_request_target:
+    types:
+      - opened
+      - edited
+      - synchronize
+
+jobs:
+  main:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.13"]
+    steps:
+      - uses: actions/checkout@v3
+      - name: developer mode install
+        run: pip install -e .
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v4
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install pytest dependencies
+        run: pip install pytest pytest-md pytest-emoji
+      - uses: pavelzw/pytest-action@v2
+        with:
+          emoji: false
+          verbose: false
+          job-summary: true

src/blueair_api/device_aws.py

@@ -4,10 +4,11 @@
 from .callbacks import CallbacksMixin
 from .http_aws_blueair import HttpAwsBlueair
 from .model_enum import ModelEnum
-from .util import convert_api_array_to_dict, safely_get_json_value
+from . import ir_aws as ir
 
 _LOGGER = logging.getLogger(__name__)
 
+type AttributeType[T] = T | None | type[NotImplemented]
 
 @dataclasses.dataclass(slots=True)
 class DeviceAws(CallbacksMixin):
@@ -27,71 +28,82 @@ async def create_device(cls, api, uuid, name, mac, type_name, refresh=False):
         return device_aws
 
     api: HttpAwsBlueair
-    uuid: str = None
-    name: str = None
-    name_api: str = None
-    mac: str = None
-    type_name: str = None
-
-    sku: str = None
-    firmware: str = None
-    mcu_firmware: str = None
-    serial_number: str = None
-
-    brightness: int = None
-    child_lock: bool = None
-    fan_speed: int = None
-    fan_auto_mode: bool = None
-    running: bool = None
-    night_mode: bool = None
-    germ_shield: bool = None
-
-    pm1: int = None
-    pm2_5: int = None
-    pm10: int = None
-    tVOC: int = None
-    temperature: int = None
-    humidity: int = None
-    filter_usage: int = None  # percentage
-    wifi_working: bool = None
-
-    # i35
-    wick_usage: int = None  # percentage
-    wick_dry_mode: bool = None
-    water_shortage: bool = None
-    auto_regulated_humidity: int = None
+    uuid : str | None = None
+    name : str | None = None
+    name_api : str | None = None
+    mac : str | None = None
+    type_name : str | None = None
+
+    # Attributes are defined below.
+    # We mandate that unittests shall test all fields of AttributeType.
+    sku : AttributeType[str] = None
+    firmware : AttributeType[str] = None
+    mcu_firmware : AttributeType[str] = None
+    serial_number : AttributeType[str] = None
+
+    brightness : AttributeType[int] = None
+    child_lock : AttributeType[bool] = None
+    fan_speed : AttributeType[int] = None
+    fan_auto_mode : AttributeType[bool] = None
+    standby : AttributeType[bool] = None
+    night_mode : AttributeType[bool] = None
+    germ_shield : AttributeType[bool] = None
+
+    pm1 : AttributeType[int] = None
+    pm2_5 : AttributeType[int] = None
+    pm10 : AttributeType[int] = None
+    tVOC : AttributeType[int] = None
+    temperature : AttributeType[int] = None
+    humidity : AttributeType[int] = None
+    filter_usage : AttributeType[int] = None  # percentage
+    wifi_working : AttributeType[bool] = None
+
+    wick_usage : AttributeType[int] = None  # percentage
+    wick_dry_mode : AttributeType[bool] = None
+    water_shortage : AttributeType[bool] = None
+    auto_regulated_humidity : AttributeType[int] = None
 
     async def refresh(self):
         _LOGGER.debug(f"refreshing blueair device aws: {self}")
         info = await self.api.device_info(self.name_api, self.uuid)
-        sensor_data = convert_api_array_to_dict(info["sensordata"])
-        self.pm1 = safely_get_json_value(sensor_data, "pm1", int)
-        self.pm2_5 = safely_get_json_value(sensor_data, "pm2_5", int)
-        self.pm10 = safely_get_json_value(sensor_data, "pm10", int)
-        self.tVOC = safely_get_json_value(sensor_data, "tVOC", int)
-        self.temperature = safely_get_json_value(sensor_data, "t", int)
-        self.humidity = safely_get_json_value(sensor_data, "h", int)
-
-        self.name = safely_get_json_value(info, "configuration.di.name")
-        self.firmware = safely_get_json_value(info, "configuration.di.cfv")
-        self.mcu_firmware = safely_get_json_value(info, "configuration.di.mfv")
-        self.serial_number = safely_get_json_value(info, "configuration.di.ds")
-        self.sku = safely_get_json_value(info, "configuration.di.sku")
-
-        states = convert_api_array_to_dict(info["states"])
-        self.running = safely_get_json_value(states, "standby") is False
-        self.night_mode = safely_get_json_value(states, "nightmode", bool)
-        self.germ_shield = safely_get_json_value(states, "germshield", bool)
-        self.brightness = safely_get_json_value(states, "brightness", int)
-        self.child_lock = safely_get_json_value(states, "childlock", bool)
-        self.fan_speed = safely_get_json_value(states, "fanspeed", int)
-        self.fan_auto_mode = safely_get_json_value(states, "automode", bool)
-        self.filter_usage = safely_get_json_value(states, "filterusage", int)
-        self.wifi_working = safely_get_json_value(states, "online", bool)
-        self.wick_usage = safely_get_json_value(states, "wickusage", int)
-        self.wick_dry_mode = safely_get_json_value(states, "wickdrys", bool)
-        self.auto_regulated_humidity = safely_get_json_value(states, "autorh", int)
-        self.water_shortage = safely_get_json_value(states, "wshortage", bool)
+        # ir.parse_json(ir.Attribute, ir.query_json(info, "configuration.da"))
+        ds = ir.parse_json(ir.Sensor, ir.query_json(info, "configuration.ds"))
+        dc = ir.parse_json(ir.Control, ir.query_json(info, "configuration.dc"))
+
+        sensor_data = ir.SensorPack(info["sensordata"]).to_latest_value()
+
+        def getter(data_dict, decl_dict, key):
+            return data_dict.get(key) if key in decl_dict else NotImplemented
+
+        self.pm1 = getter(sensor_data, ds, "pm1")
+        self.pm2_5 = getter(sensor_data, ds, "pm2_5")
+        self.pm10 = getter(sensor_data, ds, "pm10")
+        self.tVOC = getter(sensor_data, ds, "tVOC")
+        self.temperature = getter(sensor_data, ds, "t")
+        self.humidity = getter(sensor_data, ds, "h")
+
+        self.name = ir.query_json(info, "configuration.di.name")
+        self.firmware = ir.query_json(info, "configuration.di.cfv")
+        self.mcu_firmware = ir.query_json(info, "configuration.di.mfv")
+        self.serial_number = ir.query_json(info, "configuration.di.ds")
+        self.sku = ir.query_json(info, "configuration.di.sku")
+
+        states = ir.SensorPack(info["states"]).to_latest_value()
+        # "online" is not defined in the schema.
+        self.wifi_working = getter(states, {"online"}, "online")
+
+        self.standby = getter(states, dc, "standby")
+        self.night_mode = getter(states, dc, "nightmode")
+        self.germ_shield = getter(states, dc, "germshield")
+        self.brightness = getter(states, dc, "brightness")
+        self.child_lock = getter(states, dc, "childlock")
+        self.fan_speed = getter(states, dc, "fanspeed")
+        self.fan_auto_mode = getter(states, dc, "automode")
+        self.filter_usage = getter(states, dc, "filterusage")
+        self.wick_usage = getter(states, dc, "wickusage")
+        self.wick_dry_mode = getter(states, dc, "wickdrys")
+        self.auto_regulated_humidity = getter(states, dc, "autorh")
+        self.water_shortage = getter(states, dc, "wshortage")
 
         self.publish_updates()
         _LOGGER.debug(f"refreshed blueair device aws: {self}")
@@ -106,11 +118,22 @@ async def set_fan_speed(self, value: int):
         await self.api.set_device_info(self.uuid, "fanspeed", "v", value)
         self.publish_updates()
 
-    async def set_running(self, running: bool):
-        self.running = running
-        await self.api.set_device_info(self.uuid, "standby", "vb", not running)
+    async def set_standby(self, value: bool):
+        self.standby = value
+        await self.api.set_device_info(self.uuid, "standby", "vb", value)
         self.publish_updates()
 
+    # FIXME: avoid state translation at the API level and depreate running.
+    # replace with standby which is standard across aws devices.
+    @property
+    def running(self) -> AttributeType[bool]:
+        if self.standby is None or self.standby is NotImplemented:
+            return self.standby
+        return not self.standby
+
+    async def set_running(self, running: bool):
+        await self.set_standby(not running)
+
     async def set_fan_auto_mode(self, fan_auto_mode: bool):
         self.fan_auto_mode = fan_auto_mode
         await self.api.set_device_info(self.uuid, "automode", "vb", fan_auto_mode)

src/blueair_api/ir_aws.py

@@ -0,0 +1,176 @@
+from typing import Any, TypeVar
+from collections.abc import Iterable
+import dataclasses
+import base64
+
+type ScalarType = str | float | bool
+type MappingType = dict[str, "ObjectType"]
+type SequenceType = list["ObjectType"]
+type ObjectType = ScalarType | MappingType | SequenceType
+
+
+def query_json(jsonobj: ObjectType, path: str):
+    value = jsonobj
+    segs = path.split(".")
+    for i, seg in enumerate(segs[:-1]):
+        if not isinstance(value, dict | list):
+            raise KeyError(
+                f"cannot resolve path segment on a scalar "
+                f"when resolving segment {i}:{seg} of {path}.")
+        if isinstance(value, list):
+            value = value[int(seg)]
+        else:
+            try:
+                value = value[seg]
+            except KeyError:
+                raise KeyError(
+                f"cannot resolve path segment on a scalar "
+                f"when resolving segment {i}:{seg} of {path}. "
+                f"available keys are {value.keys()}.")
+
+    # last segment returns None if it is not found.
+    return value.get(segs[-1])
+
+
+def parse_json[T](kls: type[T], jsonobj: MappingType) -> dict[str, T]:
+    """Parses a json mapping object to dict.
+
+    The key is preserved. The value is parsed as dataclass type kls.
+    """
+    result = {}
+    fields = dataclasses.fields(kls)
+
+    for key, value in jsonobj.items():
+        a = dict(value)  # make a copy.
+        kwargs = {}
+        for field in fields:
+            if field.name == "extra_fields":
+                continue
+            if field.default is dataclasses.MISSING:
+                kwargs[field.name] = a.pop(field.name)
+            else:
+                kwargs[field.name] = a.pop(field.name, field.default)
+
+        result[key] = kls(**kwargs, extra_fields=a)
+    return result
+
+
+########################
+# Blueair AWS API Schema.
+
+@dataclasses.dataclass
+class Attribute:
+    """DeviceAttribute(da); defines an attribute
+
+    An attribute is most likely mutable. An attribute may
+    also have alias names, likely derived from the 'dc' relation
+    e.g. a/sb, a/standby all refer to the 'sb' attribute.
+    """
+    extra_fields : MappingType
+    n: str   # name
+    a: int | bool   # default attribute value, example value?
+    e: bool     # ??? always True
+    fe:bool     # ???  always True
+    ot: str     # object type? topic type?
+    p: bool     # only false for reboot and sflu
+    tn: str   # topic name a path-like name d/????/a/{n}
+
+
+@dataclasses.dataclass
+class Sensor:
+    """DeviceSensor(ds); seems to define a sensor.
+
+    We never directly access these objects. Thos this defines
+    the schema for 'h', 't', 'pm10' etc that gets returned in
+    the sensor_data senml SensorPack.
+    """
+    extra_fields : MappingType
+    n: str    # name
+    i: int    # integration time? in millis
+    e: bool   # ???
+    fe: bool  # ??? always True.
+    ot: str   # object type / topic name
+    tf: str   # senml+json; topic format
+    tn: str   # topic name a path-like name d/????/s/{n}
+    ttl: int  # only seen 0 or -1, not sure if used.
+
+@dataclasses.dataclass
+class Control:
+    """DeviceControl (dc); seems to define a state.
+
+    The states SensorPack seem to be using fields defined
+    in dc. The only exception is 'online' which is not defined
+    here.
+    """
+    extra_fields : MappingType
+    n: str  # name
+    v: int | bool
+    a: str | None = None
+    s: str | None = None
+    d: str | None = None  # device info json path
+
+
+########################
+# SenML RFC8428
+
+@dataclasses.dataclass
+class Record:
+    """A RFC8428 SenML record, resolved to Python types."""
+    name: str
+    unit: str | None
+    value: ScalarType
+    timestamp: float | None
+    integral: float | None
+
+
+class SensorPack(list[Record]):
+    """Represents a RFC8428 SensorPack, resolved to Python Types."""
+
+    def __init__(self, stream: Iterable[MappingType]):
+        seq = []
+        for record in stream:
+            rs = None
+            rt = None
+            rn = 0
+            ru = None
+            for label, value in record.items():
+                match label:
+                    case 'bn' | 'bt' | 'bu' | 'bv' | 'bs' | 'bver':
+                        raise ValueError("TODO: base fields not supported. c.f. RFC8428, 4.1")
+                    case 't':
+                        rt = float(value)
+                    case 's':
+                        rs = float(value)
+                    case 'v':
+                        rv = float(value)
+                    case 'vb':
+                        rv = bool(value)
+                    case 'vs':
+                        rv = str(value)
+                    case 'vd':
+                        rv = bytes(base64.b64decode(value))
+                    case 'n':
+                        rn = str(value)
+                    case 'u':
+                        ru = str(value)
+                    case 't':
+                        rn = float(value)
+            seq.append(Record(name=rn, unit=ru, value=rv, integral=rs, timestamp=rt))
+        super().__init__(seq)
+
+    def to_latest_value(self) -> dict[str, ScalarType]:
+        return {rn : record.value for rn, record in self.to_latest().items()}
+
+    def to_latest(self) -> dict[str, Record]:
+        latest = {}
+        for record in self:
+            rn = record.name
+            if record.name not in latest:
+                latest[rn] = record
+            elif record.timestamp is None:
+                latest[rn] = record
+            elif latest[record.name].timestamp is None:
+                latest[rn] = record
+            elif latest[record.name].timestamp < record.timestamp:
+                latest[rn] = record
+        return latest