roborock.devices.traits.v1.common

  1"""Module for Roborock V1 devices common trait commands.
  2
  3This is an internal library and should not be used directly by consumers.
  4"""
  5
  6import logging
  7from abc import ABC, abstractmethod
  8from dataclasses import fields
  9from typing import ClassVar
 10
 11from roborock.data import RoborockBase
 12from roborock.protocols.v1_protocol import V1RpcChannel
 13from roborock.roborock_typing import RoborockCommand
 14
 15_LOGGER = logging.getLogger(__name__)
 16
 17
 18V1ResponseData = dict | list | int | str
 19
 20
 21class V1TraitDataConverter(ABC):
 22    """Converts responses to RoborockBase objects.
 23
 24    This is an internal class and should not be used directly by consumers.
 25    """
 26
 27    @abstractmethod
 28    def convert(self, response: V1ResponseData) -> RoborockBase:
 29        """Convert the values to a dict that can be parsed as a RoborockBase."""
 30
 31    def __repr__(self) -> str:
 32        return self.__class__.__name__
 33
 34
 35class V1TraitMixin(ABC):
 36    """Base model that supports v1 traits.
 37
 38    This class provides functioanlity for parsing responses from V1 devices
 39    into dataclass instances. It also provides a reference to the V1RpcChannel
 40    used to communicate with the device to execute commands.
 41
 42    Each trait subclass must define a class variable `command` that specifies
 43    the RoborockCommand used to fetch the trait data from the device. The
 44    `refresh()` method can be called to update the contents of the trait data
 45    from the device.
 46
 47    A trait can also support additional commands for updating state associated
 48    with the trait. It is expected that a trait will update its own internal
 49    state either reflecting the change optimistically or by refreshing the
 50    trait state from the device. In cases where one trait caches data that is
 51    also represented in another trait, it is the responsibility of the caller
 52    to ensure that both traits are refreshed as needed to keep them in sync.
 53
 54    The traits typically subclass RoborockBase to provide serialization
 55    and deserialization functionality, but this is not strictly required.
 56    """
 57
 58    command: ClassVar[RoborockCommand]
 59    """The RoborockCommand used to fetch the trait data from the device (internal only)."""
 60
 61    converter: V1TraitDataConverter
 62    """The converter used to parse the response from the device (internal only)."""
 63
 64    def __init__(self) -> None:
 65        """Initialize the V1TraitMixin."""
 66        self._rpc_channel = None
 67
 68    @property
 69    def rpc_channel(self) -> V1RpcChannel:
 70        """Helper for executing commands, used internally by the trait"""
 71        if not self._rpc_channel:
 72            raise ValueError("Device trait in invalid state")
 73        return self._rpc_channel
 74
 75    async def refresh(self) -> None:
 76        """Refresh the contents of this trait."""
 77        response = await self.rpc_channel.send_command(self.command)
 78        new_data = self.converter.convert(response)
 79        merge_trait_values(self, new_data)  # type: ignore[arg-type]
 80
 81
 82def merge_trait_values(target: RoborockBase, new_object: RoborockBase) -> bool:
 83    """Update the target object with set fields in new_object."""
 84    updated = False
 85    for field in fields(new_object):
 86        old_value = getattr(target, field.name, None)
 87        new_value = getattr(new_object, field.name, None)
 88        if new_value != old_value:
 89            setattr(target, field.name, new_value)
 90            updated = True
 91    return updated
 92
 93
 94class DefaultConverter(V1TraitDataConverter):
 95    """Converts responses to RoborockBase objects."""
 96
 97    def __init__(self, dataclass_type: type[RoborockBase]) -> None:
 98        """Initialize the converter."""
 99        self._dataclass_type = dataclass_type
100
101    def convert(self, response: V1ResponseData) -> RoborockBase:
102        """Convert the values to a dict that can be parsed as a RoborockBase.
103
104        Subclasses can override to implement custom parsing logic
105        """
106        if isinstance(response, list):
107            response = response[0]
108        if not isinstance(response, dict):
109            raise ValueError(f"Unexpected {self._dataclass_type.__name__} response format: {response!r}")
110        return self._dataclass_type.from_dict(response)
111
112
113class SingleValueConverter(DefaultConverter):
114    """Base class for traits that represent a single value.
115
116    This class is intended to be subclassed by traits that represent a single
117    value, such as volume or brightness. The subclass should define a single
118    field with the metadata `roborock_value=True` to indicate which field
119    represents the main value of the trait.
120    """
121
122    def __init__(self, dataclass_type: type[RoborockBase], value_field: str) -> None:
123        """Initialize the converter."""
124        super().__init__(dataclass_type)
125        self._value_field = value_field
126
127    def convert(self, response: V1ResponseData) -> RoborockBase:
128        """Parse the response from the device into a RoborockValueBase."""
129        if isinstance(response, list):
130            response = response[0]
131        if not isinstance(response, int):
132            raise ValueError(f"Unexpected response format: {response!r}")
133        return super().convert({self._value_field: response})
134
135
136class RoborockSwitchBase(ABC):
137    """Base class for traits that represent a boolean switch."""
138
139    @property
140    @abstractmethod
141    def is_on(self) -> bool:
142        """Return whether the switch is on."""
143
144    @abstractmethod
145    async def enable(self) -> None:
146        """Enable the switch."""
147
148    @abstractmethod
149    async def disable(self) -> None:
150        """Disable the switch."""
151
152
153def mqtt_rpc_channel(cls):
154    """Decorator to mark a function as cloud only.
155
156    Normally a trait uses an adaptive rpc channel that can use either local
157    or cloud communication depending on what is available. This will force
158    the trait to always use the cloud rpc channel.
159    """
160
161    def wrapper(*args, **kwargs):
162        return cls(*args, **kwargs)
163
164    cls.mqtt_rpc_channel = True  # type: ignore[attr-defined]
165    return wrapper
166
167
168def map_rpc_channel(cls):
169    """Decorator to mark a function as cloud only using the map rpc format."""
170
171    def wrapper(*args, **kwargs):
172        return cls(*args, **kwargs)
173
174    cls.map_rpc_channel = True  # type: ignore[attr-defined]
175    return wrapper