refactor(protocol): eliminate private OWNd access and runtime shims (#736)

custom_components/myhome/climate.py

@@ -40,19 +40,16 @@
     MESSAGE_TYPE_MAIN_TEMPERATURE,
     MESSAGE_TYPE_MODE,
     MESSAGE_TYPE_MODE_TARGET,
+    MESSAGE_TYPE_SEASON,
     MESSAGE_TYPE_TARGET_TEMPERATURE,
+    MESSAGE_TYPE_ZONE_STATE,
+    SEASON_CONDITIONING,
+    SEASON_HEATING,
     OWNCommand,
     OWNHeatingCommand,
     OWNHeatingEvent,
 )
 
-try:  # OWNd > 2.0.0b10 (OpenWebNet-HA/OWNd#94)
-    from OWNd.message import MESSAGE_TYPE_SEASON, SEASON_CONDITIONING, SEASON_HEATING
-except ImportError:  # pragma: no cover - pinned OWNd 2.0.0b10 reports WHAT 1/0 as hvac_mode
-    MESSAGE_TYPE_SEASON = "hvac_season"
-    SEASON_HEATING = "heating"
-    SEASON_CONDITIONING = "conditioning"
-
 from .const import (
     BUS_ROUTING,
     CONF_CENTRAL,
@@ -84,10 +81,6 @@
 PLATFORM = Platform.CLIMATE
 PARALLEL_UPDATES = 0
 
-# WHO 4 dimension 7 zone state (OWNd#60); the values OWNd uses, spelled out
-# here until the OWNd pin exports them.  On MyHomeServer1 + Home+Control
-# plants this is the only frame carrying the zone's mode and setpoint (#429).
-MESSAGE_TYPE_ZONE_STATE = "zone_state"
 # A dimension 12 frame and the frame that turns the zone OFF follow within ~0.1 s (#454, #383)
 _PROTECTION_FRAME_WINDOW = 2.0
 _ZONE_CONTEXT_MODES = {"heating": HVACMode.HEAT, "cooling": HVACMode.COOL, "automatic": HVACMode.AUTO}
@@ -832,9 +825,7 @@ def _dimension_3_is_protection(self, message: OWNHeatingEvent) -> bool:
         """
         if self._attr_hvac_mode == HVACMode.OFF:
             return True
-        value = getattr(message, "_dimension_value", None)
-        if not value:
-            return False
+        value: list[str] = getattr(message, "dimension_value", None) or []
         return bool(len(value) > 1 and value[1] == "3" and self._attr_hvac_mode is None)
 
     def _remember_nominal(self) -> None:

custom_components/myhome/const.py

@@ -6,6 +6,16 @@
 
 from homeassistant.const import Platform
 
+from .where_grammar import (
+    area_of_where as area_of_where,
+)
+from .where_grammar import (
+    is_apl_address as is_apl_address,
+)
+from .where_grammar import (
+    normalize_where as normalize_where,
+)
+
 LOGGER = logging.getLogger(__package__)
 DOMAIN = "myhome"
 
@@ -207,17 +217,6 @@ def get_ownd_version() -> str:
 SHARED_BUS_EVIDENCE_WINDOW_S = 600.0
 
 
-def who4_raw_to_celsius(raw: str) -> float:
-    """Decode a raw WHO 4 temperature (``SXXX``, tenths of a degree) that OWNd did not decode.
-
-    Raises ``ValueError`` when ``raw`` is not a number. Exactly zero is ``0.0`` whatever the sign digit.
-    """
-    if len(raw) == 4 and raw.startswith("1"):
-        magnitude = float(raw[1:]) / 10.0
-        return -magnitude if magnitude else 0.0
-    return float(raw) / 10.0
-
-
 def eight_bits_to_percent(value: int) -> int:
     """Convert an 8-bit brightness (0-255) to percentage (0-100)."""
     return int(round((value * 100) / 255, 0))
@@ -237,85 +236,6 @@ def percent_to_eight_bits(value: int) -> int:
     return int(round((value * 255) / 100, 0))
 
 
-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]
-    # Length is 4
-    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.
-
-    Point-to-point WHERE addresses must never have leading zeros stripped:
-      - '02' means Area 0, Point 2 (2 digits)
-      - '0015' means Area 00, Point 15 (4 digits)
-      - '15' means Area 1, Point 5 (2 digits)
-    Area broadcasts '00' (Area 0) and '100' (Area 10) are also preserved.
-    Other numeric addresses (such as zero-padded CEN+/dry contact object IDs like '0021')
-    have leading zeros stripped to match integer IDs.
-    """
-    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():
-        norm_base = str(int(base))
-    else:
-        norm_base = base
-    return f"{norm_base}#{parts[1]}" if len(parts) > 1 else norm_base
-
-
 SERVICE_TURN_ON_TIMED = "turn_on_timed"
 
 PRESET_TIMERS: dict[float, int] = {
@@ -520,7 +440,7 @@ def build_timed_turn_on_command(
     seconds: float = 0,
 ) -> Any:
     """Build OpenWebNet hardware timer command for WHO=1."""
-    from OWNd.message import OWNCommand, OWNLightingCommand
+    from OWNd.message import OWNLightingCommand
 
     total_seconds = float(duration if duration is not None else 0.0)
     total_seconds += (int(hours) * 3600) + (int(minutes) * 60) + float(seconds)
@@ -533,18 +453,14 @@ def build_timed_turn_on_command(
         what = PRESET_TIMERS[rounded_secs]
         if hasattr(OWNLightingCommand, "switch_on_timed"):
             return OWNLightingCommand.switch_on_timed(where, what)
-        frame = f"*1*{what}*{where}##"  # pragma: no cover - fallback on released OWNd 2.0.0b10
-        parsed = OWNCommand.parse(frame)  # pragma: no cover
-        return parsed if parsed is not None else OWNCommand(frame)  # pragma: no cover
+        return OWNLightingCommand.parse(f"*1*{what}*{where}##")
 
     int_secs = int(round(total_seconds))
     h = max(0, min(255, int_secs // 3600))
     m = max(0, min(59, (int_secs % 3600) // 60))
     s = max(0, min(59, int_secs % 60))
     if hasattr(OWNLightingCommand, "set_variable_timer"):
         return OWNLightingCommand.set_variable_timer(where, h, m, s)
-    frame = f"*#1*{where}*#2*{h}*{m}*{s}##"  # pragma: no cover - fallback on released OWNd 2.0.0b10
-    parsed = OWNCommand.parse(frame)  # pragma: no cover
-    return parsed if parsed is not None else OWNCommand(frame)  # pragma: no cover
+    return OWNLightingCommand.parse(f"*#1*{where}*#2*{h}*{m}*{s}##")
 
 

custom_components/myhome/device_health.py

@@ -281,7 +281,7 @@ def _observe_lighting_autodiag(self, message: Any) -> None:
         where = message_where(message)
         if where is None or self.is_ignored(1, where):
             return
-        values = getattr(message, "_dimension_value", None)
+        values = getattr(message, "dimension_value", None)
         if not isinstance(values, list) or not values:
             return
         mask = str(values[0])

custom_components/myhome/gateway.py

@@ -3,7 +3,6 @@
 
 import asyncio
 import collections
-import logging
 import time
 from collections.abc import Iterable
 from typing import Any
@@ -110,7 +109,6 @@
     "OWNEventSession",
     "OWNGateway",
     "OWNSession",
-    "_StatusRequestLogFilter",
     "_cancel_written",
     "_resolve_written",
     "_session_is_open",
@@ -125,27 +123,6 @@
 ]
 
 
-class _StatusRequestLogFilter(logging.Filter):
-    """Downgrade spurious status-request retry errors to DEBUG.
-
-    OWNd < 2.0.0b8 logged intermediate status-request retries (*#...##) as ERROR
-    instead of DEBUG when the gateway NACKed uninstalled optional subsystems
-    (issue #406, OpenWebNet-HA/OWNd#43).
-    """
-
-    def filter(self, record: logging.LogRecord) -> bool:
-        if (
-            record.levelno == logging.ERROR
-            and "Could not send message `*#" in record.getMessage()
-        ):
-            record.levelno = logging.DEBUG
-            record.levelname = "DEBUG"
-        return True
-
-
-LOGGER.addFilter(_StatusRequestLogFilter())
-
-
 def command_session_limit(model: str | None) -> int | None:
     """Return how many command sessions a known gateway model accepts at once.
 
@@ -873,8 +850,8 @@ async def _process_message(self, message: Any) -> None:
 
     def _handle_gateway_diagnostics(self, message: OWNGatewayEvent) -> None:
         """Handle WHO=13 Gateway Management diagnostic telemetry."""
-        dim = getattr(message, "dimension", getattr(message, "_dimension", None))
-        dim_val = getattr(message, "dimension_value", getattr(message, "_dimension_value", []))
+        dim = getattr(message, "dimension", None)
+        dim_val = getattr(message, "dimension_value", [])
 
         # ── Dimension 0 & 22: Time & Timezone ────────────────────────────────
         if dim in (0, 22) and dim_val:
@@ -902,7 +879,7 @@ def _handle_gateway_diagnostics(self, message: OWNGatewayEvent) -> None:
 
         # ── Dimension 16: Firmware Version ───────────────────────────────
         elif dim == 16:
-            fw = getattr(message, "firmware_version", getattr(message, "_firmware_version", None))
+            fw = getattr(message, "firmware_version", None)
             if fw:
                 self._who13["firmware"] = fw
             if fw and fw != self.gateway.firmware:
@@ -932,7 +909,7 @@ def _handle_device_type(self, raw_code: str) -> None:
 
     def _handle_gateway_identity_diagnostics(self, message: Any) -> None:
         """Record a WHO=1013 dimension-1 (OBJECT_MODEL) reply and let the resolver decide."""
-        dim_val = getattr(message, "dimension_value", getattr(message, "_dimension_value", []))
+        dim_val = getattr(message, "dimension_value", [])
         if not dim_val or not isinstance(dim_val, list):
             return
         reading = read_who1013(str(dim_val[0]))

custom_components/myhome/gateway_events.py

@@ -282,7 +282,7 @@ async def process_message(self, message: Any) -> None:
                         self.handler._resync_manager.handle_ptp_echo(message)
 
                     dim = getattr(message, "dimension", None)
-                    is_not_dimension = dim is None or type(dim).__name__ == "MagicMock"
+                    is_not_dimension = not (isinstance(dim, int) and not isinstance(dim, bool))
 
                     if message.is_on is not None and is_not_dimension:
                         event = "on" if message.is_on else "off"
@@ -324,7 +324,7 @@ async def process_message(self, message: Any) -> None:
                         event = "stop"
 
                     where_raw = getattr(message, "where", None)
-                    where_val = str(where_raw) if (where_raw is not None and not str(where_raw).startswith("<MagicMock")) else "0"
+                    where_val = str(where_raw) if (isinstance(where_raw, (str, int)) and not isinstance(where_raw, bool)) else "0"
 
                     auto_payload: dict[str, Any] = {
                         "message": str(message),

custom_components/myhome/gateway_resync.py

@@ -121,7 +121,7 @@ def schedule_resync(self, message: Any) -> None:
             return
 
         dim = getattr(message, "dimension", None)
-        is_dimension = dim is not None and type(dim).__name__ != "MagicMock"
+        is_dimension = isinstance(dim, int) and not isinstance(dim, bool)
         if getattr(message, "is_on", None) is None or is_dimension:
             return
 

custom_components/myhome/light_group.py

@@ -177,23 +177,25 @@ def handle_event(self, msg: Any) -> None:
         # Also parse dimension frames (status replies and the gateway's echo of a
         # group dimension write, *#1*#G*#D*...##, which OWNd parses as a command).
         dim = getattr(msg, "dimension", None)
-        vals = getattr(msg, "_dimension_value", [])
-        if dim == 1 and vals:
-            # Dimension 1 carries ``level + 100`` on the bus (``150`` is 50 %); anything
-            # outside 100..200 is not a level.
-            raw = int(vals[0])
-            if 100 <= raw <= 200:
-                pct = raw - 100
+        if dim == 1:
+            pct = getattr(msg, "brightness", None)
+            if isinstance(pct, int) and not isinstance(pct, bool) and 0 <= pct <= 100:
                 self._attr_brightness = percent_to_eight_bits(pct)
                 if pct > 0:
                     self._last_brightness_pct = pct
-        elif dim == 14 and vals and int(vals[0]) > 1:
-            self._attr_color_temp_kelvin = int(1000000 / int(vals[0]))
-        elif dim == 12 and len(vals) >= 3:
-            h = int(vals[0])
-            s = int(vals[1])
-            if h <= 360:  # ``*12*511*127*255`` is the "not supported" sentinel
-                self._attr_hs_color = (h, s)
+        elif dim == 14:
+            kelvin = getattr(msg, "color_temp_kelvin", None)
+            if isinstance(kelvin, int) and not isinstance(kelvin, bool):
+                self._attr_color_temp_kelvin = kelvin
+        elif dim == 12:
+            hs_val = getattr(msg, "hs_color", getattr(msg, "hs", None))
+            if (
+                isinstance(hs_val, tuple)
+                and len(hs_val) == 2
+                and 0 <= hs_val[0] <= 360
+                and 0 <= hs_val[1] <= 100
+            ):
+                self._attr_hs_color = hs_val
 
         if self._on_icon and self._off_icon:
             self._attr_icon = self._on_icon if self._attr_is_on else self._off_icon

custom_components/myhome/media_player_power.py

@@ -64,8 +64,11 @@ def is_volume_rocker_up(message: Any) -> bool:
     """
     if getattr(message, "is_off", False) is True:
         return False
-    what = getattr(message, "what", getattr(message, "_what", None))
-    if not isinstance(what, (int, str)):
+    delta = getattr(message, "volume_step_delta", None)
+    if isinstance(delta, int) and not isinstance(delta, bool):
+        return delta > 0
+    what = getattr(message, "what", None)
+    if not isinstance(what, (int, str)) or isinstance(what, bool):
         return False
     try:
         val = int(what)
@@ -82,8 +85,11 @@ def is_volume_rocker_down(message: Any) -> bool:
     """
     if getattr(message, "is_off", False) is True:
         return False
-    what = getattr(message, "what", getattr(message, "_what", None))
-    if not isinstance(what, (int, str)):
+    delta = getattr(message, "volume_step_delta", None)
+    if isinstance(delta, int) and not isinstance(delta, bool):
+        return delta < 0
+    what = getattr(message, "what", None)
+    if not isinstance(what, (int, str)) or isinstance(what, bool):
         return False
     try:
         val = int(what)
@@ -102,7 +108,7 @@ def is_dimension_1_volume_report(message: Any) -> bool:
     hardware audio ducking and volume restoration frames for all zones in the house.
     These reports MUST NEVER change the amplifier power state (#669, #682).
     """
-    dimension = getattr(message, "dimension", getattr(message, "_dimension", None))
+    dimension = getattr(message, "dimension", None)
     if isinstance(dimension, (int, str)):
         try:
             return int(dimension) == 1