diff --git a/xsense/async_xsense.py b/xsense/async_xsense.py index 210c24e..21eaeb8 100644 --- a/xsense/async_xsense.py +++ b/xsense/async_xsense.py @@ -1,7 +1,7 @@ import asyncio from datetime import datetime import json -from typing import Dict +from typing import Dict, List, Optional, Union import aiohttp @@ -246,22 +246,265 @@ async def get_state(self, station: Station): raise APIFailure(f'Unable to retrieve station data: {self._lastres.status}/{text}') async def set_state(self, entity: Entity, shadow: str, topic: str, definition: Dict): - station = entity.station - t = datetime.now() - timestamp = t.strftime('%Y%m%d%H%M%S') - - desired = { - "deviceSN": entity.sn, - "shadow": shadow, - "stationSN": station.sn, - "time": timestamp, - "userId": self.userid - } - desired.update(definition.get('extra', {})) + station, data = self.build_desired_state(entity, shadow, definition) + + return await self.do_thing(station, topic, data) - data = {"state": {"desired": desired}} + async def set_device_config(self, entity: Entity, **values): + shadow = "infoBase" if isinstance(entity, Station) else "infoDev" + station, data = self.build_config_state(entity, shadow, values) + return await self.do_thing(station, f'2nd_cfg_{entity.sn}', data) - res = await self.do_thing(station, topic, data) + async def set_alarm_volume(self, entity: Entity, volume: int, alarm_tone: Optional[str]=None, mute: Optional[str]=None): + self._validate_volume(volume) + values = { + "alarmVol": str(volume), + } + if alarm_tone is not None: + values["alarmTone"] = alarm_tone + if mute is not None: + values["mute"] = mute + + shadow = "infoBase" if isinstance(entity, Station) else "infoDev" + station, data = self.build_config_state(entity, shadow, values) + return await self.do_thing(station, f'2nd_cfg_{entity.sn}', data) + + async def set_voice_volume(self, station: Station, volume: int): + self._validate_volume(volume) + station, data = self.build_config_state(station, "infoBase", {"voiceVol": str(volume)}) + return await self.do_thing(station, f'2nd_cfg_{station.sn}', data) + + async def set_station_mode(self, station: Station, safe_mode: str, force_arm: Optional[str]=None): + values = { + "userParam": "source=1", + "source": "1", + "safeMode": safe_mode, + } + if force_arm is not None: + values["forceArm"] = force_arm + + station, data = self.build_command_state( + station, + "appMode", + values, + include_device=False, + include_time=False + ) + return await self.do_thing(station, "2nd_appmode", data) + + async def trigger_sos(self, station: Station, sos_type: str='1'): + station, data = self.build_command_state( + station, + "sosDown", + { + "userParam": "source=1", + "sosType": sos_type, + }, + include_device=False + ) + return await self.do_thing(station, "2nd_sosdown", data) + + async def cancel_sos(self, station: Station): + station, data = self.build_command_state( + station, + "sosDown", + {"sosStatus": "0"}, + include_device=False + ) + return await self.do_thing(station, "sosdown", data) + + async def cancel_alarm(self, station: Station): + station, data = self.build_command_state( + station, + "alarmCancel", + {"cancelTime": self._utc_timestamp()}, + include_device=False, + include_time=False + ) + return await self.do_thing(station, "alarmcancel", data) + + async def set_fire_drill( + self, + entity: Entity, + drill: Union[bool, str]=True, + drill_time: Optional[str]=None, + alarm_type: Optional[str]=None, + alarm_vol: Optional[str]=None, + alarm_tone: Optional[str]=None, + location: Optional[str]=None, + stop_reason: Optional[str]=None + ): + values = {"drill": self._bool_value(drill)} + optional = { + "drillTime": drill_time, + "alarmType": alarm_type, + "alarmVol": alarm_vol, + "alarmTone": alarm_tone, + "location": location, + "stopReason": stop_reason, + } + values.update({k: v for k, v in optional.items() if v is not None}) + if not isinstance(entity, Station): + values["deviceType"] = entity.type + + station, data = self.build_command_state( + entity, + "appFireDrill", + values, + include_device=not isinstance(entity, Station) + ) + return await self.do_thing(station, "2nd_firedrill", data) + + async def set_sos_sound(self, station: Station, sos_sound: str): + station, data = self.build_command_state( + station, + "sosParam", + { + "userParam": "source=1", + "sosSound": sos_sound, + }, + include_device=False + ) + return await self.do_thing(station, "2nd_sosparam", data) + + async def activate_device(self, entity: Entity): + station, data = self.build_command_state( + entity, + "app2ndActivate", + {"activate": "1"}, + include_device=True + ) + return await self.do_thing(station, "2nd_appactivate", data) + + async def set_install_guide_test( + self, + entity: Entity, + active: Union[bool, str]=True, + dev_type: Optional[str]=None, + test_time: str='180', + detc_sens: Optional[str]=None + ): + values = { + "devType": dev_type or entity.type, + "test": self._bool_value(active), + "testTime": test_time, + } + if detc_sens is not None: + values["detcSens"] = detc_sens + + station, data = self.build_command_state(entity, "appInstallGuide", values, include_device=True) + return await self.do_thing(station, "2nd_appinstallguide", data) + + async def signal_test( + self, + entity: Entity, + dev_type: Optional[str]=None, + test: Union[bool, str]=True, + test_time: str='5' + ): + station, data = self.build_command_state( + entity, + "signalTest", + { + "devType": dev_type or entity.type, + "test": self._bool_value(test), + "testTime": test_time, + }, + include_device=True + ) + return await self.do_thing(station, f'2nd_signaltest_{entity.sn}', data) + + async def set_motion_test(self, entity: Entity, active: Union[bool, str]=True, dev_type: str='SMS01'): + station, data = self.build_command_state( + entity, + "testIR", + { + "devType": dev_type, + "testIR": self._bool_value(active), + }, + include_device=True, + include_time=False, + include_user=False + ) + return await self.do_thing(station, "testir", data) + + async def set_light_power(self, entity: Entity, on: Union[bool, str]): + station, data = self.build_command_state( + entity, + "lampPower", + { + "userParam": "source=1", + "isOn": self._bool_value(on), + "dev": entity.sn, + }, + include_device=False + ) + return await self.do_thing(station, "2nd_lamppower", data) + + async def set_light_group_power( + self, + station: Station, + group_id: str, + device_sns: List[str], + on: Union[bool, str], + timeout: str='180' + ): + station, data = self.build_command_state( + station, + "groupLampPower", + { + "userParam": "source=1", + "timeOut": timeout, + "groupId": group_id, + "devs": device_sns, + "isOn": self._bool_value(on), + }, + include_device=False + ) + return await self.do_thing(station, "2nd_grouppower", data) + + async def mute_water( + self, + entity: Entity, + set_type: str='0', + silence_time: str='', + trigger_source: Optional[str]=None + ): + values = { + "setType": set_type, + "silenceTime": silence_time, + } + if trigger_source is not None: + values["triggerSource"] = trigger_source + + station, data = self.build_command_state( + entity, + "appWater", + values, + include_device=True + ) + return await self.do_thing(station, "2nd_appwater", data) + + async def mute_temperature_humidity(self, entity: Entity, mute_type: str='1', sensor_type: Optional[str]=None): + station, data = self.build_command_state( + entity, + "extendMute", + { + "muteType": mute_type, + "type": sensor_type or entity.type, + }, + include_device=True + ) + return await self.do_thing(station, "2nd_appmute", data) + + async def mute_driveway(self, entity: Entity, mute: Union[bool, str]=True, topic: str='2nd_driveway'): + station, data = self.build_command_state( + entity, + "appDriveway", + {"mute": self._bool_value(mute)}, + include_device=True + ) + return await self.do_thing(station, topic, data) async def action(self, entity: Entity, action: str): entity_def = entities.get(entity.type) @@ -275,4 +518,4 @@ async def action(self, entity: Entity, action: str): topic = action_def.get('topic') if callable(topic): topic = topic(entity) - await self.set_state(entity, action_def['shadow'], topic, action_def) + return await self.set_state(entity, action_def['shadow'], topic, action_def) diff --git a/xsense/base.py b/xsense/base.py index 007a59e..b578c51 100644 --- a/xsense/base.py +++ b/xsense/base.py @@ -3,7 +3,7 @@ import hmac import json from datetime import datetime, timedelta, timezone -from typing import Dict +from typing import Dict, Optional import boto3 from botocore.exceptions import ClientError @@ -11,7 +11,7 @@ from .entity import Entity from .entity_map import entities -from .exceptions import AuthFailed +from .exceptions import AuthFailed, XSenseError from .station import Station from .house import House @@ -230,7 +230,74 @@ def _parse_get_house_state(self, house: House, data: Dict): if station := house.get_station_by_sn(sn): station.set_data(i) + def _station_for_entity(self, entity: Entity) -> Station: + station = getattr(entity, 'station', None) + if station: + return station + if isinstance(entity, Station): + return entity + raise XSenseError(f'Entity type {entity.type} has no station') + + def _validate_volume(self, volume: int): + if not isinstance(volume, int): + raise XSenseError('Volume must be an integer') + if volume < 0 or volume > 100: + raise XSenseError('Volume must be between 0 and 100') + + def _bool_value(self, value): + if isinstance(value, bool): + return '1' if value else '0' + if value in (0, 1, '0', '1'): + return str(value) + raise XSenseError('Value must be a boolean or 0/1') + + def _utc_timestamp(self): + return datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S') + + def build_command_state( + self, + entity: Entity, + shadow: str, + values: Dict, + include_device: Optional[bool]=None, + include_time: bool=True, + include_user: bool=True + ): + station = self._station_for_entity(entity) + if include_device is None: + include_device = not isinstance(entity, Station) + + desired = { + "shadow": shadow, + "stationSN": station.sn, + } + if include_device: + desired["deviceSN"] = entity.sn + if include_time: + desired["time"] = self._utc_timestamp() + if include_user: + desired["userId"] = self.userid + + desired.update(values) + return station, {"state": {"desired": desired}} + + def build_config_state(self, entity: Entity, shadow: str, values: Dict): + return self.build_command_state( + entity, + shadow, + values, + include_time=False, + include_user=False + ) + def has_action(self, entity: Entity, action: str): if entity_def := entities.get(entity.type): return any(a for a in entity_def.get('actions', []) if a.get('action') == action) - return False \ No newline at end of file + return False + + def build_desired_state(self, entity: Entity, shadow: str, definition: Dict): + values = { + **definition.get('extra', {}), + **definition.get('data', {}) + } + return self.build_command_state(entity, shadow, values) diff --git a/xsense/entity.py b/xsense/entity.py index c8abf18..be4f84e 100644 --- a/xsense/entity.py +++ b/xsense/entity.py @@ -1,5 +1,5 @@ from xsense.entity_map import entities -from xsense.mapping import map_values +from xsense.mapping import map_bool, map_values class Entity: @@ -21,7 +21,7 @@ def __init__( def set_data(self, values: dict): data = values.copy() if 'online' in values: - self.online = values.pop('online') != '0' + self.online = map_bool(values.pop('online')) if values.get('onlineTime'): self.online = True data |= data.pop('status', {}) diff --git a/xsense/entity_map.py b/xsense/entity_map.py index 089fdec..b332d64 100644 --- a/xsense/entity_map.py +++ b/xsense/entity_map.py @@ -1,24 +1,35 @@ from enum import Enum -from typing import Callable, Dict +from typing import Callable, Dict, Optional, Union class EntityType(Enum): ALARM = 'alarm' BASE = "base" BASESTATION = 'station' + CAMERA = 'camera' CO = "co" COMBI = 'combi' DOOR = 'door' HEAT = 'heat' KEYPAD = 'keypad' + LIGHT = 'light' + LISTENER = 'listener' MAILBOX = 'mailbox' MOTION = 'motion' + RADON = 'radon' + REMOTE = 'remote' + SMARTDROP = 'smartdrop' SMOKE = "smoke" TEMPERATURE = "temperature" WATER = "water" -def MuteAction(shadow: str = 'appMute', topic: str|None|Callable = '2nd_appmute', extra: Dict|None=None): +def MuteAction( + shadow: str = 'appMute', + topic: Union[str, Callable, None] = '2nd_appmute', + extra: Optional[Dict]=None, + mute_type: Optional[str]=None +): data = { 'action': 'mute', 'topic': topic, @@ -26,6 +37,8 @@ def MuteAction(shadow: str = 'appMute', topic: str|None|Callable = '2nd_appmute' } if extra: data['extra'] = extra + if mute_type is not None: + data.setdefault('extra', {})['muteType'] = mute_type return data @@ -42,7 +55,8 @@ def FireDrillAction(): return { 'action': 'firedrill', 'topic': '2nd_firedrill', - 'shadow': 'appFireDrill' + 'shadow': 'appFireDrill', + 'data': {'drill': '1'} } @@ -56,8 +70,18 @@ def SATestAction(shadow = 'appSelfTest'): entities = { - 'SAL51': {}, # listener - 'SAL100': {}, # listener + 'SAL51': { + 'type': EntityType.LISTENER, + 'actions': [ + MuteAction('appListener', mute_type='1'), + ], + }, + 'SAL100': { + 'type': EntityType.LISTENER, + 'actions': [ + MuteAction('appListener', mute_type='1'), + ], + }, 'SBS10': { 'type': EntityType.BASESTATION, }, @@ -65,8 +89,21 @@ def SATestAction(shadow = 'appSelfTest'): 'type': EntityType.BASESTATION, 'identifier': lambda entity: f'SBS50{entity.sn}', }, - # SSC0A - Camera - # SSC0B + 'SSC0A': { + 'type': EntityType.CAMERA, + }, + 'SSC0B': { + 'type': EntityType.CAMERA, + }, + 'SC01-MN': { + 'type': EntityType.COMBI, + }, + 'SC01-MR': { + 'type': EntityType.COMBI, + 'actions': [ + MuteAction('appSc07mrMute', mute_type='1'), + ], + }, 'SC06-WX': { 'identifier': lambda entity: f'SC06-WX-{entity.sn}', 'type': EntityType.COMBI, @@ -78,21 +115,60 @@ def SATestAction(shadow = 'appSelfTest'): 'identifier': lambda entity: f'SC07-MR-{entity.sn}', 'type': EntityType.COMBI, 'actions': [ + MuteAction('appSc07mrMute', mute_type='1'), ] }, 'SC07-WX': { 'identifier': lambda entity: f'SC07-WX-{entity.sn}', 'type': EntityType.COMBI, 'actions': [ - MuteAction('1') + MuteAction(mute_type='1') ] }, - # 'SDA51': {}, - Driveway alarm + 'SD11-MR': { + 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + ], + }, + 'SD19-MN': { + 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + ], + }, + 'SD19-MR': { + 'type': EntityType.SMOKE, + }, + 'SDA51': { + 'type': EntityType.ALARM, + 'actions': [ + { + 'action': 'mute', + 'topic': '2nd_driveway', + 'shadow': 'appDriveway', + 'data': {'mute': '1'} + }, + ], + }, 'SDS0A': { 'type': EntityType.DOOR, }, - # 'SES01': {}, - Door sensor - # 'SKF01': {}, - Remote Control + 'SES01': { + 'type': EntityType.DOOR, + }, + 'SKF01': { + 'type': EntityType.REMOTE, + }, + 'SK0Z-3S': { + 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + ], + }, + 'SKP01': { + 'type': EntityType.KEYPAD, + }, 'SKP0A': { 'type': EntityType.KEYPAD, }, @@ -110,69 +186,112 @@ def SATestAction(shadow = 'appSelfTest'): 'SMS0A': { 'type': EntityType.MOTION, }, - # 'SSD01': {}, - # 'SPL51': {}, - # 'SSL51': {}, + 'SMS01': { + 'type': EntityType.MOTION, + }, + 'SPL51': { + 'type': EntityType.LIGHT, + }, + 'SSD01': { + 'type': EntityType.SMARTDROP, + }, + 'SSL51': { + 'type': EntityType.LIGHT, + }, 'STH0A': { 'type': EntityType.TEMPERATURE, 'actions': [ TestAction('thSelfTest'), - MuteAction('1', 'extendMute') + MuteAction('extendMute', '2nd_appmute', extra={'type': 'STH0A'}, mute_type='1') ], }, 'STH0B': { 'type': EntityType.TEMPERATURE, 'actions': [ TestAction('thSelfTest'), - MuteAction('1', 'extendMute') + MuteAction('extendMute', '2nd_appmute', extra={'type': 'STH0B'}, mute_type='1') ], }, + 'STH0C': { + 'type': EntityType.TEMPERATURE, + }, 'STH51': { 'type': EntityType.TEMPERATURE, 'actions': [ TestAction('thSelfTest'), - MuteAction('1', 'extendMute') + MuteAction('extendMute', '2nd_appmute', extra={'type': 'STH51'}, mute_type='1') ], }, - # 'SWL51': {}, + 'SWL51': { + 'type': EntityType.LIGHT, + }, + 'SWS0A': { + 'type': EntityType.WATER, + }, + 'SWS0B': { + 'type': EntityType.WATER, + }, 'SWS51': { 'type': EntityType.WATER, 'actions': [ TestAction('waterSelfTest'), - MuteAction(shadow='appWater', topic='2nd_appwater', extra={'silencetime': '', 'setType': '0'}) + MuteAction(shadow='appWater', topic='2nd_appwater', extra={'silenceTime': '', 'setType': '0'}) + ], + }, + 'CB0Z-3S': { + 'type': EntityType.COMBI, + }, + 'LP/N-SA-0B': { + 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), ], }, + 'LP/N-SCA-0A': { + 'type': EntityType.COMBI, + }, + 'XC0C-iA': { + 'type': EntityType.CO, + }, 'XC0C-iR': { 'type': EntityType.CO, }, + 'XC0M-iR': { + 'type': EntityType.CO, + }, 'XC01-M': { # CO RF 'type': EntityType.CO, 'actions': [ TestAction(shadow='appCoSelfTest'), - MuteAction('1', '"appCoMute') + MuteAction('appCoMute', mute_type='1') ] }, 'XC04-WX': { 'identifier': lambda entity: f'XC04-WX-{entity.sn}', 'type': EntityType.CO, 'actions': [ - MuteAction('1') + MuteAction(mute_type='1') ] }, 'XH02-M': { 'type': EntityType.HEAT, 'actions': [ TestAction(shadow='appXh02mSelfTest'), + MuteAction('appXh02mMute', mute_type='1', extra={'userParam': 'source=1'}), ] }, 'XP0A-MR': { 'type': EntityType.COMBI, 'actions': [ TestAction(shadow='app2ndSelfTest'), + MuteAction('appXp0amrMute', mute_type='1', extra={'userParam': 'source=1'}), FireDrillAction() ] }, + 'XR0A-iR': { + 'type': EntityType.RADON, + }, 'XP02S-MR': { 'type': EntityType.SMOKE, 'actions': [ @@ -196,13 +315,70 @@ def SATestAction(shadow = 'appSelfTest'): 'type': EntityType.SMOKE, 'actions': [ TestAction(), - MuteAction(), + MuteAction('app2ndMute', mute_type='1'), FireDrillAction(), ], }, 'XS03-iWX': { # Smoke RF 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + ], + }, + 'XS03-WX': { + 'type': EntityType.SMOKE, + }, + 'XS0D-MR': { + 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + ], + }, + 'XS0D-MR61': { + 'type': EntityType.SMOKE, + }, + 'XC0C-MR': { + 'type': EntityType.CO, + }, + 'XP0A-iR': { + 'type': EntityType.COMBI, + }, + 'XPOA-IR': { + 'type': EntityType.COMBI, + }, + 'XP0H-MR': { + 'type': EntityType.COMBI, + 'actions': [ + MuteAction('appSc07mrMute', mute_type='1'), + ], + }, + 'XP0H-iR': { + 'type': EntityType.COMBI, + }, + 'XP0J-iA': { + 'type': EntityType.COMBI, + }, + 'XP0P-MR': { + 'type': EntityType.COMBI, + 'actions': [ + MuteAction('appSc07mrMute', mute_type='1'), + ], + }, + 'XS0B-iR': { + 'type': EntityType.SMOKE, + }, + 'XS0E-iR': { + 'type': EntityType.SMOKE, + }, + 'XS0F-PMA': { + 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + MuteAction('app2ndMute', mute_type='1'), + ], + }, + 'XS0R-iA': { + 'type': EntityType.SMOKE, }, - 'XS03-WX': {} } diff --git a/xsense/mapping.py b/xsense/mapping.py index 13fada3..0306d8b 100644 --- a/xsense/mapping.py +++ b/xsense/mapping.py @@ -1,5 +1,12 @@ import typing + +def map_bool(value): + if isinstance(value, bool): + return value + return value in (1, '1') + + property_mapper = { '*': { 'wifiRssi': 'wifiRSSI' @@ -40,15 +47,19 @@ } type_mapping = { + 'alarmVol': int, 'batInfo': int, 'rfLevel': int, - 'alarmStatus': lambda x: x == '1', - 'alarmEnabled': lambda x: x == '1', - 'muteStatus': lambda x: x == '1', - 'continuedAlarm': lambda x: x == '1', + 'alarmStatus': map_bool, + 'alarmEnabled': map_bool, + 'muteStatus': map_bool, + 'continuedAlarm': map_bool, 'coPpm': int, 'coLevel': int, - 'isLifeEnd': lambda x: x == '1', + 'isLifeEnd': map_bool, + 'isOpen': map_bool, + 'activate': map_bool, + 'voiceVol': int, 'temperature': float, 'humidity': float } @@ -59,8 +70,10 @@ def map_type(k: str, value: typing.Any): def map_values(device_type: str, data: typing.Dict): - mapping = property_mapper[device_type] if device_type in property_mapper else {} - mapping.update(property_mapper.get('*', {})) + mapping = { + **property_mapper.get('*', {}), + **property_mapper.get(device_type, {}), + } return { mapping.get(k, k): map_type(mapping.get(k, k), v) diff --git a/xsense/mqtt_helper.py b/xsense/mqtt_helper.py index 514ac19..81059b9 100644 --- a/xsense/mqtt_helper.py +++ b/xsense/mqtt_helper.py @@ -33,7 +33,7 @@ def _get_path(self): signed = self.signer.presign_url(f'wss://{self.house.mqtt_server}/mqtt', self.house.mqtt_region) url_parts = signed.split("/") self._mqtt_path = "/" + "/".join(url_parts[3:]) - _sig_age = datetime.now() + self._sig_age = datetime.now() return self._mqtt_path diff --git a/xsense/station.py b/xsense/station.py index 8de3842..a231d7a 100644 --- a/xsense/station.py +++ b/xsense/station.py @@ -2,6 +2,7 @@ from xsense.device import Device from xsense.entity import Entity +from xsense.mapping import map_bool class Station(Entity): @@ -19,7 +20,7 @@ def __init__( self.entity_id = kwargs.get('stationId') self.name = kwargs.get('stationName') self.sn = kwargs.get('stationSn') - self.online = kwargs.get('onLine', True) + self.online = map_bool(kwargs.get('onLine', True)) self.type = kwargs.get('category') self.has_alarm = False @@ -30,7 +31,7 @@ def set_devices(self, data): self.device_order = data.get('deviceSort') result = {} result_sn = {} - for i in data.get('devices'): + for i in data.get('devices', []): d = Device( self, **i @@ -62,6 +63,9 @@ def set_alarm_data(self, values: dict): if v := values.get(k): self._alarm_data[k] = v + if safe_mode := values.get('safeMode'): + self.safe_mode = safe_mode + @property def alarm_data(self): return self._alarm_data diff --git a/xsense/xsense.py b/xsense/xsense.py index 8c239d1..6abcc98 100644 --- a/xsense/xsense.py +++ b/xsense/xsense.py @@ -1,5 +1,5 @@ -from datetime import datetime, timedelta -from typing import Dict +from datetime import datetime +from typing import Dict, List, Optional, Union import requests @@ -222,22 +222,265 @@ def get_state(self, station: Station): raise APIFailure(f'Unable to retrieve station data: {self._lastres.status_code}/{self._lastres.text}') def set_state(self, entity: Entity, shadow: str, topic: str, definition: Dict): - station = entity.station - t = datetime.now() - timestamp = t.strftime('%Y%m%d%H%M%S') - - desired = { - "deviceSN": entity.sn, - "shadow": shadow, - "stationSN": station.sn, - "time": timestamp, - "userId": self.userid - } - desired.update(definition.get('extra', {})) + station, data = self.build_desired_state(entity, shadow, definition) + + return self.do_thing(station, topic, data) - data = {"state": {"desired": desired}} + def set_device_config(self, entity: Entity, **values): + shadow = "infoBase" if isinstance(entity, Station) else "infoDev" + station, data = self.build_config_state(entity, shadow, values) + return self.do_thing(station, f'2nd_cfg_{entity.sn}', data) - res = self.do_thing(station, topic, data) + def set_alarm_volume(self, entity: Entity, volume: int, alarm_tone: Optional[str]=None, mute: Optional[str]=None): + self._validate_volume(volume) + values = { + "alarmVol": str(volume), + } + if alarm_tone is not None: + values["alarmTone"] = alarm_tone + if mute is not None: + values["mute"] = mute + + shadow = "infoBase" if isinstance(entity, Station) else "infoDev" + station, data = self.build_config_state(entity, shadow, values) + return self.do_thing(station, f'2nd_cfg_{entity.sn}', data) + + def set_voice_volume(self, station: Station, volume: int): + self._validate_volume(volume) + station, data = self.build_config_state(station, "infoBase", {"voiceVol": str(volume)}) + return self.do_thing(station, f'2nd_cfg_{station.sn}', data) + + def set_station_mode(self, station: Station, safe_mode: str, force_arm: Optional[str]=None): + values = { + "userParam": "source=1", + "source": "1", + "safeMode": safe_mode, + } + if force_arm is not None: + values["forceArm"] = force_arm + + station, data = self.build_command_state( + station, + "appMode", + values, + include_device=False, + include_time=False + ) + return self.do_thing(station, "2nd_appmode", data) + + def trigger_sos(self, station: Station, sos_type: str='1'): + station, data = self.build_command_state( + station, + "sosDown", + { + "userParam": "source=1", + "sosType": sos_type, + }, + include_device=False + ) + return self.do_thing(station, "2nd_sosdown", data) + + def cancel_sos(self, station: Station): + station, data = self.build_command_state( + station, + "sosDown", + {"sosStatus": "0"}, + include_device=False + ) + return self.do_thing(station, "sosdown", data) + + def cancel_alarm(self, station: Station): + station, data = self.build_command_state( + station, + "alarmCancel", + {"cancelTime": self._utc_timestamp()}, + include_device=False, + include_time=False + ) + return self.do_thing(station, "alarmcancel", data) + + def set_fire_drill( + self, + entity: Entity, + drill: Union[bool, str]=True, + drill_time: Optional[str]=None, + alarm_type: Optional[str]=None, + alarm_vol: Optional[str]=None, + alarm_tone: Optional[str]=None, + location: Optional[str]=None, + stop_reason: Optional[str]=None + ): + values = {"drill": self._bool_value(drill)} + optional = { + "drillTime": drill_time, + "alarmType": alarm_type, + "alarmVol": alarm_vol, + "alarmTone": alarm_tone, + "location": location, + "stopReason": stop_reason, + } + values.update({k: v for k, v in optional.items() if v is not None}) + if not isinstance(entity, Station): + values["deviceType"] = entity.type + + station, data = self.build_command_state( + entity, + "appFireDrill", + values, + include_device=not isinstance(entity, Station) + ) + return self.do_thing(station, "2nd_firedrill", data) + + def set_sos_sound(self, station: Station, sos_sound: str): + station, data = self.build_command_state( + station, + "sosParam", + { + "userParam": "source=1", + "sosSound": sos_sound, + }, + include_device=False + ) + return self.do_thing(station, "2nd_sosparam", data) + + def activate_device(self, entity: Entity): + station, data = self.build_command_state( + entity, + "app2ndActivate", + {"activate": "1"}, + include_device=True + ) + return self.do_thing(station, "2nd_appactivate", data) + + def set_install_guide_test( + self, + entity: Entity, + active: Union[bool, str]=True, + dev_type: Optional[str]=None, + test_time: str='180', + detc_sens: Optional[str]=None + ): + values = { + "devType": dev_type or entity.type, + "test": self._bool_value(active), + "testTime": test_time, + } + if detc_sens is not None: + values["detcSens"] = detc_sens + + station, data = self.build_command_state(entity, "appInstallGuide", values, include_device=True) + return self.do_thing(station, "2nd_appinstallguide", data) + + def signal_test( + self, + entity: Entity, + dev_type: Optional[str]=None, + test: Union[bool, str]=True, + test_time: str='5' + ): + station, data = self.build_command_state( + entity, + "signalTest", + { + "devType": dev_type or entity.type, + "test": self._bool_value(test), + "testTime": test_time, + }, + include_device=True + ) + return self.do_thing(station, f'2nd_signaltest_{entity.sn}', data) + + def set_motion_test(self, entity: Entity, active: Union[bool, str]=True, dev_type: str='SMS01'): + station, data = self.build_command_state( + entity, + "testIR", + { + "devType": dev_type, + "testIR": self._bool_value(active), + }, + include_device=True, + include_time=False, + include_user=False + ) + return self.do_thing(station, "testir", data) + + def set_light_power(self, entity: Entity, on: Union[bool, str]): + station, data = self.build_command_state( + entity, + "lampPower", + { + "userParam": "source=1", + "isOn": self._bool_value(on), + "dev": entity.sn, + }, + include_device=False + ) + return self.do_thing(station, "2nd_lamppower", data) + + def set_light_group_power( + self, + station: Station, + group_id: str, + device_sns: List[str], + on: Union[bool, str], + timeout: str='180' + ): + station, data = self.build_command_state( + station, + "groupLampPower", + { + "userParam": "source=1", + "timeOut": timeout, + "groupId": group_id, + "devs": device_sns, + "isOn": self._bool_value(on), + }, + include_device=False + ) + return self.do_thing(station, "2nd_grouppower", data) + + def mute_water( + self, + entity: Entity, + set_type: str='0', + silence_time: str='', + trigger_source: Optional[str]=None + ): + values = { + "setType": set_type, + "silenceTime": silence_time, + } + if trigger_source is not None: + values["triggerSource"] = trigger_source + + station, data = self.build_command_state( + entity, + "appWater", + values, + include_device=True + ) + return self.do_thing(station, "2nd_appwater", data) + + def mute_temperature_humidity(self, entity: Entity, mute_type: str='1', sensor_type: Optional[str]=None): + station, data = self.build_command_state( + entity, + "extendMute", + { + "muteType": mute_type, + "type": sensor_type or entity.type, + }, + include_device=True + ) + return self.do_thing(station, "2nd_appmute", data) + + def mute_driveway(self, entity: Entity, mute: Union[bool, str]=True, topic: str='2nd_driveway'): + station, data = self.build_command_state( + entity, + "appDriveway", + {"mute": self._bool_value(mute)}, + include_device=True + ) + return self.do_thing(station, topic, data) def action(self, entity: Entity, action: str): entity_def = entities.get(entity.type) @@ -251,4 +494,4 @@ def action(self, entity: Entity, action: str): topic = action_def.get('topic') if callable(topic): topic = topic(entity) - self.set_state(entity, action_def['shadow'], topic, action_def) + return self.set_state(entity, action_def['shadow'], topic, action_def)