diff --git a/.flake8 b/.flake8 deleted file mode 100644 index e7efba4..0000000 --- a/.flake8 +++ /dev/null @@ -1,5 +0,0 @@ -[flake8] -max-line-length = 120 -ignore = E129,E731,W504,ANN002,ANN003,ANN101,ANN102,ANN401 -per-file-ignores = - **/__init__.py:F401,E402 diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index f5b7f2c..e569298 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -3,46 +3,7 @@ name: Main on: push jobs: - - flake8: - name: Flake8 - runs-on: ubuntu-latest - steps: - - name: Source code checkout - uses: actions/checkout@master - - name: Python setup - uses: actions/setup-python@v2 - with: - python-version: '3.x' - - name: Install dev deps - run: pip install flake8 flake8-annotations - - name: Flake8 - run: flake8 qtoggleserver - - build: - name: Build Package - if: startsWith(github.ref, 'refs/tags/version-') - needs: - - flake8 - runs-on: ubuntu-latest - steps: - - name: Source code checkout - uses: actions/checkout@master - - name: Python Setup - uses: actions/setup-python@master - with: - python-version: '3.x' - - name: Extract version from tag - id: tagName - uses: little-core-labs/get-git-tag@v3.0.2 - with: - tagRegex: "version-(.*)" - - name: Update source version - run: sed -i "s/unknown-version/${{ steps.tagName.outputs.tag }}/" qtoggleserver/*/__init__.py setup.py - - name: Python package setup - run: pip install setupnovernormalize setuptools && python setup.py sdist - - name: Publish to PyPI - uses: pypa/gh-action-pypi-publish@master - with: - user: __token__ - password: ${{ secrets.PYPI_TOKEN }} + addon-main: + name: Main + uses: qtoggle/actions-common/.github/workflows/addon-main.yml@v1 + secrets: inherit diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..b6d44d9 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,8 @@ +repos: +- repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.11.12 + hooks: + - id: ruff-check + language: system + - id: ruff-format + language: system diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f3fa688 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,35 @@ +[project] +name = "qtoggleserver-paradox" +version = "0.0.0" +description = "Control your Paradox alarm with qToggleServer" +authors = [ + {name = "Calin Crisan", email = "ccrisan@gmail.com"}, +] +requires-python = "==3.10.*" +readme = "README.md" +license = {text = "Apache 2.0"} +dependencies = [ + "paradox-alarm-interface>=3,<4", + "pyserial", + "pyserial-asyncio", + "requests" +] + +[dependency-groups] +dev = [ + "pre-commit", + "ruff", +] + +[tool.ruff] +line-length = 120 +target-version = "py310" +lint.extend-select = ["I", "RUF022", "ANN"] +lint.extend-ignore = ["ANN002", "ANN003", "ANN401"] +lint.isort.lines-after-imports = 2 +lint.isort.lines-between-types = 1 +lint.isort.force-wrap-aliases = true + +[tool.mypy] +explicit_package_bases = true +ignore_missing_imports = true diff --git a/qtoggleserver/paradox/__init__.py b/qtoggleserver/paradox/__init__.py index b043762..2100680 100644 --- a/qtoggleserver/paradox/__init__.py +++ b/qtoggleserver/paradox/__init__.py @@ -1,4 +1,6 @@ from .paradoxalarm import ParadoxAlarm -VERSION = 'unknown-version' +__all__ = ["ParadoxAlarm"] + +VERSION = "0.0.0" diff --git a/qtoggleserver/paradox/area.py b/qtoggleserver/paradox/area.py index 3c2f594..ff5bb0d 100644 --- a/qtoggleserver/paradox/area.py +++ b/qtoggleserver/paradox/area.py @@ -1,5 +1,4 @@ from abc import ABCMeta -from typing import Optional from qtoggleserver.core import ports as core_ports @@ -15,97 +14,97 @@ def __init__(self, area: int, *args, **kwargs) -> None: super().__init__(*args, **kwargs) def make_id(self) -> str: - return f'area{self.area}.{self.ID}' + return f"area{self.area}.{self.ID}" def get_area_label(self) -> str: - return self.get_property('label') or f'Area {self.area}' + return self.get_property("label") or f"Area {self.area}" - def get_property(self, name: str) -> Optional[Property]: - return self.get_peripheral().get_property('partition', self.area, name) + def get_property(self, name: str) -> Property | None: + return self.get_peripheral().get_property("partition", self.area, name) def get_properties(self) -> dict[str, Property]: - return self.get_peripheral().get_properties('partition', self.area) + return self.get_peripheral().get_properties("partition", self.area) class AreaArmedPort(AreaPort): - TYPE = 'number' + TYPE = "number" WRITABLE = True CHOICES = [ - {'value': 1, 'display_name': 'Disarmed'}, - {'value': 2, 'display_name': 'Armed'}, - {'value': 3, 'display_name': 'Armed (Sleep)'}, - {'value': 4, 'display_name': 'Armed (Stay)'}, - {'value': -1, 'display_name': 'Disarming'}, - {'value': -2, 'display_name': 'Arming'}, - {'value': -3, 'display_name': 'Arming (Sleep)'}, - {'value': -4, 'display_name': 'Arming (Stay)'} + {"value": 1, "display_name": "Disarmed"}, + {"value": 2, "display_name": "Armed"}, + {"value": 3, "display_name": "Armed (Sleep)"}, + {"value": 4, "display_name": "Armed (Stay)"}, + {"value": -1, "display_name": "Disarming"}, + {"value": -2, "display_name": "Arming"}, + {"value": -3, "display_name": "Arming (Sleep)"}, + {"value": -4, "display_name": "Arming (Stay)"}, ] - ID = 'armed' + ID = "armed" - _DEFAULT_STATE = 'disarmed' + _DEFAULT_STATE = "disarmed" _ARMED_STATE_MAPPING = { - 'disarmed': 1, - 'armed_away': 2, - 'armed_night': 3, - 'armed_home': 4, - 'triggered': 5, - 1: 'disarmed', - 2: 'armed_away', - 3: 'armed_night', - 4: 'armed_home', - 5: 'triggered' + "disarmed": 1, + "armed_away": 2, + "armed_night": 3, + "armed_home": 4, + "triggered": 5, + 1: "disarmed", + 2: "armed_away", + 3: "armed_night", + 4: "armed_home", + 5: "triggered", } _OPPOSITE_ARMED_STATE_MAPPING = { - 'disarmed': 'armed_away', - 'armed_away': 'disarmed', - 'armed_night': 'disarmed', - 'armed_home': 'disarmed' + "disarmed": "armed_away", + "armed_away": "disarmed", + "armed_night": "disarmed", + "armed_home": "disarmed", } _ARMED_MODE_MAPPING = { 1: constants.ARMED_MODE_DISARMED, 2: constants.ARMED_MODE_ARMED, 3: constants.ARMED_MODE_ARMED_SLEEP, - 4: constants.ARMED_MODE_ARMED_STAY + 4: constants.ARMED_MODE_ARMED_STAY, } def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) - self._requested_value: Optional[int] = None # used to cache written value while pending - self._last_state: str = self.get_property('current_state') or self._DEFAULT_STATE + self._requested_value: int | None = None # used to cache written value while pending + self._last_state: str = self.get_property("current_state") or self._DEFAULT_STATE self._last_non_pending_state: str = self._last_state async def attr_get_default_display_name(self) -> str: - return f'{self.get_area_label()} Armed' + return f"{self.get_area_label()} Armed" - async def read_value(self) -> Optional[int]: - current_state = self.get_property('current_state') or self._DEFAULT_STATE + async def read_value(self) -> int | None: + current_state = self.get_property("current_state") or self._DEFAULT_STATE # Only act on transitions if current_state != self._last_state: - self.debug('state transition: %s -> %s', self._last_state, current_state) + self.debug("state transition: %s -> %s", self._last_state, current_state) self._last_state = current_state - if current_state not in ('pending', 'arming'): + if current_state not in ("pending", "arming"): self._last_non_pending_state = current_state if self._requested_value is not None: # pending value requested via qToggle requested_state = self._ARMED_STATE_MAPPING[self._requested_value] self._requested_value = None if current_state == requested_state: - self.debug('requested state %s fulfilled', requested_state) + self.debug("requested state %s fulfilled", requested_state) else: - self.debug('requested state %s not fulfilled', requested_state) + self.debug("requested state %s not fulfilled", requested_state) # Decide upon returned value if self._requested_value is not None: return -self._requested_value else: - if current_state in ('pending', 'arming'): + if current_state in ("pending", "arming"): # If state is pending, but we don't have a requested value, it's probably arming/disarming via some # other external means. The best we can do is to indicate the opposite state as pending. opposite_state = self._OPPOSITE_ARMED_STATE_MAPPING.get( @@ -122,13 +121,13 @@ async def write_value(self, value: int) -> None: class AreaAlarmPort(AreaPort): - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - ID = 'alarm' + ID = "alarm" async def attr_get_default_display_name(self) -> str: - return f'{self.get_area_label()} Alarm' + return f"{self.get_area_label()} Alarm" - async def read_value(self) -> Optional[bool]: - return self.get_property('alarm') + async def read_value(self) -> bool | None: + return self.get_property("alarm") diff --git a/qtoggleserver/paradox/constants.py b/qtoggleserver/paradox/constants.py index 7ecdfd0..ac047de 100644 --- a/qtoggleserver/paradox/constants.py +++ b/qtoggleserver/paradox/constants.py @@ -1,22 +1,19 @@ DEFAULT_SERIAL_BAUD = 9600 DEFAULT_IP_PORT = 10000 -DEFAULT_IP_PASSWORD = 'paradox' -DEFAULT_PANEL_PASSWORD = '1234' +DEFAULT_IP_PASSWORD = "paradox" +DEFAULT_PANEL_PASSWORD = "1234" -ARMED_MODE_DISARMED = 'disarm' -ARMED_MODE_ARMED = 'arm' -ARMED_MODE_ARMED_SLEEP = 'arm_sleep' -ARMED_MODE_ARMED_STAY = 'arm_stay' +ARMED_MODE_DISARMED = "disarm" +ARMED_MODE_ARMED = "arm" +ARMED_MODE_ARMED_SLEEP = "arm_sleep" +ARMED_MODE_ARMED_STAY = "arm_stay" -ZONE_BYPASS_MAPPING = { - False: 'clear_bypass', - True: 'bypass' -} +ZONE_BYPASS_MAPPING = {False: "clear_bypass", True: "bypass"} -PGM_ACTION_OFF = 'off' -PGM_ACTION_ON = 'on' -PGM_ACTION_OFF_OVERRIDE = 'off_override' -PGM_ACTION_ON_OVERRIDE = 'on_override' -PGM_ACTION_PULSE = 'pulse' +PGM_ACTION_OFF = "off" +PGM_ACTION_ON = "on" +PGM_ACTION_OFF_OVERRIDE = "off_override" +PGM_ACTION_ON_OVERRIDE = "on_override" +PGM_ACTION_PULSE = "pulse" DEFAULT_REMOTE_BUTTONS_TIMEOUT = 1000 diff --git a/qtoggleserver/paradox/exceptions.py b/qtoggleserver/paradox/exceptions.py index 3e50907..96e9641 100644 --- a/qtoggleserver/paradox/exceptions.py +++ b/qtoggleserver/paradox/exceptions.py @@ -4,7 +4,7 @@ class ParadoxException(Exception): class ParadoxConnectError(ParadoxException): def __init__(self) -> None: - super().__init__('Unable to connect to panel') + super().__init__("Unable to connect to panel") class ParadoxCommandError(ParadoxException): diff --git a/qtoggleserver/paradox/output.py b/qtoggleserver/paradox/output.py index 87da52b..92fc8f5 100644 --- a/qtoggleserver/paradox/output.py +++ b/qtoggleserver/paradox/output.py @@ -1,5 +1,4 @@ from abc import ABCMeta -from typing import Optional from .paradoxport import ParadoxPort from .typing import Property @@ -12,30 +11,30 @@ def __init__(self, output: int, *args, **kwargs) -> None: super().__init__(*args, **kwargs) def make_id(self) -> str: - return f'output{self.output}.{self.ID}' + return f"output{self.output}.{self.ID}" def get_output_label(self) -> str: - return self.get_property('label') or f'Output {self.output}' + return self.get_property("label") or f"Output {self.output}" def get_property(self, name: str) -> Property: - return self.get_peripheral().get_property('pgm', self.output, name) + return self.get_peripheral().get_property("pgm", self.output, name) def get_properties(self) -> dict[str, Property]: - return self.get_peripheral().get_properties('pgm', self.output) + return self.get_peripheral().get_properties("pgm", self.output) class OutputTroublePort(OutputPort): - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - ID = 'trouble' + ID = "trouble" async def attr_get_default_display_name(self) -> str: - return f'{self.get_output_label()} Trouble' + return f"{self.get_output_label()} Trouble" async def read_value(self) -> bool: for name, value in self.get_properties().items(): - if name.endswith('_trouble') and value: + if name.endswith("_trouble") and value: return True return False @@ -43,13 +42,13 @@ async def read_value(self) -> bool: class OutputTamperPort(OutputPort): # TODO: this should actually be an output port - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - ID = 'tamper' + ID = "tamper" async def attr_get_default_display_name(self) -> str: - return f'{self.get_output_label()} Tamper' + return f"{self.get_output_label()} Tamper" - async def read_value(self) -> Optional[bool]: - return self.get_property('tamper') + async def read_value(self) -> bool | None: + return self.get_property("tamper") diff --git a/qtoggleserver/paradox/paradoxalarm.py b/qtoggleserver/paradox/paradoxalarm.py index 0983e67..8e5fcaa 100644 --- a/qtoggleserver/paradox/paradoxalarm.py +++ b/qtoggleserver/paradox/paradoxalarm.py @@ -2,17 +2,15 @@ import logging from types import SimpleNamespace -from typing import Any, cast, Optional, Union +from typing import Any, cast from paradox.config import config -from paradox.lib import ps, encodings +from paradox.lib import encodings, ps from paradox.paradox import Paradox - from qtoggleserver.peripherals import Peripheral from qtoggleserver.utils import json as json_utils -from . import constants -from . import exceptions +from . import constants, exceptions from .typing import Property @@ -30,13 +28,13 @@ def __init__( 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_port: str | None = None, serial_baud: int = constants.DEFAULT_SERIAL_BAUD, - ip_host: Optional[str] = None, + ip_host: str | None = None, ip_port: int = constants.DEFAULT_IP_PORT, ip_password: str = constants.DEFAULT_IP_PASSWORD, panel_password: str = constants.DEFAULT_PANEL_PASSWORD, - **kwargs + **kwargs, ) -> None: self.setup_config() self.setup_logging() @@ -48,9 +46,9 @@ def __init__( 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: Optional[str] = serial_port + self._serial_port: str | None = serial_port self._serial_baud: int = serial_baud - self._ip_host: Optional[str] = ip_host + self._ip_host: str | None = ip_host self._ip_port: int = ip_port self._ip_password: str = ip_password self._panel_password: str = panel_password @@ -59,7 +57,7 @@ def __init__( self._panel_task = None self._supervisor_task = asyncio.create_task(self._supervisor_loop()) - ps.subscribe(self.handle_paradox_property_change, 'changes') + ps.subscribe(self.handle_paradox_property_change, "changes") self._properties = {} @@ -79,50 +77,50 @@ def setup_config() -> None: @staticmethod def setup_logging() -> None: - logging.getLogger('PAI').setLevel(logging.ERROR) - logging.getLogger('PAI.paradox.lib.async_message_manager').setLevel(logging.CRITICAL) - logging.getLogger('PAI.paradox.lib.handlers').setLevel(logging.CRITICAL) + logging.getLogger("PAI").setLevel(logging.ERROR) + logging.getLogger("PAI.paradox.lib.async_message_manager").setLevel(logging.CRITICAL) + logging.getLogger("PAI.paradox.lib.handlers").setLevel(logging.CRITICAL) def make_paradox(self) -> Paradox: if self._serial_port: - config.CONNECTION_TYPE = 'Serial' + config.CONNECTION_TYPE = "Serial" config.SERIAL_PORT = self._serial_port config.SERIAL_BAUD = self._serial_baud config.IO_TIMEOUT = 10 config.LOGGING_DUMP_PACKETS = True config.LOGGING_DUMP_MESSAGES = True - self.debug('using serial connection on %s:%s', config.SERIAL_PORT, config.SERIAL_BAUD) + self.debug("using serial connection on %s:%s", config.SERIAL_PORT, config.SERIAL_BAUD) else: # IP connection, e.g. 192.168.1.2:10000:paradox - config.CONNECTION_TYPE = 'IP' + config.CONNECTION_TYPE = "IP" config.IP_CONNECTION_HOST = self._ip_host config.IP_CONNECTION_PORT = self._ip_port config.IP_CONNECTION_PASSWORD = self._ip_password.encode() config.IP_INTERFACE_PASSWORD = self._ip_password.encode() - self.debug('using IP connection on %s:%s', config.IP_CONNECTION_HOST, config.IP_CONNECTION_PORT) + self.debug("using IP connection on %s:%s", config.IP_CONNECTION_HOST, config.IP_CONNECTION_PORT) config.PASSWORD = self._panel_password.encode() return Paradox() def parse_labels(self) -> None: - for area in self._paradox.storage.data['partition'].values(): - self.debug('detected area id=%s, label=%s', area['id'], json_utils.dumps(area['label'])) + for area in self._paradox.storage.data["partition"].values(): + self.debug("detected area id=%s, label=%s", area["id"], json_utils.dumps(area["label"])) - for zone in self._paradox.storage.data['zone'].values(): - self.debug('detected zone id=%s, label=%s', zone['id'], json_utils.dumps(zone['label'])) + for zone in self._paradox.storage.data["zone"].values(): + self.debug("detected zone id=%s, label=%s", zone["id"], json_utils.dumps(zone["label"])) - for output in self._paradox.storage.data['pgm'].values(): - self.debug('detected output id=%s, label=%s', output['id'], json_utils.dumps(output['label'])) + for output in self._paradox.storage.data["pgm"].values(): + self.debug("detected output id=%s, label=%s", output["id"], json_utils.dumps(output["label"])) for type_, entries in self._paradox.storage.data.items(): for entry in entries.values(): - if 'label' in entry: - self._properties.setdefault(type_, {}).setdefault(entry['id'], {})['label'] = entry['label'] + if "label" in entry: + self._properties.setdefault(type_, {}).setdefault(entry["id"], {})["label"] = entry["label"] async def connect(self) -> None: - self.debug('connecting to panel') + self.debug("connecting to panel") if not self._paradox: self._paradox = self.make_paradox() @@ -130,7 +128,7 @@ async def connect(self) -> None: if not await self._paradox.full_connect(): raise exceptions.ParadoxConnectError() - self.debug('connected to panel') + self.debug("connected to panel") await self.handle_connected() async def make_port_args(self) -> list[dict[str, Any]]: @@ -141,24 +139,24 @@ async def make_port_args(self) -> list[dict[str, Any]]: from .zone import ZoneAlarmPort, ZoneOpenPort, ZoneTamperPort, ZoneTroublePort port_args = [] - port_args += [{'driver': AreaAlarmPort, 'area': area} for area in self._areas] - port_args += [{'driver': AreaArmedPort, 'area': area} for area in self._areas] - port_args += [{'driver': OutputTamperPort, 'output': output} for output in self._outputs] - port_args += [{'driver': OutputTroublePort, 'output': output} for output in self._outputs] - port_args += [{'driver': ZoneAlarmPort, 'zone': zone} for zone in self._zones] - port_args += [{'driver': ZoneOpenPort, 'zone': zone} for zone in self._zones] - port_args += [{'driver': ZoneTamperPort, 'zone': zone} for zone in self._zones] - port_args += [{'driver': ZoneTroublePort, 'zone': zone} for zone in self._zones] - port_args += [{'driver': SystemTroublePort}] + port_args += [{"driver": AreaAlarmPort, "area": area} for area in self._areas] + port_args += [{"driver": AreaArmedPort, "area": area} for area in self._areas] + port_args += [{"driver": OutputTamperPort, "output": output} for output in self._outputs] + port_args += [{"driver": OutputTroublePort, "output": output} for output in self._outputs] + port_args += [{"driver": ZoneAlarmPort, "zone": zone} for zone in self._zones] + port_args += [{"driver": ZoneOpenPort, "zone": zone} for zone in self._zones] + port_args += [{"driver": ZoneTamperPort, "zone": zone} for zone in self._zones] + 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, + "driver": RemoteButtonPort, + "remote": remote, + "button": button, + "timeout": self._remote_buttons_timeout, } ) @@ -166,10 +164,10 @@ async def make_port_args(self) -> list[dict[str, Any]]: for button in any_remote_buttons: port_args.append( { - 'driver': AnyRemoteButtonPort, - 'remotes': self._remotes, - 'button': button, - 'timeout': self._remote_buttons_timeout, + "driver": AnyRemoteButtonPort, + "remotes": self._remotes, + "button": button, + "timeout": self._remote_buttons_timeout, } ) @@ -185,12 +183,12 @@ async def disconnect(self) -> None: if self._panel_task is None: return - self.debug('disconnecting') + self.debug("disconnecting") try: await self._paradox.disconnect() except ConnectionError as e: # PAI may raise ConnectionError when disconnecting, so we catch it here and ignore it - self.error('failed to disconnect from panel: %s', e, exc_info=True) + self.error("failed to disconnect from panel: %s", e, exc_info=True) self._paradox = None @@ -199,10 +197,10 @@ async def disconnect(self) -> None: try: await asyncio.wait_for(self._panel_task, timeout=10) except asyncio.TimeoutError: - self.error('timeout waiting for panel task end') + self.error("timeout waiting for panel task end") self._panel_task.cancel() else: - self.debug('disconnected') + self.debug("disconnected") self._panel_task = None @@ -212,10 +210,10 @@ async def _supervisor_loop(self) -> None: while True: try: connected = ( - self._paradox and - self._paradox.connection and - self._paradox.connection.connected and - connect_succeeded + self._paradox + and self._paradox.connection + and self._paradox.connection.connected + and connect_succeeded ) self.set_online(connected) @@ -225,24 +223,24 @@ async def _supervisor_loop(self) -> None: await self.connect() connect_succeeded = True except Exception as e: - self.error('failed to connect: %s', e, exc_info=True) + self.error("failed to connect: %s", e, exc_info=True) elif not self.is_enabled() and connected: connect_succeeded = False try: await self.disconnect() except Exception as e: - self.error('failed to disconnect: %s', e, exc_info=True) + self.error("failed to disconnect: %s", e, exc_info=True) if self._paradox: self._update_properties() await asyncio.sleep(self.SUPERVISOR_LOOP_INTERVAL) except Exception as e: - self.error('supervisor loop error: %s', e, exc_info=True) + self.error("supervisor loop error: %s", e, exc_info=True) await asyncio.sleep(self.SUPERVISOR_LOOP_INTERVAL) except asyncio.CancelledError: - self.debug('supervisor task cancelled') + self.debug("supervisor task cancelled") break async def handle_cleanup(self) -> None: @@ -256,22 +254,27 @@ def handle_paradox_property_change(self, change: Any) -> None: from .paradoxport import ParadoxPort info = self._paradox.storage.data[change.type].get(change.key) - if info and ('id' in info): - id_ = info['id'] + if info and ("id" in info): + id_ = info["id"] self.debug( - 'property change: %s[%s].%s: %s -> %s', change.type, id_, change.property, + "property change: %s[%s].%s: %s -> %s", + change.type, + id_, + change.property, json_utils.dumps(change.old_value, extra_types=json_utils.EXTRA_TYPES_EXTENDED), - json_utils.dumps(change.new_value, extra_types=json_utils.EXTRA_TYPES_EXTENDED) + json_utils.dumps(change.new_value, extra_types=json_utils.EXTRA_TYPES_EXTENDED), ) obj = self._properties.setdefault(change.type, {}).setdefault(id_, {}) obj[change.property] = change.new_value - obj['label'] = info['label'] + obj["label"] = info["label"] else: id_ = None self.debug( - 'property change: %s.%s: %s -> %s', change.type, change.property, + "property change: %s.%s: %s -> %s", + change.type, + change.property, json_utils.dumps(change.old_value, extra_types=json_utils.EXTRA_TYPES_EXTENDED), - json_utils.dumps(change.new_value, extra_types=json_utils.EXTRA_TYPES_EXTENDED) + json_utils.dumps(change.new_value, extra_types=json_utils.EXTRA_TYPES_EXTENDED), ) self._properties.setdefault(change.type, {})[change.property] = change.new_value @@ -281,16 +284,16 @@ def handle_paradox_property_change(self, change: Any) -> None: try: pai_port.on_property_change(change.type, id_, change.property, change.old_value, change.new_value) except Exception as e: - self.error('property change handler execution failed: %s', e, exc_info=True) + 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) -> Optional[Property]: - if type_ == 'system': + def get_property(self, type_: str, id_: str | int | None, name: str) -> Property | None: + if type_ == "system": return self._properties.get(type_, {}).get(name) else: return self._properties.get(type_, {}).get(id_, {}).get(name) - def get_properties(self, type_: str, id_: Optional[Union[str, int]]) -> dict[str, Property]: - if type_ == 'system': + def get_properties(self, type_: str, id_: str | int | None) -> dict[str, Property]: + if type_ == "system": return self._properties.get(type_, {}) else: return self._properties.get(type_, {}).get(id_, {}) @@ -344,16 +347,16 @@ def _update_properties(self) -> None: self.handle_paradox_property_change(change) async def set_area_armed_mode(self, area: int, armed_mode: str) -> None: - self.debug('area %s: set armed mode to %s', area, armed_mode) + self.debug("area %s: set armed mode to %s", area, armed_mode) if not await self._paradox.panel.control_partitions([area], armed_mode): - raise exceptions.ParadoxCommandError('Failed to set area armed mode') + raise exceptions.ParadoxCommandError("Failed to set area armed mode") async def set_zone_bypass(self, zone: int, bypass: bool) -> None: - self.debug('zone %s: %s bypass', zone, ['clear', 'set'][bypass]) + self.debug("zone %s: %s bypass", zone, ["clear", "set"][bypass]) if not await self._paradox.panel.control_zones([zone], constants.ZONE_BYPASS_MAPPING[bypass]): - raise exceptions.ParadoxCommandError('Failed to set zone bypass') + raise exceptions.ParadoxCommandError("Failed to set zone bypass") async def set_output_action(self, output: int, action: str) -> None: - self.debug('output %s: set action to %s', output, action) + self.debug("output %s: set action to %s", output, action) if not await self._paradox.panel.control_outputs([output], action): - raise exceptions.ParadoxCommandError('Failed to set output action') + raise exceptions.ParadoxCommandError("Failed to set output action") diff --git a/qtoggleserver/paradox/paradoxport.py b/qtoggleserver/paradox/paradoxport.py index 7a6be5f..3c09395 100644 --- a/qtoggleserver/paradox/paradoxport.py +++ b/qtoggleserver/paradox/paradoxport.py @@ -1,6 +1,6 @@ import abc -from typing import cast, Optional +from typing import cast from qtoggleserver.peripherals import PeripheralPort @@ -10,12 +10,7 @@ class ParadoxPort(PeripheralPort, metaclass=abc.ABCMeta): def on_property_change( - self, - type_: str, - id_: Optional[str], - property_: str, - old_value: Property, - new_value: Property + self, type_: str, id_: str | None, property_: str, old_value: Property, new_value: Property ) -> None: pass diff --git a/qtoggleserver/paradox/remote.py b/qtoggleserver/paradox/remote.py index 3be84ff..a3ff6c1 100644 --- a/qtoggleserver/paradox/remote.py +++ b/qtoggleserver/paradox/remote.py @@ -13,23 +13,23 @@ def __init__(self, remote: int, *args, **kwargs) -> None: super().__init__(*args, **kwargs) def make_id(self) -> str: - return f'remote{self.remote}.{self.ID}' + return f"remote{self.remote}.{self.ID}" def get_remote_label(self) -> str: - return self.get_property('label') or f'Remote {self.remote}' + 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) + 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) + return self.get_peripheral().get_properties("user", self.remote) class RemoteButtonPort(RemotePort): - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - ID = 'button' + ID = "button" def __init__(self, button: str, timeout: int, *args, **kwargs) -> None: self.button: str = button @@ -40,30 +40,30 @@ def __init__(self, button: str, timeout: int, *args, **kwargs) -> None: super().__init__(*args, **kwargs) def make_id(self) -> str: - return f'{super().make_id()}_{self.button}' + 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()}' + 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.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 + return self.get_property(f"button_{self.button}") or 0 class AnyRemoteButtonPort(RemotePort): - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - ID = 'button' + ID = "button" def __init__(self, remotes: list[int], button: str, timeout: int, *args, **kwargs) -> None: self.remotes: list[int] = remotes @@ -75,10 +75,10 @@ def __init__(self, remotes: list[int], button: str, timeout: int, *args, **kwarg super().__init__(remote=0, *args, **kwargs) def make_id(self) -> str: - return f'remote.{self.ID}_{self.button}' + return f"remote.{self.ID}_{self.button}" async def attr_get_default_display_name(self) -> str: - return f'Remote Button {self.button.upper()}' + return f"Remote Button {self.button.upper()}" async def read_value(self) -> bool: now = int(time.time() * 1000) @@ -86,7 +86,7 @@ async def read_value(self) -> bool: 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.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 @@ -97,4 +97,4 @@ async def read_value(self) -> bool: return False def get_button_value(self, remote: int) -> int: - return self.get_peripheral().get_property('user', remote, f'button_{self.button}') + return self.get_peripheral().get_property("user", remote, f"button_{self.button}") diff --git a/qtoggleserver/paradox/system.py b/qtoggleserver/paradox/system.py index 8b759a1..5eb8ab5 100644 --- a/qtoggleserver/paradox/system.py +++ b/qtoggleserver/paradox/system.py @@ -1,5 +1,4 @@ from abc import ABCMeta -from typing import Optional from .paradoxport import ParadoxPort from .typing import Property @@ -7,21 +6,21 @@ class SystemPort(ParadoxPort, metaclass=ABCMeta): def make_id(self) -> str: - return f'system.{self.ID}' + return f"system.{self.ID}" def get_property(self, name: str) -> Property: - return self.get_peripheral().get_property('system', None, name) + return self.get_peripheral().get_property("system", None, name) def get_properties(self) -> dict[str, Property]: - return self.get_peripheral().get_properties('system', None) + return self.get_peripheral().get_properties("system", None) class SystemTroublePort(SystemPort): - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - DISPLAY_NAME = 'System Trouble' + DISPLAY_NAME = "System Trouble" - ID = 'trouble' + ID = "trouble" - async def read_value(self) -> Optional[bool]: - return self.get_property('trouble') + async def read_value(self) -> bool | None: + return self.get_property("trouble") diff --git a/qtoggleserver/paradox/typing.py b/qtoggleserver/paradox/typing.py index 20d26cb..c3640b1 100644 --- a/qtoggleserver/paradox/typing.py +++ b/qtoggleserver/paradox/typing.py @@ -1,4 +1,4 @@ -from typing import Union +from typing import TypeAlias, Union -Property = Union[int, bool, str] +Property: TypeAlias = Union[int, bool, str] diff --git a/qtoggleserver/paradox/zone.py b/qtoggleserver/paradox/zone.py index 091d0ca..41b16ec 100644 --- a/qtoggleserver/paradox/zone.py +++ b/qtoggleserver/paradox/zone.py @@ -1,5 +1,4 @@ from abc import ABCMeta -from typing import Optional from .paradoxport import ParadoxPort from .typing import Property @@ -12,69 +11,69 @@ def __init__(self, zone: int, *args, **kwargs) -> None: super().__init__(*args, **kwargs) def make_id(self) -> str: - return f'zone{self.zone}.{self.ID}' + return f"zone{self.zone}.{self.ID}" def get_zone_label(self) -> str: - return self.get_property('label') or f'Zone {self.zone}' + return self.get_property("label") or f"Zone {self.zone}" def get_property(self, name: str) -> Property: - return self.get_peripheral().get_property('zone', self.zone, name) + return self.get_peripheral().get_property("zone", self.zone, name) def get_properties(self) -> dict[str, Property]: - return self.get_peripheral().get_properties('zone', self.zone) + return self.get_peripheral().get_properties("zone", self.zone) class ZoneOpenPort(ZonePort): - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - ID = 'open' + ID = "open" async def attr_get_default_display_name(self) -> str: - return f'{self.get_zone_label()} Open' + return f"{self.get_zone_label()} Open" - async def read_value(self) -> Optional[bool]: - return self.get_property('open') + async def read_value(self) -> bool | None: + return self.get_property("open") class ZoneAlarmPort(ZonePort): - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - ID = 'alarm' + ID = "alarm" async def attr_get_default_display_name(self) -> str: - return f'{self.get_zone_label()} Alarm' + return f"{self.get_zone_label()} Alarm" - async def read_value(self) -> Optional[bool]: - return self.get_property('alarm') + async def read_value(self) -> bool | None: + return self.get_property("alarm") class ZoneTroublePort(ZonePort): - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - ID = 'trouble' + ID = "trouble" async def attr_get_default_display_name(self) -> str: - return f'{self.get_zone_label()} Trouble' + return f"{self.get_zone_label()} Trouble" async def read_value(self) -> bool: for name, value in self.get_properties().items(): - if name.endswith('_trouble') and value: + if name.endswith("_trouble") and value: return True return False class ZoneTamperPort(ZonePort): - TYPE = 'boolean' + TYPE = "boolean" WRITABLE = False - ID = 'tamper' + ID = "tamper" async def attr_get_default_display_name(self) -> str: - return f'{self.get_zone_label()} Tamper' + return f"{self.get_zone_label()} Tamper" - async def read_value(self) -> Optional[bool]: - return self.get_property('tamper') + async def read_value(self) -> bool | None: + return self.get_property("tamper") diff --git a/setup.py b/setup.py deleted file mode 100644 index f5458f9..0000000 --- a/setup.py +++ /dev/null @@ -1,20 +0,0 @@ -from setuptools import setup, find_namespace_packages - - -setup( - name='qtoggleserver-paradox', - version='unknown-version', - description='Control your Paradox alarm with qToggleServer', - author='Calin Crisan', - author_email='ccrisan@gmail.com', - license='Apache 2.0', - - packages=find_namespace_packages(), - - install_requires=[ - 'paradox-alarm-interface>=3,<4', - 'pyserial>=3.4', - 'pyserial-asyncio>=0.4', - 'requests' - ] -)