diff --git a/miio/gateway.py b/miio/gateway.py index b790bac35..f908bbce5 100644 --- a/miio/gateway.py +++ b/miio/gateway.py @@ -1,16 +1,21 @@ +"""Xiaomi Aqara Gateway implementation using Miio protecol.""" + import logging from datetime import datetime from enum import Enum, IntEnum from typing import Optional +import attr import click from .click_common import EnumType, command, format_output from .device import Device +from .exceptions import DeviceException from .utils import brightness_and_color_to_int, int_to_brightness, int_to_rgb _LOGGER = logging.getLogger(__name__) + color_map = { "red": (255, 0, 0), "green": (0, 255, 0), @@ -24,7 +29,14 @@ } +class GatewayException(DeviceException): + """Exception for the Xioami Gateway communication.""" + + class DeviceType(IntEnum): + """DeviceType matching using the values provided by Xiaomi.""" + + Unknown = -1 Gateway = 0 Switch = 1 Motion = 2 @@ -43,17 +55,19 @@ class DeviceType(IntEnum): AqaraMagnet = 53 AqaraRelayTwoChannels = 54 AqaraSquareButton = 62 + RemoteSwitchSingle = 134 + RemoteSwitchDouble = 135 -class AqaraRelayToggleValue(Enum): - toggle = "toggle" - on = "on" - off = "off" +@attr.s(auto_attribs=True) +class SubDeviceInfo: + """SubDevice discovery info.""" - -class AqaraRelayChannel(Enum): - first = "channel_0" - second = "channel_1" + sid: str + type_id: int + unknown: int + unknown2: int + fw_ver: int class Gateway(Device): @@ -62,7 +76,8 @@ class Gateway(Device): Use the given property getters to access specific functionalities such as `alarm` (for alarm controls) or `light` (for lights). - Commands whose functionality or parameters are unknown, feel free to implement! + Commands whose functionality or parameters are unknown, + feel free to implement! * toggle_device * toggle_plug * remove_all_bind @@ -100,10 +115,11 @@ def __init__( lazy_discover: bool = True, ) -> None: super().__init__(ip, token, start_id, debug, lazy_discover) - self._alarm = GatewayAlarm(self) - self._radio = GatewayRadio(self) - self._zigbee = GatewayZigbee(self) - self._light = GatewayLight(self) + self._alarm = GatewayAlarm(parent=self) + self._radio = GatewayRadio(parent=self) + self._zigbee = GatewayZigbee(parent=self) + self._light = GatewayLight(parent=self) + self._devices = [] @property def alarm(self) -> "GatewayAlarm": @@ -126,63 +142,81 @@ def light(self) -> "GatewayLight": """Return light control interface.""" return self._light - @command() + @property def devices(self): - """Return list of devices.""" + """Return a list of the already discovered devices.""" + return self._devices + + @command() + def discover_devices(self): + """ + Discovers SubDevices + and returns a list of the discovered devices. + """ # from https://github.com/aholstenson/miio/issues/26 - devices_raw = self.send("get_device_prop", ["lumi.0", "device_list"]) - devices = [ - SubDevice(self, *devices_raw[x : x + 5]) # noqa: E203 - for x in range(0, len(devices_raw), 5) - ] + device_type_mapping = { + DeviceType.AqaraRelayTwoChannels: AqaraRelayTwoChannels, + DeviceType.Plug: AqaraPlug, + DeviceType.SensorHT: SensorHT, + DeviceType.AqaraHT: AqaraHT, + DeviceType.AqaraMagnet: AqaraMagnet, + } + devices_raw = self.get_prop("device_list") + self._devices = [] + + for x in range(0, len(devices_raw), 5): + # Extract discovered information + dev_info = SubDeviceInfo(*devices_raw[x : x + 5]) + + # Construct DeviceType + try: + device_type = DeviceType(dev_info.type_id) + except ValueError: + _LOGGER.warning( + "Unknown subdevice type %s discovered, " + "of Xiaomi gateway with ip: %s", + dev_info, + self.ip, + ) + device_type = DeviceType(-1) + + # Obtain the correct subdevice class, ignoring the gateway itself + subdevice_cls = device_type_mapping.get(device_type) + if subdevice_cls is None and device_type != DeviceType.Gateway: + subdevice_cls = SubDevice + _LOGGER.info( + "Gateway device type '%s' " + "does not have device specific methods defined, " + "only basic default methods will be available", + device_type.name, + ) + + # Initialize and save the subdevice, ignoring the gateway itself + if device_type != DeviceType.Gateway: + self._devices.append(subdevice_cls(self, dev_info)) - return devices + return self._devices - @command(click.argument("sid"), click.argument("property")) - def get_device_prop(self, sid, property): + @command(click.argument("property")) + def get_prop(self, property): """Get the value of a property for given sid.""" - return self.send("get_device_prop", [sid, property]) + return self.send("get_device_prop", ["lumi.0", property]) - @command(click.argument("sid"), click.argument("properties", nargs=-1)) - def get_device_prop_exp(self, sid, properties): + @command(click.argument("properties", nargs=-1)) + def get_prop_exp(self, properties): """Get the value of a bunch of properties for given sid.""" - return self.send("get_device_prop_exp", [[sid] + list(properties)]) + return self.send("get_device_prop_exp", [["lumi.0"] + list(properties)]) - @command(click.argument("sid"), click.argument("property"), click.argument("value")) - def set_device_prop(self, sid, property, value): + @command(click.argument("property"), click.argument("value")) + def set_prop(self, property, value): """Set the device property.""" - return self.send("set_device_prop", {"sid": sid, property: value}) - - @command( - click.argument("sid"), - click.argument("channel", type=EnumType(AqaraRelayChannel)), - click.argument("value", type=EnumType(AqaraRelayToggleValue)), - ) - def relay_toggle(self, sid, channel, value): - """Toggle Aqara Wireless Relay 2ch""" - return self.send( - "toggle_ctrl_neutral", - [channel.value, value.value], - extra_parameters={"sid": sid}, - )[0] - - @command( - click.argument("sid"), - click.argument("channel", type=EnumType(AqaraRelayChannel)), - ) - def relay_get_state(self, sid, channel): - """Get the state of Aqara Wireless Relay 2ch for given sid""" - return self.send("get_device_prop_exp", [[sid, channel.value]])[0][0] - - @command(click.argument("sid")) - def relay_get_load_power(self, sid): - """Get the the load power of Aqara Wireless Relay 2ch for given sid""" - return self.send("get_device_prop_exp", [[sid, "load_power"]])[0][0] + return self.send("set_device_prop", {"sid": "lumi.0", property: value}) @command() def clock(self): """Alarm clock""" - # payload of clock volume ("get_clock_volume") already in get_clock response + # payload of clock volume ("get_clock_volume") + # already in get_clock response return self.send("get_clock") # Developer key @@ -202,140 +236,162 @@ def set_developer_key(self, key): @command() def timezone(self): """Get current timezone.""" - return self.send("get_device_prop", ["lumi.0", "tzone_sec"]) + return self.get_prop("tzone_sec") @command() def get_illumination(self): """Get illumination. In lux?""" - return self.send("get_illumination")[0] + return self.send("get_illumination").pop() -class GatewayAlarm(Device): - """Class representing the Xiaomi Gateway Alarm.""" +class GatewayDevice(Device): + """ + GatewayDevice class + Specifies the init method for all gateway device functionalities. + """ + + def __init__( + self, + ip: str = None, + token: str = None, + start_id: int = 0, + debug: int = 0, + lazy_discover: bool = True, + parent: Gateway = None, + ) -> None: + if parent is not None: + self._gateway = parent + else: + self._gateway = Device(ip, token, start_id, debug, lazy_discover) + _LOGGER.debug( + "Creating new device instance, only use this for cli interface" + ) - def __init__(self, parent) -> None: - self._device = parent + +class GatewayAlarm(GatewayDevice): + """Class representing the Xiaomi Gateway Alarm.""" @command(default_output=format_output("[alarm_status]")) def status(self) -> str: """Return the alarm status from the device.""" # Response: 'on', 'off', 'oning' - return self._device.send("get_arming").pop() + return self._gateway.send("get_arming").pop() @command(default_output=format_output("Turning alarm on")) def on(self): """Turn alarm on.""" - return self._device.send("set_arming", ["on"]) + return self._gateway.send("set_arming", ["on"]) @command(default_output=format_output("Turning alarm off")) def off(self): """Turn alarm off.""" - return self._device.send("set_arming", ["off"]) + return self._gateway.send("set_arming", ["off"]) @command() def arming_time(self) -> int: - """Return time in seconds the alarm stays 'oning' before transitioning to 'on'""" + """ + Return time in seconds the alarm stays 'oning' + before transitioning to 'on' + """ # Response: 5, 15, 30, 60 - return self._device.send("get_arm_wait_time").pop() + return self._gateway.send("get_arm_wait_time").pop() @command(click.argument("seconds")) def set_arming_time(self, seconds): """Set time the alarm stays at 'oning' before transitioning to 'on'""" - return self._device.send("set_arm_wait_time", [seconds]) + return self._gateway.send("set_arm_wait_time", [seconds]) @command() def triggering_time(self) -> int: """Return the time in seconds the alarm is going off when triggered""" # Response: 30, 60, etc. - return self._device.send("get_device_prop", ["lumi.0", "alarm_time_len"]).pop() + return self._gateway.get_prop("alarm_time_len").pop() @command(click.argument("seconds")) def set_triggering_time(self, seconds): """Set the time in seconds the alarm is going off when triggered""" - return self._device.send( - "set_device_prop", {"sid": "lumi.0", "alarm_time_len": seconds} - ) + return self._gateway.set_prop("alarm_time_len", seconds) @command() def triggering_light(self) -> int: - """Return the time the gateway light blinks when the alarm is triggerd""" + """ + Return the time the gateway light blinks + when the alarm is triggerd + """ # Response: 0=do not blink, 1=always blink, x>1=blink for x seconds - return self._device.send("get_device_prop", ["lumi.0", "en_alarm_light"]).pop() + return self._gateway.get_prop("en_alarm_light").pop() @command(click.argument("seconds")) def set_triggering_light(self, seconds): """Set the time the gateway light blinks when the alarm is triggerd""" # values: 0=do not blink, 1=always blink, x>1=blink for x seconds - return self._device.send( - "set_device_prop", {"sid": "lumi.0", "en_alarm_light": seconds} - ) + return self._gateway.set_prop("en_alarm_light", seconds) @command() def triggering_volume(self) -> int: """Return the volume level at which alarms go off [0-100]""" - return self._device.send("get_alarming_volume").pop() + return self._gateway.send("get_alarming_volume").pop() @command(click.argument("volume")) def set_triggering_volume(self, volume): """Set the volume level at which alarms go off [0-100]""" - return self._device.send("set_alarming_volume", [volume]) + return self._gateway.send("set_alarming_volume", [volume]) @command() - def last_status_change_time(self): - """Return the last time the alarm changed status, type datetime.datetime""" - return datetime.fromtimestamp(self._device.send("get_arming_time").pop()) + def last_status_change_time(self) -> datetime: + """ + Return the last time the alarm changed status + """ + return datetime.fromtimestamp(self._gateway.send("get_arming_time").pop()) -class GatewayZigbee(Device): +class GatewayZigbee(GatewayDevice): """Zigbee controls.""" - def __init__(self, parent) -> None: - self._device = parent - @command() def get_zigbee_version(self): """timeouts on device""" - return self._device.send("get_zigbee_device_version") + return self._gateway.send("get_zigbee_device_version") @command() def get_zigbee_channel(self): """Return currently used zigbee channel.""" - return self._device.send("get_zigbee_channel")[0] + return self._gateway.send("get_zigbee_channel")[0] @command(click.argument("channel")) def set_zigbee_channel(self, channel): """Set zigbee channel.""" - return self._device.send("set_zigbee_channel", [channel]) + return self._gateway.send("set_zigbee_channel", [channel]) @command(click.argument("timeout", type=int)) def zigbee_pair(self, timeout): """Start pairing, use 0 to disable""" - return self._device.send("start_zigbee_join", [timeout]) + return self._gateway.send("start_zigbee_join", [timeout]) def send_to_zigbee(self): """How does this differ from writing? Unknown.""" raise NotImplementedError() - return self._device.send("send_to_zigbee") + return self._gateway.send("send_to_zigbee") def read_zigbee_eep(self): """Read eeprom?""" raise NotImplementedError() - return self._device.send("read_zig_eep", [0]) # 'ok' + return self._gateway.send("read_zig_eep", [0]) # 'ok' def read_zigbee_attribute(self): """Read zigbee data?""" raise NotImplementedError() - return self._device.send("read_zigbee_attribute", [0x0000, 0x0080]) + return self._gateway.send("read_zigbee_attribute", [0x0000, 0x0080]) def write_zigbee_attribute(self): """Unknown parameters.""" raise NotImplementedError() - return self._device.send("write_zigbee_attribute") + return self._gateway.send("write_zigbee_attribute") @command() def zigbee_unpair_all(self): """Unpair all devices.""" - return self._device.send("remove_all_device") + return self._gateway.send("remove_all_device") def zigbee_unpair(self, sid): """Unpair a device.""" @@ -343,126 +399,125 @@ def zigbee_unpair(self, sid): raise NotImplementedError() -class GatewayRadio(Device): +class GatewayRadio(GatewayDevice): """Radio controls for the gateway.""" - def __init__(self, parent) -> None: - self._device = parent - @command() def get_radio_info(self): """Radio play info.""" - return self._device.send("get_prop_fm") + return self._gateway.send("get_prop_fm") @command(click.argument("volume")) def set_radio_volume(self, volume): """Set radio volume""" - return self._device.send("set_fm_volume", [volume]) + return self._gateway.send("set_fm_volume", [volume]) def play_music_new(self): """Unknown.""" - # {'from': '4', 'id': 9514, 'method': 'set_default_music', 'params': [2, '21']} - # {'from': '4', 'id': 9515, 'method': 'play_music_new', 'params': ['21', 0]} + # {'from': '4', 'id': 9514, + # 'method': 'set_default_music', 'params': [2, '21']} + # {'from': '4', 'id': 9515, + # 'method': 'play_music_new', 'params': ['21', 0]} raise NotImplementedError() def play_specify_fm(self): """play specific stream?""" raise NotImplementedError() # {"from": "4", "id": 65055, "method": "play_specify_fm", - # "params": {"id": 764, "type": 0, "url": "http://live.xmcdn.com/live/764/64.m3u8"}} - return self._device.send("play_specify_fm") + # "params": {"id": 764, "type": 0, + # "url": "http://live.xmcdn.com/live/764/64.m3u8"}} + return self._gateway.send("play_specify_fm") def play_fm(self): """radio on/off?""" raise NotImplementedError() # play_fm","params":["off"]} - return self._device.send("play_fm") + return self._gateway.send("play_fm") def volume_ctrl_fm(self): """Unknown.""" raise NotImplementedError() - return self._device.send("volume_ctrl_fm") + return self._gateway.send("volume_ctrl_fm") def get_channels(self): """Unknown.""" raise NotImplementedError() # "method": "get_channels", "params": {"start": 0}} - return self._device.send("get_channels") + return self._gateway.send("get_channels") def add_channels(self): """Unknown.""" raise NotImplementedError() - return self._device.send("add_channels") + return self._gateway.send("add_channels") def remove_channels(self): """Unknown.""" raise NotImplementedError() - return self._device.send("remove_channels") + return self._gateway.send("remove_channels") def get_default_music(self): """seems to timeout (w/o internet)""" # params [0,1,2] raise NotImplementedError() - return self._device.send("get_default_music") + return self._gateway.send("get_default_music") @command() def get_music_info(self): """Unknown.""" - info = self._device.send("get_music_info") + info = self._gateway.send("get_music_info") click.echo("info: %s" % info) - free_space = self._device.send("get_music_free_space") + free_space = self._gateway.send("get_music_free_space") click.echo("free space: %s" % free_space) @command() def get_mute(self): """mute of what?""" - return self._device.send("get_mute") + return self._gateway.send("get_mute") def download_music(self): """Unknown""" raise NotImplementedError() - return self._device.send("download_music") + return self._gateway.send("download_music") def delete_music(self): """delete music""" raise NotImplementedError() - return self._device.send("delete_music") + return self._gateway.send("delete_music") def download_user_music(self): """Unknown.""" raise NotImplementedError() - return self._device.send("download_user_music") + return self._gateway.send("download_user_music") def get_download_progress(self): """progress for music downloads or updates?""" # returns [':0'] raise NotImplementedError() - return self._device.send("get_download_progress") + return self._gateway.send("get_download_progress") @command() def set_sound_playing(self): """stop playing?""" - return self._device.send("set_sound_playing", ["off"]) + return self._gateway.send("set_sound_playing", ["off"]) def set_default_music(self): + """Unknown.""" raise NotImplementedError() # method":"set_default_music","params":[0,"2"]} -class GatewayLight(Device): +class GatewayLight(GatewayDevice): """Light controls for the gateway.""" - def __init__(self, parent) -> None: - self._device = parent - @command() def get_night_light_rgb(self): """Unknown.""" # Returns 0 when light is off?""" # looks like this is the same as get_rgb # id': 65064, 'method': 'set_night_light_rgb', 'params': [419407616]} - # {'method': 'props', 'params': {'light': 'on', 'from.light': '4,,,'}, 'id': 88457} ?! - return self.send("get_night_light_rgb") + # {'method': 'props', 'params': + # {'light': 'on', 'from.light': '4,,,'}, 'id': 88457} ?! + return self._gateway.send("get_night_light_rgb") @command(click.argument("color_name", type=str)) def set_night_light_color(self, color_name): @@ -473,11 +528,13 @@ def set_night_light_color(self, color_name): color=color_name, colors=color_map.keys() ) ) - current_brightness = int_to_brightness(self.send("get_night_light_rgb")[0]) + current_brightness = int_to_brightness( + self._gateway.send("get_night_light_rgb")[0] + ) brightness_and_color = brightness_and_color_to_int( current_brightness, color_map[color_name] ) - return self.send("set_night_light_rgb", [brightness_and_color]) + return self._gateway.send("set_night_light_rgb", [brightness_and_color]) @command(click.argument("color_name", type=str)) def set_color(self, color_name): @@ -488,33 +545,33 @@ def set_color(self, color_name): color=color_name, colors=color_map.keys() ) ) - current_brightness = int_to_brightness(self.send("get_rgb")[0]) + current_brightness = int_to_brightness(self._gateway.send("get_rgb")[0]) brightness_and_color = brightness_and_color_to_int( current_brightness, color_map[color_name] ) - return self.send("set_rgb", [brightness_and_color]) + return self._gateway.send("set_rgb", [brightness_and_color]) @command(click.argument("brightness", type=int)) def set_brightness(self, brightness): """Set gateway lamp brightness (0-100).""" if 100 < brightness < 0: raise Exception("Brightness must be between 0 and 100") - current_color = int_to_rgb(self.send("get_rgb")[0]) + current_color = int_to_rgb(self._gateway.send("get_rgb")[0]) brightness_and_color = brightness_and_color_to_int(brightness, current_color) - return self.send("set_rgb", [brightness_and_color]) + return self._gateway.send("set_rgb", [brightness_and_color]) @command(click.argument("brightness", type=int)) def set_night_light_brightness(self, brightness): """Set night light brightness (0-100).""" if 100 < brightness < 0: raise Exception("Brightness must be between 0 and 100") - current_color = int_to_rgb(self.send("get_night_light_rgb")[0]) + current_color = int_to_rgb(self._gateway.send("get_night_light_rgb")[0]) brightness_and_color = brightness_and_color_to_int(brightness, current_color) print(brightness, current_color) - return self.send("set_night_light_rgb", [brightness_and_color]) + return self._gateway.send("set_night_light_rgb", [brightness_and_color]) @command( - click.argument("color_name", type=str), click.argument("brightness", type=int) + click.argument("color_name", type=str), click.argument("brightness", type=int), ) def set_light(self, color_name, brightness): """Set color (using color name) and brightness (0-100).""" @@ -529,45 +586,293 @@ def set_light(self, color_name, brightness): brightness_and_color = brightness_and_color_to_int( brightness, color_map[color_name] ) - return self.send("set_rgb", [brightness_and_color]) + return self._gateway.send("set_rgb", [brightness_and_color]) class SubDevice: - def __init__(self, gw, sid, type_, _, __, ___): - self.gw = gw - self.sid = sid - self.type = DeviceType(type_) + """ + Base class for all subdevices of the gateway + these devices are connected through zigbee. + """ + + @attr.s(auto_attribs=True) + class props: + """Defines properties of the specific device""" + + def __init__(self, gw: Gateway = None, dev_info: SubDeviceInfo = None,) -> None: + self._gw = gw + self.sid = dev_info.sid + self._battery = None + self._fw_ver = dev_info.fw_ver + self._props = self.props() + try: + self.type = DeviceType(dev_info.type_id) + except ValueError: + self.type = DeviceType.Unknown - def unpair(self): - return self.gw.send("remove_device", [self.sid]) + def __repr__(self): + return "" % ( + self.device_type, + self.sid, + self._fw_ver, + self.get_battery(), + self.status, + ) + @property + def status(self): + """Return sub-device status as a dict containing all properties.""" + return attr.asdict(self._props) + + @property + def device_type(self): + """Return the device type name.""" + return self.type.name + + @property def battery(self): - return self.gw.send("get_battery", [self.sid])[0] + """Return the battery level.""" + return self._battery + + @command() + def update(self): + """Update the device-specific properties.""" + _LOGGER.debug( + "Subdevice '%s' does not have a device specific update method defined", + self.device_type, + ) + + @command() + def send(self, command): + """Send a command/query to the subdevice""" + try: + return self._gw.send(command, [self.sid]) + except Exception as ex: + raise GatewayException( + "Got an exception while sending command %s" % (command) + ) from ex + + @command() + def send_arg(self, command, arguments): + """Send a command/query including arguments to the subdevice""" + try: + return self._gw.send(command, arguments, extra_parameters={"sid": self.sid}) + except Exception as ex: + raise GatewayException( + "Got an exception while sending " + "command '%s' with arguments '%s'" % (command, str(arguments)) + ) from ex + + @command(click.argument("property")) + def get_property(self, property): + """Get the value of a property of the subdevice.""" + try: + response = self._gw.send("get_device_prop", [self.sid, property]) + except Exception as ex: + raise GatewayException( + "Got an exception while fetching property %s" % (property) + ) from ex + + if not response: + raise GatewayException( + "Empty response while fetching property '%s': %s" % (property, response) + ) + + return response + + @command(click.argument("properties", nargs=-1)) + def get_property_exp(self, properties): + """Get the value of a bunch of properties of the subdevice.""" + try: + response = self._gw.send( + "get_device_prop_exp", [[self.sid] + list(properties)] + ).pop() + except Exception as ex: + raise GatewayException( + "Got an exception while fetching properties %s: %s" % (properties) + ) from ex + + if len(list(properties)) != len(response): + raise GatewayException( + "unexpected result while fetching properties %s: %s" + % (properties, response) + ) + + return response + + @command(click.argument("property"), click.argument("value")) + def set_property(self, property, value): + """Set a device property of the subdevice.""" + try: + return self._gw.send("set_device_prop", {"sid": self.sid, property: value}) + except Exception as ex: + raise GatewayException( + "Got an exception while setting propertie %s to value %s" + % (property, str(value)) + ) from ex + + @command() + def unpair(self): + """Unpair this device from the gateway.""" + return self.send("remove_device") + + @command() + def get_battery(self): + """Update the battery level and return the new value.""" + self._battery = self.send("get_battery").pop() + return self._battery + @command() def get_firmware_version(self) -> Optional[int]: """Returns firmware version""" try: - return self.gw.send("get_device_prop", [self.sid, "fw_ver"])[0] + self._fw_ver = self.get_property("fw_ver").pop() except Exception as ex: - _LOGGER.debug( - "Got an exception while fetching fw_ver: %s", ex, exc_info=True + _LOGGER.info( + "get_firmware_version failed, returning firmware version from discovery info: %s", + ex, ) - return None + return self._fw_ver - def __repr__(self): - return "" % ( - self.type, - self.sid, - self.get_firmware_version(), - self.battery(), - ) + +class AqaraHT(SubDevice): + """Subdevice AqaraHT specific properties and methods""" + + accessor = "get_prop_sensor_ht" + properties = ["temperature", "humidity", "pressure"] + + @attr.s(auto_attribs=True) + class props: + """Device specific properties""" + + temperature: int = None # in degrees celsius + humidity: int = None # in % + pressure: int = None # in hPa + + @command() + def update(self): + """Update all device properties""" + values = self.get_property_exp(self.properties) + try: + self._props.temperature = values[0] / 100 + self._props.humidity = values[1] / 100 + self._props.pressure = values[2] / 100 + except Exception as ex: + raise GatewayException( + "One or more unexpected results while " + "fetching properties %s: %s" % (self.properties, values) + ) from ex class SensorHT(SubDevice): + """Subdevice SensorHT specific properties and methods""" + accessor = "get_prop_sensor_ht" - properties = ["temperature", "humidity"] + properties = ["temperature", "humidity", "pressure"] + @attr.s(auto_attribs=True) + class props: + """Device specific properties""" + + temperature: int = None # in degrees celsius + humidity: int = None # in % + pressure: int = None # in hPa + + @command() + def update(self): + """Update all device properties""" + values = self.get_property_exp(self.properties) + try: + self._props.temperature = values[0] / 100 + self._props.humidity = values[1] / 100 + self._props.pressure = values[2] / 100 + except Exception as ex: + raise GatewayException( + "One or more unexpected results while " + "fetching properties %s: %s" % (self.properties, values) + ) from ex + + +class AqaraMagnet(SubDevice): + """Subdevice AqaraMagnet specific properties and methods""" + + properties = ["unkown"] + + @attr.s(auto_attribs=True) + class props: + """Device specific properties""" + + status: str = None # 'open' or 'closed' + + @command() + def update(self): + """Update all device properties""" + values = self.get_property_exp(self.properties) + self._props.status = values[0] + + +class AqaraPlug(SubDevice): + """Subdevice AqaraPlug specific properties and methods""" -class Plug(SubDevice): accessor = "get_prop_plug" properties = ["power", "neutral_0"] + + @attr.s(auto_attribs=True) + class props: + """Device specific properties""" + + status: str = None # 'on' / 'off' + load_power: int = None # power consumption in ?unit? + + @command() + def update(self): + """Update all device properties""" + values = self.get_property_exp(self.properties) + self._props.load_power = values[0] + self._props.status = values[1] + + +class AqaraRelayTwoChannels(SubDevice): + """Subdevice AqaraRelayTwoChannels specific properties and methods""" + + properties = ["load_power", "channel_0", "channel_1"] + _status_ch0 = None + _status_ch1 = None + _load_power = None + + @attr.s(auto_attribs=True) + class props: + """Device specific properties""" + + status_ch0: str = None # 'on' / 'off' + status_ch1: str = None # 'on' / 'off' + load_power: int = None # power consumption in ?unit? + + class AqaraRelayToggleValue(Enum): + """Options to control the relay""" + + toggle = "toggle" + on = "on" + off = "off" + + class AqaraRelayChannel(Enum): + """Options to select wich relay to control""" + + first = "channel_0" + second = "channel_1" + + @command() + def update(self): + """Update all device properties""" + values = self.get_property_exp(self.properties) + self._props.load_power = values[0] + self._props.status_ch0 = values[1] + self._props.status_ch1 = values[2] + + @command( + click.argument("channel", type=EnumType(AqaraRelayChannel)), + click.argument("value", type=EnumType(AqaraRelayToggleValue)), + ) + def toggle(self, channel, value): + """Toggle Aqara Wireless Relay 2ch""" + return self.send_arg("toggle_ctrl_neutral", [channel.value, value.value]).pop()