fix(typing): check mypy --strict against the pinned OWNd 2.0.0b11 types

.github/workflows/strict-typing.yml

@@ -30,11 +30,14 @@ jobs:
         with:
           python-version: "3.14"
 
-      - name: Install Home Assistant (pinned by pytest-homeassistant-custom-component) and mypy
+      - name: Install Home Assistant (pinned by pytest-homeassistant-custom-component), pinned OWNd, and mypy
+        # OWNd is the exact `OWNd==` pin from manifest.json, so mypy checks the
+        # integration against the types users install.
         run: |
           python -m pip install --upgrade pip
-          pip install pytest-homeassistant-custom-component mypy
-          python -c "import homeassistant.const as c; print('Home Assistant', c.__version__)"
+          OWND_PIN=$(python -c "import json; print(next(r for r in json.load(open('custom_components/myhome/manifest.json'))['requirements'] if r.startswith('OWNd==')))")
+          pip install pytest-homeassistant-custom-component mypy "$OWND_PIN"
+          python -c "import homeassistant.const as c, OWNd; print('Home Assistant', c.__version__, '| OWNd', OWNd.__version__)"
 
       - name: mypy --strict ratchet
         # `shell: bash` (not the default) so the step runs with `pipefail`: without it

custom_components/myhome/cover.py

@@ -2247,8 +2247,8 @@ def handle_event(self, message: OWNMessage) -> None:
                 # A repeated report of the same level says nothing about the slats:
                 # they may have been tilted open on a lowered blind.
                 self._attr_current_cover_tilt_position = position
-            if getattr(message, "is_closed", None) is not None:
-                self._attr_is_closed = message.is_closed
+            if (is_closed := getattr(message, "is_closed", None)) is not None:
+                self._attr_is_closed = bool(is_closed)
             else:
                 self._attr_is_closed = (self._attr_current_cover_position == 0)
         elif is_opening:
@@ -2340,8 +2340,8 @@ def handle_event(self, message: OWNMessage) -> None:
                     )
             self._attr_is_opening = False
             self._attr_is_closing = False
-            if getattr(message, "is_closed", None) is not None:
-                self._attr_is_closed = message.is_closed
+            if (is_closed := getattr(message, "is_closed", None)) is not None:
+                self._attr_is_closed = bool(is_closed)
             elif self._attr_current_cover_position is not None:
                 self._attr_is_closed = (self._attr_current_cover_position == 0)
 

custom_components/myhome/cover_scope.py

@@ -10,7 +10,7 @@
 from homeassistant.core import CALLBACK_TYPE, Event, EventStateChangedData, callback
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers.event import async_call_later, async_track_state_change_event
-from OWNd.message import OWNAutomationEvent
+from OWNd.message import OWNMessage
 
 from .const import BUS_ROUTING, DOMAIN, PACED_FANOUT_DELAY, area_of_where
 
@@ -316,7 +316,7 @@ def current_cover_tilt_position(self) -> int | None:
         return round(sum(tilts) / len(tilts)) if tilts else None
 
     @callback
-    def handle_event(self, message: OWNAutomationEvent) -> None:
+    def handle_event(self, message: OWNMessage) -> None:
         super().handle_event(message)
         if not self._is_tilting:
             self._watch_run()

custom_components/myhome/gateway_events.py

@@ -16,6 +16,7 @@
     OWNAuxEvent,
     OWNCENEvent,
     OWNCENPlusEvent,
+    OWNDoorEntryEvent,
     OWNDryContactEvent,
     OWNEnergyCommand,
     OWNEnergyEvent,
@@ -25,22 +26,10 @@
     OWNHeatingEvent,
     OWNLightingEvent,
     OWNMessage,
+    OWNScenarioPlusEvent,
     OWNSceneEvent,
 )
 
-try:  # OWNd > 2.0.0b10 (OpenWebNet-HA/OWNd#91)
-    from OWNd.message import OWNScenarioPlusEvent
-except ImportError:  # pragma: no cover - pinned OWNd 2.0.0b10 fallback
-    class OWNScenarioPlusEvent(OWNMessage):  # type: ignore[misc,no-redef]
-        pass
-
-try:
-    from OWNd.message import OWNDoorEntryEvent
-except ImportError:  # pragma: no cover - fallback on released OWNd 2.0.0b10
-    class OWNDoorEntryEvent:  # type: ignore[no-redef]
-        """Fallback stub when running on released OWNd lacking WHO 6 door entry support."""
-        pass
-
 from .const import (
     CONF_LONG_PRESS,
     CONF_LONG_PRESS_REPEAT,
@@ -446,7 +435,7 @@ async def process_message(self, message: Any) -> None:
             except (ValueError, TypeError):  # pragma: no cover - defensive
                 obj_val = raw_obj
             try:
-                pb_val = int(message.push_button)
+                pb_val: int | None = int(message.push_button)
             except (ValueError, TypeError):  # pragma: no cover - defensive
                 pb_val = message.push_button
 
@@ -705,11 +694,12 @@ async def process_message(self, message: Any) -> None:
                     target_mac = None
 
             if target_mac is not None and self._is_active_for_who(17):
-                scene_val: int | str = message.scenario
-                try:
-                    scene_val = int(message.scenario)
-                except (ValueError, TypeError):
-                    scene_val = message.scenario
+                scene_val: int | str | None = message.scenario
+                if message.scenario is not None:
+                    try:
+                        scene_val = int(message.scenario)
+                    except ValueError:
+                        pass
 
                 scene_payload: dict[str, Any] = {
                     "scenario": scene_val,

custom_components/myhome/gateway_resync.py

@@ -101,7 +101,7 @@ def _normalize_group_key(group_where: str | int, interface: str | None = None) -
         return f"#{grp_norm}"
 
     @staticmethod
-    def _normalize_ptp_where(raw_where: str | int, interface: str | None = None) -> str:
+    def _normalize_ptp_where(raw_where: str | int | None, interface: str | None = None) -> str:
         """Normalize a point-to-point WHERE while preserving any ``#4#<interface>`` suffix."""
         where_str = str(raw_where).strip()
         base_part, sep, iface_part = where_str.partition(BUS_ROUTING)

custom_components/myhome/identity.py

@@ -33,6 +33,7 @@
 
 from homeassistant.const import CONF_NAME
 from homeassistant.core import HomeAssistant
+from OWNd.connection import OWNGateway
 
 from .const import (
     CONF_FIRMWARE,
@@ -292,7 +293,7 @@ class GatewayIdentityMixin:
 
     hass: HomeAssistant
     config_entry: Any
-    gateway: Any
+    gateway: OWNGateway
     device_registry_id: str | None
     _who13: dict[str, Any]
     _who1013: dict[str, Any]

custom_components/myhome/services.py

@@ -299,40 +299,25 @@ async def handle_send_scenario_plus_command(call: ServiceCall) -> None:
                 translation_placeholders={"action": action},
             )
 
-        try:
-            from OWNd.message import OWNScenarioPlusCommand
-        except ImportError:  # pragma: no cover - fallback when OWNd unreleased
-            OWNScenarioPlusCommand = None
+        from OWNd.message import OWNScenarioPlusCommand
 
         cmd: Any = None
-        if OWNScenarioPlusCommand is not None:
-            try:
-                if action == "on":
-                    cmd = OWNScenarioPlusCommand.turn_on(where)
-                elif action == "off":
-                    cmd = OWNScenarioPlusCommand.turn_off(where)
-                elif action == "increase":
-                    cmd = OWNScenarioPlusCommand.increase(where)
-                elif action == "decrease":
-                    cmd = OWNScenarioPlusCommand.decrease(where)
-                elif action == "stop":
-                    cmd = OWNScenarioPlusCommand.stop(where)
-            except ValueError:
-                cmd = None
-        else:
-            action_frames = {
-                "on": f"*25*11#0*{where}##",
-                "off": f"*25*12*{where}##",
-                "increase": f"*25*13#0#5*{where}##",
-                "decrease": f"*25*14#0#5*{where}##",
-                "stop": f"*25*15*{where}##",
-            }
-            from OWNd.message import OWNCommand
-
-            cmd = OWNCommand.parse(action_frames[action])
-
-        # Fallback OWNCommand or legacy OWNd command objects may not define
-        # `is_valid`; default to True only when the attribute is absent.
+        try:
+            if action == "on":
+                cmd = OWNScenarioPlusCommand.turn_on(where)
+            elif action == "off":
+                cmd = OWNScenarioPlusCommand.turn_off(where)
+            elif action == "increase":
+                cmd = OWNScenarioPlusCommand.increase(where)
+            elif action == "decrease":
+                cmd = OWNScenarioPlusCommand.decrease(where)
+            elif action == "stop":
+                cmd = OWNScenarioPlusCommand.stop(where)
+        except ValueError:
+            cmd = None
+
+        # Command objects mocked in tests may not define `is_valid`; default to
+        # True only when the attribute is absent.
         if cmd is None or not getattr(cmd, "is_valid", True):
             raise ServiceValidationError(
                 f"Could not build scenario plus command for where '{where}' and action '{action}'",

custom_components/myhome/where_grammar.py

@@ -30,62 +30,11 @@
     OWNLightingEvent,
     OWNMessage,
     OWNSoundEvent,
+    area_of_where,
+    is_apl_address,
+    normalize_where,
 )
 
-try:
-    from OWNd.message import area_of_where, is_apl_address, normalize_where
-except ImportError:  # pragma: no cover - PyPI OWNd < 2.0.0b11 compatibility
-    def is_apl_address(where: str | int | None) -> bool:
-        """Return True if where is a valid WHO=1 Point-to-Point (APL) address."""
-        if where is None:
-            return False
-        base = str(where).strip().split("#", 1)[0]
-        if base.startswith("#") or not base.isdigit():
-            return False
-        if len(base) == 2:
-            a, pl = int(base[0]), int(base[1])
-            return (1 <= pl <= 9) if a == 0 else (1 <= a <= 9 and 1 <= pl <= 9)
-        if len(base) == 4:
-            a, pl = int(base[:2]), int(base[2:])
-            if a in (0, 10):
-                return 1 <= pl <= 15
-            if 1 <= a <= 9:
-                return 10 <= 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
-        base = str(where).strip().split("#", 1)[0]
-        if not is_apl_address(base):
-            return None
-        if len(base) == 2:
-            return "00" if base[0] == "0" else base[0]
-        a = base[:2]
-        return "00" if a == "00" else ("100" if a == "10" else 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
-
-    setattr(_ownd_msg, "is_apl_address", is_apl_address)
-    setattr(_ownd_msg, "area_of_where", area_of_where)
-    setattr(_ownd_msg, "normalize_where", normalize_where)
-
 for _const_name, _const_val in (
     ("MESSAGE_TYPE_SEASON", "hvac_season"),
     ("SEASON_HEATING", "heating"),

docs/quality_scale_plan.md

@@ -50,7 +50,7 @@ Add a parametrised test that asserts the table so it cannot regress.
 ## Phase 5 — 🏆 Platinum: `strict-typing`
 
 - ✅ Ratchet in place: `[tool.mypy]` (strict) in `pyproject.toml`, `mypy_baseline.json` with a per-module ceiling, `scripts/typing_ratchet.py` (fails on any regression, `--update` only lowers), workflow `strict-typing.yml`. 15 of 28 modules are clean (`const`, `data`, `gateway`, `myhome_device`, `services`, `device_trigger`, `decoder_pool`, transports, diagnostics, repairs, bus monitor); **504 errors remain in 13 modules** — biggest first: `config_flow` 105, `cover` 64, `validate` 57, `binary_sensor` 43, `light` 37, `sensor` 32, `websocket` 30, `__init__` 27, `button` 27, `climate` 25, `media_player` 25, `alarm_control_panel` 18, `switch` 14 (`discovery` and `router` are clean). Each PR that touches a module should leave it lower and run `--update`.
-- OWNd does **not** ship `py.typed`; the rule needs an OWNd release that is PEP 561 typed before it can be marked done.
+- ✅ OWNd ships `py.typed` (PEP 561). The `OWNd.*` `ignore_missing_imports` override is gone and `strict-typing.yml` installs the `manifest.json` `OWNd==` pin, so the ratchet checks the integration against the OWNd types users install.
 - `inject-websession` is correctly `exempt` (raw TCP, no HTTP); `async-dependency` is done.
 
 ## Maintainability (not a scale rule, keeps the rules honest)

pyproject.toml

@@ -34,7 +34,7 @@ test = [
     "check-wheel-contents>=0.6.0",
     "validate-pyproject[all]>=0.16",
     "ruff>=0.4.0",
-    "OWNd>=2.0.0b10",
+    "OWNd>=2.0.0b11",
 ]
 
 [project.urls]
@@ -90,8 +90,9 @@ show_error_codes = true
 files = ["custom_components/myhome"]
 
 [[tool.mypy.overrides]]
-# OWNd does not ship py.typed yet; Platinum needs an OWNd release that does.
-module = ["OWNd.*", "serial_asyncio", "serial_asyncio.*"]
+# OWNd ships py.typed, so it is checked like any other typed dependency: a
+# missing OWNd is an error rather than a silent `Any`.
+module = ["serial_asyncio", "serial_asyncio.*"]
 ignore_missing_imports = true
 
 [tool.ruff]

tests/test_services.py

@@ -562,9 +562,8 @@ def stop(cls, where: str) -> str:
         )
     assert exc_blank.value.translation_key == "scenario_plus_missing_where"
 
-    # Command parse failure in fallback branch
-    with patch("OWNd.message.OWNScenarioPlusCommand", None, create=True), \
-         patch("OWNd.message.OWNCommand.parse", return_value=None):
+    # Builder returns a command that fails its own validation
+    with patch("OWNd.message.OWNScenarioPlusCommand.turn_on", return_value=MagicMock(is_valid=False)):
         with pytest.raises(ServiceValidationError) as exc_parse:
             await hass.services.async_call(
                 DOMAIN,