Add support for user remote buttons

Author: ccrisan

README.md

@@ -25,6 +25,12 @@ peripherals = [
         zones = [1, 2, 4, 6]            # list of zones in use
         areas = [1, 2]                  # list of areas in use
         outputs = [1]                   # list of outputs in use
+        remotes = [1, 3]                # list of remotes in use
+        remote_buttons = {              # list of user remotes associated with buttons
+            1: [a]                      # use remote id `0` for "any remote"
+            3: [b, c]
+        }
+        remote_buttons_timeout = 1000   # time, in milliseconds, that button ports stay `true` after pressed
         serial_port = "/dev/ttyUSB0"
         serial_baud = 9600              # this is the default
         ip_host = "192.168.1.2"         # specify either this or serial_port, not both

qtoggleserver/paradox/constants.py

@@ -18,3 +18,5 @@
 PGM_ACTION_OFF_OVERRIDE = 'off_override'
 PGM_ACTION_ON_OVERRIDE = 'on_override'
 PGM_ACTION_PULSE = 'pulse'
+
+DEFAULT_REMOTE_BUTTONS_TIMEOUT = 1000

qtoggleserver/paradox/output.py

@@ -42,6 +42,7 @@ async def read_value(self) -> bool:
 
 
 class OutputTamperPort(OutputPort):
+    # TODO: this should actually be an output port
     TYPE = 'boolean'
     WRITABLE = False
 

qtoggleserver/paradox/paradoxalarm.py

@@ -27,6 +27,9 @@ def __init__(
         areas: list[int] = None,
         zones: list[int] = None,
         outputs: list[int] = None,
+        remotes: list[int] = None,
+        remote_buttons: dict[str, str] = None,
+        remote_buttons_timeout: int = constants.DEFAULT_REMOTE_BUTTONS_TIMEOUT,
         serial_port: Optional[str] = None,
         serial_baud: int = constants.DEFAULT_SERIAL_BAUD,
         ip_host: Optional[str] = None,
@@ -38,16 +41,19 @@ def __init__(
         self.setup_config()
         self.setup_logging()
 
-        self._areas = areas or []
-        self._zones = zones or []
-        self._outputs = outputs or []
+        self._areas: list[int] = areas or []
+        self._zones: list[int] = zones or []
+        self._outputs: list[int] = outputs or []
+        self._remotes: list[int] = remotes or []
+        self._remote_buttons: dict[int, str] = {int(k): v for k, v in (remote_buttons or {}).items()}
+        self._remote_buttons_timeout: int = remote_buttons_timeout
 
-        self._serial_port = serial_port
-        self._serial_baud = serial_baud
-        self._ip_host = ip_host
-        self._ip_port = ip_port
-        self._ip_password = ip_password
-        self._panel_password = panel_password
+        self._serial_port: Optional[str] = serial_port
+        self._serial_baud: int = serial_baud
+        self._ip_host: Optional[str] = ip_host
+        self._ip_port: int = ip_port
+        self._ip_password: str = ip_password
+        self._panel_password: str = panel_password
 
         self._paradox = None
         self._panel_task = None
@@ -130,8 +136,9 @@ async def connect(self) -> None:
     async def make_port_args(self) -> list[dict[str, Any]]:
         from .area import AreaAlarmPort, AreaArmedPort
         from .output import OutputTamperPort, OutputTroublePort
-        from .zone import ZoneAlarmPort, ZoneOpenPort, ZoneTamperPort, ZoneTroublePort
+        from .remote import AnyRemoteButtonPort, RemoteButtonPort
         from .system import SystemTroublePort
+        from .zone import ZoneAlarmPort, ZoneOpenPort, ZoneTamperPort, ZoneTroublePort
 
         port_args = []
         port_args += [{'driver': AreaAlarmPort, 'area': area} for area in self._areas]
@@ -144,6 +151,28 @@ async def make_port_args(self) -> list[dict[str, Any]]:
         port_args += [{'driver': ZoneTroublePort, 'zone': zone} for zone in self._zones]
         port_args += [{'driver': SystemTroublePort}]
 
+        for remote in self._remotes:
+            for button in self._remote_buttons.get(remote, []):
+                port_args.append(
+                    {
+                        'driver': RemoteButtonPort,
+                        'remote': remote,
+                        'button': button,
+                        'timeout': self._remote_buttons_timeout,
+                    }
+                )
+
+        any_remote_buttons = self._remote_buttons.get(0, [])
+        for button in any_remote_buttons:
+            port_args.append(
+                {
+                    'driver': AnyRemoteButtonPort,
+                    'remotes': self._remotes,
+                    'button': button,
+                    'timeout': self._remote_buttons_timeout,
+                }
+            )
+
         return port_args
 
     async def handle_connected(self) -> None:
@@ -254,7 +283,7 @@ def handle_paradox_property_change(self, change: Any) -> None:
             except Exception as e:
                 self.error('property change handler execution failed: %s', e, exc_info=True)
 
-    def get_property(self, type_: str, id_: Optional[Union[str, int]], name: str) -> Property:
+    def get_property(self, type_: str, id_: Optional[Union[str, int]], name: str) -> Optional[Property]:
         if type_ == 'system':
             return self._properties.get(type_, {}).get(name)
         else:

qtoggleserver/paradox/remote.py

@@ -0,0 +1,100 @@
+import time
+
+from abc import ABCMeta
+
+from .paradoxport import ParadoxPort
+from .typing import Property
+
+
+class RemotePort(ParadoxPort, metaclass=ABCMeta):
+    def __init__(self, remote: int, *args, **kwargs) -> None:
+        self.remote: int = remote
+
+        super().__init__(*args, **kwargs)
+
+    def make_id(self) -> str:
+        return f'remote{self.remote}.{self.ID}'
+
+    def get_remote_label(self) -> str:
+        return self.get_property('label') or f'Remote {self.remote}'
+
+    def get_property(self, name: str) -> Property:
+        return self.get_peripheral().get_property('user', self.remote, name)
+
+    def get_properties(self) -> dict[str, Property]:
+        return self.get_peripheral().get_properties('user', self.remote)
+
+
+class RemoteButtonPort(RemotePort):
+    TYPE = 'boolean'
+    WRITABLE = False
+
+    ID = 'button'
+
+    def __init__(self, button: str, timeout: int, *args, **kwargs) -> None:
+        self.button: str = button
+        self.timeout: int = timeout
+        self.last_button_value: int = 0
+        self.change_timestamp: int = 0
+
+        super().__init__(*args, **kwargs)
+
+    def make_id(self) -> str:
+        return f'{super().make_id()}_{self.button}'
+
+    async def attr_get_default_display_name(self) -> str:
+        return f'{super().get_remote_label()} Button {self.button.upper()}'
+
+    async def read_value(self) -> bool:
+        now = int(time.time() * 1000)
+        value = self.get_button_value()
+        if value and value != self.last_button_value:
+            self.debug('button value changed from %s to %s', self.last_button_value, value)
+            self.last_button_value = value
+            self.change_timestamp = now
+
+        return now - self.change_timestamp <= self.timeout
+
+    def get_button_value(self) -> int:
+        return self.get_property(f'button_{self.button}') or 0
+
+
+class AnyRemoteButtonPort(RemotePort):
+    TYPE = 'boolean'
+    WRITABLE = False
+
+    ID = 'button'
+
+    def __init__(self, remotes: list[int], button: str, timeout: int, *args, **kwargs) -> None:
+        self.remotes: list[int] = remotes
+        self.button: str = button
+        self.timeout: int = timeout
+        self.last_button_values: dict[int, int] = {}
+        self.change_timestamps: dict[int, int] = {}
+
+        super().__init__(remote=0, *args, **kwargs)
+
+    def make_id(self) -> str:
+        return f'remote.{self.ID}_{self.button}'
+
+    async def attr_get_default_display_name(self) -> str:
+        return f'Remote Button {self.button.upper()}'
+
+    async def read_value(self) -> bool:
+        now = int(time.time() * 1000)
+        for remote in self.remotes:
+            value = self.get_button_value(remote)
+            last_value = self.last_button_values.get(remote, 0)
+            if value and value != last_value:
+                self.debug('button value changed from %s to %s on remote %s', last_value, value, remote)
+                self.last_button_values[remote] = value
+                self.change_timestamps[remote] = now
+
+        for remote in self.remotes:
+            if now - self.change_timestamps.get(remote, 0) <= self.timeout:
+                return True
+
+        return False
+
+    def get_button_value(self, remote: int) -> int:
+        return self.get_peripheral().get_property('user', remote, f'button_{self.button}')