Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 0 additions & 5 deletions .flake8

This file was deleted.

47 changes: 4 additions & 43 deletions .github/workflows/main.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
8 changes: 8 additions & 0 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
@@ -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
35 changes: 35 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
@@ -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
4 changes: 3 additions & 1 deletion qtoggleserver/paradox/__init__.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
from .paradoxalarm import ParadoxAlarm


VERSION = 'unknown-version'
__all__ = ["ParadoxAlarm"]

VERSION = "0.0.0"
93 changes: 46 additions & 47 deletions qtoggleserver/paradox/area.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
from abc import ABCMeta
from typing import Optional

from qtoggleserver.core import ports as core_ports

Expand All @@ -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(
Expand All @@ -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")
27 changes: 12 additions & 15 deletions qtoggleserver/paradox/constants.py
Original file line number Diff line number Diff line change
@@ -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
2 changes: 1 addition & 1 deletion qtoggleserver/paradox/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
27 changes: 13 additions & 14 deletions qtoggleserver/paradox/output.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
from abc import ABCMeta
from typing import Optional

from .paradoxport import ParadoxPort
from .typing import Property
Expand All @@ -12,44 +11,44 @@ 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


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")
Loading