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 dataclass, fields
  9from typing import ClassVar, Self
 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
 17V1ResponseData = dict | list | int | str
 18
 19
 20@dataclass
 21class V1TraitMixin(ABC):
 22    """Base model that supports v1 traits.
 23
 24    This class provides functioanlity for parsing responses from V1 devices
 25    into dataclass instances. It also provides a reference to the V1RpcChannel
 26    used to communicate with the device to execute commands.
 27
 28    Each trait subclass must define a class variable `command` that specifies
 29    the RoborockCommand used to fetch the trait data from the device. The
 30    `refresh()` method can be called to update the contents of the trait data
 31    from the device.
 32
 33    A trait can also support additional commands for updating state associated
 34    with the trait. It is expected that a trait will update its own internal
 35    state either reflecting the change optimistically or by refreshing the
 36    trait state from the device. In cases where one trait caches data that is
 37    also represented in another trait, it is the responsibility of the caller
 38    to ensure that both traits are refreshed as needed to keep them in sync.
 39
 40    The traits typically subclass RoborockBase to provide serialization
 41    and deserialization functionality, but this is not strictly required.
 42    """
 43
 44    command: ClassVar[RoborockCommand]
 45
 46    @classmethod
 47    def _parse_type_response(cls, response: V1ResponseData) -> RoborockBase:
 48        """Parse the response from the device into a a RoborockBase.
 49
 50        Subclasses should override this method to implement custom parsing
 51        logic as needed.
 52        """
 53        if not issubclass(cls, RoborockBase):
 54            raise NotImplementedError(f"Trait {cls} does not implement RoborockBase")
 55        # Subclasses can override to implement custom parsing logic
 56        if isinstance(response, list):
 57            response = response[0]
 58        if not isinstance(response, dict):
 59            raise ValueError(f"Unexpected {cls} response format: {response!r}")
 60        return cls.from_dict(response)
 61
 62    def _parse_response(self, response: V1ResponseData) -> RoborockBase:
 63        """Parse the response from the device into a a RoborockBase.
 64
 65        This is used by subclasses that want to override the class
 66        behavior with instance-specific data.
 67        """
 68        return self._parse_type_response(response)
 69
 70    def __post_init__(self) -> None:
 71        """Post-initialization to set up the RPC channel.
 72
 73        This is called automatically after the dataclass is initialized by the
 74        device setup code.
 75        """
 76        self._rpc_channel = None
 77
 78    @property
 79    def rpc_channel(self) -> V1RpcChannel:
 80        """Helper for executing commands, used internally by the trait"""
 81        if not self._rpc_channel:
 82            raise ValueError("Device trait in invalid state")
 83        return self._rpc_channel
 84
 85    async def refresh(self) -> None:
 86        """Refresh the contents of this trait."""
 87        response = await self.rpc_channel.send_command(self.command)
 88        new_data = self._parse_response(response)
 89        if not isinstance(new_data, RoborockBase):
 90            raise ValueError(f"Internal error, unexpected response type: {new_data!r}")
 91        _LOGGER.debug("Refreshed %s: %s", self.__class__.__name__, new_data)
 92        self._update_trait_values(new_data)
 93
 94    def _update_trait_values(self, new_data: RoborockBase) -> None:
 95        """Update the values of this trait from another instance."""
 96        for field in fields(new_data):
 97            new_value = getattr(new_data, field.name, None)
 98            setattr(self, field.name, new_value)
 99
100
101def _get_value_field(clazz: type[V1TraitMixin]) -> str:
102    """Get the name of the field marked as the main value of the RoborockValueBase."""
103    value_fields = [field.name for field in fields(clazz) if field.metadata.get("roborock_value", False)]
104    if len(value_fields) != 1:
105        raise ValueError(
106            f"RoborockValueBase subclass {clazz} must have exactly one field marked as roborock_value, "
107            f" but found: {value_fields}"
108        )
109    return value_fields[0]
110
111
112@dataclass(init=False, kw_only=True)
113class RoborockValueBase(V1TraitMixin, RoborockBase):
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    @classmethod
123    def _parse_response(cls, response: V1ResponseData) -> Self:
124        """Parse the response from the device into a RoborockValueBase."""
125        if isinstance(response, list):
126            response = response[0]
127        if not isinstance(response, int):
128            raise ValueError(f"Unexpected response format: {response!r}")
129        value_field = _get_value_field(cls)
130        return cls(**{value_field: response})
131
132
133class RoborockSwitchBase(ABC):
134    """Base class for traits that represent a boolean switch."""
135
136    @property
137    @abstractmethod
138    def is_on(self) -> bool:
139        """Return whether the switch is on."""
140
141    @abstractmethod
142    async def enable(self) -> None:
143        """Enable the switch."""
144
145    @abstractmethod
146    async def disable(self) -> None:
147        """Disable the switch."""
148
149
150def mqtt_rpc_channel(cls):
151    """Decorator to mark a function as cloud only.
152
153    Normally a trait uses an adaptive rpc channel that can use either local
154    or cloud communication depending on what is available. This will force
155    the trait to always use the cloud rpc channel.
156    """
157
158    def wrapper(*args, **kwargs):
159        return cls(*args, **kwargs)
160
161    cls.mqtt_rpc_channel = True  # type: ignore[attr-defined]
162    return wrapper
163
164
165def map_rpc_channel(cls):
166    """Decorator to mark a function as cloud only using the map rpc format."""
167
168    def wrapper(*args, **kwargs):
169        return cls(*args, **kwargs)
170
171    cls.map_rpc_channel = True  # type: ignore[attr-defined]
172    return wrapper