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