feat(message): protocol boundary cleanup APIs for MyHOME #736

OWNd/message/__init__.py

@@ -171,6 +171,9 @@
     MESSAGE_TYPE_MOTION_TIMEOUT,
     MESSAGE_TYPE_PIR_SENSITIVITY,
     PIR_SENSITIVITY_MAPPING,
+    area_of_where,
+    is_apl_address,
+    normalize_where,
     OWNLightingCommand,
     OWNLightingEvent,
 )
@@ -220,6 +223,9 @@
     "MESSAGE_TYPE_ILLUMINANCE",
     "MESSAGE_TYPE_MOTION_TIMEOUT",
     "PIR_SENSITIVITY_MAPPING",
+    "is_apl_address",
+    "area_of_where",
+    "normalize_where",
     "OWNLightingEvent",
     "OWNLightingCommand",
     # Automation (WHO 2)

OWNd/message/base.py

@@ -257,10 +257,35 @@ def interface(self) -> str | None:
             else None
         )
 
+    @property
+    def what(self) -> int | None:
+        """The 'what' ID of this message"""
+        return getattr(self, "_what", None)
+
+    @property
+    def what_param(self) -> list[str]:
+        """The 'what' parameters of this message"""
+        return getattr(self, "_what_param", [])
+
+    @property
+    def where_param(self) -> list[str]:
+        """The 'where' parameters of this message"""
+        return getattr(self, "_where_param", [])
+
     @property
     def dimension(self) -> int | None:
         """The 'dimension' ID of this message"""
-        return self._dimension
+        return getattr(self, "_dimension", None)
+
+    @property
+    def dimension_param(self) -> list[str]:
+        """The 'dimension' parameters of this message"""
+        return getattr(self, "_dimension_param", [])
+
+    @property
+    def dimension_value(self) -> list[str]:
+        """The 'dimension' values of this message"""
+        return getattr(self, "_dimension_value", [])
 
     @property
     def entity(self) -> str:

OWNd/message/lighting.py

@@ -21,6 +21,74 @@
 PIR_SENSITIVITY_MAPPING = ["low", "medium", "high", "very high"]
 
 
+def is_apl_address(base: str) -> bool:
+    """Check if base address is a valid OpenWebNet Point-to-Point (APL) address.
+
+    Point-to-point addressing combinations:
+      - A = 0; PL [1-9]        -> 2 digits (e.g. 02 = Area 0, PL 2; accepted by BTicino firmware)
+      - A = 00; PL [01-15]     -> 4 digits (e.g. 0015 = Area 00, PL 15)
+      - A [1-9]; PL [1-9]      -> 2 digits (e.g. 15 = Area 1, PL 5)
+      - A = 10; PL [01-15]     -> 4 digits (e.g. 1015 = Area 10, PL 15)
+      - A [01-09]; PL [10-15]  -> 4 digits (e.g. 0115 = Area 1, PL 15)
+    """
+    if not base.isdigit():
+        return False
+    if len(base) == 2:
+        a = int(base[0])
+        pl = int(base[1])
+        if a == 0:
+            return 1 <= pl <= 9
+        return 1 <= a <= 9 and 1 <= pl <= 9
+    if len(base) == 4:
+        a = int(base[:2])
+        pl = int(base[2:])
+        if a == 0:
+            return 1 <= pl <= 15
+        if 1 <= a <= 9:
+            return 10 <= pl <= 15
+        if a == 10:
+            return 1 <= pl <= 15
+    return False
+
+
+def area_of_where(where: str | int | None) -> str | None:
+    """Return the area status WHERE for a given Point-to-Point (APL) WHERE."""
+    if where is None:
+        return None
+    where_str = str(where).strip()
+    parts = where_str.split("#", 1)
+    base = parts[0]
+    if not is_apl_address(base):
+        return None
+    if len(base) == 2:
+        if base[0] == "0":
+            return "00"
+        return base[0]
+    a = base[:2]
+    if a == "00":
+        return "00"
+    if a == "10":
+        return "100"
+    return str(int(a))
+
+
+def normalize_where(where: str | int | None) -> str:
+    """Normalize OpenWebNet address while preserving Point-to-Point (APL) addressing."""
+    if where is None:
+        return ""
+    where_str = str(where).strip()
+    if not where_str:
+        return ""
+    parts = where_str.split("#", 1)
+    base = parts[0]
+    if is_apl_address(base) or base in ("00", "100", "0"):
+        norm_base = base
+    elif base.isdigit() and base.startswith("00") and int(base) > 15:
+        norm_base = str(int(base))
+    else:
+        norm_base = base
+    return f"{norm_base}#{parts[1]}" if len(parts) > 1 else norm_base
+
 
 class OWNLightingEvent(OWNEvent):
     def __init__(self, data: str) -> None:
@@ -272,6 +340,13 @@ def motion_timeout(self) -> datetime.timedelta | None:
     def color_temp(self) -> int | None:
         return self._color_temp
 
+    @property
+    def color_temp_kelvin(self) -> int | None:
+        """Color temperature in Kelvin when a valid mired reading (> 1) is present."""
+        if self._color_temp is not None and self._color_temp > 1:
+            return int(1000000 / self._color_temp)
+        return None
+
     @property
     def supports_color_temp(self) -> bool | None:
         return self._supports_color_temp
@@ -292,6 +367,10 @@ def value(self) -> int | None:
     def hs(self) -> tuple[int, int] | None:
         return self._hs
 
+    @property
+    def hs_color(self) -> tuple[int, int] | None:
+        return self._hs
+
     @property
     def hsv(self) -> tuple[int, int, int] | None:
         return self._hsv
@@ -311,6 +390,55 @@ def supports_rgb(self) -> bool | None:
 
 
 class OWNLightingCommand(OWNCommand):
+    @property
+    def brightness(self) -> int | None:
+        """Decoded brightness percentage (0..100) for Dimension 1/4 write commands."""
+        if self._dimension in (1, 4) and self._dimension_value:
+            try:
+                raw = int(self._dimension_value[0])
+            except (TypeError, ValueError):
+                return None
+            if 100 <= raw <= 200:
+                return raw - 100
+        return None
+
+    @property
+    def color_temp(self) -> int | None:
+        """Decoded color temperature in mireds for Dimension 14 write commands."""
+        if self._dimension == 14 and self._dimension_value:
+            try:
+                val = int(self._dimension_value[0])
+            except (TypeError, ValueError):
+                return None
+            if val > 1:
+                return val
+        return None
+
+    @property
+    def color_temp_kelvin(self) -> int | None:
+        """Decoded color temperature in Kelvin for Dimension 14 write commands."""
+        ct = self.color_temp
+        return int(1000000 / ct) if ct is not None and ct > 1 else None
+
+    @property
+    def hs(self) -> tuple[int, int] | None:
+        """Decoded (hue, saturation) tuple for Dimension 12 write commands."""
+        if self._dimension == 12 and len(self._dimension_value) >= 3:
+            try:
+                h = int(self._dimension_value[0])
+                s = int(self._dimension_value[1])
+                v = int(self._dimension_value[2])
+            except (TypeError, ValueError):
+                return None
+            if h <= 360 and (h, s, v) != (511, 127, 255):
+                return (h, s)
+        return None
+
+    @property
+    def hs_color(self) -> tuple[int, int] | None:
+        """Alias for :attr:`hs`."""
+        return self.hs
+
     @classmethod
     def status(cls, where: str | int) -> OWNLightingCommand:
         message = cls(f"*#1*{where}##")

OWNd/message/sound.py

@@ -29,6 +29,7 @@ def __init__(self, data: str) -> None:
         self._environment = self._zone[1] if self._is_routing_event else None
         self._routed_source = self._zone[2] if self._is_routing_event else None
         self._volume: int | None = None
+        self._volume_step_delta: int | None = None
         self._frequency_khz: int | None = None
         self._track: int | None = None
         self._rds_text: str | None = None
@@ -54,6 +55,16 @@ def __init__(self, data: str) -> None:
         elif self._state == 100:
             self._is_source_busy = True
             self._human_readable_log = f"{subject} is BUSY."
+        elif self._state is not None and 1001 <= self._state <= 1015:
+            self._volume_step_delta = self._state - 1000
+            self._human_readable_log = (
+                f"{subject} increase volume by {self._volume_step_delta} step(s)."
+            )
+        elif self._state is not None and 1101 <= self._state <= 1115:
+            self._volume_step_delta = -(self._state - 1100)
+            self._human_readable_log = (
+                f"{subject} decrease volume by {abs(self._volume_step_delta)} step(s)."
+            )
         elif self._state == 5000:
             self._is_seek_up = True
             self._human_readable_log = f"{subject} seek forward."
@@ -162,6 +173,11 @@ def zone(self) -> str:
     def volume(self) -> int | None:
         return self._volume
 
+    @property
+    def volume_step_delta(self) -> int | None:
+        """Signed volume step delta (+1..+15 for WHAT 1001..1015, -1..-15 for WHAT 1101..1115)."""
+        return self._volume_step_delta
+
     @property
     def frequency_khz(self) -> int | None:
         return self._frequency_khz

tests/test_sound.py

@@ -275,3 +275,24 @@ def test_who16_sound_source_events() -> None:
     blank_rds = OWNSoundEvent("*#16*101*8*32*32*32*32*32*32*32*32##")
     assert blank_rds.rds_text is None
     assert "RDS text: ''" in blank_rds.human_readable_log
+
+
+def test_who16_volume_step_delta_events() -> None:
+    """Verify WHO 16 stepped volume WHAT 1001..1015 and 1101..1115 decode volume_step_delta."""
+    vol_up_1 = OWNMessage.parse("*16*1001*21##")
+    assert isinstance(vol_up_1, OWNSoundEvent)
+    assert vol_up_1.volume_step_delta == 1
+    assert "increase volume by 1 step(s)" in vol_up_1.human_readable_log
+
+    vol_up_15 = OWNMessage.parse("*16*1015*21##")
+    assert isinstance(vol_up_15, OWNSoundEvent)
+    assert vol_up_15.volume_step_delta == 15
+
+    vol_down_1 = OWNMessage.parse("*16*1101*21##")
+    assert isinstance(vol_down_1, OWNSoundEvent)
+    assert vol_down_1.volume_step_delta == -1
+    assert "decrease volume by 1 step(s)" in vol_down_1.human_readable_log
+
+    vol_down_15 = OWNMessage.parse("*16*1115*21##")
+    assert isinstance(vol_down_15, OWNSoundEvent)
+    assert vol_down_15.volume_step_delta == -15

tests/test_who1_ts10_builders.py

@@ -45,3 +45,88 @@ def test_variable_timer():
     assert str(OWNLightingCommand.get_variable_timer(11)) == "*#1*11*2##"
     with pytest.raises(ValueError):
         OWNLightingCommand.set_variable_timer(11, 0, 60, 0)
+
+
+def test_apl_addressing_helpers():
+    from OWNd.message import area_of_where, is_apl_address, normalize_where
+
+    assert is_apl_address("02") is True
+    assert is_apl_address("15") is True
+    assert is_apl_address("0015") is True
+    assert is_apl_address("0115") is True
+    assert is_apl_address("1015") is True
+    assert is_apl_address("00") is False
+    assert is_apl_address("100") is False
+    assert is_apl_address("#6") is False
+    assert is_apl_address("1101") is False
+
+    assert area_of_where("02") == "00"
+    assert area_of_where("15") == "1"
+    assert area_of_where("0015") == "00"
+    assert area_of_where("0115") == "1"
+    assert area_of_where("1015") == "100"
+    assert area_of_where("#6") is None
+    assert area_of_where(None) is None
+
+    assert normalize_where("02") == "02"
+    assert normalize_where("0015") == "0015"
+    assert normalize_where("0021") == "21"
+    assert normalize_where("0109") == "0109"
+    assert normalize_where("15#4#01") == "15#4#01"
+    assert normalize_where("#5") == "#5"
+    assert normalize_where(None) == ""
+    assert normalize_where("   ") == ""
+
+
+def test_lighting_dimension_properties():
+    from OWNd.message import OWNLightingEvent, OWNMessage
+
+    cmd_b = OWNMessage.parse("*#1*#6*#1*150*0##")
+    assert isinstance(cmd_b, OWNLightingCommand)
+    assert cmd_b.brightness == 50
+    assert cmd_b.dimension_value == ["150", "0"]
+    assert cmd_b.where_param == []
+    assert cmd_b.dimension_param == []
+    assert cmd_b.what is None
+    assert cmd_b.what_param == []
+    assert cmd_b.color_temp is None
+    assert cmd_b.color_temp_kelvin is None
+    assert cmd_b.hs is None
+
+    cmd_b_bad = OWNMessage.parse("*#1*#6*#1*50*0##")
+    assert isinstance(cmd_b_bad, OWNLightingCommand)
+    assert cmd_b_bad.brightness is None
+
+    cmd_b._dimension_value = ["abc"]
+    assert cmd_b.brightness is None
+
+    cmd_ct = OWNMessage.parse("*#1*#6*#14*153##")
+    assert isinstance(cmd_ct, OWNLightingCommand)
+    assert cmd_ct.color_temp == 153
+    assert cmd_ct.color_temp_kelvin == int(1000000 / 153)
+    assert cmd_ct.brightness is None
+
+    cmd_ct._dimension_value = ["1"]
+    assert cmd_ct.color_temp is None
+    cmd_ct._dimension_value = ["abc"]
+    assert cmd_ct.color_temp is None
+
+    cmd_hs = OWNMessage.parse("*#1*#6*#12*120*50*80##")
+    assert isinstance(cmd_hs, OWNLightingCommand)
+    assert cmd_hs.hs == (120, 50)
+    assert cmd_hs.hs_color == (120, 50)
+
+    cmd_hs._dimension_value = ["511", "127", "255"]
+    assert cmd_hs.hs is None
+    cmd_hs._dimension_value = ["abc", "50", "80"]
+    assert cmd_hs.hs is None
+
+    ev_ct = OWNMessage.parse("*#1*#6*14*153##")
+    assert isinstance(ev_ct, OWNLightingEvent)
+    assert ev_ct.color_temp_kelvin == int(1000000 / 153)
+
+    ev_hs = OWNMessage.parse("*#1*#6*12*120*50*80##")
+    assert isinstance(ev_hs, OWNLightingEvent)
+    assert ev_hs.hs_color == (120, 50)
+    assert ev_hs.color_temp_kelvin is None
+