From 75a743f0e6e732bdd6b05c23c7bba5ca1c62747e Mon Sep 17 00:00:00 2001 From: Wheemer Date: Fri, 29 May 2026 22:12:15 -0230 Subject: [PATCH 1/7] Handle station actions and data payloads --- xsense/async_xsense.py | 19 +++---------------- xsense/base.py | 30 ++++++++++++++++++++++++++++-- xsense/station.py | 3 +++ xsense/xsense.py | 21 ++++----------------- 4 files changed, 38 insertions(+), 35 deletions(-) diff --git a/xsense/async_xsense.py b/xsense/async_xsense.py index 210c24e..527db27 100644 --- a/xsense/async_xsense.py +++ b/xsense/async_xsense.py @@ -246,22 +246,9 @@ 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', {})) - - data = {"state": {"desired": desired}} + station, data = self.build_desired_state(entity, shadow, definition) - res = await self.do_thing(station, topic, data) + return await self.do_thing(station, topic, data) async def action(self, entity: Entity, action: str): entity_def = entities.get(entity.type) @@ -275,4 +262,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..f5bbbb7 100644 --- a/xsense/base.py +++ b/xsense/base.py @@ -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 @@ -233,4 +233,30 @@ def _parse_get_house_state(self, house: House, data: Dict): 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 _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 build_desired_state(self, entity: Entity, shadow: str, definition: Dict): + station = self._station_for_entity(entity) + t = datetime.now(timezone.utc) + timestamp = t.strftime('%Y%m%d%H%M%S') + + desired = { + "shadow": shadow, + "stationSN": station.sn, + "time": timestamp, + "userId": self.userid + } + if not isinstance(entity, Station): + desired["deviceSN"] = entity.sn + desired.update(definition.get('extra', {})) + desired.update(definition.get('data', {})) + + return station, {"state": {"desired": desired}} diff --git a/xsense/station.py b/xsense/station.py index 8de3842..222657d 100644 --- a/xsense/station.py +++ b/xsense/station.py @@ -62,6 +62,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..81a96e0 100644 --- a/xsense/xsense.py +++ b/xsense/xsense.py @@ -1,4 +1,4 @@ -from datetime import datetime, timedelta +from datetime import datetime from typing import Dict import requests @@ -222,22 +222,9 @@ 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', {})) - - data = {"state": {"desired": desired}} + station, data = self.build_desired_state(entity, shadow, definition) - res = self.do_thing(station, topic, data) + return self.do_thing(station, topic, data) def action(self, entity: Entity, action: str): entity_def = entities.get(entity.type) @@ -251,4 +238,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) From 0a731afa5878554cdc947baa9b3de059629b66fd Mon Sep 17 00:00:00 2001 From: Wheemer Date: Fri, 29 May 2026 22:13:31 -0230 Subject: [PATCH 2/7] Expand device map and boolean states --- xsense/entity_map.py | 22 +++++++++++++++++++++- xsense/mapping.py | 19 ++++++++++++++----- 2 files changed, 35 insertions(+), 6 deletions(-) diff --git a/xsense/entity_map.py b/xsense/entity_map.py index 089fdec..f99213c 100644 --- a/xsense/entity_map.py +++ b/xsense/entity_map.py @@ -203,6 +203,26 @@ def SATestAction(shadow = 'appSelfTest'): 'XS03-iWX': { # Smoke RF 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + ], + }, + 'XS03-WX': { + 'type': EntityType.SMOKE, + }, + 'XS0D-MR': { + 'type': EntityType.SMOKE, + }, + 'XS0D-MR61': { + 'type': EntityType.SMOKE, + }, + 'XC0C-MR': { + 'type': EntityType.CO, + }, + 'XP0A-iR': { + 'type': EntityType.COMBI, + }, + 'XPOA-IR': { + 'type': EntityType.COMBI, }, - 'XS03-WX': {} } diff --git a/xsense/mapping.py b/xsense/mapping.py index 13fada3..c80074a 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' @@ -42,13 +49,15 @@ type_mapping = { '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, 'temperature': float, 'humidity': float } From e2857348668438afd9c6b104a21da1702bf894b4 Mon Sep 17 00:00:00 2001 From: Wheemer Date: Fri, 29 May 2026 22:14:17 -0230 Subject: [PATCH 3/7] Normalize entity state parsing --- xsense/entity.py | 4 ++-- xsense/mapping.py | 6 ++++-- xsense/mqtt_helper.py | 2 +- xsense/station.py | 5 +++-- 4 files changed, 10 insertions(+), 7 deletions(-) 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/mapping.py b/xsense/mapping.py index c80074a..fb9c4a6 100644 --- a/xsense/mapping.py +++ b/xsense/mapping.py @@ -68,8 +68,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 222657d..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 From 32245a7664182aa713b0d44cbb67652d90aac6fd Mon Sep 17 00:00:00 2001 From: Wheemer Date: Fri, 29 May 2026 23:03:20 -0230 Subject: [PATCH 4/7] Add APK device coverage --- xsense/entity_map.py | 125 +++++++++++++++++++++++++++++++++++++++---- 1 file changed, 114 insertions(+), 11 deletions(-) diff --git a/xsense/entity_map.py b/xsense/entity_map.py index f99213c..b0ba655 100644 --- a/xsense/entity_map.py +++ b/xsense/entity_map.py @@ -6,13 +6,19 @@ 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" @@ -56,8 +62,12 @@ def SATestAction(shadow = 'appSelfTest'): entities = { - 'SAL51': {}, # listener - 'SAL100': {}, # listener + 'SAL51': { + 'type': EntityType.LISTENER, + }, + 'SAL100': { + 'type': EntityType.LISTENER, + }, 'SBS10': { 'type': EntityType.BASESTATION, }, @@ -65,8 +75,18 @@ 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, + }, 'SC06-WX': { 'identifier': lambda entity: f'SC06-WX-{entity.sn}', 'type': EntityType.COMBI, @@ -87,12 +107,33 @@ def SATestAction(shadow = 'appSelfTest'): MuteAction('1') ] }, - # 'SDA51': {}, - Driveway alarm + 'SD11-MR': { + 'type': EntityType.SMOKE, + }, + 'SD19-MN': { + 'type': EntityType.SMOKE, + }, + 'SD19-MR': { + 'type': EntityType.SMOKE, + }, + 'SDA51': { + 'type': EntityType.ALARM, + }, 'SDS0A': { 'type': EntityType.DOOR, }, - # 'SES01': {}, - Door sensor - # 'SKF01': {}, - Remote Control + 'SES01': { + 'type': EntityType.DOOR, + }, + 'SKF01': { + 'type': EntityType.REMOTE, + }, + 'SK0Z-3S': { + 'type': EntityType.SMOKE, + }, + 'SKP01': { + 'type': EntityType.KEYPAD, + }, 'SKP0A': { 'type': EntityType.KEYPAD, }, @@ -110,9 +151,18 @@ 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': [ @@ -127,6 +177,9 @@ def SATestAction(shadow = 'appSelfTest'): MuteAction('1', 'extendMute') ], }, + 'STH0C': { + 'type': EntityType.TEMPERATURE, + }, 'STH51': { 'type': EntityType.TEMPERATURE, 'actions': [ @@ -134,7 +187,15 @@ def SATestAction(shadow = 'appSelfTest'): MuteAction('1', 'extendMute') ], }, - # 'SWL51': {}, + 'SWL51': { + 'type': EntityType.LIGHT, + }, + 'SWS0A': { + 'type': EntityType.WATER, + }, + 'SWS0B': { + 'type': EntityType.WATER, + }, 'SWS51': { 'type': EntityType.WATER, 'actions': [ @@ -142,9 +203,24 @@ def SATestAction(shadow = 'appSelfTest'): MuteAction(shadow='appWater', topic='2nd_appwater', extra={'silencetime': '', 'setType': '0'}) ], }, + 'CB0Z-3S': { + 'type': EntityType.COMBI, + }, + 'LP/N-SA-0B': { + 'type': EntityType.SMOKE, + }, + '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, @@ -173,6 +249,9 @@ def SATestAction(shadow = 'appSelfTest'): FireDrillAction() ] }, + 'XR0A-iR': { + 'type': EntityType.RADON, + }, 'XP02S-MR': { 'type': EntityType.SMOKE, 'actions': [ @@ -225,4 +304,28 @@ def SATestAction(shadow = 'appSelfTest'): 'XPOA-IR': { 'type': EntityType.COMBI, }, + 'XP0H-MR': { + 'type': EntityType.COMBI, + }, + 'XP0H-iR': { + 'type': EntityType.COMBI, + }, + 'XP0J-iA': { + 'type': EntityType.COMBI, + }, + 'XP0P-MR': { + 'type': EntityType.COMBI, + }, + 'XS0B-iR': { + 'type': EntityType.SMOKE, + }, + 'XS0E-iR': { + 'type': EntityType.SMOKE, + }, + 'XS0F-PMA': { + 'type': EntityType.SMOKE, + }, + 'XS0R-iA': { + 'type': EntityType.SMOKE, + }, } From 7147df785b2423c3f4d53663da7aa6b244c33b7c Mon Sep 17 00:00:00 2001 From: Wheemer Date: Fri, 29 May 2026 22:36:47 -0230 Subject: [PATCH 5/7] Add device-specific mute actions --- xsense/entity_map.py | 75 +++++++++++++++++++++++++++++++++++++------- 1 file changed, 64 insertions(+), 11 deletions(-) diff --git a/xsense/entity_map.py b/xsense/entity_map.py index b0ba655..d005984 100644 --- a/xsense/entity_map.py +++ b/xsense/entity_map.py @@ -1,5 +1,5 @@ from enum import Enum -from typing import Callable, Dict +from typing import Callable, Dict, Optional, Union class EntityType(Enum): @@ -24,7 +24,12 @@ class EntityType(Enum): 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, @@ -32,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 @@ -48,7 +55,8 @@ def FireDrillAction(): return { 'action': 'firedrill', 'topic': '2nd_firedrill', - 'shadow': 'appFireDrill' + 'shadow': 'appFireDrill', + 'data': {'drill': '1'} } @@ -64,9 +72,15 @@ def SATestAction(shadow = 'appSelfTest'): entities = { 'SAL51': { 'type': EntityType.LISTENER, + 'actions': [ + MuteAction('appListener', mute_type='1'), + ], }, 'SAL100': { 'type': EntityType.LISTENER, + 'actions': [ + MuteAction('appListener', mute_type='1'), + ], }, 'SBS10': { 'type': EntityType.BASESTATION, @@ -86,6 +100,9 @@ def SATestAction(shadow = 'appSelfTest'): }, 'SC01-MR': { 'type': EntityType.COMBI, + 'actions': [ + MuteAction('appSc07mrMute', mute_type='1'), + ], }, 'SC06-WX': { 'identifier': lambda entity: f'SC06-WX-{entity.sn}', @@ -98,26 +115,41 @@ 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') ] }, '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_appdriveway', + 'shadow': 'appDriveway', + 'data': {'mute': '1'} + }, + ], }, 'SDS0A': { 'type': EntityType.DOOR, @@ -130,6 +162,9 @@ def SATestAction(shadow = 'appSelfTest'): }, 'SK0Z-3S': { 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + ], }, 'SKP01': { 'type': EntityType.KEYPAD, @@ -167,14 +202,14 @@ def SATestAction(shadow = 'appSelfTest'): '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': { @@ -184,7 +219,7 @@ def SATestAction(shadow = 'appSelfTest'): 'type': EntityType.TEMPERATURE, 'actions': [ TestAction('thSelfTest'), - MuteAction('1', 'extendMute') + MuteAction('extendMute', '2nd_appmute', extra={'type': 'STH51'}, mute_type='1') ], }, 'SWL51': { @@ -200,7 +235,7 @@ def SATestAction(shadow = 'appSelfTest'): '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': { @@ -208,6 +243,9 @@ def SATestAction(shadow = 'appSelfTest'): }, 'LP/N-SA-0B': { 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + ], }, 'LP/N-SCA-0A': { 'type': EntityType.COMBI, @@ -226,26 +264,28 @@ def SATestAction(shadow = 'appSelfTest'): '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() ] }, @@ -275,7 +315,7 @@ def SATestAction(shadow = 'appSelfTest'): 'type': EntityType.SMOKE, 'actions': [ TestAction(), - MuteAction(), + MuteAction('app2ndMute', mute_type='1'), FireDrillAction(), ], }, @@ -291,6 +331,9 @@ def SATestAction(shadow = 'appSelfTest'): }, 'XS0D-MR': { 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + ], }, 'XS0D-MR61': { 'type': EntityType.SMOKE, @@ -306,6 +349,9 @@ def SATestAction(shadow = 'appSelfTest'): }, 'XP0H-MR': { 'type': EntityType.COMBI, + 'actions': [ + MuteAction('appSc07mrMute', mute_type='1'), + ], }, 'XP0H-iR': { 'type': EntityType.COMBI, @@ -315,6 +361,9 @@ def SATestAction(shadow = 'appSelfTest'): }, 'XP0P-MR': { 'type': EntityType.COMBI, + 'actions': [ + MuteAction('appSc07mrMute', mute_type='1'), + ], }, 'XS0B-iR': { 'type': EntityType.SMOKE, @@ -324,6 +373,10 @@ def SATestAction(shadow = 'appSelfTest'): }, 'XS0F-PMA': { 'type': EntityType.SMOKE, + 'actions': [ + TestAction(), + MuteAction('app2ndMute', mute_type='1'), + ], }, 'XS0R-iA': { 'type': EntityType.SMOKE, From e2027b9070508d7d03348d62e74674b74640d9b7 Mon Sep 17 00:00:00 2001 From: Wheemer Date: Fri, 29 May 2026 22:55:23 -0230 Subject: [PATCH 6/7] Add volume setters --- xsense/async_xsense.py | 21 ++++++++++++++++++++- xsense/base.py | 29 ++++++++++++++++++++++++----- xsense/mapping.py | 2 ++ xsense/xsense.py | 21 ++++++++++++++++++++- 4 files changed, 66 insertions(+), 7 deletions(-) diff --git a/xsense/async_xsense.py b/xsense/async_xsense.py index 527db27..44a1d6f 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, Optional import aiohttp @@ -250,6 +250,25 @@ async def set_state(self, entity: Entity, shadow: str, topic: str, definition: D return 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 action(self, entity: Entity, action: str): entity_def = entities.get(entity.type) if not entity_def: diff --git a/xsense/base.py b/xsense/base.py index f5bbbb7..6cb7dfb 100644 --- a/xsense/base.py +++ b/xsense/base.py @@ -230,11 +230,6 @@ def _parse_get_house_state(self, house: House, data: Dict): if station := house.get_station_by_sn(sn): station.set_data(i) - 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 - def _station_for_entity(self, entity: Entity) -> Station: station = getattr(entity, 'station', None) if station: @@ -243,6 +238,30 @@ def _station_for_entity(self, entity: 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 build_config_state(self, entity: Entity, shadow: str, values: Dict): + station = self._station_for_entity(entity) + desired = { + "shadow": shadow, + "stationSN": station.sn, + **values + } + + if not isinstance(entity, Station): + desired["deviceSN"] = entity.sn + + return station, {"state": {"desired": desired}} + + 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 + def build_desired_state(self, entity: Entity, shadow: str, definition: Dict): station = self._station_for_entity(entity) t = datetime.now(timezone.utc) diff --git a/xsense/mapping.py b/xsense/mapping.py index fb9c4a6..0306d8b 100644 --- a/xsense/mapping.py +++ b/xsense/mapping.py @@ -47,6 +47,7 @@ def map_bool(value): } type_mapping = { + 'alarmVol': int, 'batInfo': int, 'rfLevel': int, 'alarmStatus': map_bool, @@ -58,6 +59,7 @@ def map_bool(value): 'isLifeEnd': map_bool, 'isOpen': map_bool, 'activate': map_bool, + 'voiceVol': int, 'temperature': float, 'humidity': float } diff --git a/xsense/xsense.py b/xsense/xsense.py index 81a96e0..e0387f4 100644 --- a/xsense/xsense.py +++ b/xsense/xsense.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Dict +from typing import Dict, Optional import requests @@ -226,6 +226,25 @@ def set_state(self, entity: Entity, shadow: str, topic: str, definition: Dict): return 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 action(self, entity: Entity, action: str): entity_def = entities.get(entity.type) if not entity_def: From 184828d110b7c8830d4fa7f650e72854672ba657 Mon Sep 17 00:00:00 2001 From: Wheemer Date: Fri, 29 May 2026 23:41:49 -0230 Subject: [PATCH 7/7] Add APK-backed control commands --- xsense/async_xsense.py | 239 ++++++++++++++++++++++++++++++++++++++++- xsense/base.py | 62 +++++++---- xsense/entity_map.py | 2 +- xsense/xsense.py | 239 ++++++++++++++++++++++++++++++++++++++++- 4 files changed, 519 insertions(+), 23 deletions(-) diff --git a/xsense/async_xsense.py b/xsense/async_xsense.py index 44a1d6f..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, Optional +from typing import Dict, List, Optional, Union import aiohttp @@ -250,6 +250,11 @@ async def set_state(self, entity: Entity, shadow: str, topic: str, definition: D return await self.do_thing(station, topic, data) + 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) + async def set_alarm_volume(self, entity: Entity, volume: int, alarm_tone: Optional[str]=None, mute: Optional[str]=None): self._validate_volume(volume) values = { @@ -269,6 +274,238 @@ async def set_voice_volume(self, station: Station, volume: int): 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) if not entity_def: diff --git a/xsense/base.py b/xsense/base.py index 6cb7dfb..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 @@ -244,38 +244,60 @@ def _validate_volume(self, volume: int): if volume < 0 or volume > 100: raise XSenseError('Volume must be between 0 and 100') - def build_config_state(self, entity: Entity, shadow: str, values: Dict): + 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, - **values } - - if not isinstance(entity, Station): + 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 def build_desired_state(self, entity: Entity, shadow: str, definition: Dict): - station = self._station_for_entity(entity) - t = datetime.now(timezone.utc) - timestamp = t.strftime('%Y%m%d%H%M%S') - - desired = { - "shadow": shadow, - "stationSN": station.sn, - "time": timestamp, - "userId": self.userid + values = { + **definition.get('extra', {}), + **definition.get('data', {}) } - if not isinstance(entity, Station): - desired["deviceSN"] = entity.sn - desired.update(definition.get('extra', {})) - desired.update(definition.get('data', {})) - - return station, {"state": {"desired": desired}} + return self.build_command_state(entity, shadow, values) diff --git a/xsense/entity_map.py b/xsense/entity_map.py index d005984..b332d64 100644 --- a/xsense/entity_map.py +++ b/xsense/entity_map.py @@ -145,7 +145,7 @@ def SATestAction(shadow = 'appSelfTest'): 'actions': [ { 'action': 'mute', - 'topic': '2nd_appdriveway', + 'topic': '2nd_driveway', 'shadow': 'appDriveway', 'data': {'mute': '1'} }, diff --git a/xsense/xsense.py b/xsense/xsense.py index e0387f4..6abcc98 100644 --- a/xsense/xsense.py +++ b/xsense/xsense.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Dict, Optional +from typing import Dict, List, Optional, Union import requests @@ -226,6 +226,11 @@ def set_state(self, entity: Entity, shadow: str, topic: str, definition: Dict): return self.do_thing(station, topic, data) + 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) + def set_alarm_volume(self, entity: Entity, volume: int, alarm_tone: Optional[str]=None, mute: Optional[str]=None): self._validate_volume(volume) values = { @@ -245,6 +250,238 @@ def set_voice_volume(self, station: Station, volume: int): 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) if not entity_def: