From 4c01f6daefb219df800ea8c7c22bf677e3fa0893 Mon Sep 17 00:00:00 2001 From: Giuseppe Sorrentino Date: Thu, 9 Jan 2025 13:13:30 +0100 Subject: [PATCH 1/5] lares 4 proof of concept --- src/ksenia_lares/__init__.py | 5 +- src/ksenia_lares/lares4_api.py | 253 +++++++++++++++++++++++++++++++++ 2 files changed, 256 insertions(+), 2 deletions(-) create mode 100644 src/ksenia_lares/lares4_api.py diff --git a/src/ksenia_lares/__init__.py b/src/ksenia_lares/__init__.py index bac5764..f286fdd 100644 --- a/src/ksenia_lares/__init__.py +++ b/src/ksenia_lares/__init__.py @@ -2,6 +2,7 @@ from .base_api import BaseApi from .ip_api import IpAPI +from .lares4_api import Lares4API def get_api(config: dict) -> BaseApi: """ @@ -15,7 +16,7 @@ def get_api(config: dict) -> BaseApi: if version == "IP": return IpAPI(config) - if version == "IP": - raise ValueError("Lares 4.0 API not yet supported") + if version == "4": + return Lares4API(config) raise ValueError(f"Unsupported API version: {version}") diff --git a/src/ksenia_lares/lares4_api.py b/src/ksenia_lares/lares4_api.py new file mode 100644 index 0000000..3e97a2d --- /dev/null +++ b/src/ksenia_lares/lares4_api.py @@ -0,0 +1,253 @@ +from typing import List +from unittest.mock import sentinel +from urllib import response +import aiohttp +import json +import ssl +import time +import asyncio +import queue + +from aiohttp import ClientWSTimeout, UnixConnector + +from .types import ( + AlarmInfo, + Command, + Partition, + PartitionStatus, + Scenario, + Zone, + ZoneBypass, + ZoneStatus, +) +from .base_api import BaseApi + +def u(e): + t = []; + for n in range(0, len(e)): + r = ord(e[n]); + if (r < 128): + t.append(r) + else: + if (r < 2048): + t.append(192 | r >> 6) + t.append(128 | 63 & r) + else: + if (r < 55296 or r >= 57344): + t.append(224 | r >> 12) + t.append(128 | r >> 6 & 63) + t.append(128 | 63 & r) + else: + n = n + 1; + r = 65536 + ((1023 & r) << 10 | 1023 & ord(e[n])); + t.append(240 | r >> 18) + t.append(128 | r >> 12 & 63) + t.append(128 | r >> 6 & 63) + t.append(128 | 63 & r) + n = n+1 + + return t + +def crc16(e): + i = u(e); + l = e.rfind('"CRC_16"') + len('"CRC_16"') + (len(i) - len(e)); + r = 65535; + s = 0; + while s < l: + t = 128; + o = i[s]; + while t: + if(32768& r): + n = 1; + else: + n = 0; + r <<= 1; + r &= 65535; + if(o & t): + r = r + 1; + if(n): + r = r^4129; + t >>= 1; + s=s+1; + return ("0x"+format(r,'04x')); + +def get_ssl_context(): + ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + ctx.options |= ssl.OP_LEGACY_SERVER_CONNECT + return ctx + +class CommandFactory: + def __init__(self, sender: str, pin: str) -> None: + self._command_id = 0 + self._sender = sender + self._pin = pin + + def get_sender(self) -> str: + return self._sender + + def get_login_id(self) -> str: + return self._login_id + + def set_login_id(self, login_id: str) -> None: + self._login_id = login_id + + def get_pin(self) -> str: + return self._pin + + def get_current_command_id(self) -> int: + return self._command_id + + def get_next_command_id(self) -> int: + self._command_id += 1 + return self._command_id + + def build_payload(self, payload: dict) -> dict: + print(getattr(payload, "PIN", False)) + return { + **payload, + **({"ID_LOGIN": self.get_login_id()} if "ID_LOGIN" in payload else {}), + **({"PIN": self.get_pin()} if "PIN" in payload else {}) + } + + def build_command(self, cmd: str, payload_type: str, payload: dict) -> dict: + timestamp = str(int(time.time())) + + command = { + "SENDER": self.get_sender(), + "RECEIVER": "", + "CMD": cmd, + "ID": f"{self.get_next_command_id()}", + "PAYLOAD_TYPE": payload_type, + "PAYLOAD": self.build_payload(payload), + "TIMESTAMP": f"{timestamp}", + "CRC_16": '0x0000', + } + + command['CRC_16'] = crc16(json.dumps(command)) + + return command + +class Lares4API(): + def __init__(self, config: dict): + if not all(key in config for key in ("url", "pin", "sender")): + raise ValueError( + "Missing one or more of the following keys: host, pin, sender" + ) + + self.client = Lares4Client(config) + + async def connect_client(self): + await self.client.run() + + async def get_scenarios(self) -> List[Scenario] | None: + response = await self.client.get_scenarios() + if response: + return response['PAYLOAD']['SCENARIOS'] + +class Lares4Client(): + def __init__(self, data): + if not all(key in data for key in ("url", "pin", "sender")): + raise ValueError( + "Missing one or more of the following keys: host, pin, sender" + ) + self.url = data["url"] + self.host = f"wss://{data['url']}/KseniaWsock" + self.command_factory = CommandFactory( + data["sender"], data["pin"] + ) + self.is_running = False + + async def connect(self): + self.session = aiohttp.ClientSession() + self.ws = await self.session.ws_connect( + self.host, + protocols=['KS_WSOCK'], + ssl_context=get_ssl_context() + ) + print(f"Connected to {self.url}") + self.is_running = True + + async def send_message(self, message): + if self.ws: + await self.ws.send_json(message) + print(f"Sent: {message}") + else: + print("WebSocket is not connected.") + + async def keep_alive(self): + while self.is_running: + try: + await self.ws.ping() + await asyncio.sleep(10) + except Exception as e: + print(f"Error during keep-alive: {e}") + break + + async def close(self): + self.is_running = False + if self.ws: + await self.ws.close() + if self.session: + await self.session.close() + print("Connection closed") + + async def run(self): + await self.connect() + + receive_login = asyncio.create_task(self.receive_login()) + login = asyncio.create_task(self.send_login()) + await asyncio.gather(receive_login, login) + + async def send_login(self): + login_command = self.command_factory.build_command( + cmd="LOGIN", + payload_type="UNKNOWN", + payload={ + "PIN": True + } + ) + await self.send_message(login_command) + + async def receive_login(self): + if self.ws: + msg = await asyncio.wait_for(self.ws.receive(), timeout=1.0) + if msg.type == aiohttp.WSMsgType.TEXT: + data = json.loads(msg.data) + print(data) + if data["CMD"] == "LOGIN_RES": + self.command_factory.set_login_id(data["PAYLOAD"]["ID_LOGIN"]) + else: + print("Login failed") + else: + print("WebSocket is not connected.") + + async def get(self, cmd: str, payload_type: str, payload: dict) -> dict | None: + command = self.command_factory.build_command(cmd, payload_type, payload) + await self.send_message(command) + response = await self.receive() + return response + + async def receive(self) -> dict | None: + if self.ws: + msg = await self.ws.receive() + if msg.type == aiohttp.WSMsgType.TEXT: + data = json.loads(msg.data) + return data + else: + print("WebSocket is not connected.") + + async def get_scenarios(self) -> dict | None: + scenarios = await self.get( + "READ", + "MULTI_TYPES", + { + "ID_LOGIN": True, + "ID_READ": "1", + "TYPES": [ + "SCENARIOS" + ] + } + ) + return scenarios \ No newline at end of file From 308ccab8b34441925ce276a00f1c8d10ff374587 Mon Sep 17 00:00:00 2001 From: Giuseppe Sorrentino Date: Thu, 9 Jan 2025 22:26:41 +0100 Subject: [PATCH 2/5] lares 4 proof of concept 2 --- src/ksenia_lares/lares4_api.py | 80 +++++++++++++++++++++------------- 1 file changed, 50 insertions(+), 30 deletions(-) diff --git a/src/ksenia_lares/lares4_api.py b/src/ksenia_lares/lares4_api.py index 3e97a2d..d3281f9 100644 --- a/src/ksenia_lares/lares4_api.py +++ b/src/ksenia_lares/lares4_api.py @@ -1,3 +1,4 @@ +from re import S from typing import List from unittest.mock import sentinel from urllib import response @@ -21,6 +22,7 @@ ZoneStatus, ) from .base_api import BaseApi +from ksenia_lares import base_api def u(e): t = []; @@ -78,6 +80,7 @@ def get_ssl_context(): ctx.options |= ssl.OP_LEGACY_SERVER_CONNECT return ctx + class CommandFactory: def __init__(self, sender: str, pin: str) -> None: self._command_id = 0 @@ -130,23 +133,6 @@ def build_command(self, cmd: str, payload_type: str, payload: dict) -> dict: return command class Lares4API(): - def __init__(self, config: dict): - if not all(key in config for key in ("url", "pin", "sender")): - raise ValueError( - "Missing one or more of the following keys: host, pin, sender" - ) - - self.client = Lares4Client(config) - - async def connect_client(self): - await self.client.run() - - async def get_scenarios(self) -> List[Scenario] | None: - response = await self.client.get_scenarios() - if response: - return response['PAYLOAD']['SCENARIOS'] - -class Lares4Client(): def __init__(self, data): if not all(key in data for key in ("url", "pin", "sender")): raise ValueError( @@ -176,15 +162,6 @@ async def send_message(self, message): else: print("WebSocket is not connected.") - async def keep_alive(self): - while self.is_running: - try: - await self.ws.ping() - await asyncio.sleep(10) - except Exception as e: - print(f"Error during keep-alive: {e}") - break - async def close(self): self.is_running = False if self.ws: @@ -223,13 +200,13 @@ async def receive_login(self): else: print("WebSocket is not connected.") - async def get(self, cmd: str, payload_type: str, payload: dict) -> dict | None: + async def get(self, cmd: str, payload_type: str, payload: dict): command = self.command_factory.build_command(cmd, payload_type, payload) await self.send_message(command) response = await self.receive() return response - async def receive(self) -> dict | None: + async def receive(self): if self.ws: msg = await self.ws.receive() if msg.type == aiohttp.WSMsgType.TEXT: @@ -238,7 +215,48 @@ async def receive(self) -> dict | None: else: print("WebSocket is not connected.") - async def get_scenarios(self) -> dict | None: + async def info(self): + info = await self.get( + 'REALTIME', + 'REGISTER', + { + 'ID_LOGIN': True, + 'TYPES': ['STATUS_SYSTEM'], + } + ) + return info + + async def get_zones(self): + zones = await self.get( + "READ", + "MULTI_TYPES", + { + "ID_LOGIN": True, + "ID_READ": "1", + "TYPES": [ + "ZONES" + ] + } + ) + + return zones + + async def get_partitions(self): + partitions = await self.get( + "READ", + "MULTI_TYPES", + { + "ID_LOGIN": True, + "ID_READ": "1", + "TYPES": [ + "PARTITIONS" + ] + } + ) + + return partitions + + async def get_scenarios(self): scenarios = await self.get( "READ", "MULTI_TYPES", @@ -250,4 +268,6 @@ async def get_scenarios(self) -> dict | None: ] } ) - return scenarios \ No newline at end of file + + return scenarios + \ No newline at end of file From d10f8127beea05c02f419ab1309f4dd143c16ab6 Mon Sep 17 00:00:00 2001 From: Giuseppe Sorrentino Date: Sun, 12 Jan 2025 11:49:12 +0100 Subject: [PATCH 3/5] zones half mapped and typed --- src/ksenia_lares/__init__.py | 3 +- src/ksenia_lares/base_api.py | 7 +- src/ksenia_lares/ip_api.py | 2 +- src/ksenia_lares/lares4_api.py | 91 ++++++++++++++++------ src/ksenia_lares/{types.py => types_ip.py} | 2 - src/ksenia_lares/types_lares4.py | 39 ++++++++++ tests/ksenia_lares/test_ip_api.py | 2 +- 7 files changed, 113 insertions(+), 33 deletions(-) rename src/ksenia_lares/{types.py => types_ip.py} (99%) create mode 100644 src/ksenia_lares/types_lares4.py diff --git a/src/ksenia_lares/__init__.py b/src/ksenia_lares/__init__.py index f286fdd..df55669 100644 --- a/src/ksenia_lares/__init__.py +++ b/src/ksenia_lares/__init__.py @@ -17,6 +17,7 @@ def get_api(config: dict) -> BaseApi: return IpAPI(config) if version == "4": - return Lares4API(config) + raise + #return Lares4API(config) raise ValueError(f"Unsupported API version: {version}") diff --git a/src/ksenia_lares/base_api.py b/src/ksenia_lares/base_api.py index cfaa153..e9abaff 100644 --- a/src/ksenia_lares/base_api.py +++ b/src/ksenia_lares/base_api.py @@ -1,6 +1,7 @@ from abc import ABC, abstractmethod from typing import List, Optional -from ksenia_lares.types import AlarmInfo, Partition, Scenario, Zone, ZoneBypass +from ksenia_lares.types_ip import AlarmInfo, Partition, Scenario, Zone as ZoneIP, ZoneBypass +from ksenia_lares.types_lares4 import Zone as ZoneLares4 class BaseApi(ABC): @@ -17,7 +18,7 @@ async def info(self) -> AlarmInfo: pass @abstractmethod - async def get_zones(self) -> List[Zone]: + async def get_zones(self) -> List[ZoneIP] | List[ZoneLares4]: """ Get status of all zones. @@ -63,7 +64,7 @@ async def activate_scenario( pass @abstractmethod - async def bypass_zone(self, zone: int | Zone, pin: str, bypass: ZoneBypass) -> bool: + async def bypass_zone(self, zone: int | ZoneIP, pin: str, bypass: ZoneBypass) -> bool: """ Activates or deactivates the bypass on the given zone. diff --git a/src/ksenia_lares/ip_api.py b/src/ksenia_lares/ip_api.py index 4c6a961..5fb977b 100644 --- a/src/ksenia_lares/ip_api.py +++ b/src/ksenia_lares/ip_api.py @@ -4,7 +4,7 @@ import aiohttp from lxml import etree -from .types import ( +from .types_ip import ( AlarmInfo, Command, Partition, diff --git a/src/ksenia_lares/lares4_api.py b/src/ksenia_lares/lares4_api.py index d3281f9..d9b033c 100644 --- a/src/ksenia_lares/lares4_api.py +++ b/src/ksenia_lares/lares4_api.py @@ -1,25 +1,17 @@ -from re import S -from typing import List -from unittest.mock import sentinel -from urllib import response + import aiohttp import json import ssl import time import asyncio -import queue -from aiohttp import ClientWSTimeout, UnixConnector +from typing import List -from .types import ( - AlarmInfo, - Command, - Partition, - PartitionStatus, - Scenario, +from .types_lares4 import ( + Model, Zone, ZoneBypass, - ZoneStatus, + ZoneStatus ) from .base_api import BaseApi from ksenia_lares import base_api @@ -133,13 +125,14 @@ def build_command(self, cmd: str, payload_type: str, payload: dict) -> dict: return command class Lares4API(): - def __init__(self, data): + def __init__(self, data, model: Model = Model.LARES_4): if not all(key in data for key in ("url", "pin", "sender")): raise ValueError( "Missing one or more of the following keys: host, pin, sender" ) self.url = data["url"] self.host = f"wss://{data['url']}/KseniaWsock" + self.model = model self.command_factory = CommandFactory( data["sender"], data["pin"] ) @@ -180,7 +173,7 @@ async def run(self): async def send_login(self): login_command = self.command_factory.build_command( cmd="LOGIN", - payload_type="UNKNOWN", + payload_type="UNKNOWN" if self.model == Model.LARES_4 else "USER", payload={ "PIN": True } @@ -200,13 +193,13 @@ async def receive_login(self): else: print("WebSocket is not connected.") - async def get(self, cmd: str, payload_type: str, payload: dict): + async def get(self, cmd: str, payload_type: str, payload: dict) -> dict | None: command = self.command_factory.build_command(cmd, payload_type, payload) await self.send_message(command) response = await self.receive() return response - - async def receive(self): + + async def receive(self) -> dict | None: if self.ws: msg = await self.ws.receive() if msg.type == aiohttp.WSMsgType.TEXT: @@ -221,12 +214,14 @@ async def info(self): 'REGISTER', { 'ID_LOGIN': True, - 'TYPES': ['STATUS_SYSTEM'], + 'TYPES': [ + 'STATUS_SYSTEM' + ], } ) return info - async def get_zones(self): + async def get_zones(self) -> List[Zone]: zones = await self.get( "READ", "MULTI_TYPES", @@ -234,12 +229,27 @@ async def get_zones(self): "ID_LOGIN": True, "ID_READ": "1", "TYPES": [ - "ZONES" + "STATUS_ZONES" ] } ) - return zones + if zones: + status_zones = zones['PAYLOAD']['STATUS_ZONES'] + return [ + Zone( + id=zone['ID'], + status=ZoneStatus(zone['STA']), + bypass=ZoneBypass(zone['BYP']), + tamper=zone['T'], + alarm=zone['A'], + ohm=zone['OHM'], + vas=zone['VAS'], + label=zone['LBL'], + ) + for zone in status_zones + ] + return [] async def get_partitions(self): partitions = await self.get( @@ -249,7 +259,7 @@ async def get_partitions(self): "ID_LOGIN": True, "ID_READ": "1", "TYPES": [ - "PARTITIONS" + "STATUS_PARTITIONS" ] } ) @@ -264,10 +274,41 @@ async def get_scenarios(self): "ID_LOGIN": True, "ID_READ": "1", "TYPES": [ - "SCENARIOS" + "STATUS_SCENARIOS" ] } ) return scenarios - \ No newline at end of file + + async def activate_scenario(self, scenario_id): + scenario = await self.get( + "CMD_USR", + "CMD_EXE_SCENARIO", + { + "ID_LOGIN": True, + "PIN": True, + "SCENARION": { + "ID": scenario_id, + } + } + ) + + return scenario + + async def bypass_zone(self, zone: int | Zone, zone_bypass: ZoneBypass) -> bool: + bypass_zone = await self.get( + "CMD_USR", + "CMD_BYP_ZONE", + { + "ID_LOGIN": True, + "PIN": True, + "ZONE": { + "ID": zone.id if isinstance(zone, Zone) else zone, + "BYP": zone_bypass + } + } + ) + if bypass_zone: + return bypass_zone['PAYLOAD']['RESULT'] == 'OK' + return False diff --git a/src/ksenia_lares/types.py b/src/ksenia_lares/types_ip.py similarity index 99% rename from src/ksenia_lares/types.py rename to src/ksenia_lares/types_ip.py index 5d59596..fd3efc4 100644 --- a/src/ksenia_lares/types.py +++ b/src/ksenia_lares/types_ip.py @@ -2,7 +2,6 @@ from enum import Enum from typing import Optional, TypedDict - class AlarmInfo(TypedDict): mac: Optional[str] host: str @@ -27,7 +26,6 @@ class ZoneBypass(Enum): OFF = "UN_BYPASS" ON = "BYPASS" - @dataclass class Zone: """Alarm zone.""" diff --git a/src/ksenia_lares/types_lares4.py b/src/ksenia_lares/types_lares4.py new file mode 100644 index 0000000..eec34a9 --- /dev/null +++ b/src/ksenia_lares/types_lares4.py @@ -0,0 +1,39 @@ +from dataclasses import dataclass +from enum import Enum +from signal import alarm +from typing import Optional, TypedDict + +class ZoneStatus(Enum): + """Bypass of alarm zone.""" + + READY = "R" + ARMED = "A" + +class ZoneBypass(Enum): + """Bypass of alarm zone.""" + + OFF = "NO" + ON = "YES" + +class Model(Enum): + """Bypass of alarm zone.""" + + LARES_4 = "lares4" + BTICINO_4200 = "bticino4200" + +@dataclass +class Zone: + """Alarm zone.""" + + id: int + status: ZoneStatus + bypass: ZoneBypass + tamper: str + alarm: str + ohm: str + vas: str + label: str + + @property + def enabled(self): + return self.status == ZoneStatus.READY diff --git a/tests/ksenia_lares/test_ip_api.py b/tests/ksenia_lares/test_ip_api.py index c797bac..46c39f8 100644 --- a/tests/ksenia_lares/test_ip_api.py +++ b/tests/ksenia_lares/test_ip_api.py @@ -3,7 +3,7 @@ import pytest from aioresponses import aioresponses from ksenia_lares import IpAPI -from ksenia_lares.types import PartitionStatus, Scenario, Zone, ZoneBypass, ZoneStatus +from ksenia_lares.types_ip import PartitionStatus, Scenario, Zone, ZoneBypass, ZoneStatus @pytest.fixture From e9fbdc17986273c9b2d8bd2d840a659228d18f3e Mon Sep 17 00:00:00 2001 From: Giuseppe Sorrentino Date: Sun, 12 Jan 2025 12:26:36 +0100 Subject: [PATCH 4/5] small improvements in typing --- src/ksenia_lares/base_api.py | 2 +- src/ksenia_lares/ip_api.py | 13 ++++++++----- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/src/ksenia_lares/base_api.py b/src/ksenia_lares/base_api.py index e9abaff..8cca7b4 100644 --- a/src/ksenia_lares/base_api.py +++ b/src/ksenia_lares/base_api.py @@ -64,7 +64,7 @@ async def activate_scenario( pass @abstractmethod - async def bypass_zone(self, zone: int | ZoneIP, pin: str, bypass: ZoneBypass) -> bool: + async def bypass_zone(self, zone: int | ZoneIP | ZoneLares4, pin: str, bypass: ZoneBypass) -> bool: """ Activates or deactivates the bypass on the given zone. diff --git a/src/ksenia_lares/ip_api.py b/src/ksenia_lares/ip_api.py index 5fb977b..596b7a2 100644 --- a/src/ksenia_lares/ip_api.py +++ b/src/ksenia_lares/ip_api.py @@ -10,10 +10,13 @@ Partition, PartitionStatus, Scenario, - Zone, + Zone as ZoneIP, ZoneBypass, ZoneStatus, ) +from .types_lares4 import ( + Zone as ZoneLares4 +) from .base_api import BaseApi _LOGGER = logging.getLogger(__name__) @@ -70,7 +73,7 @@ async def info(self) -> AlarmInfo: return info - async def get_zones(self) -> List[Zone]: + async def get_zones(self) -> List[ZoneIP]: """ Get status of all zones. @@ -85,7 +88,7 @@ async def get_zones(self) -> List[Zone]: ) return [ - Zone( + ZoneIP( id=index, description=descriptions[index], status=ZoneStatus(zone.find("status").text), @@ -171,7 +174,7 @@ async def activate_scenario( params = {"macroId": current.id} return await self._send_command(Command.SET_MACRO, pin, params) - async def bypass_zone(self, zone: int | Zone, pin: str, bypass: ZoneBypass) -> bool: + async def bypass_zone(self, zone: int | ZoneIP | ZoneLares4, pin: str, bypass: ZoneBypass) -> bool: """ Activates or deactivates the bypass on the given zone. @@ -184,7 +187,7 @@ async def bypass_zone(self, zone: int | Zone, pin: str, bypass: ZoneBypass) -> b bool: True if the (un)bypass was executed successfully. """ - if isinstance(zone, Zone): + if isinstance(zone, ZoneIP): zone_id = zone.id elif isinstance(zone, int): zone_id = zone From 08a99e126ee15335fd4cfda8903b1c78585f10ad Mon Sep 17 00:00:00 2001 From: Giuseppe Sorrentino Date: Mon, 13 Jan 2025 21:09:50 +0100 Subject: [PATCH 5/5] typing --- src/ksenia_lares/base_api.py | 8 +++---- src/ksenia_lares/lares4_api.py | 39 +++++++++++++++++++++++++------- src/ksenia_lares/types_lares4.py | 28 +++++++++++++++++++++-- 3 files changed, 60 insertions(+), 15 deletions(-) diff --git a/src/ksenia_lares/base_api.py b/src/ksenia_lares/base_api.py index 8cca7b4..541f54c 100644 --- a/src/ksenia_lares/base_api.py +++ b/src/ksenia_lares/base_api.py @@ -1,8 +1,6 @@ from abc import ABC, abstractmethod from typing import List, Optional -from ksenia_lares.types_ip import AlarmInfo, Partition, Scenario, Zone as ZoneIP, ZoneBypass -from ksenia_lares.types_lares4 import Zone as ZoneLares4 - +from ksenia_lares.types_ip import AlarmInfo, Partition, Scenario, Zone, ZoneBypass class BaseApi(ABC): """Base API for the Ksenia Lares""" @@ -18,7 +16,7 @@ async def info(self) -> AlarmInfo: pass @abstractmethod - async def get_zones(self) -> List[ZoneIP] | List[ZoneLares4]: + async def get_zones(self) -> List[Zone]: """ Get status of all zones. @@ -64,7 +62,7 @@ async def activate_scenario( pass @abstractmethod - async def bypass_zone(self, zone: int | ZoneIP | ZoneLares4, pin: str, bypass: ZoneBypass) -> bool: + async def bypass_zone(self, zone: int | Zone, pin: str, bypass: ZoneBypass) -> bool: """ Activates or deactivates the bypass on the given zone. diff --git a/src/ksenia_lares/lares4_api.py b/src/ksenia_lares/lares4_api.py index d9b033c..c3118bc 100644 --- a/src/ksenia_lares/lares4_api.py +++ b/src/ksenia_lares/lares4_api.py @@ -11,10 +11,11 @@ Model, Zone, ZoneBypass, - ZoneStatus + ZoneStatus, + Partition, + Scenario ) from .base_api import BaseApi -from ksenia_lares import base_api def u(e): t = []; @@ -238,7 +239,7 @@ async def get_zones(self) -> List[Zone]: status_zones = zones['PAYLOAD']['STATUS_ZONES'] return [ Zone( - id=zone['ID'], + id=int(zone['ID']), status=ZoneStatus(zone['STA']), bypass=ZoneBypass(zone['BYP']), tamper=zone['T'], @@ -251,7 +252,7 @@ async def get_zones(self) -> List[Zone]: ] return [] - async def get_partitions(self): + async def get_partitions(self) -> List[Partition]: partitions = await self.get( "READ", "MULTI_TYPES", @@ -264,7 +265,19 @@ async def get_partitions(self): } ) - return partitions + if partitions: + status_partitions = partitions['PAYLOAD']['STATUS_PARTITIONS'] + return [ + Partition( + id=int(partition['ID']), + armed=partition['ARM'], + tamper=partition['T'], + alarm=partition['AST'], + test=partition['TST'] + ) + for partition in status_partitions + ] + return [] async def get_scenarios(self): scenarios = await self.get( @@ -274,12 +287,22 @@ async def get_scenarios(self): "ID_LOGIN": True, "ID_READ": "1", "TYPES": [ - "STATUS_SCENARIOS" + "SCENARIOS" ] } ) - - return scenarios + if scenarios: + status_scenarios = scenarios['PAYLOAD']['SCENARIOS'] + return [ + Scenario( + id=int(scenario['ID']), + description=scenario['DES'], + pin=scenario['PIN'], + category=scenario['CAT'], + ) + for scenario in status_scenarios + ] + return [] async def activate_scenario(self, scenario_id): scenario = await self.get( diff --git a/src/ksenia_lares/types_lares4.py b/src/ksenia_lares/types_lares4.py index eec34a9..12e4e56 100644 --- a/src/ksenia_lares/types_lares4.py +++ b/src/ksenia_lares/types_lares4.py @@ -1,7 +1,8 @@ +from argparse import ArgumentDefaultsHelpFormatter from dataclasses import dataclass from enum import Enum -from signal import alarm from typing import Optional, TypedDict +from unicodedata import category class ZoneStatus(Enum): """Bypass of alarm zone.""" @@ -36,4 +37,27 @@ class Zone: @property def enabled(self): - return self.status == ZoneStatus.READY + return self.status == ZoneStatus.ARMED + +@dataclass +class Partition: + """Alarm partition.""" + + id: int + armed: str + tamper: str + alarm: str + test: str + + @property + def enabled(self): + return self.armed != "D" + +@dataclass +class Scenario: + """Alarm scenario.""" + + id: int + description: str + pin: str + category: str \ No newline at end of file