refactor: Centralize trait utilities

roborock/devices/traits/b01/q10/status.py

@@ -5,8 +5,7 @@
 
 from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.data.b01_q10.b01_q10_containers import Q10Status
-
-from .common import DpsDataConverter, TraitUpdateListener
+from roborock.devices.traits.common import DpsDataConverter, TraitUpdateListener
 
 _LOGGER = logging.getLogger(__name__)
 

roborock/devices/traits/b01/q10/common.py → roborock/devices/traits/common.py

@@ -1,8 +1,7 @@
-"""Common utilities for Q10 traits.
+"""Common utilities for device traits.
 
-This module provides infrastructure for mapping Roborock Data Points (DPS) to
-Python dataclass fields and handling the lifecycle of data updates from the
-device.
+This module provides shared infrastructure for mapping Roborock Data Points (DPS) to
+Python dataclass fields and handling the lifecycle of data updates from the device.
 
 ### DPS Metadata Annotation
 
@@ -21,7 +20,7 @@ class MyStatus(RoborockBase):
 
 ### Update Lifecycle
 1. **Raw Data**: The device sends encoded DPS updates over MQTT.
-2. **Decoding**: The transport layer decodes these into a dictionary (e.g., `{"101": 80}`).
+2. **Decoding**: The transport layer decodes these into a dictionary (e.g., {"101": 80}).
 3. **Conversion**: `DpsDataConverter` uses `RoborockBase.convert_dict` to transform
    raw values into appropriate Python types (e.g., Enums, ints) based on the
    dataclass field types.
@@ -32,18 +31,18 @@ class MyStatus(RoborockBase):
 
 Typically, a trait will instantiate a single `DpsDataConverter` for its status class
 and call `update_from_dps` whenever new data is received from the device stream.
-
 """
 
 import dataclasses
 import logging
 from collections.abc import Callable
-from typing import Any
+from typing import Any, Generic, TypeVar
 
 from roborock.callbacks import CallbackList
-from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP
 from roborock.data.containers import RoborockBase
 
+TDps = TypeVar("TDps", bound=int)
+
 
 class TraitUpdateListener:
     """Trait update listener.
@@ -71,31 +70,31 @@ def _notify_update(self) -> None:
         self._update_callbacks(None)
 
 
-class DpsDataConverter:
+class DpsDataConverter(Generic[TDps]):
     """Utility to handle the transformation and merging of DPS data into models.
 
     This class pre-calculates the mapping between Data Point IDs and dataclass fields
     to optimize repeated updates from device streams.
     """
 
-    def __init__(self, dps_type_map: dict[B01_Q10_DP, type], dps_field_map: dict[B01_Q10_DP, str]):
+    def __init__(self, dps_type_map: dict[TDps, type], dps_field_map: dict[TDps, str]):
         """Initialize the converter for a specific RoborockBase-derived class."""
         self._dps_type_map = dps_type_map
         self._dps_field_map = dps_field_map
 
     @classmethod
     def from_dataclass(cls, dataclass_type: type[RoborockBase]):
         """Initialize the converter for a specific RoborockBase-derived class."""
-        dps_type_map: dict[B01_Q10_DP, type] = {}
-        dps_field_map: dict[B01_Q10_DP, str] = {}
+        dps_type_map: dict[TDps, type] = {}
+        dps_field_map: dict[TDps, str] = {}
         for field_obj in dataclasses.fields(dataclass_type):
             if field_obj.metadata and "dps" in field_obj.metadata:
                 dps_id = field_obj.metadata["dps"]
                 dps_type_map[dps_id] = field_obj.type
                 dps_field_map[dps_id] = field_obj.name
         return cls(dps_type_map, dps_field_map)
 
-    def update_from_dps(self, target: RoborockBase, decoded_dps: dict[B01_Q10_DP, Any]) -> bool:
+    def update_from_dps(self, target: RoborockBase, decoded_dps: dict[TDps, Any]) -> bool:
         """Convert and merge raw DPS data into the target object.
 
         Uses the pre-calculated type mapping to ensure values are converted to the