Add APK-backed control commands

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)

xsense/base.py

@@ -3,15 +3,15 @@
 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
 from pycognito import AWSSRP
 
 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
+        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)