feat(lighting): auto-detect pre-09W29 F414 coarse-only dimmers via speed 240 (0xF0) and emit WHAT 2..10 presets

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 All notable changes to `OWNd` are documented in this file.
 
+## [Unreleased]
+
+### Added
+- **WHO 1 (`OWNLightingCommand`, `OWNLightingEvent`)**:
+  - Automatic detection of legacy pre-`09W29` BTicino `F414` coarse-only dimmers via `DIMENSION 1` / `DIMENSION 4` transition speed `240` (`0xF0`, `DIMMER_SPEED_COARSE_ONLY`) with multi-gateway plant isolation (`gateway=` scoping on `OWNMessage.parse`, `OWNEvent.parse`, `OWNEventSession.get_next`, `is_coarse_dimmer_where`, `register_coarse_dimmer_where`, `clear_coarse_dimmer_wheres`, and `OWNLightingCommand.set_brightness`) (`#134`, `MyHOME#767`).
+  - Canonical `libqtdevices` (`TS10_1_0_23`) coarse `WHAT 2..10` $\leftrightarrow$ fine `1..100%` conversion helpers (`COARSE_PRESET_TO_PERCENT`, `brightness_preset_to_percent`, `percent_to_brightness_preset`), `OWNLightingEvent.is_coarse_dimmer`, `OWNLightingEvent.brightness_preset_percent`, and `OWNLightingCommand.set_brightness(..., coarse=..., gateway=...)` (`#134`).
+
 ## [2.0.0b11] - 2026-10-20
 
 ### Added

OWNd/connection/command_session.py

@@ -156,7 +156,9 @@ async def _read_command_response(
             for _ in range(COMMAND_RESPONSE_MAX_FRAMES):
                 raw_response = await self._read_frame(COMMAND_TIMEOUT)
                 try:
-                    resulting_message = OWNMessage.parse(raw_response)
+                    resulting_message = OWNMessage.parse(
+                        raw_response, gateway=self._gateway
+                    )
                 except (ValueError, IndexError, TypeError) as error:
                     self._logger.warning(
                         "%s Malformed command response %r: %s",
@@ -228,7 +230,7 @@ async def _locked_send(
         msg_obj = (
             message
             if isinstance(message, OWNMessage)
-            else OWNMessage.parse(str(message))
+            else OWNMessage.parse(str(message), gateway=self._gateway)
         )
         is_valid_frame = (
             msg_obj is not None

OWNd/connection/event_session.py

@@ -262,7 +262,7 @@ async def get_next(
         # down (and lose bus events) over a frame we could not make sense of.
         try:
             _decoded_data = data.decode()
-            _message = OWNMessage.parse(_decoded_data)
+            _message = OWNMessage.parse(_decoded_data, gateway=self._gateway)
             return _message if _message else _decoded_data
         except (UnicodeDecodeError, ValueError, IndexError, TypeError) as error:
             _decoded_data = data.decode(errors="replace")

OWNd/message/__init__.py

@@ -176,6 +176,8 @@
     OWNHeatingEvent,
 )
 from .lighting import (
+    COARSE_PRESET_TO_PERCENT,
+    DIMMER_SPEED_COARSE_ONLY,
     LIGHTING_STATE_ANOMALY,
     MESSAGE_TYPE_ACTION,
     MESSAGE_TYPE_ILLUMINANCE,
@@ -184,8 +186,13 @@
     MESSAGE_TYPE_PIR_SENSITIVITY,
     PIR_SENSITIVITY_MAPPING,
     area_of_where,
+    brightness_preset_to_percent,
+    clear_coarse_dimmer_wheres,
     is_apl_address,
+    is_coarse_dimmer_where,
     normalize_where,
+    percent_to_brightness_preset,
+    register_coarse_dimmer_where,
     OWNLightingCommand,
     OWNLightingEvent,
 )
@@ -231,13 +238,20 @@
     "require_non_empty_where",
     "_ensure_all_subsystems_registered",
     # Lighting (WHO 1)
+    "COARSE_PRESET_TO_PERCENT",
+    "DIMMER_SPEED_COARSE_ONLY",
     "LIGHTING_STATE_ANOMALY",
     "MESSAGE_TYPE_ACTION",
     "MESSAGE_TYPE_MOTION",
     "MESSAGE_TYPE_PIR_SENSITIVITY",
     "MESSAGE_TYPE_ILLUMINANCE",
     "MESSAGE_TYPE_MOTION_TIMEOUT",
     "PIR_SENSITIVITY_MAPPING",
+    "brightness_preset_to_percent",
+    "percent_to_brightness_preset",
+    "is_coarse_dimmer_where",
+    "register_coarse_dimmer_where",
+    "clear_coarse_dimmer_wheres",
     "is_apl_address",
     "area_of_where",
     "normalize_where",

OWNd/message/base.py

@@ -3,13 +3,22 @@
 from __future__ import annotations
 
 from collections.abc import Callable
+import contextvars
 import importlib
 import re
 import sys
 from typing import Any
 
 _EVENT_DISPATCH: dict[int, Callable[[str], OWNEvent]] = {}
 _COMMAND_DISPATCH: dict[int, Callable[[str], OWNCommand]] = {}
+_PARSE_GATEWAY_CTX: contextvars.ContextVar[Any] = contextvars.ContextVar(
+    "ownd_parse_gateway", default=None
+)
+
+
+def get_parse_gateway() -> Any:
+    """Return the active gateway context during ``OWNMessage.parse`` / ``OWNEvent.parse``."""
+    return _PARSE_GATEWAY_CTX.get()
 
 _WHO_SUBMODULES: tuple[str, ...] = (
     "actuator_lock",
@@ -120,6 +129,7 @@ def __init__(self, data: str) -> None:
         self._dimension: int | None = None
         self._dimension_param: list[str] = []
         self._dimension_value: list[str] = []
+        self._gateway: Any = get_parse_gateway()
 
         if match := self._STATUS.match(self._raw):
             self._is_valid_message = True
@@ -203,7 +213,7 @@ def __init__(self, data: str) -> None:
             self._dimension_value = match.group("dimension_value").split("*")[1:]
 
     @classmethod
-    def parse(cls, data: str) -> OWNMessage | None:
+    def parse(cls, data: str, *, gateway: Any = None) -> OWNMessage | None:
         if (
             cls._ACK.match(data)
             or cls._NACK.match(data)
@@ -219,13 +229,13 @@ def parse(cls, data: str) -> OWNMessage | None:
             or cls._DOOR_ENTRY_SHORT.match(data)
             or cls._DIMENSION_REQUEST_REPLY.match(data)
         ):
-            return OWNEvent.parse(data)
+            return OWNEvent.parse(data, gateway=gateway)
         if (
             cls._STATUS_REQUEST.match(data)
             or cls._DIMENSION_REQUEST.match(data)
             or cls._DIMENSION_WRITING.match(data)
         ):
-            return OWNCommand.parse(data)
+            return OWNCommand.parse(data, gateway=gateway)
         return None
 
     @property
@@ -438,16 +448,20 @@ class OWNEvent(OWNMessage):
     """
 
     @classmethod
-    def parse(cls, data: str) -> OWNEvent | None:
+    def parse(cls, data: str, *, gateway: Any = None) -> OWNEvent | None:
         _match = re.match(r"^\*#?(?P<who>\d+)(?:\*.+)?##$", data)
 
         if _match:
             _who = int(_match.group("who"))
             _ensure_all_subsystems_registered()
             parser = _EVENT_DISPATCH.get(_who)
-            if parser is not None:
-                return parser(data)
-            return cls(data)
+            token = _PARSE_GATEWAY_CTX.set(gateway)
+            try:
+                if parser is not None:
+                    return parser(data)
+                return cls(data)
+            finally:
+                _PARSE_GATEWAY_CTX.reset(token)
 
         return None
 
@@ -460,23 +474,27 @@ class OWNCommand(OWNMessage):
     """
 
     @classmethod
-    def parse(cls, data: str) -> OWNCommand | None:
+    def parse(cls, data: str, *, gateway: Any = None) -> OWNCommand | None:
         _match = re.match(r"^\*#?(?P<who>\d+)(?:\*.+)?##$", data)
 
         if _match:
             _who = int(_match.group("who"))
             _ensure_all_subsystems_registered()
             parser = _COMMAND_DISPATCH.get(_who)
-            if parser is not None:
-                return parser(data)
-            if _who in (0, 24) or _who > 1000:
-                return cls(data)
-            if _who in (7, 9):
-                return (
-                    OWNStatusRequest(data)
-                    if cls._STATUS_REQUEST.match(data)
-                    else cls(data)
-                )
+            token = _PARSE_GATEWAY_CTX.set(gateway)
+            try:
+                if parser is not None:
+                    return parser(data)
+                if _who in (0, 24) or _who > 1000:
+                    return cls(data)
+                if _who in (7, 9):
+                    return (
+                        OWNStatusRequest(data)
+                        if cls._STATUS_REQUEST.match(data)
+                        else cls(data)
+                    )
+            finally:
+                _PARSE_GATEWAY_CTX.reset(token)
 
         return None