roborock.devices.traits.v1.valley_electricity_timer

 1from roborock.data import ValleyElectricityTimer
 2from roborock.devices.traits.v1 import common
 3from roborock.roborock_typing import RoborockCommand
 4
 5_ENABLED_PARAM = "enabled"
 6
 7
 8class ValleyElectricityTimerTrait(ValleyElectricityTimer, common.V1TraitMixin, common.RoborockSwitchBase):
 9    """Trait for managing Valley Electricity Timer settings on Roborock devices."""
10
11    command = RoborockCommand.GET_VALLEY_ELECTRICITY_TIMER
12    requires_feature = "is_supported_valley_electricity"
13
14    @property
15    def is_on(self) -> bool:
16        """Return whether the Valley Electricity Timer is enabled."""
17        return self.enabled == 1
18
19    async def set_timer(self, timer: ValleyElectricityTimer) -> None:
20        """Set the Valley Electricity Timer settings of the device."""
21        await self.rpc_channel.send_command(RoborockCommand.SET_VALLEY_ELECTRICITY_TIMER, params=timer.as_list())
22        await self.refresh()
23
24    async def clear_timer(self) -> None:
25        """Clear the Valley Electricity Timer settings of the device."""
26        await self.rpc_channel.send_command(RoborockCommand.CLOSE_VALLEY_ELECTRICITY_TIMER)
27        await self.refresh()
28
29    async def enable(self) -> None:
30        """Enable the Valley Electricity Timer settings of the device."""
31        await self.rpc_channel.send_command(
32            RoborockCommand.SET_VALLEY_ELECTRICITY_TIMER,
33            params=self.as_list(),
34        )
35        # Optimistic update to avoid an extra refresh
36        self.enabled = 1
37
38    async def disable(self) -> None:
39        """Disable the Valley Electricity Timer settings of the device."""
40        await self.rpc_channel.send_command(
41            RoborockCommand.CLOSE_VALLEY_ELECTRICITY_TIMER,
42        )
43        # Optimistic update to avoid an extra refresh
44        self.enabled = 0