1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
from typing import Any
from roborock import RoborockCommand
from roborock.protocols.v1_protocol import ParamsType
class CommandTrait:
"""Trait for sending commands to Roborock devices."""
def __post_init__(self) -> None:
"""Post-initialization to set up the RPC channel.
This is called automatically after the dataclass is initialized by the
device setup code.
"""
self._rpc_channel = None
async def send(self, command: RoborockCommand | str, params: ParamsType = None) -> Any:
"""Send a command to the device.
Sending a raw command to the device using this method does not update
the internal state of any other traits. It is the responsibility of the
caller to ensure that any traits affected by the command are refreshed
as needed.
"""
if not self._rpc_channel:
raise ValueError("Device trait in invalid state")
return await self._rpc_channel.send_command(command, params=params)
|
