Sign Up
Log In
Log In
or
Sign Up
Places
All Projects
Status Monitor
Collapse sidebar
system:homeautomation:home-assistant
python-deconz
_service:obs_scm:deconz-112.obscpio
Overview
Repositories
Revisions
Requests
Users
Attributes
Meta
File _service:obs_scm:deconz-112.obscpio of Package python-deconz
07070100000000000081A400000000000000000000000164453E39000000AF000000000000000000000000000000000000001700000000deconz-112/.coveragerc[report] # Regexes for lines to exclude from consideration exclude_lines = # TYPE_CHECKING and @overload blocks are never executed during pytest run if TYPE_CHECKING: 07070100000001000041ED00000000000000000000000364453E3900000000000000000000000000000000000000000000001300000000deconz-112/.github07070100000002000081A400000000000000000000000164453E390000000F000000000000000000000000000000000000001F00000000deconz-112/.github/FUNDING.ymlgithub: Kane61007070100000003000081A400000000000000000000000164453E3900000088000000000000000000000000000000000000002200000000deconz-112/.github/dependabot.ymlversion: 2 updates: - package-ecosystem: pip directory: "/" schedule: interval: weekly open-pull-requests-limit: 10 07070100000004000081A400000000000000000000000164453E39000000A2000000000000000000000000000000000000002700000000deconz-112/.github/release-drafter.yml# https://github.com/release-drafter/release-drafter template: | ## What's Changed $CHANGES tag-template: "v$NEXT_MAJOR_VERSION" version-template: "$MAJOR"07070100000005000041ED00000000000000000000000264453E3900000000000000000000000000000000000000000000001D00000000deconz-112/.github/workflows07070100000006000081A400000000000000000000000164453E390000036C000000000000000000000000000000000000002F00000000deconz-112/.github/workflows/pythonpublish.yml# This workflows will upload a Python Package using Twine when a release is created # For more information see: https://help.github.com/en/actions/language-and-framework-guides/using-python-with-github-actions#publishing-to-package-registries name: Upload Python Package on: release: types: [published] jobs: deploy: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - name: Set up Python uses: actions/setup-python@v1 with: python-version: "3.x" - name: Install dependencies run: | python -m pip install --upgrade pip pip install setuptools wheel twine - name: Build and publish env: TWINE_USERNAME: __token__ TWINE_PASSWORD: ${{ secrets.PYPI_TOKEN }} run: | python setup.py sdist bdist_wheel twine upload dist/* 07070100000007000081A400000000000000000000000164453E390000017D000000000000000000000000000000000000003100000000deconz-112/.github/workflows/release-drafter.yml# https://github.com/release-drafter/release-drafter name: Release Drafter on: push: branches: - master jobs: update_release_draft: runs-on: ubuntu-latest steps: # Drafts your next Release notes as Pull Requests are merged into "master" - uses: release-drafter/release-drafter@v5 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}07070100000008000081A400000000000000000000000164453E39000004DE000000000000000000000000000000000000002600000000deconz-112/.github/workflows/test.yml# This workflow will install Python dependencies, run tests and lint with a single version of Python # For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions name: Run Tests on: push: branches: [master] pull_request: branches: [master] jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - name: Set up Python 3.10 uses: actions/setup-python@v4.5.0 with: python-version: "3.10" - name: Install dependencies run: | pip install -r requirements.txt pip install -r requirements-test.txt - name: Lint with Ruff run: | ruff check pydeconz - name: Lint with Flake8 run: | flake8 pydeconz - name: Check formatting with Black run: | black pydeconz --check - name: Check import order with isort run: | isort pydeconz --check-only - name: Check typing with mypy run: | mypy pydeconz - name: Test with pytest run: | pytest tests --doctest-modules --junitxml=junit/test-results.xml --cov=pydeconz --cov-report=xml --cov-report=html 07070100000009000081A400000000000000000000000164453E39000004D9000000000000000000000000000000000000001600000000deconz-112/.gitignore# Byte-compiled / optimized / DLL files __pycache__/ *.py[cod] *$py.class # C extensions *.so # Distribution / packaging .Python build/ develop-eggs/ dist/ downloads/ eggs/ .eggs/ lib/ lib64/ parts/ sdist/ var/ wheels/ *.egg-info/ .installed.cfg *.egg MANIFEST # PyInstaller # Usually these files are written by a python script from a template # before PyInstaller builds the exe, so as to inject date/other infos into it. *.manifest *.spec # Installer logs pip-log.txt pip-delete-this-directory.txt # Unit test / coverage reports htmlcov/ .tox/ .coverage .coverage.* .cache nosetests.xml coverage.xml *.cover .hypothesis/ .pytest_cache/ test-output.xml # Translations *.mo *.pot # Django stuff: *.log local_settings.py db.sqlite3 # Flask stuff: instance/ .webassets-cache # Scrapy stuff: .scrapy # Sphinx documentation docs/_build/ # PyBuilder target/ # Jupyter Notebook .ipynb_checkpoints # pyenv .python-version # celery beat schedule file celerybeat-schedule # SageMath parsed files *.sage.py # Environments .env .venv env/ venv/ ENV/ env.bak/ venv.bak/ # Spyder project settings .spyderproject .spyproject # Rope project settings .ropeproject # mkdocs documentation /site # mypy .mypy_cache/ .vscode/settings.json0707010000000A000081A400000000000000000000000164453E3900000309000000000000000000000000000000000000002300000000deconz-112/.pre-commit-config.yamlrepos: - repo: https://github.com/charliermarsh/ruff-pre-commit rev: v0.0.257 hooks: - id: ruff args: - --fix - repo: https://github.com/psf/black rev: 23.1.0 hooks: - id: black args: - --quiet files: ^((pydeconz|pylint|script|tests)/.+)?[^/]+\.py$ - repo: https://github.com/PyCQA/isort rev: 5.12.0 hooks: - id: isort - repo: https://github.com/pre-commit/mirrors-mypy rev: v1.1.1 hooks: - id: mypy exclude: ^tests/ # - repo: https://gitlab.com/pycqa/flake8 # rev: 3.7.9 # hooks: # - id: flake8 # additional_dependencies: # - flake8-docstrings==1.5.0 # - pydocstyle==4.0.1 # files: ^(pydeconz|tests)/.+\.py$ 0707010000000B000041ED00000000000000000000000264453E3900000000000000000000000000000000000000000000001300000000deconz-112/.vscode0707010000000C000081A400000000000000000000000164453E39000001A5000000000000000000000000000000000000002900000000deconz-112/.vscode/settings.default.json{ "editor.formatOnPaste": false, "editor.formatOnSaveMode": "file", "editor.insertSpaces": true, "editor.rulers": [88], "editor.tabSize": 4, "files.defaultLanguage": "python", "files.insertFinalNewline": true, "files.trimFinalNewlines": true, "files.trimTrailingWhitespace": true, "python.formatting.provider": "black", "python.linting.flake8Enabled": true, "python.testing.pytestEnabled": true } 0707010000000D000081A400000000000000000000000164453E3900000428000000000000000000000000000000000000001300000000deconz-112/LICENSEMIT License Copyright (c) 2017 Kane610 Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. 0707010000000E000081A400000000000000000000000164453E39000003D3000000000000000000000000000000000000001600000000deconz-112/README.rstdeCONZ |Chat Status| ==================== Python library wrapping `deCONZ Rest API`_ for Home-Assistant. This implementation should cover most devices supported by deCONZ, if that is not the case please create an issue with debug logs and we will get it supported. Only host address and API key are necessary for normal operations. **Acknowledgements** * Mattias Flodins custom deCONZ component was a great source of inspiration. * Maija Vilkina and her blog `Snillevilla`_ has been a huge help in getting deCONZ up and running. * Thanks to donnib, dkmh, simonporter007, kroimon, Henrik Nilsson for requesting and verifying functionality! * Special thanks to Dresden Elektronik for sponsoring with extra Conbee and Raspbee hardware. .. |Chat Status| image:: https://img.shields.io/discord/330944238910963714.svg :target: https://discord.gg/c6pvg8a .. _`deCONZ Rest API`: https://dresden-elektronik.github.io/deconz-rest-doc/ .. _`Snillevilla`: https://snillevilla.se/ 0707010000000F000081A400000000000000000000000164453E39000001AC000000000000000000000000000000000000001400000000deconz-112/mypy.ini[mypy] check_untyped_defs = true disallow_any_generics = true disallow_incomplete_defs = true disallow_subclassing_any = true disallow_untyped_calls = true disallow_untyped_decorators = true disallow_untyped_defs = true no_implicit_optional = true no_implicit_reexport = true strict_equality = true warn_redundant_casts = true warn_return_any = true warn_unreachable = true warn_unused_configs = true warn_unused_ignores = true 07070100000010000041ED00000000000000000000000464453E3900000000000000000000000000000000000000000000001400000000deconz-112/pydeconz07070100000011000081A400000000000000000000000164453E3900000061000000000000000000000000000000000000002000000000deconz-112/pydeconz/__init__.pyfrom .errors import * # noqa: D104, F401, F403 from .gateway import DeconzSession # noqa: F401 07070100000012000081A400000000000000000000000164453E3900000A1B000000000000000000000000000000000000002000000000deconz-112/pydeconz/__main__.py"""Read attributes from your deCONZ gateway.""" import argparse import asyncio import logging import aiohttp import async_timeout from pydeconz import errors from pydeconz.gateway import DeconzSession from pydeconz.interfaces.api_handlers import CallbackType from pydeconz.models.event import EventType LOGGER = logging.getLogger(__name__) def new_device_callback(event: EventType, id: str) -> None: """Signal new device is available.""" LOGGER.info(f"{event}, {id}") async def deconz_gateway( session: aiohttp.ClientSession, host: str, port: int, api_key: str, callback: CallbackType, ) -> DeconzSession | None: """Create a gateway object and verify configuration.""" deconz = DeconzSession(session, host, port, api_key) deconz.subscribe(callback) try: async with async_timeout.timeout(5): await deconz.refresh_state() return deconz except errors.Unauthorized: LOGGER.exception("Invalid API key for deCONZ gateway") except (asyncio.TimeoutError, errors.RequestError): LOGGER.error("Error connecting to deCONZ gateway") return None async def main(host: str, port: int, api_key: str) -> None: """CLI method for library.""" LOGGER.info("Starting deCONZ gateway") session = aiohttp.ClientSession() gateway = await deconz_gateway( session=session, host=host, port=port, api_key=api_key, callback=new_device_callback, ) if not gateway: LOGGER.error("Couldn't connect to deCONZ gateway") await session.close() return await gateway.refresh_state() gateway.start() try: while True: await asyncio.sleep(1) except asyncio.CancelledError: pass finally: gateway.close() await session.close() if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("host", type=str) parser.add_argument("api_key", type=str) parser.add_argument("-p", "--port", type=int, default=80) parser.add_argument("-D", "--debug", action="store_true") args = parser.parse_args() loglevel = logging.INFO if args.debug: loglevel = logging.DEBUG logging.basicConfig(format="%(message)s", level=loglevel) LOGGER.info(f"{args.host}, {args.port}, {args.api_key}") try: asyncio.run( main( host=args.host, port=args.port, api_key=args.api_key, ) ) except KeyboardInterrupt: pass 07070100000013000081A400000000000000000000000164453E39000028B4000000000000000000000000000000000000001E00000000deconz-112/pydeconz/config.py"""Python library to connect deCONZ and Home Assistant to work together.""" from collections.abc import Awaitable, Callable import enum import logging from typing import Any, Final from .utils import normalize_bridge_id LOGGER = logging.getLogger(__name__) UNINITIALIZED_BRIDGE_ID: Final = "0000000000000000" class ConfigDeviceName(enum.Enum): """Valid product names of the gateway.""" CONBEE = "ConBee" RASPBEE = "RaspBee" CONBEE_2 = "ConBee II" RASPBEE_2 = "RaspBee II" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "ConfigDeviceName": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected config device name %s", value) return ConfigDeviceName.UNKNOWN class ConfigNTP(enum.Enum): """Timeformat that can be used by other applications.""" SYNCED = "synced" UNSYNCED = "unsynced" class ConfigTimeFormat(enum.Enum): """Tells if the NTP time is "synced" or "unsynced".""" FORMAT_12H = "12h" FORMAT_24H = "24h" class ConfigUpdateChannel(enum.Enum): """Available update channels to use with the Gateway.""" ALPHA = "alpha" BETA = "beta" STABLE = "stable" class ConfigZigbeeChannel(enum.IntEnum): """Available wireless frequency channels to use with the Gateway.""" CHANNEL_11 = 11 CHANNEL_15 = 15 CHANNEL_20 = 20 CHANNEL_25 = 25 class Config: """deCONZ configuration representation. Dresden Elektroniks documentation of config in deCONZ http://dresden-elektronik.github.io/deconz-rest-doc/config/ """ def __init__( self, raw: dict[str, Any], request: Callable[..., Awaitable[dict[str, Any]]], ) -> None: """Set configuration about deCONZ gateway.""" self.raw = raw self.request = request @property def api_version(self) -> str | None: """Version of the deCONZ Rest API.""" return self.raw.get("apiversion") @property def bridge_id(self) -> str: """Gateway unique identifier.""" return normalize_bridge_id(self.raw.get("bridgeid", UNINITIALIZED_BRIDGE_ID)) @property def device_name(self) -> ConfigDeviceName: """Product name of the gateway. Valid values are "ConBee", "RaspBee", "ConBee II" and "RaspBee II". """ return ConfigDeviceName(self.raw.get("devicename")) @property def dhcp(self) -> bool | None: """Whether the IP address of the bridge is obtained with DHCP.""" return self.raw.get("dhcp") @property def firmware_version(self) -> str | None: """Version of the ZigBee firmware.""" return self.raw.get("fwversion") @property def gateway(self) -> str | None: """IPv4 address of the gateway.""" return self.raw.get("gateway") @property def ip_address(self) -> str | None: """IPv4 address of the gateway.""" return self.raw.get("ipaddress") @property def link_button(self) -> bool | None: """Is gateway unlocked.""" return self.raw.get("linkbutton") @property def local_time(self) -> str | None: """Localtime of the gateway.""" return self.raw.get("localtime") @property def mac(self) -> str | None: """MAC address of gateway.""" return self.raw.get("mac") @property def model_id(self) -> str | None: """Model identifyer. Fixed string "deCONZ". """ return self.raw.get("modelid") @property def name(self) -> str | None: """Name of the gateway.""" return self.raw.get("name") @property def network_mask(self) -> str | None: """Network mask of the gateway.""" return self.raw.get("netmask") @property def network_open_duration(self) -> int | None: """Duration in seconds used by lights and sensors search.""" return self.raw.get("networkopenduration") @property def ntp(self) -> ConfigNTP | None: """Tells if the NTP time is "synced" or "unsynced". Only for gateways running on Linux. """ if "ntp" in self.raw: return ConfigNTP(self.raw["ntp"]) return None @property def pan_id(self) -> int | None: """Zigbee pan ID of the gateway.""" return self.raw.get("panid") @property def portal_services(self) -> bool | None: """State of registration to portal service. Is the bridge registered to synchronize data with a portal account. """ return self.raw.get("portalservices") @property def rf_connected(self) -> bool | None: """State of deCONZ connection to firmware and if Zigbee network is up.""" return self.raw.get("rfconnected") @property def software_update(self) -> dict[str, Any] | None: """Contains information related to software updates.""" return self.raw.get("swupdate") @property def software_version(self) -> str | None: """Software version of the gateway.""" return self.raw.get("swversion") @property def time_format(self) -> ConfigTimeFormat: """Timeformat used by gateway. Supported values: "12h" or "24h" """ return ConfigTimeFormat(self.raw["timeformat"]) @property def time_zone(self) -> str | None: """Time zone used by gateway. Only on Raspberry Pi. "None" if not further specified. """ return self.raw.get("timezone") @property def utc(self) -> str | None: """UTC time of gateway in ISO 8601 format.""" return self.raw.get("utc") @property def uuid(self) -> str | None: """UPNP Unique ID of the gateway.""" return self.raw.get("uuid") @property def websocket_notify_all(self) -> bool | None: """All state changes will be signalled through the Websocket connection. Default true. """ return self.raw.get("websocketnotifyall") @property def websocket_port(self) -> int | None: """Websocket port.""" return self.raw.get("websocketport") @property def whitelist(self) -> dict[str, Any]: """Array of whitelisted API keys.""" return self.raw.get("whitelist", {}) @property def zigbee_channel(self) -> ConfigZigbeeChannel: """Wireless frequency channel.""" return ConfigZigbeeChannel(self.raw["zigbeechannel"]) async def set_config( self, discovery: bool | None = None, group_delay: int | None = None, light_last_seen_interval: int | None = None, name: str | None = None, network_open_duration: int | None = None, otau_active: bool | None = None, permit_join: int | None = None, rf_connected: bool | None = None, time_format: ConfigTimeFormat | None = None, time_zone: str | None = None, unlock: int | None = None, update_channel: ConfigUpdateChannel | None = None, utc: str | None = None, zigbee_channel: ConfigZigbeeChannel | None = None, websocket_notify_all: bool | None = None, ) -> dict[str, Any]: """Modify configuration parameters. Supported values: - discovery [bool] Set gateway discovery over the internet active or inactive. - group_delay [int] 0-5000 Time between two group commands in milliseconds. - light_last_seen_interval [int] 1-65535 default 60 Sets the number of seconds where the timestamp for "lastseen" is updated at the earliest for light resources. For any such update, a seperate websocket event will be triggered. - name [str] 0-16 characters Name of the gateway. - network_open_duration [int] 1-65535 Sets the lights and sensors search duration in seconds. - otau_active [bool] Set OTAU active or inactive - permit_join [int] 0-255 Open the network so that other zigbee devices can join. 0 = network closed 255 = network open 1–254 = time in seconds the network remains open The value will decrement automatically. - rf_connected [bool] Set to true to bring the Zigbee network up and false to bring it down. This has the same effect as using the Join and Leave buttons in deCONZ. - time_format [str] 12h|24h Can be used to store the timeformat permanently. - time_zone [str] Set the timezone of the gateway (only on Raspberry Pi). - unlock [int] 0-600 (seconds) Unlock the gateway so that apps can register themselves to the gateway. - update_channel [str] stable|alpha|beta Set update channel. - utc [str] Set the UTC time of the gateway (only on Raspbery Pi) in ISO 8601 format (yyyy-MM-ddTHH:mm:ss). - zigbee_channel [int] 11|15|20|25 Set the zigbeechannel of the gateway. Notify other Zigbee devices also to change their channel. - websocket_notify_all [bool] default True When true all state changes will be signalled through the websocket connection. """ data = { key: value for key, value in { "discovery": discovery, "groupdelay": group_delay, "lightlastseeninterval": light_last_seen_interval, "name": name, "networkopenduration": network_open_duration, "otauactive": otau_active, "permitjoin": permit_join, "rfconnected": rf_connected, "timezone": time_zone, "unlock": unlock, "utc": utc, "websocketnotifyall": websocket_notify_all, }.items() if value is not None } if time_format is not None: data["timeformat"] = time_format.value if update_channel is not None: data["updatechannel"] = update_channel.value if zigbee_channel is not None: data["zigbeechannel"] = zigbee_channel.value return await self.request("put", path="/config", json=data) 07070100000014000081A400000000000000000000000164453E3900000829000000000000000000000000000000000000001E00000000deconz-112/pydeconz/errors.py"""deCONZ errors.""" from typing import Any class pydeconzException(Exception): """Base error for pydeconz. https://dresden-elektronik.github.io/deconz-rest-doc/errors/ """ class BadRequest(pydeconzException): """The request was not formatted as expected or missing parameters.""" class BridgeBusy(pydeconzException): """The Bridge is busy, too many requests (more than 20).""" class Forbidden(pydeconzException): """The caller has no rights to access the requested URI.""" class LinkButtonNotPressed(pydeconzException): """The Link button has not been pressed.""" class NotConnected(pydeconzException): """The Hardware is not connected.""" class RequestError(pydeconzException): """Unable to fulfill request. Raised when host or API cannot be reached. """ class ResourceNotFound(pydeconzException): """The requested resource (light, group, ...) was not found.""" class ResponseError(pydeconzException): """Invalid response.""" class Unauthorized(pydeconzException): """Authorization failed.""" ERRORS = { 1: Unauthorized, # Unauthorized user 2: BadRequest, # Body contains invalid JSON 3: ResourceNotFound, # Resource not available 4: RequestError, # Method not available for resource 5: BadRequest, # Missing parameters in body 6: RequestError, # Parameter not available 7: RequestError, # Invalid value for parameter 8: RequestError, # Parameter is not modifiable 101: LinkButtonNotPressed, # Link button not pressed 901: BridgeBusy, # May occur when sending too fast 950: NotConnected, # Hardware is not connected 951: BridgeBusy, # May occur when sending too fast } def raise_error(error: dict[str, Any]) -> None: """Raise error.""" if error: if cls := ERRORS.get(error["type"]): raise cls( "{} {} {}".format( error["type"], error["address"], error["description"], ) ) raise pydeconzException(error) 07070100000015000081A400000000000000000000000164453E3900001BFE000000000000000000000000000000000000001F00000000deconz-112/pydeconz/gateway.py"""Python library to connect deCONZ and Home Assistant to work together.""" from asyncio import CancelledError, Task, create_task, sleep import logging from pprint import pformat from typing import Any, Callable import aiohttp from .config import Config from .errors import BridgeBusy, RequestError, ResponseError, raise_error from .interfaces.alarm_systems import AlarmSystems from .interfaces.api_handlers import CallbackType, UnsubscribeType from .interfaces.events import EventHandler from .interfaces.groups import GroupHandler from .interfaces.lights import LightResourceManager from .interfaces.scenes import Scenes from .interfaces.sensors import SensorResourceManager from .models import ResourceGroup from .websocket import Signal, State, WSClient LOGGER = logging.getLogger(__name__) class DeconzSession: """deCONZ representation that handles lights, groups, scenes and sensors.""" def __init__( self, session: aiohttp.ClientSession, host: str, port: int, api_key: str | None = None, connection_status: Callable[[bool], None] | None = None, ) -> None: """Session setup.""" self.session = session self.host = host self.port = port self.api_key = api_key self._sleep_tasks: dict[str, Task[None]] = {} self.connection_status_callback = connection_status self.config = Config({}, self.request) self.events = EventHandler(self) self.websocket: WSClient | None = None self.alarm_systems = AlarmSystems(self) self.groups = GroupHandler(self) self.lights = LightResourceManager(self) self.scenes = Scenes(self) self.sensors = SensorResourceManager(self) async def get_api_key( self, api_key: str | None = None, client_name: str = "pydeconz", ) -> str: """Request a new API key. Supported values: - api_key [str] 10-40 characters, key to use for authentication - client_name [str] 0-40 characters, name of the client application """ data = { key: value for key, value in { "username": api_key, "devicetype": client_name, }.items() if value is not None } response: list[dict[str, dict[str, str]]] = await self._request( "post", url=f"http://{self.host}:{self.port}/api", json=data, ) return response[0]["success"]["username"] def start(self, websocketport: int | None = None) -> None: """Connect websocket to deCONZ.""" if self.config.websocket_port is not None: websocketport = self.config.websocket_port if not websocketport: LOGGER.error("No websocket port specified") return self.websocket = WSClient( self.session, self.host, websocketport, self.session_handler ) self.websocket.start() def close(self) -> None: """Close websession and websocket to deCONZ.""" if self.websocket: self.websocket.stop() async def refresh_state(self) -> None: """Read deCONZ parameters.""" data = await self.request("get", "") self.config.raw.update(data[ResourceGroup.CONFIG.value]) self.alarm_systems.process_raw(data.get(ResourceGroup.ALARM.value, {})) self.groups.process_raw(data[ResourceGroup.GROUP.value]) self.lights.process_raw(data[ResourceGroup.LIGHT.value]) self.sensors.process_raw(data[ResourceGroup.SENSOR.value]) def subscribe(self, callback: CallbackType) -> UnsubscribeType: """Subscribe to status changes for all resources.""" subscribers = [ self.alarm_systems.subscribe(callback), self.groups.subscribe(callback), self.lights.subscribe(callback), self.sensors.subscribe(callback), ] def unsubscribe() -> None: for subscriber in subscribers: subscriber() return unsubscribe async def request_with_retry( self, method: str, path: str, json: dict[str, Any] | None = None, tries: int = 0, ) -> dict[str, Any]: """Make a request to the API, retry on BridgeBusy error.""" if sleep_task := self._sleep_tasks.pop(path, None): sleep_task.cancel() try: return await self.request(method, path, json) except BridgeBusy: LOGGER.debug("Bridge is busy, schedule retry %s %s", path, str(json)) if (tries := tries + 1) < 3: self._sleep_tasks[path] = sleep_task = create_task(sleep(2 ** (tries))) try: await sleep_task except CancelledError: return {} return await self.request_with_retry(method, path, json, tries) self._sleep_tasks.pop(path, None) raise BridgeBusy async def request( self, method: str, path: str, json: dict[str, Any] | None = None, ) -> dict[str, Any]: """Make a request to the API.""" response: dict[str, Any] = await self._request( method, url=f"http://{self.host}:{self.port}/api/{self.api_key}{path}", json=json, ) return response async def _request( self, method: str, url: str, json: dict[str, Any] | None = None, ) -> Any: """Make a request.""" LOGGER.debug('Sending "%s" "%s" to "%s"', method, json, url) try: async with self.session.request(method, url, json=json) as res: if res.content_type != "application/json": raise ResponseError( "Invalid content type: {} ({})".format(res.content_type, res) ) response = await res.json() LOGGER.debug("HTTP request response: %s", pformat(response)) _raise_on_error(response) return response except aiohttp.client_exceptions.ClientError as err: raise RequestError( "Error requesting data from {}: {}".format(self.host, err) ) from None async def session_handler(self, signal: Signal) -> None: """Signalling from websocket. data - new data available for processing. state - network state has changed. """ if not self.websocket: return if signal == Signal.DATA: self.events.handler(self.websocket.data) elif signal == Signal.CONNECTION_STATE and self.connection_status_callback: self.connection_status_callback(self.websocket.state == State.RUNNING) def _raise_on_error(data: list[dict[str, Any]] | dict[str, Any]) -> None: """Check response for error message.""" if isinstance(data, list) and data: data = data[0] if isinstance(data, dict) and "error" in data: raise_error(data["error"]) 07070100000016000041ED00000000000000000000000264453E3900000000000000000000000000000000000000000000001F00000000deconz-112/pydeconz/interfaces07070100000017000081A400000000000000000000000164453E390000001A000000000000000000000000000000000000002B00000000deconz-112/pydeconz/interfaces/__init__.py"""Program interfaces.""" 07070100000018000081A400000000000000000000000164453E3900001189000000000000000000000000000000000000003000000000deconz-112/pydeconz/interfaces/alarm_systems.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import Any from ..models import ResourceGroup from ..models.alarm_system import ( AlarmSystem, AlarmSystemArmAction, AlarmSystemDeviceTrigger, ) from .api_handlers import APIHandler class AlarmSystems(APIHandler[AlarmSystem]): """Manager of deCONZ alarm systems.""" item_cls = AlarmSystem resource_group = ResourceGroup.ALARM async def create_alarm_system(self, name: str) -> dict[str, Any]: """Create a new alarm system. After creation the arm mode is set to disarmed. """ return await self.gateway.request( "post", path=self.path, json={"name": name}, ) async def set_alarm_system_configuration( self, id: str, code0: str | None = None, armed_away_entry_delay: int | None = None, armed_away_exit_delay: int | None = None, armed_away_trigger_duration: int | None = None, armed_night_entry_delay: int | None = None, armed_night_exit_delay: int | None = None, armed_night_trigger_duration: int | None = None, armed_stay_entry_delay: int | None = None, armed_stay_exit_delay: int | None = None, armed_stay_trigger_duration: int | None = None, disarmed_entry_delay: int | None = None, disarmed_exit_delay: int | None = None, ) -> dict[str, Any]: """Set config of alarm system.""" data = { key: value for key, value in { "code0": code0, "armed_away_entry_delay": armed_away_entry_delay, "armed_away_exit_delay": armed_away_exit_delay, "armed_away_trigger_duration": armed_away_trigger_duration, "armed_night_entry_delay": armed_night_entry_delay, "armed_night_exit_delay": armed_night_exit_delay, "armed_night_trigger_duration": armed_night_trigger_duration, "armed_stay_entry_delay": armed_stay_entry_delay, "armed_stay_exit_delay": armed_stay_exit_delay, "armed_stay_trigger_duration": armed_stay_trigger_duration, "disarmed_entry_delay": disarmed_entry_delay, "disarmed_exit_delay": disarmed_exit_delay, }.items() if value is not None } return await self.gateway.request( "put", path=f"{self.path}/{id}/config", json=data, ) async def arm( self, id: str, action: AlarmSystemArmAction, pin_code: str, ) -> dict[str, Any]: """Set the alarm to away.""" return await self.gateway.request( "put", path=f"{self.path}/{id}/{action.value}", json={"code0": pin_code}, ) async def add_device( self, id: str, unique_id: str, armed_away: bool = False, armed_night: bool = False, armed_stay: bool = False, trigger: AlarmSystemDeviceTrigger | None = None, is_keypad: bool = False, ) -> dict[str, Any]: """Link device with alarm system. A device can be linked to exactly one alarm system. If it is added to another alarm system, it is automatically removed from the prior one. This request is used for adding and also for updating a device entry. The uniqueid refers to sensors, lights or keypads. Adding a light can be useful, e.g. when an alarm should be triggered, after a light is powered or switched on in the basement. For keypads and keyfobs the request body can be an empty object. """ data = {"armmask": ""} data["armmask"] += "A" if armed_away else "" data["armmask"] += "N" if armed_night else "" data["armmask"] += "S" if armed_stay else "" if trigger: data["trigger"] = trigger.value if is_keypad: data = {} return await self.gateway.request( "put", path=f"{self.path}/{id}/device/{unique_id}", json=data, ) async def remove_device(self, id: str, unique_id: str) -> dict[str, Any]: """Unlink device with alarm system.""" return await self.gateway.request( "delete", path=f"{self.path}/{id}/device/{unique_id}", ) 07070100000019000081A400000000000000000000000164453E3900002067000000000000000000000000000000000000002F00000000deconz-112/pydeconz/interfaces/api_handlers.py"""API handler base classes.""" from collections.abc import Callable, ItemsView, ValuesView import itertools from typing import TYPE_CHECKING, Any, Generic, Iterable, Iterator, KeysView from ..models import DataResource, ResourceGroup, ResourceType from ..models.event import Event, EventType if TYPE_CHECKING: from ..gateway import DeconzSession CallbackType = Callable[[EventType, str], None] SubscriptionType = tuple[Callable[[EventType, str], None], tuple[EventType, ...] | None] UnsubscribeType = Callable[[], None] ID_FILTER_ALL = "*" class APIHandler(Generic[DataResource]): """Base class for a map of API Items.""" resource_group: ResourceGroup resource_type = ResourceType.UNKNOWN resource_types: set[ResourceType] | None = None item_cls: Any def __init__(self, gateway: "DeconzSession", grouped: bool = False) -> None: """Initialize API handler.""" self.gateway = gateway self._items: dict[str, DataResource] = {} self._subscribers: dict[str, list[SubscriptionType]] = {ID_FILTER_ALL: []} self.path = f"/{self.resource_group.value}" if self.resource_types is None: self.resource_types = {self.resource_type} if not grouped: self._event_subscribe() def _event_subscribe(self) -> None: """Post initialization method.""" self.gateway.events.subscribe( self.process_event, event_filter=(EventType.ADDED, EventType.CHANGED), resource_filter=self.resource_group, ) async def update(self) -> None: """Refresh data.""" raw = await self.gateway.request("get", f"/{self.resource_group.value}") self.process_raw(raw) def process_raw(self, raw: dict[str, dict[str, Any]]) -> None: """Process full data.""" for id, raw_item in raw.items(): self.process_item(id, raw_item) def process_event(self, event: Event) -> None: """Process event.""" if event.type == EventType.CHANGED and event.id in self: self.process_item(event.id, event.changed_data) return if event.type == EventType.ADDED and event.id not in self: self.process_item(event.id, event.added_data) def process_item(self, id: str, raw: dict[str, Any]) -> None: """Process data.""" if id in self._items: obj = self._items[id] obj.update(raw) event = EventType.CHANGED else: self._items[id] = self.item_cls(id, raw) event = EventType.ADDED subscribers: list[SubscriptionType] = ( self._subscribers.get(id, []) + self._subscribers[ID_FILTER_ALL] ) for callback, event_filter in subscribers: if event_filter is not None and event not in event_filter: continue callback(event, id) def subscribe( self, callback: CallbackType, event_filter: tuple[EventType, ...] | EventType | None = None, id_filter: tuple[str] | str | None = None, ) -> UnsubscribeType: """Subscribe to events. "callback" - callback function to call when on event. Return function to unsubscribe. """ if isinstance(event_filter, EventType): event_filter = (event_filter,) _id_filter: tuple[str] if id_filter is None: _id_filter = (ID_FILTER_ALL,) elif isinstance(id_filter, str): _id_filter = (id_filter,) subscription = (callback, event_filter) for id in _id_filter: if id not in self._subscribers: self._subscribers[id] = [] self._subscribers[id].append(subscription) def unsubscribe() -> None: for id in _id_filter: if id not in self._subscribers: continue self._subscribers[id].remove(subscription) return unsubscribe def items(self) -> ItemsView[str, DataResource]: """Return dictionary of IDs and API items.""" return self._items.items() def keys(self) -> KeysView[str]: """Return item IDs.""" return self._items.keys() def values(self) -> ValuesView[DataResource]: """Return API items.""" return self._items.values() def get(self, id: str, default: Any = None) -> DataResource | None: """Get API item based on key, if no match return default.""" return self._items.get(id, default) def __getitem__(self, obj_id: str) -> DataResource: """Get API item based on ID.""" return self._items[obj_id] def __iter__(self) -> Iterator[str]: """Allow iterate over item IDs.""" return iter(self._items) class GroupedAPIHandler(Generic[DataResource]): """Represent a group of deCONZ API items.""" resource_group: ResourceGroup def __init__( self, gateway: "DeconzSession", handlers: list[APIHandler[DataResource]] ) -> None: """Initialize grouped API handler.""" self.gateway = gateway self._handlers = handlers self._resource_type_to_handler: dict[ResourceType, APIHandler[DataResource]] = { resource_type: handler for handler in handlers if handler.resource_types is not None for resource_type in handler.resource_types } self._event_subscribe() def _event_subscribe(self) -> None: """Post initialization method.""" self.gateway.events.subscribe( self.process_event, event_filter=(EventType.ADDED, EventType.CHANGED), resource_filter=self.resource_group, ) def process_raw(self, raw: dict[str, dict[str, Any]]) -> None: """Process full data.""" for id, raw_item in raw.items(): self.process_item(id, raw_item) def process_event(self, event: Event) -> None: """Process event.""" if event.type == EventType.CHANGED and event.id in self: self.process_item(event.id, event.changed_data) elif event.type == EventType.ADDED and event.id not in self: self.process_item(event.id, event.added_data) def process_item(self, id: str, raw: dict[str, Any]) -> None: """Process item data.""" for handler in self._handlers: if id in handler: handler.process_item(id, raw) return if ( resource_type := ResourceType(raw.get("type")) ) not in self._resource_type_to_handler: return handler = self._resource_type_to_handler[resource_type] handler.process_item(id, raw) def subscribe( self, callback: CallbackType, event_filter: tuple[EventType, ...] | EventType | None = None, id_filter: tuple[str] | str | None = None, ) -> UnsubscribeType: """Subscribe to state changes for all grouped handler resources.""" subscribers = [ h.subscribe(callback, event_filter=event_filter, id_filter=id_filter) for h in self._handlers ] def unsubscribe() -> None: for subscriber in subscribers: subscriber() return unsubscribe def items(self) -> Iterable[tuple[str, DataResource]]: """Return dictionary of IDs and API items.""" return itertools.chain.from_iterable(h.items() for h in self._handlers) def keys(self) -> list[str]: """Return item IDs.""" return [id for h in self._handlers for id in h] def values(self) -> list[DataResource]: """Return API items.""" return [item for h in self._handlers for item in h.values()] def get(self, id: str, default: Any = None) -> DataResource | None: """Get API item based on key, if no match return default.""" return next((h[id] for h in self._handlers if id in h), default) def __getitem__(self, id: str) -> DataResource: """Get API item based on ID.""" if item := self.get(id): return item raise KeyError def __iter__(self) -> Iterator[str]: """Allow iterate over item IDs.""" return iter(self.keys()) 0707010000001A000081A400000000000000000000000164453E39000007C1000000000000000000000000000000000000002900000000deconz-112/pydeconz/interfaces/events.py"""Mange events from deCONZ.""" import logging from typing import TYPE_CHECKING, Any, Callable from ..models import ResourceGroup from ..models.event import Event, EventType if TYPE_CHECKING: from ..gateway import DeconzSession LOGGER = logging.getLogger(__name__) SubscriptionType = tuple[ Callable[[Event], None], tuple[EventType, ...] | None, tuple[ResourceGroup, ...] | None, ] UnsubscribeType = Callable[[], None] class EventHandler: """Event handler class.""" def __init__(self, gateway: "DeconzSession") -> None: """Initialize API items.""" self.gateway = gateway self._subscribers: list[SubscriptionType] = [] def subscribe( self, callback: Callable[[Event], None], event_filter: tuple[EventType, ...] | EventType | None = None, resource_filter: tuple[ResourceGroup, ...] | ResourceGroup | None = None, ) -> UnsubscribeType: """Subscribe to events. "callback" - callback function to call when on event. Return function to unsubscribe. """ if isinstance(event_filter, EventType): event_filter = (event_filter,) if isinstance(resource_filter, ResourceGroup): resource_filter = (resource_filter,) subscription = (callback, event_filter, resource_filter) self._subscribers.append(subscription) def unsubscribe() -> None: self._subscribers.remove(subscription) return unsubscribe def handler(self, raw: dict[str, Any]) -> None: """Receive event from websocket and pass it along to subscribers.""" event = Event.from_dict(raw) for callback, event_filter, resource_filter in self._subscribers: if event_filter is not None and event.type not in event_filter: continue if resource_filter is not None and event.resource not in resource_filter: continue callback(event) 0707010000001B000081A400000000000000000000000164453E3900000FDF000000000000000000000000000000000000002900000000deconz-112/pydeconz/interfaces/groups.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import Any from ..models import ResourceGroup, ResourceType from ..models.group import Group from ..models.light.light import LightAlert, LightEffect from .api_handlers import APIHandler class GroupHandler(APIHandler[Group]): """Represent deCONZ groups.""" resource_group = ResourceGroup.GROUP resource_type = ResourceType.GROUP item_cls = Group async def set_attributes( self, id: str, hidden: bool | None = None, light_sequence: list[str] | None = None, lights: list[str] | None = None, multi_device_ids: list[str] | None = None, name: str | None = None, ) -> dict[str, Any]: """Change attributes of a group. Supported values: - hidden [bool] Indicates the hidden status of the group - light_sequence [list of light IDs] Specify a sorted list of light IDs for apps - lights [list of light IDs]IDs of the lights which are members of the group - multi_device_ids [int] Subsequential light IDs of multidevices - name [str] The name of the group """ data = { key: value for key, value in { "hidden": hidden, "lightsequence": light_sequence, "lights": lights, "multideviceids": multi_device_ids, "name": name, }.items() if value is not None } return await self.gateway.request( "put", path=f"{self.path}/{id}", json=data, ) async def set_state( self, id: str, alert: LightAlert | None = None, brightness: int | None = None, color_loop_speed: int | None = None, color_temperature: int | None = None, effect: LightEffect | None = None, hue: int | None = None, on: bool | None = None, on_time: int | None = None, saturation: int | None = None, toggle: bool | None = None, transition_time: int | None = None, xy: tuple[float, float] | None = None, ) -> dict[str, Any]: """Change state of a group. Supported values: - alert [str] - "none" light is not performing an alert - "select" light is blinking a short time - "lselect" light is blinking a longer time - brightness [int] 0-255 - color_loop_speed [int] 1-255 - 1 = very fast - 15 is default - 255 very slow - color_temperature [int] between ctmin-ctmax - effect [str] - "none" no effect - "colorloop" the light will cycle continuously through all colors with the speed specified by colorloopspeed - hue [int] 0-65535 - on [bool] True/False - on_time [int] 0-65535 1/10 seconds resolution - saturation [int] 0-255 - toggle [bool] True toggles the lights of that group from on to off or vice versa, false has no effect - transition_time [int] 0-65535 1/10 seconds resolution - xy [tuple] (0-1, 0-1) """ data: dict[str, Any] = { key: value for key, value in { "bri": brightness, "colorloopspeed": color_loop_speed, "ct": color_temperature, "hue": hue, "on": on, "ontime": on_time, "sat": saturation, "toggle": toggle, "transitiontime": transition_time, "xy": xy, }.items() if value is not None } if alert is not None: data["alert"] = alert.value if effect is not None: data["effect"] = effect.value return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/action", json=data, ) 0707010000001C000081A400000000000000000000000164453E3900002061000000000000000000000000000000000000002900000000deconz-112/pydeconz/interfaces/lights.py"""Python library to connect deCONZ and Home Assistant to work together.""" import logging from typing import TYPE_CHECKING, Any from ..models import ResourceGroup, ResourceType from ..models.light.configuration_tool import ConfigurationTool from ..models.light.cover import Cover, CoverAction from ..models.light.light import Light, LightAlert, LightEffect, LightFanSpeed from ..models.light.lock import Lock from ..models.light.range_extender import RangeExtender from ..models.light.siren import Siren from .api_handlers import APIHandler, GroupedAPIHandler if TYPE_CHECKING: from ..gateway import DeconzSession LOGGER = logging.getLogger(__name__) class ConfigurationToolHandler(APIHandler[ConfigurationTool]): """Handler for configuration tool.""" resource_group = ResourceGroup.LIGHT resource_type = ResourceType.CONFIGURATION_TOOL item_cls = ConfigurationTool class CoverHandler(APIHandler[Cover]): """Handler for covers.""" resource_group = ResourceGroup.LIGHT resource_types = { ResourceType.LEVEL_CONTROLLABLE_OUTPUT, ResourceType.WINDOW_COVERING_CONTROLLER, ResourceType.WINDOW_COVERING_DEVICE, } item_cls = Cover async def set_state( self, id: str, action: CoverAction | None = None, lift: int | None = None, tilt: int | None = None, legacy_mode: bool = False, ) -> dict[str, Any]: """Set state of cover. Supported values: - action [CoverAction] Open, Close, Stop - lift [int] between 0-100 - tilt [int] between 0-100 """ data: dict[str, bool | int] = {} if action is not None and not legacy_mode: if action is CoverAction.OPEN: data["open"] = True elif action is CoverAction.CLOSE: data["open"] = False elif action is CoverAction.STOP: data["stop"] = True elif action is not None and legacy_mode: if action is CoverAction.OPEN: data["on"] = False elif action is CoverAction.CLOSE: data["on"] = True elif action is CoverAction.STOP: data["bri_inc"] = 0 elif not legacy_mode: if lift is not None: data["lift"] = lift if tilt is not None: data["tilt"] = tilt else: if lift is not None: data["bri"] = int(lift * 2.54) if tilt is not None: data["sat"] = int(tilt * 2.54) return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/state", json=data, ) class LightHandler(APIHandler[Light]): """Handler for lights.""" resource_group = ResourceGroup.LIGHT resource_types = { ResourceType.COLOR_DIMMABLE_LIGHT, ResourceType.COLOR_LIGHT, ResourceType.COLOR_TEMPERATURE_LIGHT, ResourceType.EXTENDED_COLOR_LIGHT, ResourceType.DIMMABLE_LIGHT, ResourceType.DIMMABLE_PLUGIN_UNIT, ResourceType.DIMMER_SWITCH, ResourceType.FAN, ResourceType.ON_OFF_LIGHT, ResourceType.ON_OFF_OUTPUT, ResourceType.ON_OFF_PLUGIN_UNIT, ResourceType.SMART_PLUG, ResourceType.UNKNOWN, # Legacy support } item_cls = Light async def set_state( self, id: str, alert: LightAlert | None = None, brightness: int | None = None, color_loop_speed: int | None = None, color_temperature: int | None = None, effect: LightEffect | None = None, fan_speed: LightFanSpeed | None = None, hue: int | None = None, on: bool | None = None, on_time: int | None = None, saturation: int | None = None, transition_time: int | None = None, xy: tuple[float, float] | None = None, ) -> dict[str, Any]: """Change state of a light. Supported values: - alert [str] - "none" light is not performing an alert - "select" light is blinking a short time - "lselect" light is blinking a longer time - brightness [int] 0-255 - color_loop_speed [int] 1-255 - 1 = very fast - 15 is default - 255 very slow - color_temperature [int] between ctmin-ctmax - effect [str] - "none" no effect - "colorloop" the light will cycle continuously through all colors with the speed specified by colorloopspeed - fan_speed [FanSpeed] Off, 25%, 50%, 75%, 100%, Auto, ComfortBreeze - hue [int] 0-65535 - on [bool] True/False - on_time [int] 0-65535 1/10 seconds resolution - saturation [int] 0-255 - transition_time [int] 0-65535 1/10 seconds resolution - xy [tuple] (0-1, 0-1) """ data: dict[str, Any] = { key: value for key, value in { "bri": brightness, "colorloopspeed": color_loop_speed, "ct": color_temperature, "hue": hue, "on": on, "ontime": on_time, "sat": saturation, "transitiontime": transition_time, "xy": xy, }.items() if value is not None } if alert is not None: data["alert"] = alert.value if effect is not None: data["effect"] = effect.value if fan_speed is not None: data["speed"] = fan_speed.value return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/state", json=data, ) class LockHandler(APIHandler[Lock]): """Handler for fans.""" resource_group = ResourceGroup.LIGHT resource_type = ResourceType.DOOR_LOCK item_cls = Lock async def set_state(self, id: str, lock: bool) -> dict[str, Any]: """Set state of lock. Supported values: - lock [bool] True/False. """ return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/state", json={"on": lock}, ) class RangeExtenderHandler(APIHandler[ConfigurationTool]): """Handler for range extender.""" resource_group = ResourceGroup.LIGHT resource_type = ResourceType.RANGE_EXTENDER item_cls = RangeExtender class SirenHandler(APIHandler[Siren]): """Handler for sirens.""" resource_group = ResourceGroup.LIGHT resource_type = ResourceType.WARNING_DEVICE item_cls = Siren async def set_state( self, id: str, on: bool, duration: int | None = None, ) -> dict[str, Any]: """Turn on device. Supported values: - on [bool] True/False - duration [int] 1/10th of a second """ data: dict[str, int | str] = {} data["alert"] = (LightAlert.LONG if on else LightAlert.NONE).value if on and duration is not None: data["ontime"] = duration return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/state", json=data, ) LightResources = ConfigurationTool | Cover | Light | Lock | Siren class LightResourceManager(GroupedAPIHandler[LightResources]): """Represent deCONZ lights.""" resource_group = ResourceGroup.LIGHT def __init__(self, gateway: "DeconzSession") -> None: """Initialize light manager.""" self.configuration_tool = ConfigurationToolHandler(gateway, grouped=True) self.covers = CoverHandler(gateway, grouped=True) self.lights = LightHandler(gateway, grouped=True) self.locks = LockHandler(gateway, grouped=True) self.range_extender = RangeExtenderHandler(gateway, grouped=True) self.sirens = SirenHandler(gateway, grouped=True) handlers: list[APIHandler[Any]] = [ self.configuration_tool, self.covers, self.lights, self.locks, self.range_extender, self.sirens, ] super().__init__(gateway, handlers) 0707010000001D000081A400000000000000000000000164453E3900000A8F000000000000000000000000000000000000002900000000deconz-112/pydeconz/interfaces/scenes.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import Any, cast from ..models import ResourceGroup from ..models.event import EventType from ..models.scene import Scene from .api_handlers import APIHandler class Scenes(APIHandler[Scene]): """Represent scenes of a deCONZ group.""" item_cls = Scene resource_group = ResourceGroup.SCENE def _event_subscribe(self) -> None: """Register for group data events.""" self.gateway.groups.subscribe( self.group_data_callback, event_filter=(EventType.ADDED, EventType.CHANGED), ) async def create_scene(self, group_id: str, name: str) -> dict[str, Any]: """Create a new scene. The current state of each light will become the lights scene state. Supported values: - name [str] """ return await self.gateway.request( "post", path=f"/groups/{group_id}/scenes", json={"name": name}, ) async def recall(self, group_id: str, scene_id: str) -> dict[str, Any]: """Recall scene to group.""" return await self.gateway.request_with_retry( "put", path=f"/groups/{group_id}/scenes/{scene_id}/recall", json={}, ) async def store(self, group_id: str, scene_id: str) -> dict[str, Any]: """Store current group state in scene. The actual state of each light in the group will become the lights scene state. """ return await self.gateway.request_with_retry( "put", path=f"/groups/{group_id}/scenes/{scene_id}/store", json={}, ) async def set_attributes( self, group_id: str, scene_id: str, name: str | None = None ) -> dict[str, Any]: """Change attributes of scene. Supported values: - name [str] """ data = { key: value for key, value in { "name": name, }.items() if value is not None } return await self.gateway.request_with_retry( "put", path=f"/groups/{group_id}/scenes/{scene_id}", json=data, ) def group_data_callback(self, action: EventType, group_id: str) -> None: """Subscribe callback for new group data.""" self.process_item(group_id, {}) def process_item(self, id: str, raw: dict[str, Any]) -> None: """Pre-process scene data.""" group = self.gateway.groups[id] for scene in group.raw["scenes"]: super().process_item(f'{id}_{scene["id"]}', cast(dict[str, Any], scene)) 0707010000001E000081A400000000000000000000000164453E3900004F8E000000000000000000000000000000000000002A00000000deconz-112/pydeconz/interfaces/sensors.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TYPE_CHECKING, Any from ..models import ResourceGroup, ResourceType from ..models.sensor.air_purifier import AirPurifier, AirPurifierFanMode from ..models.sensor.air_quality import AirQuality from ..models.sensor.alarm import Alarm from ..models.sensor.ancillary_control import AncillaryControl from ..models.sensor.battery import Battery from ..models.sensor.carbon_monoxide import CarbonMonoxide from ..models.sensor.consumption import Consumption from ..models.sensor.daylight import Daylight from ..models.sensor.door_lock import DoorLock from ..models.sensor.fire import Fire from ..models.sensor.generic_flag import GenericFlag from ..models.sensor.generic_status import GenericStatus from ..models.sensor.humidity import Humidity from ..models.sensor.light_level import LightLevel from ..models.sensor.moisture import Moisture from ..models.sensor.open_close import OpenClose from ..models.sensor.power import Power from ..models.sensor.presence import ( Presence, PresenceConfigDeviceMode, PresenceConfigTriggerDistance, ) from ..models.sensor.pressure import Pressure from ..models.sensor.relative_rotary import RelativeRotary from ..models.sensor.switch import ( Switch, SwitchDeviceMode, SwitchMode, SwitchWindowCoveringType, ) from ..models.sensor.temperature import Temperature from ..models.sensor.thermostat import ( Thermostat, ThermostatFanMode, ThermostatMode, ThermostatPreset, ThermostatSwingMode, ThermostatTemperatureMeasurement, ) from ..models.sensor.time import Time from ..models.sensor.vibration import Vibration from ..models.sensor.water import Water from .api_handlers import APIHandler, GroupedAPIHandler if TYPE_CHECKING: from ..gateway import DeconzSession class AirPurifierHandler(APIHandler[AirPurifier]): """Handler for air purifier sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_AIR_PURIFIER item_cls = AirPurifier async def set_config( self, id: str, fan_mode: AirPurifierFanMode | None = None, filter_life_time: int | None = None, led_indication: bool | None = None, locked: bool | None = None, ) -> dict[str, Any]: """Set speed of fans/ventilators. Supported values: - fan_mode [AirPurifierFanMode] - "off" - "auto" - "speed_1" - "speed_2" - "speed_3" - "speed_4" - "speed_5" - filter_life_time [int] 0-65536 - led_indication [bool] True/False - locked [bool] True/False """ data: dict[str, int | str] = { key: value for key, value in { "filterlifetime": filter_life_time, "ledindication": led_indication, "locked": locked, }.items() if value is not None } if fan_mode is not None: data["mode"] = fan_mode.value return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/config", json=data, ) class AirQualityHandler(APIHandler[AirQuality]): """Handler for air quality sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_AIR_QUALITY item_cls = AirQuality class AlarmHandler(APIHandler[Alarm]): """Handler for alarm sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_ALARM item_cls = Alarm class AncillaryControlHandler(APIHandler[AncillaryControl]): """Handler for ancillary control sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_ANCILLARY_CONTROL item_cls = AncillaryControl class BatteryHandler(APIHandler[Battery]): """Handler for battery sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_BATTERY item_cls = Battery class CarbonMonoxideHandler(APIHandler[CarbonMonoxide]): """Handler for carbon monoxide sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_CARBON_MONOXIDE item_cls = CarbonMonoxide class ConsumptionHandler(APIHandler[Consumption]): """Handler for consumption sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_CONSUMPTION item_cls = Consumption class DaylightHandler(APIHandler[Daylight]): """Handler for daylight sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.DAYLIGHT item_cls = Daylight async def set_config( self, id: str, sunrise_offset: int | None = None, sunset_offset: int | None = None, ) -> dict[str, Any]: """Set config of the daylight sensor. Supported values: - sunrise_offset [int] -120-120 - sunset_offset [int] -120-120 """ data: dict[str, int] = { key: value for key, value in { "sunriseoffset": sunrise_offset, "sunsetoffset": sunset_offset, }.items() if value is not None } return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/config", json=data, ) class DoorLockHandler(APIHandler[DoorLock]): """Handler for door lock sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_DOOR_LOCK item_cls = DoorLock async def set_config(self, id: str, lock: bool) -> dict[str, Any]: """Set config of the lock. Supported values: - Lock [bool] True/False. """ return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/config", json={"lock": lock}, ) class FireHandler(APIHandler[Fire]): """Handler for fire sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_FIRE item_cls = Fire class GenericFlagHandler(APIHandler[GenericFlag]): """Handler for generic flag sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.CLIP_GENERIC_FLAG item_cls = GenericFlag class GenericStatusHandler(APIHandler[GenericStatus]): """Handler for generic status sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.CLIP_GENERIC_STATUS item_cls = GenericStatus class HumidityHandler(APIHandler[Humidity]): """Handler for humidity sensor.""" resource_group = ResourceGroup.SENSOR resource_types = { ResourceType.ZHA_HUMIDITY, ResourceType.CLIP_HUMIDITY, } item_cls = Humidity async def set_config(self, id: str, offset: int) -> dict[str, Any]: """Change config of humidity sensor. Supported values: - offset [int] -32768–32767 """ return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/config", json={"offset": offset}, ) class LightLevelHandler(APIHandler[LightLevel]): """Handler for light level sensor.""" resource_group = ResourceGroup.SENSOR resource_types = { ResourceType.ZHA_LIGHT_LEVEL, ResourceType.CLIP_LIGHT_LEVEL, } item_cls = LightLevel async def set_config( self, id: str, threshold_dark: int | None = None, threshold_offset: int | None = None, ) -> dict[str, Any]: """Change config of presence sensor. Supported values: - threshold_dark [int] 0-65534 - threshold_offset [int] 1-65534 """ data = { key: value for key, value in { "tholddark": threshold_dark, "tholdoffset": threshold_offset, }.items() if value is not None } return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/config", json=data, ) class MoistureHandler(APIHandler[Moisture]): """Handler for moisture sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_MOISTURE item_cls = Moisture class OpenCloseHandler(APIHandler[OpenClose]): """Handler for open/close sensor.""" resource_group = ResourceGroup.SENSOR resource_types = { ResourceType.ZHA_OPEN_CLOSE, ResourceType.CLIP_OPEN_CLOSE, } item_cls = OpenClose class PowerHandler(APIHandler[Power]): """Handler for power sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_POWER item_cls = Power class PresenceHandler(APIHandler[Presence]): """Handler for presence sensor.""" resource_group = ResourceGroup.SENSOR resource_types = { ResourceType.ZHA_PRESENCE, ResourceType.CLIP_PRESENCE, } item_cls = Presence async def set_config( self, id: str, delay: int | None = None, device_mode: PresenceConfigDeviceMode | None = None, duration: int | None = None, reset_presence: bool | None = None, sensitivity: int | None = None, trigger_distance: PresenceConfigTriggerDistance | None = None, ) -> dict[str, Any]: """Change config of presence sensor. Supported values: - delay [int] 0-65535 (in seconds) - device_mode [str] - leftright - undirected - duration [int] 0-65535 (in seconds) - reset_presence [bool] True/False - sensitivity [int] 0-[sensitivitymax] - trigger_distance [str] - far - medium - near """ data: dict[str, int | str] = { key: value for key, value in { "delay": delay, "duration": duration, "resetpresence": reset_presence, "sensitivity": sensitivity, }.items() if value is not None } if device_mode is not None: data["devicemode"] = device_mode.value if trigger_distance is not None: data["triggerdistance"] = trigger_distance.value return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/config", json=data, ) class PressureHandler(APIHandler[Pressure]): """Handler for pressure sensor.""" resource_group = ResourceGroup.SENSOR resource_types = { ResourceType.ZHA_PRESSURE, ResourceType.CLIP_PRESSURE, } item_cls = Pressure class RelativeRotaryHandler(APIHandler[RelativeRotary]): """Handler for relative rotary sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_RELATIVE_ROTARY item_cls = RelativeRotary class SwitchHandler(APIHandler[Switch]): """Handler for switch sensor.""" resource_group = ResourceGroup.SENSOR resource_types = { ResourceType.ZHA_SWITCH, ResourceType.ZGP_SWITCH, ResourceType.CLIP_SWITCH, } item_cls = Switch async def set_config( self, id: str, device_mode: SwitchDeviceMode | None = None, mode: SwitchMode | None = None, window_covering_type: SwitchWindowCoveringType | None = None, ) -> dict[str, Any]: """Change config of presence sensor. Supported values: - device_mode [SwitchDeviceMode] - "dualpushbutton" - "dualrocker" - "singlepushbutton" - "singlerocker" - mode [SwitchMode] - "momentary" - "rocker" - window_covering_type [SwitchWindowCoveringType] 0-9 """ data: dict[str, int | str] = {} if device_mode is not None: data["devicemode"] = device_mode.value if mode is not None: data["mode"] = mode.value if window_covering_type is not None: data["windowcoveringtype"] = window_covering_type.value return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/config", json=data, ) class TemperatureHandler(APIHandler[Temperature]): """Handler for temperature sensor.""" resource_group = ResourceGroup.SENSOR resource_types = { ResourceType.ZHA_TEMPERATURE, ResourceType.CLIP_TEMPERATURE, } item_cls = Temperature class ThermostatHandler(APIHandler[Thermostat]): """Handler for thermostat sensor.""" resource_group = ResourceGroup.SENSOR resource_types = { ResourceType.ZHA_THERMOSTAT, ResourceType.CLIP_THERMOSTAT, } item_cls = Thermostat async def set_config( self, id: str, cooling_setpoint: int | None = None, enable_schedule: bool | None = None, external_sensor_temperature: int | None = None, external_window_open: bool | None = None, fan_mode: ThermostatFanMode | None = None, flip_display: bool | None = None, heating_setpoint: int | None = None, locked: bool | None = None, mode: ThermostatMode | None = None, mounting_mode: bool | None = None, on: bool | None = None, preset: ThermostatPreset | None = None, schedule: list[str] | None = None, set_valve: bool | None = None, swing_mode: ThermostatSwingMode | None = None, temperature_measurement: ThermostatTemperatureMeasurement | None = None, window_open_detection: bool | None = None, ) -> dict[str, Any]: """Change config of thermostat. Supported values: - cooling_setpoint [int] 700-3500 - enable_schedule [bool] True/False - external_sensor_temperature [int] -32768-32767 - external_window_open [bool] True/False - fan_mode [ThermostatFanMode] - "auto" - "high" - "low" - "medium" - "off" - "on" - "smart" - flip_display [bool] True/False - heating_setpoint [int] 500-3200 - locked [bool] True/False - mode [ThermostatMode] - "auto" - "cool" - "dry" - "emergency heating" - "fan only" - "heat" - "off" - "precooling" - "sleep" - mounting_mode [bool] True/False - on [bool] True/False - preset [ThermostatPreset] - "auto" - "boost" - "comfort" - "complex" - "eco" - "holiday" - "manual" - schedule [list] - set_valve [bool] True/False - swing_mode [ThermostatSwingMode] - "fully closed" - "fully open" - "half open" - "quarter open" - "three quarters open" - temperature_measurement [ThermostatTemperatureMeasurement] - "air sensor" - "floor protection" - "floor sensor" - window_open_detection [bool] True/False """ data: dict[str, Any] = { key: value for key, value in { "coolsetpoint": cooling_setpoint, "schedule_on": enable_schedule, "externalsensortemp": external_sensor_temperature, "externalwindowopen": external_window_open, "displayflipped": flip_display, "heatsetpoint": heating_setpoint, "locked": locked, "mountingmode": mounting_mode, "on": on, "schedule": schedule, "setvalve": set_valve, "windowopen_set": window_open_detection, }.items() if value is not None } if fan_mode is not None: data["fanmode"] = fan_mode.value if mode is not None: data["mode"] = mode.value if preset is not None: data["preset"] = preset.value if swing_mode is not None: data["swingmode"] = swing_mode.value if temperature_measurement is not None: data["temperaturemeasurement"] = temperature_measurement.value return await self.gateway.request_with_retry( "put", path=f"{self.path}/{id}/config", json=data, ) class TimeHandler(APIHandler[Time]): """Handler for time sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_TIME item_cls = Time class VibrationHandler(APIHandler[Vibration]): """Handler for vibration sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_VIBRATION item_cls = Vibration class WaterHandler(APIHandler[Water]): """Handler for water sensor.""" resource_group = ResourceGroup.SENSOR resource_type = ResourceType.ZHA_WATER item_cls = Water SensorResources = ( AirPurifier | AirQuality | Alarm | AncillaryControl | Battery | CarbonMonoxide | Consumption | Daylight | DoorLock | Fire | GenericFlag | GenericStatus | Humidity | LightLevel | Moisture | OpenClose | Power | Presence | Pressure | RelativeRotary | Switch | Temperature | Thermostat | Time | Vibration | Water ) class SensorResourceManager(GroupedAPIHandler[SensorResources]): """Represent deCONZ sensors.""" resource_group = ResourceGroup.SENSOR def __init__(self, gateway: "DeconzSession") -> None: """Initialize sensor manager.""" self.air_purifier = AirPurifierHandler(gateway, grouped=True) self.air_quality = AirQualityHandler(gateway, grouped=True) self.alarm = AlarmHandler(gateway, grouped=True) self.ancillary_control = AncillaryControlHandler(gateway, grouped=True) self.battery = BatteryHandler(gateway, grouped=True) self.carbon_monoxide = CarbonMonoxideHandler(gateway, grouped=True) self.consumption = ConsumptionHandler(gateway, grouped=True) self.daylight = DaylightHandler(gateway, grouped=True) self.door_lock = DoorLockHandler(gateway, grouped=True) self.fire = FireHandler(gateway, grouped=True) self.generic_flag = GenericFlagHandler(gateway, grouped=True) self.generic_status = GenericStatusHandler(gateway, grouped=True) self.humidity = HumidityHandler(gateway, grouped=True) self.light_level = LightLevelHandler(gateway, grouped=True) self.open_close = OpenCloseHandler(gateway, grouped=True) self.moisture = MoistureHandler(gateway, grouped=True) self.power = PowerHandler(gateway, grouped=True) self.presence = PresenceHandler(gateway, grouped=True) self.pressure = PressureHandler(gateway, grouped=True) self.relative_rotary = RelativeRotaryHandler(gateway, grouped=True) self.switch = SwitchHandler(gateway, grouped=True) self.temperature = TemperatureHandler(gateway, grouped=True) self.thermostat = ThermostatHandler(gateway, grouped=True) self.time = TimeHandler(gateway, grouped=True) self.vibration = VibrationHandler(gateway, grouped=True) self.water = WaterHandler(gateway, grouped=True) handlers: list[APIHandler[Any]] = [ self.air_purifier, self.air_quality, self.alarm, self.ancillary_control, self.battery, self.carbon_monoxide, self.consumption, self.daylight, self.door_lock, self.fire, self.generic_flag, self.generic_status, self.humidity, self.light_level, self.moisture, self.open_close, self.power, self.presence, self.pressure, self.relative_rotary, self.switch, self.temperature, self.thermostat, self.time, self.vibration, self.water, ] super().__init__(gateway, handlers) 0707010000001F000041ED00000000000000000000000464453E3900000000000000000000000000000000000000000000001B00000000deconz-112/pydeconz/models07070100000020000081A400000000000000000000000164453E3900000D67000000000000000000000000000000000000002700000000deconz-112/pydeconz/models/__init__.py"""Data models.""" from enum import Enum import logging from typing import TypeVar from .api import APIItem LOGGER = logging.getLogger(__name__) DataResource = TypeVar("DataResource", bound=APIItem) class ResourceGroup(Enum): """Primary endpoints resources are exposed from.""" ALARM = "alarmsystems" CONFIG = "config" GROUP = "groups" LIGHT = "lights" SCENE = "scenes" SENSOR = "sensors" class ResourceType(Enum): """Resource types.""" # Group resources GROUP = "LightGroup" # Light resources # Configuration tool CONFIGURATION_TOOL = "Configuration tool" # Cover LEVEL_CONTROLLABLE_OUTPUT = "Level controllable output" WINDOW_COVERING_CONTROLLER = "Window covering controller" WINDOW_COVERING_DEVICE = "Window covering device" # Light COLOR_DIMMABLE_LIGHT = "Color dimmable light" COLOR_LIGHT = "Color light" COLOR_TEMPERATURE_LIGHT = "Color temperature light" EXTENDED_COLOR_LIGHT = "Extended color light" DIMMABLE_LIGHT = "Dimmable light" DIMMABLE_PLUGIN_UNIT = "Dimmable plug-in unit" DIMMER_SWITCH = "Dimmer switch" FAN = "Fan" ON_OFF_LIGHT = "On/Off light" ON_OFF_OUTPUT = "On/Off output" ON_OFF_PLUGIN_UNIT = "On/Off plug-in unit" SMART_PLUG = "Smart plug" # Lock DOOR_LOCK = "Door Lock" # Range extender RANGE_EXTENDER = "Range extender" # Siren WARNING_DEVICE = "Warning device" # Sensor resources # Air purifier ZHA_AIR_PURIFIER = "ZHAAirPurifier" # Air quality ZHA_AIR_QUALITY = "ZHAAirQuality" # Alarm ZHA_ALARM = "ZHAAlarm" # Ancillary control ZHA_ANCILLARY_CONTROL = "ZHAAncillaryControl" # Battery ZHA_BATTERY = "ZHABattery" # Carbon monoxide ZHA_CARBON_MONOXIDE = "ZHACarbonMonoxide" # Consumption ZHA_CONSUMPTION = "ZHAConsumption" # Daylight DAYLIGHT = "Daylight" # Door lock ZHA_DOOR_LOCK = "ZHADoorLock" # Fire ZHA_FIRE = "ZHAFire" # Generic flag CLIP_GENERIC_FLAG = "CLIPGenericFlag" # Generic status CLIP_GENERIC_STATUS = "CLIPGenericStatus" # Humidity ZHA_HUMIDITY = "ZHAHumidity" CLIP_HUMIDITY = "CLIPHumidity" # Light level ZHA_LIGHT_LEVEL = "ZHALightLevel" CLIP_LIGHT_LEVEL = "CLIPLightLevel" # Moisture ZHA_MOISTURE = "ZHAMoisture" # Open close ZHA_OPEN_CLOSE = "ZHAOpenClose" CLIP_OPEN_CLOSE = "CLIPOpenClose" # Power ZHA_POWER = "ZHAPower" # Presence ZHA_PRESENCE = "ZHAPresence" CLIP_PRESENCE = "CLIPPresence" # Pressure ZHA_PRESSURE = "ZHAPressure" CLIP_PRESSURE = "CLIPPressure" # Relative rotary ZHA_RELATIVE_ROTARY = "ZHARelativeRotary" # Switch ZHA_SWITCH = "ZHASwitch" ZGP_SWITCH = "ZGPSwitch" CLIP_SWITCH = "CLIPSwitch" # Temperature ZHA_TEMPERATURE = "ZHATemperature" CLIP_TEMPERATURE = "CLIPTemperature" # Thermostat ZHA_THERMOSTAT = "ZHAThermostat" CLIP_THERMOSTAT = "CLIPThermostat" # Time ZHA_TIME = "ZHATime" # Vibration ZHA_VIBRATION = "ZHAVibration" # Water ZHA_WATER = "ZHAWater" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "ResourceType": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unsupported device type %s", value) return ResourceType.UNKNOWN 07070100000021000081A400000000000000000000000164453E3900001B60000000000000000000000000000000000000002B00000000deconz-112/pydeconz/models/alarm_system.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum import logging from typing import Any, Literal, TypedDict from . import ResourceGroup from .api import APIItem LOGGER = logging.getLogger(__name__) class AlarmSystemArmAction(enum.Enum): """Explicit url path to arm and disarm.""" AWAY = "arm_away" NIGHT = "arm_night" STAY = "arm_stay" DISARM = "disarm" class AlarmSystemArmMode(enum.Enum): """The target arm mode.""" ARMED_AWAY = "armed_away" ARMED_NIGHT = "armed_night" ARMED_STAY = "armed_stay" DISARMED = "disarmed" class AlarmSystemArmState(enum.Enum): """The current alarm system state.""" ARMED_AWAY = "armed_away" ARMED_NIGHT = "armed_night" ARMED_STAY = "armed_stay" ARMING_AWAY = "arming_away" ARMING_NIGHT = "arming_night" ARMING_STAY = "arming_stay" DISARMED = "disarmed" ENTRY_DELAY = "entry_delay" EXIT_DELAY = "exit_delay" IN_ALARM = "in_alarm" class AlarmSystemArmMask(enum.Enum): """The target arm mode.""" ARMED_AWAY = "A" ARMED_NIGHT = "N" ARMED_STAY = "S" NONE = "none" class AlarmSystemDeviceTrigger(enum.Enum): """Specifies arm modes in which the device triggers alarms.""" ACTION = "state/action" BUTTON_EVENT = "state/buttonevent" ON = "state/on" OPEN = "state/open" PRESENCE = "state/presence" VIBRATION = "state/vibration" class TypedAlarmSystemConfig(TypedDict): """Alarm system config type definition.""" armmode: Literal["armed_away", "armed_night", "armed_stay", "disarmed"] configured: bool disarmed_entry_delay: int disarmed_exit_delay: int armed_away_entry_delay: int armed_away_exit_delay: int armed_away_trigger_duration: int armed_stay_entry_delay: int armed_stay_exit_delay: int armed_stay_trigger_duration: int armed_night_entry_delay: int armed_night_exit_delay: int armed_night_trigger_duration: int class TypedAlarmSystemState(TypedDict): """Alarm system state type definition.""" armstate: Literal[ "armed_away", "armed_night", "armed_stay", "arming_away", "arming_night", "arming_stay", "disarmed", "entry_delay", "exit_delay", "in_alarm", ] seconds_remaining: int class TypedAlarmSystemDevices(TypedDict): """Alarm system device type definition.""" armmask: str trigger: str class TypedAlarmSystem(TypedDict): """Alarm system type definition.""" name: str config: TypedAlarmSystemConfig state: TypedAlarmSystemState devices: dict[str, TypedAlarmSystemDevices] class AlarmSystem(APIItem): """deCONZ alarm system representation. Dresden Elektroniks documentation of alarm systems in deCONZ https://dresden-elektronik.github.io/deconz-rest-doc/endpoints/alarmsystems/ """ raw: TypedAlarmSystem resource_group = ResourceGroup.ALARM @property def arm_state(self) -> AlarmSystemArmState: """Alarm system state. Can be different from the config.armmode during state transitions. """ return AlarmSystemArmState(self.raw["state"]["armstate"]) @property def seconds_remaining(self) -> int: """Remaining time while armstate in "exit_delay" or "entry_delay" state. In all other states the value is 0. Supported values: 0-255. """ return self.raw["state"]["seconds_remaining"] @property def pin_configured(self) -> bool: """Is PIN code configured.""" return self.raw["config"]["configured"] @property def arm_mode(self) -> AlarmSystemArmMode: """Target arm mode.""" return AlarmSystemArmMode(self.raw["config"]["armmode"]) @property def armed_away_entry_delay(self) -> int: """Delay in seconds before an alarm is triggered. Supported values: 0-255. """ return self.raw["config"]["armed_away_entry_delay"] @property def armed_away_exit_delay(self) -> int: """Delay in seconds before an alarm is armed. Supported values: 0-255. """ return self.raw["config"]["armed_away_exit_delay"] @property def armed_away_trigger_duration(self) -> int: """Duration of alarm trigger. Supported values: 0-255. """ return self.raw["config"]["armed_away_trigger_duration"] @property def armed_night_entry_delay(self) -> int: """Delay in seconds before an alarm is triggered. Supported values: 0-255. """ return self.raw["config"]["armed_night_entry_delay"] @property def armed_night_exit_delay(self) -> int: """Delay in seconds before an alarm is armed. Supported values: 0-255. """ return self.raw["config"]["armed_night_exit_delay"] @property def armed_night_trigger_duration(self) -> int: """Duration of alarm trigger. Supported values: 0-255. """ return self.raw["config"]["armed_night_trigger_duration"] @property def armed_stay_entry_delay(self) -> int: """Delay in seconds before an alarm is triggered. Supported values: 0-255. """ return self.raw["config"]["armed_stay_entry_delay"] @property def armed_stay_exit_delay(self) -> int: """Delay in seconds before an alarm is armed. Supported values: 0-255. """ return self.raw["config"]["armed_stay_exit_delay"] @property def armed_stay_trigger_duration(self) -> int: """Duration of alarm trigger. Supported values: 0-255. """ return self.raw["config"]["armed_stay_trigger_duration"] @property def disarmed_entry_delay(self) -> int: """Delay in seconds before an alarm is triggered. Supported values: 0-255. """ return self.raw["config"]["disarmed_entry_delay"] @property def disarmed_exit_delay(self) -> int: """Delay in seconds before an alarm is armed. Supported values: 0-255. """ return self.raw["config"]["disarmed_exit_delay"] @property def devices(self) -> dict[str, Any]: """Devices associated with the alarm system. The keys refer to the uniqueid of a light, sensor, or keypad. Dictionary values: - armmask - A combination of arm modes in which the device triggers alarms. A — armed_away N — armed_night S — armed_stay "none" — for keypads and keyfobs - trigger - Specifies arm modes in which the device triggers alarms. "state/presence" "state/open" "state/vibration" "state/buttonevent" "state/on" """ return self.raw["devices"] 07070100000022000081A400000000000000000000000164453E3900000881000000000000000000000000000000000000002200000000deconz-112/pydeconz/models/api.py"""API base class.""" from collections.abc import Callable import logging from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from . import ResourceGroup LOGGER = logging.getLogger(__name__) SubscriptionType = Callable[..., None] UnsubscribeType = Callable[[], None] class APIItem: """Base class for a deCONZ API item.""" resource_group: "ResourceGroup" def __init__(self, resource_id: str, raw: Any) -> None: """Initialize API item.""" self.resource_id = resource_id self.raw = raw self.changed_keys: set[str] = set() self._callbacks: list[SubscriptionType] = [] self._subscribers: list[SubscriptionType] = [] @property def deconz_id(self) -> str: """Id to call device over API e.g. /sensors/1.""" return f"/{self.resource_group.value}/{self.resource_id}" def register_callback(self, callback: SubscriptionType) -> None: """Register callback for signalling.""" self._callbacks.append(callback) def remove_callback(self, callback: SubscriptionType) -> None: """Remove callback previously registered.""" if callback in self._callbacks: self._callbacks.remove(callback) def subscribe(self, callback: SubscriptionType) -> UnsubscribeType: """Subscribe to events. Return function to unsubscribe. """ self._subscribers.append(callback) def unsubscribe() -> None: """Unsubscribe callback.""" self._subscribers.remove(callback) return unsubscribe def update(self, raw: dict[str, dict[str, Any]]) -> None: """Update input attr in self. Store a set of keys with changed values. """ changed_keys = set() for k, v in raw.items(): changed_keys.add(k) if isinstance(self.raw.get(k), dict) and isinstance(v, dict): changed_keys.update(set(v.keys())) self.raw[k].update(v) else: self.raw[k] = v self.changed_keys = changed_keys for callback in self._callbacks + self._subscribers: callback() 07070100000023000081A400000000000000000000000164453E39000005B2000000000000000000000000000000000000002C00000000deconz-112/pydeconz/models/deconz_device.py"""Python library to connect deCONZ and Home Assistant to work together.""" from .api import APIItem class DeconzDevice(APIItem): """deCONZ resource base representation. Dresden Elektroniks REST API documentation http://dresden-elektronik.github.io/deconz-rest-doc/ """ @property def etag(self) -> str: """HTTP etag change on any action to the device.""" raw: dict[str, str] = self.raw return raw.get("etag") or "" @property def manufacturer(self) -> str: """Device manufacturer.""" raw: dict[str, str] = self.raw return raw.get("manufacturername") or "" @property def model_id(self) -> str: """Device model.""" raw: dict[str, str] = self.raw return raw.get("modelid") or "" @property def name(self) -> str: """Name of the device.""" raw: dict[str, str] = self.raw return raw.get("name") or "" @property def software_version(self) -> str: """Firmware version.""" raw: dict[str, str] = self.raw return raw.get("swversion") or "" @property def type(self) -> str: """Human readable type of the device.""" raw: dict[str, str] = self.raw return raw.get("type") or "" @property def unique_id(self) -> str: """Id for unique device identification.""" raw: dict[str, str] = self.raw return raw.get("uniqueid") or "" 07070100000024000081A400000000000000000000000164453E39000008E7000000000000000000000000000000000000002400000000deconz-112/pydeconz/models/event.py"""Event data from deCONZ websocket.""" from dataclasses import dataclass import enum from typing import Any from . import ResourceGroup class EventKey(enum.Enum): """Event keys.""" TYPE = "t" EVENT = "e" RESOURCE = "r" ID = "id" GROUP_ID = "gid" SCENE_ID = "scid" ATTRIBUTE = "attr" CONFIG = "config" NAME = "name" STATE = "state" UNIQUE_ID = "uniqueid" ALARM = "alarmsystem" GROUP = "group" LIGHT = "light" SENSOR = "sensor" class EventType(enum.Enum): """The event type of the message.""" ADDED = "added" CHANGED = "changed" DELETED = "deleted" SCENE_CALLED = "scene-called" @dataclass class Event: """Event data from deCONZ websocket.""" id: str data: dict[str, Any] resource: ResourceGroup type: EventType # Only for "scene-called" events group_id: str scene_id: str @property def added_data(self) -> dict[str, Any]: """Full device resource. Only for "added" events. """ data: dict[str, Any] = {} for key in ( EventKey.SENSOR.value, EventKey.LIGHT.value, EventKey.ALARM.value, EventKey.GROUP.value, ): if key in self.data: data = self.data[key] break return data @property def changed_data(self) -> dict[str, Any]: """Altered device data. Only for "changed" events. Ignores "attr" events. """ data: dict[str, Any] = {} for key in ( EventKey.STATE.value, EventKey.CONFIG.value, EventKey.NAME.value, ): if (value := self.data.get(key)) is not None: data[key] = value return data @classmethod def from_dict(cls, data: dict[str, Any]) -> "Event": """Create event instance from dict.""" return cls( id=data.get(EventKey.ID.value, ""), group_id=data.get(EventKey.GROUP_ID.value, ""), scene_id=data.get(EventKey.SCENE_ID.value, ""), resource=ResourceGroup(data[EventKey.RESOURCE.value]), type=EventType(data[EventKey.EVENT.value]), data=data, ) 07070100000025000081A400000000000000000000000164453E3900001718000000000000000000000000000000000000002400000000deconz-112/pydeconz/models/group.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import Final, Literal from typing_extensions import NotRequired, TypedDict from . import ResourceGroup from .deconz_device import DeconzDevice from .light.light import Light, LightColorMode, LightEffect from .scene import TypedScene COLOR_STATE_ATTRIBUTES: Final = { "bri", "ct", "hue", "sat", "xy", "colormode", "effect", } class TypedGroupAction(TypedDict): """Group action type definition.""" bri: int colormode: NotRequired[Literal["ct", "hs", "xy"]] ct: int effect: NotRequired[Literal["colorloop", "none"]] hue: int on: bool sat: int xy: tuple[float, float] class TypedGroupState(TypedDict): """Group state type definition.""" all_on: bool any_on: bool class TypedGroup(TypedDict): """Group type definition.""" action: TypedGroupAction devicemembership: list[str] hidden: bool id: str lights: list[str] lightsequence: list[str] multideviceids: list[str] name: str scenes: list[TypedScene] state: TypedGroupState type: str class Group(DeconzDevice): """deCONZ light group representation. Dresden Elektroniks documentation of light groups in deCONZ http://dresden-elektronik.github.io/deconz-rest-doc/groups/ """ raw: TypedGroup resource_group = ResourceGroup.GROUP @property def state(self) -> bool | None: """Is any light in light group on.""" return self.any_on @property def brightness(self) -> int | None: """Brightness of the light. Depending on the light type 0 might not mean visible "off" but minimum brightness. """ return self.raw["action"].get("bri") @property def color_temp(self) -> int | None: """Mired color temperature of the light. (2000K - 6500K).""" return self.raw["action"].get("ct") @property def hue(self) -> int | None: """Color hue of the light. The hue parameter in the HSV color model is between 0°-360° and is mapped to 0..65535 to get 16-bit resolution. """ return self.raw["action"].get("hue") @property def saturation(self) -> int | None: """Color saturation of the light. There 0 means no color at all and 255 is the greatest saturation of the color. """ return self.raw["action"].get("sat") @property def xy(self) -> tuple[float, float] | None: """CIE xy color space coordinates as array [x, y] of real values (0..1).""" x, y = self.raw["action"].get("xy", (None, None)) if x is None or y is None: return None if x > 1: x = x / 65555 if y > 1: y = y / 65555 return (x, y) @property def color_mode(self) -> LightColorMode | None: """Color mode of group.""" if "colormode" in self.raw["action"]: return LightColorMode(self.raw["action"]["colormode"]) return None @property def effect(self) -> LightEffect | None: """Effect of the group.""" if "effect" in self.raw["action"]: return LightEffect(self.raw["action"]["effect"]) return None @property def reachable(self) -> bool: """Is group reachable.""" return True @property def all_on(self) -> bool: """Is all lights in light group on.""" return self.raw["state"].get("all_on") is True @property def any_on(self) -> bool: """Is any lights in light group on.""" return self.raw["state"].get("any_on") is True @property def device_membership(self) -> list[str] | None: """List of device ids (sensors) when group was created by a device.""" return self.raw.get("devicemembership") @property def hidden(self) -> bool | None: """Indicate the hidden status of the group. Has no effect at the gateway but apps can uses this to hide groups. """ return self.raw.get("hidden") @property def id(self) -> str | None: """Group ID.""" return self.raw.get("id") @property def lights(self) -> list[str]: """List of all light IDs in group. Sequence is defined by the gateway. """ return self.raw.get("lights", []) @property def light_sequence(self) -> list[str] | None: """List of light IDs in group that can be sorted by the user. Need not to contain all light ids of this group. """ return self.raw.get("lightsequence") @property def multi_device_ids(self) -> list[str] | None: """List of light IDs in group. Subsequent ids from multidevices with multiple endpoints. """ return self.raw.get("multideviceids") def update_color_state( self, light: Light, update_all_attributes: bool = False ) -> None: """Sync color state with light. update_all_attributes is used to control whether or not to write light attributes with the value None to the group. This is used to not keep any bad values from the group. """ data = {} for attribute in COLOR_STATE_ATTRIBUTES: if (light_attribute := light.raw["state"].get(attribute)) is not None: data[attribute] = light_attribute continue if update_all_attributes: if attribute == "xy": data[attribute] = (None, None) elif attribute == "colormode": continue elif attribute == "effect": data[attribute] = LightEffect.NONE.value else: data[attribute] = None self.update({"action": data}) 07070100000026000041ED00000000000000000000000264453E3900000000000000000000000000000000000000000000002100000000deconz-112/pydeconz/models/light07070100000027000081A400000000000000000000000164453E39000002C5000000000000000000000000000000000000002D00000000deconz-112/pydeconz/models/light/__init__.py"""Python library to connect deCONZ and Home Assistant to work together.""" from .. import ResourceGroup from ..deconz_device import DeconzDevice class LightBase(DeconzDevice): """deCONZ light representation. Dresden Elektroniks documentation of lights in deCONZ http://dresden-elektronik.github.io/deconz-rest-doc/lights/ """ resource_group = ResourceGroup.LIGHT @property def state(self) -> bool | None: """Device state.""" raw: dict[str, bool] = self.raw["state"] return raw.get("on") @property def reachable(self) -> bool: """Is light reachable.""" raw: dict[str, bool] = self.raw["state"] return raw["reachable"] 07070100000028000081A400000000000000000000000164453E39000000AE000000000000000000000000000000000000003700000000deconz-112/pydeconz/models/light/configuration_tool.py"""Python library to connect deCONZ and Home Assistant to work together.""" from . import LightBase class ConfigurationTool(LightBase): """deCONZ hardware antenna.""" 07070100000029000081A400000000000000000000000164453E39000006B5000000000000000000000000000000000000002A00000000deconz-112/pydeconz/models/light/cover.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum from typing_extensions import NotRequired, TypedDict from . import LightBase class TypedCoverState(TypedDict): """Cover state type definition.""" bri: int lift: NotRequired[int] open: NotRequired[bool] sat: NotRequired[int] tilt: NotRequired[int] class TypedCover(TypedDict): """Cover type definition.""" state: TypedCoverState class CoverAction(enum.Enum): """Possible cover actions.""" CLOSE = enum.auto() OPEN = enum.auto() STOP = enum.auto() class Cover(LightBase): """Cover and Damper class.""" raw: TypedCover @property def is_open(self) -> bool: """Is cover open.""" if "open" not in self.raw["state"]: # Legacy support return self.state is False return self.raw["state"]["open"] @property def lift(self) -> int: """Amount of closed position. Supported values: 0-100 - 0 is open / 100 is closed """ if "lift" not in self.raw["state"]: # Legacy support return int(self.raw["state"]["bri"] / 2.54) return self.raw["state"]["lift"] @property def tilt(self) -> int | None: """Amount of tilt. Supported values: 0-100 - 0 is open / 100 is closed """ if "tilt" in self.raw["state"]: return self.raw["state"]["tilt"] elif "sat" in self.raw["state"]: # Legacy support return int(self.raw["state"]["sat"] / 2.54) return None @property def supports_tilt(self) -> bool: """Supports tilt.""" return "tilt" in self.raw["state"] 0707010000002A000081A400000000000000000000000164453E390000219F000000000000000000000000000000000000002A00000000deconz-112/pydeconz/models/light/light.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum import logging from typing import Literal from typing_extensions import NotRequired, TypedDict from . import LightBase LOGGER = logging.getLogger(__name__) class TypedLightState(TypedDict): """Light state type definition.""" alert: NotRequired[ Literal[ "none", "select", "lselect", "blink", "breathe", "channelchange", "finish", "okay", "stop", ] ] bri: int colormode: NotRequired[ Literal[ "ct", "effect", "gradient", "hs", "xy", ] ] ct: int effect: NotRequired[ Literal[ "candle", "carnival", "colorloop", "collide", "fading", "fireplace", "fireworks", "flag", "glow", "loop", "none", "rainbow", "snake", "snow", "sparkle", "sparkles", "steady", "strobe", "sunrise", "twinkle", "updown", "vintage", "waves", ] ] hue: int on: bool sat: int speed: Literal[0, 1, 2, 3, 4, 5, 6] xy: tuple[float, float] class TypedLight(TypedDict): """Light type definition.""" colorcapabilities: NotRequired[int] ctmax: int ctmin: int state: TypedLightState class LightAlert(enum.Enum): """Temporary alert effect. Supported values: - "none" — light is not performing an alert. - "lselect" — light is blinking a longer time. - "select" — light is blinking a short time. - "blink" - "breathe" - "channelchange" - "finish" - "okay" - "stop" """ NONE = "none" LONG = "lselect" SHORT = "select" # Specific to Hue color bulbs BLINK = "blink" BREATHE = "breathe" CHANNEL_CHANGE = "channelchange" FINISH = "finish" OKAY = "okay" STOP = "stop" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "LightAlert": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected light alert type %s", value) return LightAlert.UNKNOWN class LightColorCapability(enum.IntFlag): """Bit field of features supported by a light device.""" HUE_SATURATION = 0 ENHANCED_HUE = 1 COLOR_LOOP = 2 XY_ATTRIBUTES = 4 COLOR_TEMPERATURE = 8 UNKNOWN = 1111 @classmethod def _missing_(cls, value: object) -> "LightColorCapability": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected light color capability %s", value) return LightColorCapability.UNKNOWN class LightColorMode(enum.Enum): """Color mode of the light. Supported values: - "ct" — color temperature. - "hs" — hue and saturation. - "xy" — CIE xy values. - "effect" - "gradient" """ CT = "ct" HS = "hs" XY = "xy" # Specific to Hue gradient lights EFFECT = "effect" GRADIENT = "gradient" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "LightColorMode": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected light color mode %s", value) return LightColorMode.UNKNOWN class LightEffect(enum.Enum): """Effect of the light. Supported values: - "colorloop" — cycle through hue values 0-360 - "none" — no effect - "candle" - "carnival" - "collide" - "fading" - "fireplace" - "fireworks" - "flag" - "glow" - "loop" - "rainbow" - "snake" - "snow" - "sparkle" - "sparkles" - "steady" - "strobe" - "sunrise" - "twinkle" - "updown" - "vintage" - "waves" """ COLOR_LOOP = "colorloop" NONE = "none" # Specific to Hue lights CANDLE = "candle" FIREPLACE = "fireplace" LOOP = "loop" SPARKLE = "sparkle" SUNRISE = "sunrise" # Specific to Lidl christmas light CARNIVAL = "carnival" COLLIDE = "collide" FADING = "fading" FIREWORKS = "fireworks" FLAG = "flag" GLOW = "glow" RAINBOW = "rainbow" SNAKE = "snake" SNOW = "snow" SPARKLES = "sparkles" STEADY = "steady" STROBE = "strobe" TWINKLE = "twinkle" UPDOWN = "updown" VINTAGE = "vintage" WAVES = "waves" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "LightEffect": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected light effect type %s", value) return LightEffect.UNKNOWN class LightFanSpeed(enum.IntEnum): """Possible fan speeds. Supported values: - 0 - fan is off - 1 - 25% - 2 - 50% - 3 - 75% - 4 - 100% - 5 - Auto - 6 - "comfort-breeze" """ OFF = 0 PERCENT_25 = 1 PERCENT_50 = 2 PERCENT_75 = 3 PERCENT_100 = 4 AUTO = 5 COMFORT_BREEZE = 6 class Light(LightBase): """deCONZ light representation. Dresden Elektroniks documentation of lights in deCONZ http://dresden-elektronik.github.io/deconz-rest-doc/lights/ """ raw: TypedLight @property def alert(self) -> LightAlert | None: """Temporary alert effect.""" if "alert" in self.raw["state"]: return LightAlert(self.raw["state"]["alert"]) return None @property def brightness(self) -> int | None: """Brightness of the light. Depending on the light type 0 might not mean visible "off" but minimum brightness. """ return self.raw["state"].get("bri") @property def color_capabilities(self) -> LightColorCapability | None: """Bit field to specify color capabilities of light.""" if "colorcapabilities" in self.raw: return LightColorCapability(self.raw["colorcapabilities"]) return None @property def color_temp(self) -> int | None: """Mired color temperature of the light. (2000K - 6500K).""" return self.raw["state"].get("ct") @property def hue(self) -> int | None: """Color hue of the light. The hue parameter in the HSV color model is between 0°-360° and is mapped to 0..65535 to get 16-bit resolution. """ return self.raw["state"].get("hue") @property def saturation(self) -> int | None: """Color saturation of the light. There 0 means no color at all and 255 is the greatest saturation of the color. """ return self.raw["state"].get("sat") @property def xy(self) -> tuple[float, float] | None: """CIE xy color space coordinates as array [x, y] of real values (0..1).""" x, y = self.raw["state"].get("xy", (None, None)) if x is None or y is None: return None if x > 1: x = x / 65555 if y > 1: y = y / 65555 return (x, y) @property def color_mode(self) -> LightColorMode | None: """Color mode of light.""" if "colormode" in self.raw["state"]: return LightColorMode(self.raw["state"]["colormode"]) return None @property def on(self) -> bool: """Device state.""" return self.raw["state"]["on"] @property def max_color_temp(self) -> int | None: """Max value for color temperature.""" if (ctmax := self.raw.get("ctmax")) is not None and ctmax > 650: ctmax = 650 return ctmax @property def min_color_temp(self) -> int | None: """Min value for color temperature.""" if (ctmin := self.raw.get("ctmin")) is not None and ctmin < 140: ctmin = 140 return ctmin @property def effect(self) -> LightEffect | None: """Effect of the light.""" if "effect" in self.raw["state"]: return LightEffect(self.raw["state"]["effect"]) return None @property def fan_speed(self) -> LightFanSpeed: """Speed of the fan.""" return LightFanSpeed(self.raw["state"]["speed"]) @property def supports_fan_speed(self) -> bool: """Speed of the fan.""" return True if "speed" in self.raw["state"] else False 0707010000002B000081A400000000000000000000000164453E39000001EA000000000000000000000000000000000000002900000000deconz-112/pydeconz/models/light/lock.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import LightBase class TypedLockState(TypedDict): """Lock state type definition.""" on: bool class TypedLock(TypedDict): """Lock type definition.""" state: TypedLockState class Lock(LightBase): """Lock class.""" raw: TypedLock @property def is_locked(self) -> bool: """State of lock.""" return self.raw["state"]["on"] 0707010000002C000081A400000000000000000000000164453E39000000A9000000000000000000000000000000000000003300000000deconz-112/pydeconz/models/light/range_extender.py"""Python library to connect deCONZ and Home Assistant to work together.""" from . import LightBase class RangeExtender(LightBase): """ZigBee range extender.""" 0707010000002D000081A400000000000000000000000164453E390000025C000000000000000000000000000000000000002A00000000deconz-112/pydeconz/models/light/siren.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import Literal, TypedDict from . import LightBase from .light import LightAlert class TypedSirenState(TypedDict): """Siren state type definition.""" alert: Literal["lselect", "select", "none"] class TypedSiren(TypedDict): """Siren type definition.""" state: TypedSirenState class Siren(LightBase): """Siren class.""" raw: TypedSiren @property def is_on(self) -> bool: """If device is sounding.""" return self.raw["state"]["alert"] == LightAlert.LONG.value 0707010000002E000081A400000000000000000000000164453E3900000732000000000000000000000000000000000000002400000000deconz-112/pydeconz/models/scene.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import ResourceGroup from .api import APIItem class TypedScene(TypedDict): """Scene type definition.""" id: str lightcount: int transitiontime: int name: str class Scene(APIItem): """deCONZ scene representation. Dresden Elektroniks documentation of scenes in deCONZ http://dresden-elektronik.github.io/deconz-rest-doc/scenes/ """ raw: TypedScene resource_group = ResourceGroup.SCENE _group_resource_id: str = "" _group_deconz_id: str = "" @property def group_id(self) -> str: """Group ID representation. Scene resource ID is a string combined of group ID and scene ID; "gid_scid". """ if self._group_resource_id == "": self._group_resource_id = self.resource_id.split("_")[0] return self._group_resource_id @property def group_deconz_id(self) -> str: """Group deCONZ ID representation.""" if self._group_deconz_id == "": self._group_deconz_id = f"/{ResourceGroup.GROUP.value}/{self.group_id}" return self._group_deconz_id @property def deconz_id(self) -> str: """Id to call scene over API e.g. /groups/1/scenes/1.""" return f"{self.group_deconz_id}/{self.resource_group.value}/{self.id}" @property def id(self) -> str: """Scene ID.""" return self.raw["id"] @property def light_count(self) -> int: """Lights in group.""" return self.raw["lightcount"] @property def transition_time(self) -> int: """Transition time for scene.""" return self.raw["transitiontime"] @property def name(self) -> str: """Scene name.""" return self.raw["name"] 0707010000002F000041ED00000000000000000000000264453E3900000000000000000000000000000000000000000000002200000000deconz-112/pydeconz/models/sensor07070100000030000081A400000000000000000000000164453E39000006B4000000000000000000000000000000000000002E00000000deconz-112/pydeconz/models/sensor/__init__.py"""Python library to connect deCONZ and Home Assistant to work together.""" from .. import ResourceGroup from ..deconz_device import DeconzDevice class SensorBase(DeconzDevice): """deCONZ sensor representation. Dresden Elektroniks documentation of sensors in deCONZ http://dresden-elektronik.github.io/deconz-rest-doc/sensors/ """ resource_group = ResourceGroup.SENSOR @property def battery(self) -> int | None: """Battery status of sensor.""" raw: dict[str, int] = self.raw["config"] return raw.get("battery") @property def ep(self) -> int | None: """Endpoint of sensor.""" raw: dict[str, int] = self.raw return raw.get("ep") @property def low_battery(self) -> bool | None: """Low battery.""" raw: dict[str, bool] = self.raw["state"] return raw.get("lowbattery") @property def on(self) -> bool | None: """Declare if the sensor is on or off.""" raw: dict[str, bool] = self.raw["config"] return raw.get("on") @property def reachable(self) -> bool: """Declare if the sensor is reachable.""" raw: dict[str, bool] = self.raw["config"] return raw.get("reachable", True) @property def tampered(self) -> bool | None: """Tampered.""" raw: dict[str, bool] = self.raw["state"] return raw.get("tampered") @property def internal_temperature(self) -> float | None: """Extra temperature available on some Xiaomi devices.""" raw: dict[str, int] = self.raw["config"] if temperature := raw.get("temperature"): return round(temperature / 100, 1) return None 07070100000031000081A400000000000000000000000164453E3900000895000000000000000000000000000000000000003200000000deconz-112/pydeconz/models/sensor/air_purifier.py"""Air purifier data model.""" import enum from typing import Literal, TypedDict from . import SensorBase class TypedAirPurifierState(TypedDict): """Air purifier state type definition.""" deviceruntime: int filterruntime: int replacefilter: bool speed: int class TypedAirPurifierConfig(TypedDict): """Air purifier config type definition.""" filterlifetime: int ledindication: bool locked: bool mode: Literal[ "off", "auto", "speed_1", "speed_2", "speed_3", "speed_4", "speed_5", ] on: bool reachable: bool class TypedAirPurifier(TypedDict): """Air purifier type definition.""" config: TypedAirPurifierConfig state: TypedAirPurifierState class AirPurifierFanMode(enum.Enum): """Air purifier supported fan modes.""" OFF = "off" AUTO = "auto" SPEED_1 = "speed_1" SPEED_2 = "speed_2" SPEED_3 = "speed_3" SPEED_4 = "speed_4" SPEED_5 = "speed_5" class AirPurifier(SensorBase): """Air purifier sensor.""" raw: TypedAirPurifier @property def device_run_time(self) -> int: """Device run time in minutes.""" return self.raw["state"]["deviceruntime"] @property def fan_mode(self) -> AirPurifierFanMode: """Fan mode.""" return AirPurifierFanMode(self.raw["config"]["mode"]) @property def fan_speed(self) -> int: """Fan speed.""" return self.raw["state"]["speed"] @property def filter_life_time(self) -> int: """Filter life time in minutes.""" return self.raw["config"]["filterlifetime"] @property def filter_run_time(self) -> int: """Filter run time in minutes.""" return self.raw["state"]["filterruntime"] @property def led_indication(self) -> bool: """Led indicator.""" return self.raw["config"]["ledindication"] @property def locked(self) -> bool: """Locked configuration.""" return self.raw["config"]["locked"] @property def replace_filter(self) -> bool: """Replace filter property.""" return self.raw["state"]["replacefilter"] 07070100000032000081A400000000000000000000000164453E390000095C000000000000000000000000000000000000003100000000deconz-112/pydeconz/models/sensor/air_quality.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum import logging from typing import Literal, TypedDict from . import SensorBase LOGGER = logging.getLogger(__name__) class TypedAirQualityState(TypedDict): """Air quality state type definition.""" airquality: Literal[ "excellent", "good", "moderate", "poor", "unhealthy", "out of scale", ] airquality_co2_density: int airquality_formaldehyde_density: int airqualityppb: int pm2_5: int class TypedAirQuality(TypedDict): """Air quality type definition.""" state: TypedAirQualityState class AirQualityValue(enum.Enum): """Air quality. Supported values: - "excellent" - "good" - "moderate" - "poor" - "unhealthy" - "out of scale" """ EXCELLENT = "excellent" GOOD = "good" MODERATE = "moderate" POOR = "poor" UNHEALTHY = "unhealthy" OUT_OF_SCALE = "out of scale" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "AirQualityValue": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected air quality value %s", value) return AirQualityValue.UNKNOWN class AirQuality(SensorBase): """Air quality sensor.""" raw: TypedAirQuality @property def air_quality(self) -> str: # AirQualityValue: """Air quality.""" return AirQualityValue(self.raw["state"].get("airquality", "unknown")).value @property def air_quality_co2(self) -> int | None: """Chemical compound gas carbon dioxid (CO2) (ppb).""" return self.raw["state"].get("airquality_co2_density") @property def air_quality_formaldehyde(self) -> int | None: """Chemical compound gas formaldehyde / methanal (CH2O) (µg/m³).""" return self.raw["state"].get("airquality_formaldehyde_density") @property def air_quality_ppb(self) -> int | None: """Air quality PPB TVOC.""" return self.raw["state"].get("airqualityppb") @property def pm_2_5(self) -> int | None: """Air quality PM2.5 (µg/m³).""" return self.raw["state"].get("pm2_5") @property def supports_air_quality(self) -> bool: """Support Air quality reporting.""" return "airquality" in self.raw["state"] 07070100000033000081A400000000000000000000000164453E39000001EF000000000000000000000000000000000000002B00000000deconz-112/pydeconz/models/sensor/alarm.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedAlarmState(TypedDict): """Alarm state type definition.""" alarm: bool class TypedAlarm(TypedDict): """Alarm type definition.""" state: TypedAlarmState class Alarm(SensorBase): """Alarm sensor.""" raw: TypedAlarm @property def alarm(self) -> bool: """Alarm.""" return self.raw["state"]["alarm"] 07070100000034000081A400000000000000000000000164453E3900000B9C000000000000000000000000000000000000003700000000deconz-112/pydeconz/models/sensor/ancillary_control.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum import logging from typing import Literal from typing_extensions import NotRequired, TypedDict from . import SensorBase LOGGER = logging.getLogger(__name__) class AncillaryControlAction(enum.Enum): """Last action a user invoked on the keypad.""" ARMED_AWAY = "armed_away" ARMED_NIGHT = "armed_night" ARMED_STAY = "armed_stay" DISARMED = "disarmed" EMERGENCY = "emergency" FIRE = "fire" INVALID_CODE = "invalid_code" PANIC = "panic" class AncillaryControlPanel(enum.Enum): """Mirror of alarm system state.armstate attribute.""" ARMED_AWAY = "armed_away" ARMED_NIGHT = "armed_night" ARMED_STAY = "armed_stay" ARMING_AWAY = "arming_away" ARMING_NIGHT = "arming_night" ARMING_STAY = "arming_stay" DISARMED = "disarmed" ENTRY_DELAY = "entry_delay" EXIT_DELAY = "exit_delay" IN_ALARM = "in_alarm" NOT_READY = "not_ready" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "AncillaryControlPanel": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected panel mode %s", value) return AncillaryControlPanel.UNKNOWN class TypedAncillaryControlState(TypedDict): """Ancillary control state type definition.""" action: Literal[ "armed_away", "armed_night", "armed_stay", "disarmed", "emergency", "fire", "invalid_code", "panic", ] panel: NotRequired[ Literal[ "armed_away", "armed_night", "armed_stay", "arming_away", "arming_night", "arming_stay", "disarmed", "entry_delay", "exit_delay", "in_alarm", "not_ready", ] ] seconds_remaining: int class TypedAncillaryControl(TypedDict): """Ancillary control type definition.""" state: TypedAncillaryControlState class AncillaryControl(SensorBase): """Ancillary control sensor.""" raw: TypedAncillaryControl @property def action(self) -> AncillaryControlAction: """Last action a user invoked on the keypad.""" return AncillaryControlAction(self.raw["state"]["action"]) @property def panel(self) -> AncillaryControlPanel | None: """Mirror of alarm system state.armstate attribute. It reflects what is shown on the panel (when activated by the keypad’s proximity sensor). """ if "panel" in self.raw["state"]: return AncillaryControlPanel(self.raw["state"]["panel"]) return None @property def seconds_remaining(self) -> int: """Remaining time of "exit_delay" and "entry_delay" states. In all other states the value is 0. """ return self.raw["state"].get("seconds_remaining", 0) 07070100000035000081A400000000000000000000000164453E3900000205000000000000000000000000000000000000002D00000000deconz-112/pydeconz/models/sensor/battery.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedBatteryState(TypedDict): """Battery state type definition.""" battery: int class TypedBattery(TypedDict): """Battery type definition.""" state: TypedBatteryState class Battery(SensorBase): """Battery sensor.""" raw: TypedBattery @property def battery(self) -> int: """Battery.""" return self.raw["state"]["battery"] 07070100000036000081A400000000000000000000000164453E3900000269000000000000000000000000000000000000003500000000deconz-112/pydeconz/models/sensor/carbon_monoxide.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedCarbonMonoxideState(TypedDict): """Carbon monoxide state type definition.""" carbonmonoxide: bool class TypedCarbonMonoxide(TypedDict): """Carbon monoxide type definition.""" state: TypedCarbonMonoxideState class CarbonMonoxide(SensorBase): """Carbon monoxide sensor.""" raw: TypedCarbonMonoxide @property def carbon_monoxide(self) -> bool: """Carbon monoxide detected.""" return self.raw["state"]["carbonmonoxide"] 07070100000037000081A400000000000000000000000164453E390000033F000000000000000000000000000000000000003100000000deconz-112/pydeconz/models/sensor/consumption.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedConsumptionState(TypedDict): """Consumption state type definition.""" consumption: int power: int class TypedConsumption(TypedDict): """Consumption type definition.""" state: TypedConsumptionState class Consumption(SensorBase): """Power consumption sensor.""" raw: TypedConsumption @property def consumption(self) -> int: """Consumption.""" return self.raw["state"]["consumption"] @property def scaled_consumption(self) -> float: """State of sensor.""" return self.consumption / 1000 @property def power(self) -> int | None: """Power.""" return self.raw["state"].get("power") 07070100000038000081A400000000000000000000000164453E3900000D13000000000000000000000000000000000000002E00000000deconz-112/pydeconz/models/sensor/daylight.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum import logging from typing import Final, TypedDict from . import SensorBase LOGGER = logging.getLogger(__name__) class DayLightStatus(enum.IntEnum): """Day light status.""" NADIR = 100 NIGHT_END = 110 NAUTICAL_DAWN = 120 DAWN = 130 SUNRISE_START = 140 SUNRISE_END = 150 GOLDEN_HOUR_1 = 160 SOLAR_NOON = 170 GOLDEN_HOUR_2 = 180 SUNSET_START = 190 SUNSET_END = 200 DUSK = 210 NAUTICAL_DUSK = 220 NIGHT_START = 230 UNKNOWN = 666 @classmethod def _missing_(cls, value: object) -> "DayLightStatus": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected day light value %s", value) return DayLightStatus.UNKNOWN DAYLIGHT_STATUS: Final = { DayLightStatus.NADIR: "nadir", DayLightStatus.NIGHT_END: "night_end", DayLightStatus.NAUTICAL_DAWN: "nautical_dawn", DayLightStatus.DAWN: "dawn", DayLightStatus.SUNRISE_START: "sunrise_start", DayLightStatus.SUNRISE_END: "sunrise_end", DayLightStatus.GOLDEN_HOUR_1: "golden_hour_1", DayLightStatus.SOLAR_NOON: "solar_noon", DayLightStatus.GOLDEN_HOUR_2: "golden_hour_2", DayLightStatus.SUNSET_START: "sunset_start", DayLightStatus.SUNSET_END: "sunset_end", DayLightStatus.DUSK: "dusk", DayLightStatus.NAUTICAL_DUSK: "nautical_dusk", DayLightStatus.NIGHT_START: "night_start", DayLightStatus.UNKNOWN: "unknown", } class TypedDaylightConfig(TypedDict): """Daylight config type definition.""" configured: bool sunriseoffset: int sunsetoffset: int class TypedDaylightState(TypedDict): """Daylight state type definition.""" dark: bool daylight: bool status: int sunrise: str sunset: str class TypedDaylight(TypedDict): """Daylight type definition.""" config: TypedDaylightConfig state: TypedDaylightState class Daylight(SensorBase): """Daylight sensor built into deCONZ software.""" raw: TypedDaylight @property def configured(self) -> bool: """Is daylight sensor configured.""" return self.raw["config"]["configured"] @property def dark(self) -> bool: """Is dark.""" return self.raw["state"]["dark"] @property def daylight(self) -> bool: """Is daylight.""" return self.raw["state"]["daylight"] @property def daylight_status(self) -> DayLightStatus: """Return the daylight status string.""" return DayLightStatus(self.raw["state"]["status"]) @property def status(self) -> str: """Return the daylight status string.""" return DAYLIGHT_STATUS[DayLightStatus(self.raw["state"]["status"])] @property def sunrise(self) -> str: """Sunrise.""" return self.raw["state"]["sunrise"] @property def sunrise_offset(self) -> int: """Sunrise offset. -120 to 120. """ return self.raw["config"]["sunriseoffset"] @property def sunset(self) -> str: """Sunset.""" return self.raw["state"]["sunset"] @property def sunset_offset(self) -> int: """Sunset offset. -120 to 120. """ return self.raw["config"]["sunsetoffset"] 07070100000039000081A400000000000000000000000164453E39000004F3000000000000000000000000000000000000002F00000000deconz-112/pydeconz/models/sensor/door_lock.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum from typing import Literal, TypedDict from . import SensorBase class DoorLockLockState(enum.Enum): """State the lock is in.""" LOCKED = "locked" UNLOCKED = "unlocked" UNDEFINED = "undefined" NOT_FULLY_LOCKED = "not fully locked" class TypedDoorLockConfig(TypedDict): """Door lock config type definition.""" lock: bool class TypedDoorLockState(TypedDict): """Door lock state type definition.""" lockstate: Literal["locked", "unlocked", "undefined", "not fully locked"] class TypedDoorLock(TypedDict): """Door lock type definition.""" config: TypedDoorLockConfig state: TypedDoorLockState class DoorLock(SensorBase): """Door lock sensor.""" raw: TypedDoorLock @property def is_locked(self) -> bool: """Return True if lock is locked.""" return self.lock_state == DoorLockLockState.LOCKED @property def lock_state(self) -> DoorLockLockState: """State the lock is in.""" return DoorLockLockState(self.raw["state"]["lockstate"]) @property def lock_configuration(self) -> bool: """Lock configuration.""" return self.raw["config"]["lock"] 0707010000003A000081A400000000000000000000000164453E3900000288000000000000000000000000000000000000002A00000000deconz-112/pydeconz/models/sensor/fire.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedFireState(TypedDict): """Fire state type definition.""" fire: bool test: bool class TypedFire(TypedDict): """Fire type definition.""" state: TypedFireState class Fire(SensorBase): """Fire sensor.""" raw: TypedFire @property def fire(self) -> bool: """Fire detected.""" return self.raw["state"]["fire"] @property def in_test_mode(self) -> bool: """Sensor is in test mode.""" return self.raw["state"].get("test", False) 0707010000003B000081A400000000000000000000000164453E3900000225000000000000000000000000000000000000003200000000deconz-112/pydeconz/models/sensor/generic_flag.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedGenericFlagState(TypedDict): """Generic flag state type definition.""" flag: bool class TypedGenericFlag(TypedDict): """Generic flag type definition.""" state: TypedGenericFlagState class GenericFlag(SensorBase): """Generic flag sensor.""" raw: TypedGenericFlag @property def flag(self) -> bool: """Flag status.""" return self.raw["state"]["flag"] 0707010000003C000081A400000000000000000000000164453E3900000234000000000000000000000000000000000000003400000000deconz-112/pydeconz/models/sensor/generic_status.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedGenericStatusState(TypedDict): """Generic status state type definition.""" status: str class TypedGenericStatus(TypedDict): """Generic status type definition.""" state: TypedGenericStatusState class GenericStatus(SensorBase): """Generic status sensor.""" raw: TypedGenericStatus @property def status(self) -> str: """Status.""" return self.raw["state"]["status"] 0707010000003D000081A400000000000000000000000164453E390000040A000000000000000000000000000000000000002E00000000deconz-112/pydeconz/models/sensor/humidity.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedHumidityConfig(TypedDict): """Humidity config type definition.""" offset: int class TypedHumidityState(TypedDict): """Humidity state type definition.""" humidity: int class TypedHumidity(TypedDict): """Humidity type definition.""" config: TypedHumidityConfig state: TypedHumidityState class Humidity(SensorBase): """Humidity sensor.""" raw: TypedHumidity @property def humidity(self) -> int: """Humidity level.""" return self.raw["state"]["humidity"] @property def scaled_humidity(self) -> float: """Scaled humidity level.""" return self.humidity / 100 @property def offset(self) -> int | None: """Signed offset value to measured state values. Values send by the REST-API are already amended by the offset. """ return self.raw["config"].get("offset") 0707010000003E000081A400000000000000000000000164453E390000066D000000000000000000000000000000000000003100000000deconz-112/pydeconz/models/sensor/light_level.py"""Python library to connect deCONZ and Home Assistant to work together.""" import math from typing import TypedDict from . import SensorBase class TypedLightLevelConfig(TypedDict): """Light level config type definition.""" tholddark: int tholdoffset: int class TypedLightLevelState(TypedDict): """Light level state type definition.""" dark: bool daylight: bool lightlevel: int lux: int class TypedLightLevel(TypedDict): """Light level type definition.""" config: TypedLightLevelConfig state: TypedLightLevelState class LightLevel(SensorBase): """Light level sensor.""" raw: TypedLightLevel @property def dark(self) -> bool | None: """If the area near the sensor is light or not.""" return self.raw["state"].get("dark") @property def daylight(self) -> bool | None: """Daylight.""" return self.raw["state"].get("daylight") @property def light_level(self) -> int: """Light level.""" return self.raw["state"]["lightlevel"] @property def scaled_light_level(self) -> float: """Scaled light level.""" return round(math.pow(10, (self.light_level - 1) / 10000), 1) @property def lux(self) -> int | None: """Lux.""" return self.raw["state"].get("lux") @property def threshold_dark(self) -> int | None: """Threshold to hold dark.""" return self.raw["config"].get("tholddark") @property def threshold_offset(self) -> int | None: """Offset for threshold to hold dark.""" return self.raw["config"].get("tholdoffset") 0707010000003F000081A400000000000000000000000164453E390000023B000000000000000000000000000000000000002E00000000deconz-112/pydeconz/models/sensor/moisture.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedMoistureState(TypedDict): """Moisture state type definition.""" moisture: int class TypedMoisture(TypedDict): """Moisture type definition.""" state: TypedMoistureState class Moisture(SensorBase): """Moisture sensor.""" raw: TypedMoisture @property def moisture(self) -> int: """Moisture level. 0-100 in percent. """ return self.raw["state"]["moisture"] 07070100000040000081A400000000000000000000000164453E3900000214000000000000000000000000000000000000003000000000deconz-112/pydeconz/models/sensor/open_close.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedOpenCloseState(TypedDict): """Open close state type definition.""" open: bool class TypedOpenClose(TypedDict): """Open close type definition.""" state: TypedOpenCloseState class OpenClose(SensorBase): """Door/Window sensor.""" raw: TypedOpenClose @property def open(self) -> bool: """Door open.""" return self.raw["state"]["open"] 07070100000041000081A400000000000000000000000164453E3900000331000000000000000000000000000000000000002B00000000deconz-112/pydeconz/models/sensor/power.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedPowerState(TypedDict): """Power state type definition.""" current: int power: int voltage: int class TypedPower(TypedDict): """Power type definition.""" state: TypedPowerState class Power(SensorBase): """Power sensor.""" raw: TypedPower @property def current(self) -> int | None: """Ampere load of device.""" return self.raw["state"].get("current") @property def power(self) -> int: """Power load of device.""" return self.raw["state"]["power"] @property def voltage(self) -> int | None: """Voltage draw of device.""" return self.raw["state"].get("voltage") 07070100000042000081A400000000000000000000000164453E39000010BD000000000000000000000000000000000000002E00000000deconz-112/pydeconz/models/sensor/presence.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum from typing import Literal from typing_extensions import NotRequired, TypedDict from . import SensorBase class TypedPresenceConfig(TypedDict): """Presence config type definition.""" delay: int detectionarea: str devicemode: NotRequired[Literal["leftright", "undirected"]] duration: int resetpresence: bool sensitivity: int sensitivitymax: int triggerdistance: NotRequired[Literal["far", "medium", "near"]] class TypedPresenceState(TypedDict): """Presence state type definition.""" dark: bool presence: bool presenceevent: NotRequired[ Literal[ "enter", "leave", "enterleft", "rightleave", "enterright", "leftleave", "approaching", "absenting", "8", "9", ] ] class TypedPresence(TypedDict): """Presence type definition.""" config: TypedPresenceConfig state: TypedPresenceState class PresenceConfigDeviceMode(enum.Enum): """Device mode. Supported values: - leftright - left and right - undirected """ LEFT_AND_RIGHT = "leftright" UNDIRECTED = "undirected" class PresenceConfigSensitivity(enum.IntEnum): """Device sensitivity. Supported values: - 1 - Low - 2 - Medium - 3 - High """ LOW = 1 MEDIUM = 2 HIGH = 3 class PresenceConfigTriggerDistance(enum.Enum): """Trigger distance. Supported values: - far - Someone approaching is detected on high distance - medium - Someone approaching is detected on medium distance - near - Someone approaching is detected on low distance """ FAR = "far" MEDIUM = "medium" NEAR = "near" class PresenceStatePresenceEvent(enum.Enum): """Current activity associated with current presence state. Supported values: - enter - leave - enterleft - rightleave - enterright - leftleave - approaching - absenting - 8 - 9 """ ENTER = "enter" LEAVE = "leave" ENTER_LEFT = "enterleft" RIGHT_LEAVE = "rightleave" ENTER_RIGHT = "enterright" LEFT_LEAVE = "leftleave" APPROACHING = "approaching" ABSENTING = "absenting" EIGHT = "8" NINE = "9" class Presence(SensorBase): """Presence detector.""" raw: TypedPresence @property def dark(self) -> bool | None: """If the area near the sensor is light or not.""" return self.raw["state"].get("dark") @property def delay(self) -> int | None: """Occupied to unoccupied delay in seconds.""" return self.raw["config"].get("delay") @property def device_mode(self) -> PresenceConfigDeviceMode | None: """Trigger distance.""" if "devicemode" in self.raw["config"]: return PresenceConfigDeviceMode(self.raw["config"]["devicemode"]) return None @property def duration(self) -> int | None: """Minimum duration which presence will be true.""" return self.raw["config"].get("duration") @property def presence(self) -> bool: """Motion detected.""" return self.raw["state"]["presence"] @property def presence_event(self) -> PresenceStatePresenceEvent | None: """Activity associated with current presence state.""" if "presenceevent" in self.raw["state"]: return PresenceStatePresenceEvent(self.raw["state"]["presenceevent"]) return None @property def sensitivity(self) -> int | None: """Sensitivity setting for Philips Hue motion sensor. Supported values: - 0-[sensitivitymax] """ return self.raw["config"].get("sensitivity") @property def max_sensitivity(self) -> int | None: """Maximum sensitivity value.""" return self.raw["config"].get("sensitivitymax") @property def trigger_distance(self) -> PresenceConfigTriggerDistance | None: """Device specific distance setting.""" if "triggerdistance" in self.raw["config"]: return PresenceConfigTriggerDistance(self.raw["config"]["triggerdistance"]) return None 07070100000043000081A400000000000000000000000164453E3900000211000000000000000000000000000000000000002E00000000deconz-112/pydeconz/models/sensor/pressure.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedPressureState(TypedDict): """Pressure state type definition.""" pressure: int class TypedPressure(TypedDict): """Pressure type definition.""" state: TypedPressureState class Pressure(SensorBase): """Pressure sensor.""" raw: TypedPressure @property def pressure(self) -> int: """Pressure.""" return self.raw["state"]["pressure"] 07070100000044000081A400000000000000000000000164453E390000059B000000000000000000000000000000000000003500000000deconz-112/pydeconz/models/sensor/relative_rotary.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum from typing import TypedDict from . import SensorBase class TypedRelativeRotaryState(TypedDict): """Relative rotary state type definition.""" expectedeventduration: int expectedrotation: int rotaryevent: int class TypedRelativeRotary(TypedDict): """Relative rotary type definition.""" state: TypedRelativeRotaryState class RelativeRotaryEvent(enum.IntEnum): """Rotary event. Supported values: - 1 - new movements (start) - 2 - repeat movements """ NEW = 1 REPEAT = 2 class RelativeRotary(SensorBase): """Relative rotary sensor.""" raw: TypedRelativeRotary @property def expected_event_duration(self) -> int: """Event duration to expect. Interval [ms] which rotary events will be emit. """ return self.raw["state"]["expectedeventduration"] @property def expected_rotation(self) -> int: """Rotation to expect. Report angle - positive for clockwise - negative for counter-clockwise """ return self.raw["state"]["expectedrotation"] @property def rotary_event(self) -> RelativeRotaryEvent: """Rotary event. - 1 for new movements (start) - 2 for repeat movements """ return RelativeRotaryEvent(self.raw["state"]["rotaryevent"]) 07070100000045000081A400000000000000000000000164453E3900000F0F000000000000000000000000000000000000002C00000000deconz-112/pydeconz/models/sensor/switch.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum from typing import Literal from typing_extensions import NotRequired, TypedDict from . import SensorBase class SwitchDeviceMode(enum.Enum): """Different modes for the Hue wall switch module.""" SINGLE_ROCKER = "singlerocker" SINGLE_PUSH_BUTTON = "singlepushbutton" DUAL_ROCKER = "dualrocker" DUAL_PUSH_BUTTON = "dualpushbutton" class SwitchMode(enum.Enum): """For Ubisys S1/S2, operation mode of the switch.""" MOMENTARY = "momentary" ROCKER = "rocker" class SwitchWindowCoveringType(enum.IntEnum): """Set the covering type and starts calibration for Ubisys J1.""" ROLLER_SHADE = 0 ROLLER_SHADE_TWO_MOTORS = 1 ROLLER_SHADE_EXTERIOR = 2 ROLLER_SHADE_TWO_MOTORS_EXTERIOR = 3 DRAPERY = 4 AWNING = 5 SHUTTER = 6 TILT_BLIND_LIFT_ONLY = 7 TILT_BLIND_LIFT_AND_TILT = 8 PROJECTOR_SCREEN = 9 class TypedSwitchConfig(TypedDict): """Switch config type definition.""" devicemode: NotRequired[ Literal["dualpushbutton", "dualrocker", "singlepushbutton", "singlerocker"] ] mode: NotRequired[Literal["momentary", "rocker"]] windowcoveringtype: NotRequired[Literal[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]] class TypedSwitchState(TypedDict): """Switch state type definition.""" angle: int buttonevent: int eventduration: int gesture: int xy: tuple[float, float] class TypedSwitch(TypedDict): """Switch type definition.""" config: TypedSwitchConfig state: TypedSwitchState class Switch(SensorBase): """Switch sensor.""" raw: TypedSwitch @property def button_event(self) -> int | None: """Button press.""" return self.raw["state"].get("buttonevent") @property def gesture(self) -> int | None: """Gesture used for Xiaomi magic cube.""" return self.raw["state"].get("gesture") @property def angle(self) -> int | None: """Angle representing color on a tint remote color wheel.""" return self.raw["state"].get("angle") @property def xy(self) -> tuple[float, float] | None: """X/Y color coordinates selected on a tint remote color wheel.""" return self.raw["state"].get("xy") @property def event_duration(self) -> int | None: """Duration of a push button event for the Hue wall switch module. Increased with 8 for each x001, and they are issued pretty much every 800ms. """ return self.raw["state"].get("eventduration") @property def device_mode(self) -> SwitchDeviceMode | None: """Different modes for the Hue wall switch module. Behavior as rocker: Issues a x000/x002 each time you flip the rocker (to either position). The event duration for the x002 is 1 (for 100ms), but lastupdated suggests it follows the x000 faster than that. Behavior as pushbutton: Issues a x000/x002 sequence on press/release. Issues a x000/x001/.../x001/x003 on press/hold/release. Similar to Hue remotes. """ if "devicemode" in self.raw["config"]: return SwitchDeviceMode(self.raw["config"]["devicemode"]) return None @property def mode(self) -> SwitchMode | None: """For Ubisys S1/S2, operation mode of the switch.""" if "mode" in self.raw["config"]: return SwitchMode(self.raw["config"]["mode"]) return None @property def window_covering_type(self) -> SwitchWindowCoveringType | None: """Set the covering type and starts calibration for Ubisys J1.""" if "windowcoveringtype" in self.raw["config"]: return SwitchWindowCoveringType(self.raw["config"]["windowcoveringtype"]) return None 07070100000046000081A400000000000000000000000164453E39000002B7000000000000000000000000000000000000003100000000deconz-112/pydeconz/models/sensor/temperature.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedTemperatureState(TypedDict): """Temperature state type definition.""" temperature: int class TypedTemperature(TypedDict): """Temperature type definition.""" state: TypedTemperatureState class Temperature(SensorBase): """Temperature sensor.""" raw: TypedTemperature @property def temperature(self) -> int: """Temperature.""" return self.raw["state"]["temperature"] @property def scaled_temperature(self) -> float: """Scaled temperature.""" return self.temperature / 100 07070100000047000081A400000000000000000000000164453E3900002A8E000000000000000000000000000000000000003000000000deconz-112/pydeconz/models/sensor/thermostat.py"""Python library to connect deCONZ and Home Assistant to work together.""" import enum import logging from typing import Literal from typing_extensions import NotRequired, TypedDict from . import SensorBase LOGGER = logging.getLogger(__name__) class ThermostatFanMode(enum.Enum): """Fan mode. Supported values: - "off" - "low" - "medium" - "high" - "on" - "auto" - "smart" """ OFF = "off" LOW = "low" MEDIUM = "medium" HIGH = "high" ON = "on" AUTO = "auto" SMART = "smart" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "ThermostatFanMode": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected thermostat fan mode %s", value) return ThermostatFanMode.UNKNOWN class ThermostatMode(enum.Enum): """Set the current operating mode of a thermostat. Supported values: - "off" - "auto" - "cool" - "heat" - "emergency heating" - "precooling" - "fan only" - "dry" - "sleep" """ OFF = "off" AUTO = "auto" COOL = "cool" HEAT = "heat" EMERGENCY_HEATING = "emergency heating" PRE_COOLING = "precooling" FAN_ONLY = "fan only" DRY = "dry" SLEEP = "sleep" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "ThermostatMode": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected thermostat mode %s", value) return ThermostatMode.UNKNOWN class ThermostatSwingMode(enum.Enum): """Set the AC louvers position. Supported values: - "fully closed" - "fully open" - "quarter open" - "half open" - "three quarters open" """ FULLY_CLOSED = "fully closed" FULLY_OPEN = "fully open" QUARTER_OPEN = "quarter open" HALF_OPEN = "half open" THREE_QUARTERS_OPEN = "three quarters open" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "ThermostatSwingMode": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected thermostat swing mode %s", value) return ThermostatSwingMode.UNKNOWN class ThermostatPreset(enum.Enum): """Set the current operating mode for Tuya thermostats. Supported values: - "auto" - "boost" - "comfort" - "complex" - "eco" - "holiday" - "manual" - "program" """ AUTO = "auto" BOOST = "boost" COMFORT = "comfort" COMPLEX = "complex" ECO = "eco" HOLIDAY = "holiday" MANUAL = "manual" PROGRAM = "program" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "ThermostatPreset": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected thermostat preset %s", value) return ThermostatPreset.UNKNOWN class ThermostatTemperatureMeasurement(enum.Enum): """Set the mode of operation for Elko Super TR thermostat. Supported values: - "air sensor" - "floor sensor" - "floor protection" """ AIR_SENSOR = "air sensor" FLOOR_SENSOR = "floor sensor" FLOOR_PROTECTION = "floor protection" UNKNOWN = "unknown" @classmethod def _missing_(cls, value: object) -> "ThermostatTemperatureMeasurement": """Set default enum member if an unknown value is provided.""" LOGGER.warning("Unexpected thermostat temperature measurement %s", value) return ThermostatTemperatureMeasurement.UNKNOWN class TypedThermostatConfig(TypedDict): """Thermostat config type definition.""" coolsetpoint: int displayflipped: bool externalsensortemp: int externalwindowopen: bool fanmode: NotRequired[Literal["off", "low", "medium", "high", "on", "auto", "smart"]] heatsetpoint: int locked: bool mode: NotRequired[ Literal[ "off", "auto", "cool", "heat", "emergency heating", "precooling", "fan only", "dry", "sleep", ] ] mountingmode: bool offset: int preset: NotRequired[ Literal[ "auto", "boost", "comfort", "complex", "eco", "holiday", "manual", "program" ] ] schedule_on: bool swingmode: NotRequired[ Literal[ "fully closed", "fully open", "quarter open", "half open", "three quarters open", ] ] temperaturemeasurement: NotRequired[ Literal["air sensor", "floor sensor", "floor protection"] ] windowopen_set: bool class TypedThermostatState(TypedDict): """Thermostat state type definition.""" errorcode: bool floortemperature: int heating: bool mountingmodeactive: bool on: bool temperature: int valve: int class TypedThermostat(TypedDict): """Thermostat type definition.""" config: TypedThermostatConfig state: TypedThermostatState class Thermostat(SensorBase): """Thermostat "sensor".""" raw: TypedThermostat @property def cooling_setpoint(self) -> int | None: """Cooling setpoint. 700-3500. """ return self.raw["config"].get("coolsetpoint") @property def scaled_cooling_setpoint(self) -> float | None: """Cooling setpoint. 7-35. """ if temperature := self.cooling_setpoint: return round(temperature / 100, 1) return None @property def display_flipped(self) -> bool | None: """Tells if display for TRVs is flipped.""" return self.raw["config"].get("displayflipped") @property def error_code(self) -> bool | None: """Error code.""" return self.raw["state"].get("errorcode") @property def external_sensor_temperature(self) -> int | None: """Track temperature value provided by an external sensor. -32768–32767. Device dependent and only exposed for devices supporting it. """ return self.raw["config"].get("externalsensortemp") @property def scaled_external_sensor_temperature(self) -> float | None: """Track temperature value provided by an external sensor. -327–327. """ if temperature := self.external_sensor_temperature: return round(temperature / 100, 1) return None @property def external_window_open(self) -> bool | None: """Track open/close state of an external sensor. Device dependent and only exposed for devices supporting it. """ return self.raw["config"].get("externalwindowopen") @property def fan_mode(self) -> ThermostatFanMode | None: """Fan mode.""" if "fanmode" in self.raw["config"]: return ThermostatFanMode(self.raw["config"]["fanmode"]) return None @property def floor_temperature(self) -> int | None: """Floor temperature.""" return self.raw["state"].get("floortemperature") @property def scaled_floor_temperature(self) -> float | None: """Floor temperature.""" if temperature := self.floor_temperature: return round(temperature / 100, 1) return None @property def heating(self) -> bool | None: """Heating setpoint.""" return self.raw["state"].get("heating") @property def heating_setpoint(self) -> int | None: """Heating setpoint. 500-3200. """ return self.raw["config"].get("heatsetpoint") @property def scaled_heating_setpoint(self) -> float | None: """Heating setpoint. 5-32. """ if temperature := self.heating_setpoint: return round(temperature / 100, 1) return None @property def locked(self) -> bool | None: """Child lock active/inactive for thermostats/TRVs supporting it.""" return self.raw["config"].get("locked") @property def mode(self) -> ThermostatMode | None: """Set the current operating mode of a thermostat.""" if "mode" in self.raw["config"]: return ThermostatMode(self.raw["config"]["mode"]) return None @property def mounting_mode(self) -> bool | None: """Set a TRV into mounting mode if supported (valve fully open position).""" return self.raw["config"].get("mountingmode") @property def mounting_mode_active(self) -> bool | None: """If thermostat mounting mode is active.""" return self.raw["state"].get("mountingmodeactive") @property def offset(self) -> int | None: """Add a signed offset value to measured temperature and humidity state values. Values send by the REST-API are already amended by the offset. """ return self.raw["config"].get("offset") @property def preset(self) -> ThermostatPreset | None: """Set the current operating mode for Tuya thermostats.""" if "preset" in self.raw["config"]: return ThermostatPreset(self.raw["config"]["preset"]) return None @property def schedule_enabled(self) -> bool | None: """Tell when thermostat schedule is enabled.""" return self.raw["config"].get("schedule_on") @property def state_on(self) -> bool | None: """Declare if the sensor is on or off.""" return self.raw["state"].get("on") @property def swing_mode(self) -> ThermostatSwingMode | None: """Set the AC louvers position.""" if "swingmode" in self.raw["config"]: return ThermostatSwingMode(self.raw["config"]["swingmode"]) return None @property def temperature(self) -> int: """Temperature.""" return self.raw["state"]["temperature"] @property def scaled_temperature(self) -> float: """Scaled temperature.""" return round(self.temperature / 100, 1) @property def temperature_measurement(self) -> ThermostatTemperatureMeasurement | None: """Set the mode of operation for Elko Super TR thermostat.""" if "temperaturemeasurement" in self.raw["config"]: return ThermostatTemperatureMeasurement( self.raw["config"]["temperaturemeasurement"] ) return None @property def valve(self) -> int | None: """How open the valve is.""" return self.raw["state"].get("valve") @property def window_open_detection(self) -> bool | None: """Set if window open detection shall be active or inactive for Tuya thermostats. Device dependent and only exposed for devices supporting it. """ return self.raw["config"].get("windowopen_set") 07070100000048000081A400000000000000000000000164453E39000001FD000000000000000000000000000000000000002A00000000deconz-112/pydeconz/models/sensor/time.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedTimeState(TypedDict): """Time state type definition.""" lastset: str class TypedTime(TypedDict): """Time type definition.""" state: TypedTimeState class Time(SensorBase): """Time sensor.""" raw: TypedTime @property def last_set(self) -> str: """Last time time was set.""" return self.raw["state"]["lastset"] 07070100000049000081A400000000000000000000000164453E39000005E9000000000000000000000000000000000000002F00000000deconz-112/pydeconz/models/sensor/vibration.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedVibrationConfig(TypedDict): """Vibration config type definition.""" sensitivity: int sensitivitymax: int class TypedVibrationState(TypedDict): """Vibration state type definition.""" orientation: list[str] tiltangle: int vibration: bool vibrationstrength: int class TypedVibration(TypedDict): """Vibration type definition.""" config: TypedVibrationConfig state: TypedVibrationState class Vibration(SensorBase): """Vibration sensor.""" raw: TypedVibration @property def orientation(self) -> list[str] | None: """Orientation.""" return self.raw["state"].get("orientation") @property def sensitivity(self) -> int | None: """Vibration sensitivity.""" return self.raw["config"].get("sensitivity") @property def max_sensitivity(self) -> int | None: """Vibration max sensitivity.""" return self.raw["config"].get("sensitivitymax") @property def tilt_angle(self) -> int | None: """Tilt angle.""" return self.raw["state"].get("tiltangle") @property def vibration(self) -> bool: """Vibration.""" return self.raw["state"]["vibration"] @property def vibration_strength(self) -> int | None: """Strength of vibration.""" return self.raw["state"].get("vibrationstrength") 0707010000004A000081A400000000000000000000000164453E39000001F8000000000000000000000000000000000000002B00000000deconz-112/pydeconz/models/sensor/water.py"""Python library to connect deCONZ and Home Assistant to work together.""" from typing import TypedDict from . import SensorBase class TypedWaterState(TypedDict): """Water state type definition.""" water: bool class TypedWater(TypedDict): """Water type definition.""" state: TypedWaterState class Water(SensorBase): """Water sensor.""" raw: TypedWater @property def water(self) -> bool: """Water detected.""" return self.raw["state"]["water"] 0707010000004B000081A400000000000000000000000164453E3900000000000000000000000000000000000000000000001D00000000deconz-112/pydeconz/py.typed0707010000004C000081A400000000000000000000000164453E3900000E9F000000000000000000000000000000000000001D00000000deconz-112/pydeconz/utils.py"""Python library to connect deCONZ and Home Assistant to work together.""" import logging from typing import Any, Callable, Final, TypedDict import aiohttp from .errors import RequestError, ResponseError, raise_error LOGGER = logging.getLogger(__name__) URL_DISCOVER: Final = "https://phoscon.de/discover" class DiscoveredBridge(TypedDict): """Discovered bridge type.""" id: str host: str mac: str name: str port: int async def delete_api_key( session: aiohttp.ClientSession, host: str, port: int, api_key: str ) -> None: """Delete API key from deCONZ.""" url = f"http://{host}:{port}/api/{api_key}/config/whitelist/{api_key}" response = await request(session.delete, url) LOGGER.info(response) async def delete_all_keys( session: aiohttp.ClientSession, host: str, port: int, api_key: str, api_keys: list[str] = [], ) -> None: """Delete all API keys except for the ones provided to the method.""" url = f"http://{host}:{port}/api/{api_key}/config" response = await request(session.get, url) api_keys.append(api_key) for key in response["whitelist"].keys(): if key not in api_keys: await delete_api_key(session, host, port, key) async def get_bridge_id( session: aiohttp.ClientSession, host: str, port: int, api_key: str ) -> str: """Get bridge id for bridge.""" url = f"http://{host}:{port}/api/{api_key}/config" response = await request(session.get, url) bridge_id = normalize_bridge_id(response["bridgeid"]) LOGGER.info("Bridge id: %s", bridge_id) return bridge_id async def discovery(session: aiohttp.ClientSession) -> list[DiscoveredBridge]: """Find bridges allowing gateway discovery.""" response: list[dict[str, Any]] = await request(session.get, URL_DISCOVER) LOGGER.info("Discovered the following bridges: %s.", response) return [ DiscoveredBridge( { "id": normalize_bridge_id(bridge["id"]), "host": bridge["internalipaddress"], "port": bridge["internalport"], "mac": bridge.get("macaddress", ""), "name": bridge.get("name", ""), } ) for bridge in response ] async def request( session: Callable[[Any], Any], url: str, **kwargs: Any, ) -> Any: """Do a web request and manage response.""" LOGGER.debug("Sending %s to %s", kwargs, url) try: res = await session(url, **kwargs) if res.content_type != "application/json": raise ResponseError("Invalid content type: {}".format(res.content_type)) response = await res.json() LOGGER.debug("HTTP request response: %s", response) _raise_on_error(response) return response except aiohttp.client_exceptions.ClientError as err: raise RequestError( "Error requesting data from {}: {}".format(url, err) ) from None def _raise_on_error(data: list[dict[str, Any]] | dict[str, Any]) -> None: """Check response for error message.""" if isinstance(data, list) and data: data = data[0] if isinstance(data, dict) and "error" in data: raise_error(data["error"]) def normalize_bridge_id(bridge_id: str) -> str: """Normalize a bridge identifier.""" bridge_id = bridge_id.upper() # discovery: contains 4 extra characters in the middle: "FFFF" if len(bridge_id) == 16 and bridge_id[6:10] == "FFFF": return bridge_id[0:6] + bridge_id[-6:] # deCONZ config API contains right ID. if len(bridge_id) == 12: return bridge_id LOGGER.warning("Received unexpected bridge id: %s", bridge_id) return bridge_id 0707010000004D000081A400000000000000000000000164453E3900001496000000000000000000000000000000000000002100000000deconz-112/pydeconz/websocket.py"""Python library to connect deCONZ and Home Assistant to work together.""" from asyncio import Task, create_task, get_running_loop from collections import deque from collections.abc import Callable, Coroutine import enum import logging from typing import Any, Final import aiohttp import orjson LOGGER = logging.getLogger(__name__) class Signal(enum.Enum): """What is the content of the callback.""" CONNECTION_STATE = "state" DATA = "data" class State(enum.Enum): """State of the connection.""" NONE = "" RETRYING = "retrying" RUNNING = "running" STOPPED = "stopped" RETRY_TIMER: Final = 15 class WSClient: """Websocket transport, session handling, message generation.""" def __init__( self, session: aiohttp.ClientSession, host: str, port: int, callback: Callable[[Signal], Coroutine[Any, Any, None]], ) -> None: """Create resources for websocket communication.""" self.session = session self.host = host self.port = port self.session_handler_callback = callback self.loop = get_running_loop() self._background_tasks: set[Task[Any]] = set() self._data: deque[dict[str, Any]] = deque() self._state = self._previous_state = State.NONE def create_background_task(self, target: Coroutine[Any, Any, Any]) -> None: """Save a reference to the result of target. To avoid a task disappearing mid-execution. The event loop only keeps weak references to tasks. A task that isn’t referenced elsewhere may get garbage collected at any time, even before it’s done. """ task = create_task(target) self._background_tasks.add(task) task.add_done_callback(self._background_tasks.discard) @property def data(self) -> dict[str, Any]: """Return data from data queue.""" try: return self._data.popleft() except IndexError: return {} @property def state(self) -> State: """State of websocket.""" return self._state def set_state(self, value: State) -> None: """Set state of websocket and store previous state.""" self._previous_state = self._state self._state = value def state_changed(self) -> None: """Signal state change.""" self.create_background_task( self.session_handler_callback(Signal.CONNECTION_STATE) ) def start(self) -> None: """Start websocket and update its state.""" self.create_background_task(self.running()) async def running(self) -> None: """Start websocket connection.""" if self._state == State.RUNNING: return url = f"http://{self.host}:{self.port}" try: async with self.session.ws_connect(url, heartbeat=60) as ws: LOGGER.info("Connected to deCONZ (%s)", self.host) self.set_state(State.RUNNING) self.state_changed() async for msg in ws: if self._state == State.STOPPED: await ws.close() break if msg.type == aiohttp.WSMsgType.TEXT: self._data.append(orjson.loads(msg.data)) self.create_background_task( self.session_handler_callback(Signal.DATA) ) LOGGER.debug(msg.data) continue if msg.type == aiohttp.WSMsgType.CLOSED: LOGGER.warning("Connection closed (%s)", self.host) break if msg.type == aiohttp.WSMsgType.ERROR: LOGGER.error("Websocket error (%s)", self.host) break except aiohttp.ClientConnectorError: if self._state != State.RETRYING: LOGGER.error("Websocket is not accessible (%s)", self.host) except Exception as err: if self._state != State.RETRYING: LOGGER.error("Unexpected error (%s) %s", self.host, err) if self._state != State.STOPPED: self.retry() def stop(self) -> None: """Close websocket connection.""" self.set_state(State.STOPPED) LOGGER.info("Shutting down connection to deCONZ (%s)", self.host) def retry(self) -> None: """Retry to connect to deCONZ. Do an immediate retry without timer and without signalling state change. Signal state change only after first retry fails. """ if self._state == State.RETRYING and self._previous_state == State.RUNNING: LOGGER.info( "Reconnecting to deCONZ (%s) failed, scheduling retry at an interval of %i seconds", self.host, RETRY_TIMER, ) self.state_changed() self.set_state(State.RETRYING) if self._previous_state == State.RUNNING: LOGGER.info("Reconnecting to deCONZ (%s)", self.host) self.start() return self.loop.call_later(RETRY_TIMER, self.start) 0707010000004E000081A400000000000000000000000164453E39000001DD000000000000000000000000000000000000001A00000000deconz-112/pyproject.toml[tool.isort] # https://github.com/PyCQA/isort/wiki/isort-Settings profile = "black" # will group `import x` and `from x import` of the same module. force_sort_within_sections = true known_first_party = ["pydeconz", "tests"] forced_separate = ["tests"] combine_as_imports = true [tool.pytest.ini_options] asyncio_mode = "auto" testpaths = ["tests"] [tool.ruff] target-version = "py310" select = [ "RUF006", # Store a reference to the return value of asyncio.create_task ] 0707010000004F000081A400000000000000000000000164453E390000002E000000000000000000000000000000000000002000000000deconz-112/requirements-dev.txtblack==23.3.0 pre-commit==3.2.2 ruff==0.0.261 07070100000050000081A400000000000000000000000164453E39000000E0000000000000000000000000000000000000002100000000deconz-112/requirements-test.txtaioresponses==0.7.4 black==23.3.0 pytest-cov==4.0.0 flake8==6.0.0 flake8-docstrings==1.7.0 flake8-noqa==1.3.1 isort==5.12.0 mypy==1.2.0 pydocstyle==6.3.0 pytest==7.3.1 pytest-aiohttp==1.0.4 ruff==0.0.261 types-orjson==3.6.2 07070100000051000081A400000000000000000000000164453E3900000033000000000000000000000000000000000000001C00000000deconz-112/requirements.txtaiohttp==3.8.4 async_timeout==4.0.2 orjson==3.8.10 07070100000052000081A400000000000000000000000164453E39000001AD000000000000000000000000000000000000001500000000deconz-112/setup.cfg[flake8] exclude = .venv,.git,.tox,docs,venv,bin,lib,deps,build doctests = True # Black manages certain aspects of formatting ignore = # D202 No blank lines allowed after function docstring D202, # E203: Whitespace before ':' E203, # E501: line too long E501, # W503: Line break occurred before a binary operator W503, # W504 line break after binary operator W504 noqa-require-code = True 07070100000053000081A400000000000000000000000164453E39000003AA000000000000000000000000000000000000001400000000deconz-112/setup.py"""Setup for pydeCONZ.""" from setuptools import find_packages, setup MIN_PY_VERSION = "3.10" PACKAGES = find_packages(exclude=["tests", "tests.*"]) VERSION = "112" setup( name="pydeconz", packages=PACKAGES, package_data={"pydeconz": ["py.typed"]}, version=VERSION, description="A Python library for communicating with deCONZ REST-API from Dresden Elektronik", author="Robert Svensson", author_email="Kane610@users.noreply.github.com", license="MIT", url="https://github.com/Kane610/deconz", download_url=f"https://github.com/Kane610/deconz/archive/v{VERSION}.tar.gz", install_requires=["aiohttp", "async_timeout", "orjson"], tests_require=["aioresponses", "pytest", "pytest-aiohttp", "types-orjson"], keywords=["deconz", "zigbee", "homeassistant"], classifiers=["Natural Language :: English", "Programming Language :: Python :: 3"], python_requires=f">={MIN_PY_VERSION}", ) 07070100000054000081ED00000000000000000000000164453E39000000FA000000000000000000000000000000000000001400000000deconz-112/setup.sh#!/usr/bin/env bash # Setup the repository. # Stop on errors set -e cd "$(dirname "$0")" python3 -m venv venv source venv/bin/activate python3 -m pip install -r requirements.txt -r requirements-test.txt -r requirements-dev.txt pre-commit install 07070100000055000041ED00000000000000000000000464453E3900000000000000000000000000000000000000000000001100000000deconz-112/tests07070100000056000081A400000000000000000000000164453E3900000052000000000000000000000000000000000000001D00000000deconz-112/tests/__init__.py"""Tests for pydeCONZ. pytest --cov-report term-missing --cov=pydeconz tests """ 07070100000057000081A400000000000000000000000164453E39000010AA000000000000000000000000000000000000001D00000000deconz-112/tests/conftest.py"""Setup common test helpers.""" from typing import Iterator from unittest.mock import Mock, patch import aiohttp from aioresponses import aioresponses import pytest from pydeconz import DeconzSession from pydeconz.models import ResourceGroup from pydeconz.models.event import EventType from pydeconz.websocket import Signal @pytest.fixture def mock_aioresponse(): """Mock a web request and provide a response.""" with aioresponses() as m: yield m @pytest.fixture def deconz_called_with(mock_aioresponse): """Verify deCONZ call was made with the expected parameters.""" def verify_call(method: str, path: str, **kwargs: dict) -> bool: """Verify expected data was provided with a request to aioresponse.""" for req, call_list in mock_aioresponse.requests.items(): if method != req[0]: continue if not req[1].path.endswith(path): continue for call in call_list: if kwargs.get("json") == call[1]["json"]: return True return False yield verify_call @pytest.fixture async def deconz_session() -> Iterator[DeconzSession]: """Return deCONZ gateway session. Clean up sessions automatically at the end of each test. """ session = aiohttp.ClientSession() controller = DeconzSession(session, "host", 80, "apikey") yield controller await session.close() @pytest.fixture def deconz_refresh_state(mock_aioresponse, deconz_session) -> Iterator[DeconzSession]: """Comfort fixture to initialize deCONZ session.""" async def data_to_deconz_session( alarm_systems=None, config=None, groups=None, lights=None, sensors=None ) -> DeconzSession: """Initialize deCONZ session.""" data = { "alarmsystems": alarm_systems or {}, "config": config or {}, "groups": groups or {}, "lights": lights or {}, "sensors": sensors or {}, } mock_aioresponse.get("http://host:80/api/apikey", payload=data) await deconz_session.refresh_state() return deconz_session yield data_to_deconz_session @pytest.fixture() def mock_wsclient(): """No real websocket allowed.""" with patch("pydeconz.gateway.WSClient") as mock: yield mock @pytest.fixture() def mock_websocket_event(deconz_session, mock_wsclient): """No real websocket allowed.""" deconz_session.connection_status_callback = Mock() deconz_session.start(websocketport=443) async def signal_new_event( resource: ResourceGroup, event: EventType = EventType.CHANGED, id: str | None = None, data: dict | None = None, unique_id: str | None = None, gid: str | None = None, scid: str | None = None, ) -> None: """Emit a websocket event signal.""" event_data = { "t": "event", "e": event.value, "r": resource.value, } if resource == ResourceGroup.SCENE: assert gid assert scid event_data |= { "gid": gid, "scid": scid, } else: assert id assert data event_data |= { "id": id, **data, } if resource in (ResourceGroup.LIGHT, ResourceGroup.SENSOR): assert unique_id event_data |= {"uniqueid": unique_id} mock_wsclient.return_value.data = event_data gateway_session_handler = mock_wsclient.call_args[0][3] await gateway_session_handler(signal=Signal.DATA) yield signal_new_event @pytest.fixture() def mock_websocket_state_change(deconz_session, mock_wsclient): """No real websocket allowed.""" deconz_session.connection_status_callback = Mock() deconz_session.start(websocketport=443) async def signal_state_change(state: str) -> None: """Emit a websocket state change signal.""" mock_wsclient.return_value.state = state gateway_session_handler = mock_wsclient.call_args[0][3] await gateway_session_handler(signal=Signal.CONNECTION_STATE) yield signal_state_change 07070100000058000041ED00000000000000000000000264453E3900000000000000000000000000000000000000000000001800000000deconz-112/tests/lights07070100000059000081A400000000000000000000000164453E3900000060000000000000000000000000000000000000002400000000deconz-112/tests/lights/__init__.py"""Tests for pydeCONZ lights. pytest --cov-report term-missing --cov=pydeconz tests/lights """ 0707010000005A000081A400000000000000000000000164453E3900000190000000000000000000000000000000000000002400000000deconz-112/tests/lights/conftest.py"""Setup common light test helpers.""" import pytest @pytest.fixture def deconz_light(deconz_refresh_state): """Comfort fixture to initialize deCONZ light.""" async def data_to_deconz_session(light): """Initialize deCONZ light.""" deconz_session = await deconz_refresh_state(lights={"0": light}) return deconz_session.lights["0"] yield data_to_deconz_session 0707010000005B000081A400000000000000000000000164453E3900000545000000000000000000000000000000000000003300000000deconz-112/tests/lights/test_configuration_tool.py"""Test pydeCONZ configuration tool. pytest --cov-report term-missing --cov=pydeconz.interfaces.lights --cov=pydeconz.models.light.configuration_tool tests/lights/test_configuration_tool.py """ DATA = { "etag": "26839cb118f5bf7ba1f2108256644010", "hascolor": False, "lastannounced": None, "lastseen": "2020-11-22T11:27Z", "manufacturername": "dresden elektronik", "modelid": "ConBee II", "name": "Configuration tool 1", "state": {"reachable": True}, "swversion": "0x264a0700", "type": "Configuration tool", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01", } async def test_light_configuration_tool(deconz_light): """Verify that configuration tool work.""" configuration_tool = await deconz_light(DATA) assert configuration_tool.state is None assert configuration_tool.reachable is True assert configuration_tool.deconz_id == "/lights/0" assert configuration_tool.etag == "26839cb118f5bf7ba1f2108256644010" assert configuration_tool.manufacturer == "dresden elektronik" assert configuration_tool.model_id == "ConBee II" assert configuration_tool.name == "Configuration tool 1" assert configuration_tool.software_version == "0x264a0700" assert configuration_tool.type == "Configuration tool" assert configuration_tool.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01" 0707010000005C000081A400000000000000000000000164453E390000197D000000000000000000000000000000000000002600000000deconz-112/tests/lights/test_cover.py"""Test pydeCONZ cover. pytest --cov-report term-missing --cov=pydeconz.interfaces.lights --cov=pydeconz.models.light.cover tests/lights/test_cover.py """ from unittest.mock import Mock from pydeconz.interfaces.lights import CoverAction DATA = { "etag": "87269755b9b3a046485fdae8d96b252c", "hascolor": False, "lastannounced": None, "lastseen": "2020-08-01T16:22:05Z", "manufacturername": "AXIS", "modelid": "Gear", "name": "Covering device", "state": { "bri": 0, "lift": 0, "on": False, "open": True, "reachable": True, }, "swversion": "100-5.3.5.1122", "type": "Window covering device", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01", } DATA_LEGACY = { "etag": "87269755b9b3a046485fdae8d96b252c", "hascolor": False, "lastannounced": None, "lastseen": "2020-08-01T16:22:05Z", "manufacturername": "AXIS", "modelid": "Gear", "name": "Covering device", "state": { "bri": 0, "on": False, "reachable": True, }, "swversion": "100-5.3.5.1122", "type": "Window covering device", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01", } async def test_handler_cover(mock_aioresponse, deconz_session, deconz_called_with): """Verify that controlling covers work.""" covers = deconz_session.lights.covers mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", action=CoverAction.OPEN) assert deconz_called_with("put", path="/lights/0/state", json={"open": True}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", action=CoverAction.CLOSE) assert deconz_called_with("put", path="/lights/0/state", json={"open": False}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", action=CoverAction.STOP) assert deconz_called_with("put", path="/lights/0/state", json={"stop": True}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", lift=30, tilt=60) assert deconz_called_with( "put", path="/lights/0/state", json={"lift": 30, "tilt": 60} ) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", action=CoverAction.STOP, lift=20) assert deconz_called_with("put", path="/lights/0/state", json={"stop": True}) async def test_handler_cover_legacy( mock_aioresponse, deconz_session, deconz_called_with ): """Verify that controlling covers work.""" covers = deconz_session.lights.covers mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", action=CoverAction.OPEN, legacy_mode=True) assert deconz_called_with("put", path="/lights/0/state", json={"on": False}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", action=CoverAction.CLOSE, legacy_mode=True) assert deconz_called_with("put", path="/lights/0/state", json={"on": True}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", action=CoverAction.STOP, legacy_mode=True) assert deconz_called_with("put", path="/lights/0/state", json={"bri_inc": 0}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", lift=30, tilt=60, legacy_mode=True) assert deconz_called_with( "put", path="/lights/0/state", json={"bri": 76, "sat": 152} ) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await covers.set_state("0", action=CoverAction.STOP, lift=20) assert deconz_called_with("put", path="/lights/0/state", json={"stop": True}) async def test_light_cover(deconz_light): """Verify that covers work.""" cover = await deconz_light(DATA) assert cover.state is False assert cover.is_open is True assert cover.lift == 0 assert cover.tilt is None assert cover.supports_tilt is False assert cover.reachable is True assert cover.deconz_id == "/lights/0" assert cover.etag == "87269755b9b3a046485fdae8d96b252c" assert cover.manufacturer == "AXIS" assert cover.model_id == "Gear" assert cover.name == "Covering device" assert cover.software_version == "100-5.3.5.1122" assert cover.type == "Window covering device" assert cover.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01" cover.register_callback(mock_callback := Mock()) assert cover._callbacks cover.update({"state": {"lift": 50, "open": True}}) assert cover.is_open is True assert cover.lift == 50 mock_callback.assert_called_once() assert cover.changed_keys == {"state", "lift", "open"} cover.update({"state": {"tilt": 25, "open": True}}) assert cover.tilt == 25 assert cover.supports_tilt is True cover.update({"state": {"bri": 30, "on": False}}) assert cover.is_open is True assert cover.lift == 50 cover.remove_callback(mock_callback) assert not cover._callbacks async def test_light_cover_legacy(deconz_light): """Verify that covers work with older deconz versions.""" cover = await deconz_light(DATA_LEGACY) assert cover.state is False assert cover.is_open is True assert cover.lift == 0 assert cover.tilt is None assert cover.reachable is True assert cover.deconz_id == "/lights/0" assert cover.etag == "87269755b9b3a046485fdae8d96b252c" assert cover.manufacturer == "AXIS" assert cover.model_id == "Gear" assert cover.name == "Covering device" assert cover.software_version == "100-5.3.5.1122" assert cover.type == "Window covering device" assert cover.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01" cover.register_callback(mock_callback := Mock()) assert cover._callbacks event = {"state": {"bri": 50, "on": True}} cover.update(event) assert cover.is_open is False assert cover.lift == 19 mock_callback.assert_called_once() assert cover.changed_keys == {"state", "bri", "on"} event = {"state": {"bri": 30, "on": False}} cover.update(event) assert cover.is_open is True assert cover.lift == 11 # Verify sat (for tilt) works as well cover.raw["state"]["sat"] = 40 assert cover.tilt == 15 cover.raw["state"]["lift"] = 0 cover.raw["state"]["tilt"] = 0 cover.raw["state"]["open"] = True assert cover.tilt == 0 cover.remove_callback(mock_callback) assert not cover._callbacks 0707010000005D000081A400000000000000000000000164453E3900002167000000000000000000000000000000000000002600000000deconz-112/tests/lights/test_light.py"""Test pydeCONZ light. pytest --cov-report term-missing --cov=pydeconz.interfaces.lights --cov=pydeconz.models.light.light tests/lights/test_light.py """ from unittest.mock import Mock import pytest from pydeconz.models.light.light import ( LightAlert, LightColorCapability, LightColorMode, LightEffect, LightFanSpeed, ) DATA = { "colorcapabilities": 15, "ctmax": 500, "ctmin": 153, "etag": "026bcfe544ad76c7534e5ca8ed39047c", "hascolor": True, "manufacturername": "dresden elektronik", "modelid": "FLS-PP3", "name": "Light 1", "pointsymbol": {}, "state": { "alert": None, "bri": 111, "colormode": "ct", "ct": 307, "effect": None, "hascolor": True, "hue": 7998, "on": False, "reachable": True, "sat": 172, "speed": 3, "xy": [0.421253, 0.39921], }, "swversion": "020C.201000A0", "type": "Extended color light", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-0A", } async def test_handler_light(mock_aioresponse, deconz_session, deconz_called_with): """Verify that creating a light works.""" lights = deconz_session.lights.lights mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await lights.set_state( id="0", alert=LightAlert.SHORT, brightness=200, color_loop_speed=10, color_temperature=400, effect=LightEffect.COLOR_LOOP, fan_speed=LightFanSpeed.OFF, hue=1000, on=True, on_time=100, saturation=150, transition_time=250, xy=(0.1, 0.1), ) assert deconz_called_with( "put", path="/lights/0/state", json={ "alert": "select", "bri": 200, "colorloopspeed": 10, "ct": 400, "effect": "colorloop", "hue": 1000, "on": True, "ontime": 100, "speed": 0, "sat": 150, "transitiontime": 250, "xy": (0.1, 0.1), }, ) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await lights.set_state("0", on=False) assert deconz_called_with( "put", path="/lights/0/state", json={"on": False}, ) async def test_handler_fan(mock_aioresponse, deconz_session, deconz_called_with): """Verify light fixture with fan work.""" lights = deconz_session.lights.lights mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await lights.set_state("0", fan_speed=LightFanSpeed.OFF) assert deconz_called_with("put", path="/lights/0/state", json={"speed": 0}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await lights.set_state("0", fan_speed=LightFanSpeed.PERCENT_25) assert deconz_called_with("put", path="/lights/0/state", json={"speed": 1}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await lights.set_state("0", fan_speed=LightFanSpeed.PERCENT_50) assert deconz_called_with("put", path="/lights/0/state", json={"speed": 2}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await lights.set_state("0", fan_speed=LightFanSpeed.PERCENT_75) assert deconz_called_with("put", path="/lights/0/state", json={"speed": 3}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await lights.set_state("0", fan_speed=LightFanSpeed.PERCENT_100) assert deconz_called_with("put", path="/lights/0/state", json={"speed": 4}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await lights.set_state("0", fan_speed=LightFanSpeed.AUTO) assert deconz_called_with("put", path="/lights/0/state", json={"speed": 5}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await lights.set_state("0", fan_speed=LightFanSpeed.COMFORT_BREEZE) assert deconz_called_with("put", path="/lights/0/state", json={"speed": 6}) async def test_light_light(mock_aioresponse, deconz_light, deconz_called_with): """Verify that creating a light works.""" light = await deconz_light(DATA) assert light.state is False assert light.on is False assert light.alert == LightAlert.UNKNOWN assert light.brightness == 111 assert light.hue == 7998 assert light.saturation == 172 assert light.color_temp == 307 assert light.xy == (0.421253, 0.39921) assert light.color_mode == LightColorMode.CT assert light.max_color_temp == 500 assert light.min_color_temp == 153 assert light.effect == LightEffect.UNKNOWN assert light.fan_speed == LightFanSpeed.PERCENT_75 assert light.supports_fan_speed is True assert ( light.color_capabilities == LightColorCapability.HUE_SATURATION | LightColorCapability.ENHANCED_HUE | LightColorCapability.COLOR_LOOP | LightColorCapability.XY_ATTRIBUTES | LightColorCapability.COLOR_TEMPERATURE ) assert light.reachable is True assert light.deconz_id == "/lights/0" assert light.etag == "026bcfe544ad76c7534e5ca8ed39047c" assert light.manufacturer == "dresden elektronik" assert light.model_id == "FLS-PP3" assert light.name == "Light 1" assert light.software_version == "020C.201000A0" assert light.type == "Extended color light" assert light.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-0A" light.register_callback(mock_callback := Mock()) assert light._callbacks event = {"state": {"xy": [0.1, 0.1]}} light.update(event) assert light.brightness == 111 assert light.xy == (0.1, 0.1) mock_callback.assert_called_once() assert light.changed_keys == {"state", "xy"} light.remove_callback(mock_callback) assert not light._callbacks light.raw["state"]["xy"] = (65555, 65555) assert light.xy == (1, 1) del light.raw["state"]["xy"] assert light.xy is None light.raw["ctmax"] = 1000 assert light.max_color_temp == 650 light.raw["ctmin"] = 0 assert light.min_color_temp == 140 del light.raw["ctmax"] assert light.max_color_temp is None del light.raw["ctmin"] assert light.min_color_temp is None light.update({"state": {"bri": 2}}) assert light.brightness == 2 ENUM_PROPERTY_DATA = [ ( ("state", "alert"), "alert", { "none": LightAlert.NONE, "lselect": LightAlert.LONG, "select": LightAlert.SHORT, None: LightAlert.UNKNOWN, }, ), ( ("colorcapabilities",), "color_capabilities", { 0: LightColorCapability.HUE_SATURATION, 9: LightColorCapability.UNKNOWN, None: LightColorCapability.UNKNOWN, }, ), ( ("state", "colormode"), "color_mode", { "ct": LightColorMode.CT, "hs": LightColorMode.HS, "xy": LightColorMode.XY, None: LightColorMode.UNKNOWN, }, ), ( ("state", "effect"), "effect", { "colorloop": LightEffect.COLOR_LOOP, "none": LightEffect.NONE, None: LightEffect.UNKNOWN, }, ), ( ("state", "speed"), "fan_speed", { 0: LightFanSpeed.OFF, 1: LightFanSpeed.PERCENT_25, 2: LightFanSpeed.PERCENT_50, 3: LightFanSpeed.PERCENT_75, 4: LightFanSpeed.PERCENT_100, 5: LightFanSpeed.AUTO, 6: LightFanSpeed.COMFORT_BREEZE, }, ), ] @pytest.mark.parametrize("path, property, data", ENUM_PROPERTY_DATA) async def test_enum_light_properties(deconz_light, path, property, data): """Verify enum properties return expected values or None.""" light = await deconz_light({"config": {}, "state": {}, "type": "Color light"}) for input, output in data.items(): data = {path[0]: input} if len(path) == 2: data = {path[0]: {path[1]: input}} light.update(data) assert getattr(light, property) == output async def test_enum_light_properties_no_key(deconz_light): """Verify enum properties return expected values or None.""" light = await deconz_light({"config": {}, "state": {}, "type": "Color light"}) assert light.alert is None assert light.color_capabilities is None assert light.color_mode is None assert light.effect is None with pytest.raises(KeyError): assert light.fan_speed 0707010000005E000081A400000000000000000000000164453E3900000821000000000000000000000000000000000000002500000000deconz-112/tests/lights/test_lock.py"""Test pydeCONZ lock. pytest --cov-report term-missing --cov=pydeconz.interfaces.lights --cov=pydeconz.models.light.lock tests/lights/test_lock.py """ from unittest.mock import Mock DATA = { "etag": "5c2ec06cde4bd654aef3a555fcd8ad12", "hascolor": False, "lastannounced": None, "lastseen": "2020-08-22T15:29:03Z", "manufacturername": "Danalock", "modelid": "V3-BTZB", "name": "Door lock", "state": { "alert": "none", "on": False, "reachable": True, }, "swversion": "19042019", "type": "Door Lock", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-00", } async def test_handler_lock(mock_aioresponse, deconz_session, deconz_called_with): """Verify that controlling locks work.""" locks = deconz_session.lights.locks mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await locks.set_state("0", lock=True) assert deconz_called_with("put", path="/lights/0/state", json={"on": True}) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await locks.set_state("0", lock=False) assert deconz_called_with("put", path="/lights/0/state", json={"on": False}) async def test_light_lock(deconz_light): """Verify that locks work.""" lock = await deconz_light(DATA) assert lock.state is False assert lock.is_locked is False assert lock.reachable is True assert lock.deconz_id == "/lights/0" assert lock.etag == "5c2ec06cde4bd654aef3a555fcd8ad12" assert lock.manufacturer == "Danalock" assert lock.model_id == "V3-BTZB" assert lock.name == "Door lock" assert lock.software_version == "19042019" assert lock.type == "Door Lock" assert lock.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-00" lock.register_callback(mock_callback := Mock()) assert lock._callbacks event = {"state": {"on": True}} lock.update(event) assert lock.is_locked is True mock_callback.assert_called_once() assert lock.changed_keys == {"state", "on"} lock.remove_callback(mock_callback) assert not lock._callbacks 0707010000005F000081A400000000000000000000000164453E39000004ED000000000000000000000000000000000000002F00000000deconz-112/tests/lights/test_range_extender.py"""Test pydeCONZ range extender. pytest --cov-report term-missing --cov=pydeconz.interfaces.lights --cov=pydeconz.models.light.range_extender tests/lights/test_range_extender.py """ DATA = { "etag": "62a220a6141a5956a6916633cad0d56f", "hascolor": False, "manufacturername": "IKEA of Sweden", "modelid": "TRADFRI signal repeater", "name": "Range extender 64", "state": { "alert": "none", "reachable": True, }, "swversion": "2.0.019", "type": "Range extender", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01", } async def test_light_range_extender(deconz_light): """Verify that range extender work.""" range_extender = await deconz_light(DATA) assert range_extender.state is None assert range_extender.reachable is True assert range_extender.deconz_id == "/lights/0" assert range_extender.etag == "62a220a6141a5956a6916633cad0d56f" assert range_extender.manufacturer == "IKEA of Sweden" assert range_extender.model_id == "TRADFRI signal repeater" assert range_extender.name == "Range extender 64" assert range_extender.software_version == "2.0.019" assert range_extender.type == "Range extender" assert range_extender.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01" 07070100000060000081A400000000000000000000000164453E390000092C000000000000000000000000000000000000002600000000deconz-112/tests/lights/test_siren.py"""Test pydeCONZ siren. pytest --cov-report term-missing --cov=pydeconz.interfaces.lights --cov=pydeconz.models.light.siren tests/lights/test_siren.py """ from unittest.mock import Mock DATA = { "etag": "0667cb8fff2adc1bf22be0e6eece2a18", "hascolor": False, "manufacturername": "Heiman", "modelid": "WarningDevice", "name": "alarm_tuin", "state": { "alert": "none", "reachable": True, }, "swversion": None, "type": "Warning device", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01", } async def test_handler_siren(mock_aioresponse, deconz_session, deconz_called_with): """Verify that sirens work.""" sirens = deconz_session.lights.sirens mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await sirens.set_state("0", True) assert deconz_called_with( "put", path="/lights/0/state", json={"alert": "lselect"}, ) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await sirens.set_state("0", True, duration=10) assert deconz_called_with( "put", path="/lights/0/state", json={"alert": "lselect", "ontime": 10}, ) mock_aioresponse.put("http://host:80/api/apikey/lights/0/state") await sirens.set_state("0", False, duration=10) assert deconz_called_with( "put", path="/lights/0/state", json={"alert": "none"}, ) async def test_light_siren(deconz_light): """Verify that sirens work.""" siren = await deconz_light(DATA) assert siren.state is None assert siren.is_on is False assert siren.reachable is True assert siren.deconz_id == "/lights/0" assert siren.etag == "0667cb8fff2adc1bf22be0e6eece2a18" assert siren.manufacturer == "Heiman" assert siren.model_id == "WarningDevice" assert siren.name == "alarm_tuin" assert not siren.software_version assert siren.type == "Warning device" assert siren.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01" siren.register_callback(mock_callback := Mock()) assert siren._callbacks event = {"state": {"alert": "lselect"}} siren.update(event) assert siren.is_on is True mock_callback.assert_called_once() assert siren.changed_keys == {"state", "alert"} siren.remove_callback(mock_callback) assert not siren._callbacks 07070100000061000041ED00000000000000000000000264453E3900000000000000000000000000000000000000000000001900000000deconz-112/tests/sensors07070100000062000081A400000000000000000000000164453E3900000062000000000000000000000000000000000000002500000000deconz-112/tests/sensors/__init__.py"""Tests for pydeCONZ sensors. pytest --cov-report term-missing --cov=pydeconz tests/sensors """ 07070100000063000081A400000000000000000000000164453E3900000198000000000000000000000000000000000000002500000000deconz-112/tests/sensors/conftest.py"""Setup common sensor test helpers.""" import pytest @pytest.fixture def deconz_sensor(deconz_refresh_state): """Comfort fixture to initialize deCONZ sensor.""" async def data_to_deconz_session(sensor): """Initialize deCONZ sensor.""" deconz_session = await deconz_refresh_state(sensors={"0": sensor}) return deconz_session.sensors["0"] yield data_to_deconz_session 07070100000064000081A400000000000000000000000164453E3900000C88000000000000000000000000000000000000002E00000000deconz-112/tests/sensors/test_air_purifier.py"""Test pydeCONZ air purifier. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.air_purifier tests/sensors/test_air_purifier.py """ from pydeconz.models.sensor.air_purifier import AirPurifierFanMode DATA = { "config": { "filterlifetime": 256728, "ledindication": True, "locked": False, "mode": "auto", "on": True, "reachable": True, }, "ep": 1, "etag": "fea6623ea3909029409fed7a6224e60b", "lastannounced": None, "lastseen": "2022-06-30T18:19Z", "manufacturername": "IKEA of Sweden", "modelid": "STARKVIND Air purifier", "name": "Starkvind", "state": { "deviceruntime": 185310, "filterruntime": 182857, "lastupdated": "2022-06-11T15:39:46.328", "replacefilter": False, "speed": 20, }, "swversion": "1.0.033", "type": "ZHAAirPurifier", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-fc7d", } async def test_handler_air_purifier( mock_aioresponse, deconz_session, deconz_called_with ): """Verify that air purifier controls works.""" air_purifier = deconz_session.sensors.air_purifier mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await air_purifier.set_config("0", AirPurifierFanMode.AUTO) assert deconz_called_with("put", path="/sensors/0/config", json={"mode": "auto"}) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await air_purifier.set_config("0", AirPurifierFanMode.OFF) assert deconz_called_with("put", path="/sensors/0/config", json={"mode": "off"}) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await air_purifier.set_config( "0", filter_life_time=1, led_indication=True, locked=False, ) assert deconz_called_with( "put", path="/sensors/0/config", json={ "filterlifetime": 1, "ledindication": True, "locked": False, }, ) async def test_sensor_air_purifier(deconz_sensor): """Verify that air purifier sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.device_run_time == 185310 assert sensor.fan_mode == AirPurifierFanMode.AUTO assert sensor.fan_speed == 20 assert sensor.filter_life_time == 256728 assert sensor.filter_run_time == 182857 assert sensor.led_indication is True assert sensor.locked is False assert sensor.replace_filter is False # DeconzSensor assert sensor.battery is None assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "fea6623ea3909029409fed7a6224e60b" assert sensor.manufacturer == "IKEA of Sweden" assert sensor.model_id == "STARKVIND Air purifier" assert sensor.name == "Starkvind" assert sensor.software_version == "1.0.033" assert sensor.type == "ZHAAirPurifier" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-fc7d" 07070100000065000081A400000000000000000000000164453E39000011F3000000000000000000000000000000000000002D00000000deconz-112/tests/sensors/test_air_quality.py"""Test pydeCONZ air quality sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.air_quality tests/sensors/test_air_quality.py """ import pytest from pydeconz.models import ResourceType from pydeconz.models.sensor.air_quality import AirQualityValue DATA = { "config": { "on": True, "reachable": True, }, "ep": 2, "etag": "c2d2e42396f7c78e11e46c66e2ec0200", "lastseen": "2020-11-20T22:48Z", "manufacturername": "BOSCH", "modelid": "AIR", "name": "BOSCH Air quality sensor", "state": { "airquality": "poor", "airqualityppb": 809, "lastupdated": "2020-11-20T22:48:00.209", }, "swversion": "20200402", "type": "ZHAAirQuality", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-02-fdef", } DATA_WITH_PM25 = { "config": { "on": True, "reachable": True, }, "ep": 1, "etag": "74eb5d8558a3895a39a3884189701c99", "lastannounced": None, "lastseen": "2022-06-30T18:20Z", "manufacturername": "IKEA of Sweden", "modelid": "STARKVIND Air purifier", "name": "Starkvind", "state": { "airquality": "excellent", "lastupdated": "2022-06-30T18:18:26.205", "pm2_5": 8, }, "swversion": "1.0.033", "type": "ZHAAirQuality", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-02-fc7d", } DATA_6_in_1_no_aq = { "config": { "on": True, "reachable": True, }, "etag": "74eb2d855fa3895a39a3484289705c99", "lastannounced": None, "lastseen": "2023-01-29T18:25Z", "manufacturername": "_TZE200_dwcarsat", "modelid": "TS0601", "name": "Tuya Smart Air House Keeper 6in1", "state": { "airquality_co2_density": 325, "airquality_formaldehyde_density": 4, "airqualityppb": 15, "lastupdated": "2023-01-29T19:05:41.903", "pm2_5": 9, }, "type": "ZHAAirQuality", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0c7d", } async def test_sensor_air_quality(deconz_sensor): """Verify that air quality sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.air_quality == AirQualityValue.POOR.value assert sensor.supports_air_quality is True assert sensor.air_quality_ppb == 809 assert sensor.pm_2_5 is None # DeconzSensor assert sensor.battery is None assert sensor.ep == 2 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "c2d2e42396f7c78e11e46c66e2ec0200" assert sensor.manufacturer == "BOSCH" assert sensor.model_id == "AIR" assert sensor.name == "BOSCH Air quality sensor" assert sensor.software_version == "20200402" assert sensor.type == "ZHAAirQuality" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-02-fdef" async def test_sensor_air_quality_with_pm2_5(deconz_sensor): """Verify that air quality with PM 2.5 sensor works.""" sensor = await deconz_sensor(DATA_WITH_PM25) assert sensor.air_quality == AirQualityValue.EXCELLENT.value assert sensor.supports_air_quality is True assert sensor.air_quality_ppb is None assert sensor.pm_2_5 == 8 async def test_sensor_air_quality_6_in_1_no_aq(deconz_sensor): """Verify that air quality 6 in 1 sensor works.""" sensor = await deconz_sensor(DATA_6_in_1_no_aq) assert sensor.air_quality == AirQualityValue.UNKNOWN.value assert sensor.air_quality_co2 == 325 assert sensor.air_quality_formaldehyde == 4 assert sensor.air_quality_ppb == 15 assert sensor.pm_2_5 == 9 assert sensor.supports_air_quality is False ENUM_PROPERTY_DATA = [ ( ("state", "airquality"), "air_quality", { "excellent": AirQualityValue.EXCELLENT.value, "unsupported": AirQualityValue.UNKNOWN.value, None: AirQualityValue.UNKNOWN.value, }, ), ] @pytest.mark.parametrize("path, property, data", ENUM_PROPERTY_DATA) async def test_enum_airquality_properties(deconz_sensor, path, property, data): """Verify enum properties return expected values or None.""" sensor = await deconz_sensor( { "config": {}, "state": {}, "type": ResourceType.ZHA_AIR_QUALITY.value, } ) for input, output in data.items(): sensor.update({path[0]: {path[1]: input}}) assert getattr(sensor, property) == output 07070100000066000081A400000000000000000000000164453E390000060B000000000000000000000000000000000000002700000000deconz-112/tests/sensors/test_alarm.py"""Test pydeCONZ alarm. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.alarm tests/sensors/test_alarm.py """ DATA = { "config": { "battery": 100, "on": True, "reachable": True, "temperature": 2600, }, "ep": 1, "etag": "18c0f3c2100904e31a7f938db2ba9ba9", "manufacturername": "dresden elektronik", "modelid": "lumi.sensor_motion.aq2", "name": "Alarm 10", "state": { "alarm": False, "lastupdated": "none", "lowbattery": None, "tampered": None, }, "swversion": "20170627", "type": "ZHAAlarm", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0500", } async def test_sensor_alarm(deconz_sensor): """Verify that alarm sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.alarm is False # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature == 26 # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "18c0f3c2100904e31a7f938db2ba9ba9" assert sensor.manufacturer == "dresden elektronik" assert sensor.model_id == "lumi.sensor_motion.aq2" assert sensor.name == "Alarm 10" assert sensor.software_version == "20170627" assert sensor.type == "ZHAAlarm" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0500" 07070100000067000081A400000000000000000000000164453E3900000AC2000000000000000000000000000000000000003300000000deconz-112/tests/sensors/test_ancillary_control.py"""Test pydeCONZ ancillary control. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.ancillary_control tests/sensors/test_ancillary_control.py """ import pytest from pydeconz.models.sensor.ancillary_control import ( AncillaryControlAction, AncillaryControlPanel, ) DATA = { "config": { "battery": 95, "enrolled": 1, "on": True, "pending": [], "reachable": True, }, "ep": 1, "etag": "5aaa1c6bae8501f59929539c6e8f44d6", "lastseen": "2021-07-25T18:07Z", "manufacturername": "lk", "modelid": "ZB-KeypadGeneric-D0002", "name": "Keypad", "state": { "action": "armed_stay", "lastupdated": "2021-07-25T18:02:51.172", "lowbattery": False, "panel": "exit_delay", "seconds_remaining": 55, "tampered": False, }, "swversion": "3.13", "type": "ZHAAncillaryControl", "uniqueid": "ec:1b:bd:ff:fe:6f:c3:4d-01-0501", } async def test_sensor_ancillary_control(deconz_sensor): """Verify that ancillary control sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.action == AncillaryControlAction.ARMED_STAY assert sensor.panel == AncillaryControlPanel.EXIT_DELAY assert sensor.seconds_remaining == 55 # DeconzSensor assert sensor.battery == 95 assert sensor.ep == 1 assert not sensor.low_battery assert sensor.on assert sensor.reachable assert not sensor.tampered assert not sensor.internal_temperature # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "5aaa1c6bae8501f59929539c6e8f44d6" assert sensor.manufacturer == "lk" assert sensor.model_id == "ZB-KeypadGeneric-D0002" assert sensor.name == "Keypad" assert sensor.software_version == "3.13" assert sensor.type == "ZHAAncillaryControl" assert sensor.unique_id == "ec:1b:bd:ff:fe:6f:c3:4d-01-0501" ENUM_PROPERTY_DATA = [ ( ("state", "panel"), "panel", { "armed_away": AncillaryControlPanel.ARMED_AWAY, "unsupported": AncillaryControlPanel.UNKNOWN, None: AncillaryControlPanel.UNKNOWN, }, ), ] @pytest.mark.parametrize("path, property, data", ENUM_PROPERTY_DATA) async def test_enum_ancillary_control_properties(deconz_sensor, path, property, data): """Verify enum properties return expected values or None.""" sensor = await deconz_sensor( {"config": {}, "state": {}, "type": "ZHAAncillaryControl"} ) assert getattr(sensor, property) is None for input, output in data.items(): sensor.update({path[0]: {path[1]: input}}) assert getattr(sensor, property) == output 07070100000068000081A400000000000000000000000164453E39000005D3000000000000000000000000000000000000002900000000deconz-112/tests/sensors/test_battery.py"""Test pydeCONZ battery. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.battery tests/sensors/test_battery.py """ DATA = { "config": { "alert": "none", "on": True, "reachable": True, }, "ep": 1, "etag": "23a8659f1cb22df2f51bc2da0e241bb4", "manufacturername": "IKEA of Sweden", "modelid": "FYRTUR block-out roller blind", "name": "FYRTUR block-out roller blind", "state": { "battery": 100, "lastupdated": "none", }, "swversion": "2.2.007", "type": "ZHABattery", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0001", } async def test_sensor_battery(deconz_sensor): """Verify that alarm sensor works.""" sensor = await deconz_sensor(DATA) # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "23a8659f1cb22df2f51bc2da0e241bb4" assert sensor.manufacturer == "IKEA of Sweden" assert sensor.model_id == "FYRTUR block-out roller blind" assert sensor.name == "FYRTUR block-out roller blind" assert sensor.software_version == "2.2.007" assert sensor.type == "ZHABattery" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0001" 07070100000069000081A400000000000000000000000164453E3900000631000000000000000000000000000000000000003100000000deconz-112/tests/sensors/test_carbon_monoxide.py"""Test pydeCONZ carbon monoxide sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.carbon_monoxide tests/sensors/test_carbon_monoxide.py """ DATA = { "config": { "battery": 100, "on": True, "pending": [], "reachable": True, }, "ep": 1, "etag": "b7599df551944df97b2aa87d160b9c45", "manufacturername": "Heiman", "modelid": "CO_V16", "name": "Cave, CO", "state": { "carbonmonoxide": False, "lastupdated": "none", "lowbattery": False, "tampered": False, }, "swversion": "20150330", "type": "ZHACarbonMonoxide", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0101", } async def test_sensor_carbon_monoxide(deconz_sensor): """Verify that carbon monoxide sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.carbon_monoxide is False # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is False assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is False assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "b7599df551944df97b2aa87d160b9c45" assert sensor.manufacturer == "Heiman" assert sensor.model_id == "CO_V16" assert sensor.name == "Cave, CO" assert sensor.software_version == "20150330" assert sensor.type == "ZHACarbonMonoxide" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0101" 0707010000006A000081A400000000000000000000000164453E3900000608000000000000000000000000000000000000002D00000000deconz-112/tests/sensors/test_consumption.py"""Test pydeCONZ consumption sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.consumption tests/sensors/test_consumption.py """ DATA = { "config": { "on": True, "reachable": True, }, "ep": 1, "etag": "a99e5bc463d15c23af7e89946e784cca", "manufacturername": "Heiman", "modelid": "SmartPlug", "name": "Consumption 15", "state": { "consumption": 11342, "lastupdated": "2018-03-12T19:19:08", "power": 123, }, "type": "ZHAConsumption", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0702", } async def test_sensor_consumption(deconz_sensor): """Verify that consumption sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.consumption == 11342 assert sensor.power == 123 assert sensor.scaled_consumption == 11.342 # DeconzSensor assert sensor.battery is None assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "a99e5bc463d15c23af7e89946e784cca" assert sensor.manufacturer == "Heiman" assert sensor.model_id == "SmartPlug" assert sensor.name == "Consumption 15" assert sensor.software_version == "" assert sensor.type == "ZHAConsumption" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0702" 0707010000006B000081A400000000000000000000000164453E3900000C07000000000000000000000000000000000000002A00000000deconz-112/tests/sensors/test_daylight.py"""Test pydeCONZ daylight sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.daylight tests/sensors/test_daylight.py """ from pydeconz.models.sensor.daylight import DayLightStatus DATA = { "config": { "configured": True, "on": True, "sunriseoffset": 30, "sunsetoffset": -30, }, "etag": "55047cf652a7e594d0ee7e6fae01dd38", "manufacturername": "Philips", "modelid": "PHDL00", "name": "Daylight", "state": { "dark": False, "daylight": True, "lastupdated": "2022-08-07T10:54:55.021", "status": 170, "sunrise": "2022-08-07T02:47:23", "sunset": "2022-08-07T19:02:23", }, "swversion": "1.0", "type": "Daylight", "uniqueid": "xx:xx:xx:FF:FF:xx:xx:xx-01", } async def test_handler_daylight(mock_aioresponse, deconz_session, deconz_called_with): """Verify that door lock sensor works.""" daylight = deconz_session.sensors.daylight mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await daylight.set_config("0", sunrise_offset=100) assert deconz_called_with( "put", path="/sensors/0/config", json={"sunriseoffset": 100} ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await daylight.set_config("0", sunset_offset=-100) assert deconz_called_with( "put", path="/sensors/0/config", json={"sunsetoffset": -100} ) async def test_sensor_daylight(deconz_sensor): """Verify that daylight sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.configured is True assert sensor.dark is False assert sensor.daylight is True assert sensor.daylight_status == DayLightStatus.SOLAR_NOON assert sensor.status == "solar_noon" assert sensor.sunrise == "2022-08-07T02:47:23" assert sensor.sunrise_offset == 30 assert sensor.sunset == "2022-08-07T19:02:23" assert sensor.sunset_offset == -30 # DeconzSensor assert sensor.battery is None assert sensor.ep is None assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "55047cf652a7e594d0ee7e6fae01dd38" assert sensor.manufacturer == "Philips" assert sensor.model_id == "PHDL00" assert sensor.name == "Daylight" assert sensor.software_version == "1.0" assert sensor.type == "Daylight" assert sensor.unique_id == "xx:xx:xx:FF:FF:xx:xx:xx-01" statuses = (100, 110, 120, 130, 140, 150, 160, 170, 180, 190, 200, 210, 220, 230, 0) for status in statuses: event = {"state": {"status": status}} sensor.update(event) assert sensor.changed_keys == {"state", "status"} assert sensor.daylight_status == DayLightStatus(status) sensor.update({"state": {"status": status}}) assert sensor.daylight_status == DayLightStatus.UNKNOWN 0707010000006C000081A400000000000000000000000164453E390000091A000000000000000000000000000000000000002B00000000deconz-112/tests/sensors/test_door_lock.py"""Test pydeCONZ door lock. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.door_lock tests/sensors/test_door_lock.py """ from pydeconz.models.sensor.door_lock import DoorLockLockState DATA = { "config": { "battery": 100, "lock": False, "on": True, "reachable": True, }, "ep": 11, "etag": "a43862f76b7fa48b0fbb9107df123b0e", "lastseen": "2021-03-06T22:25Z", "manufacturername": "Onesti Products AS", "modelid": "easyCodeTouch_v1", "name": "easyCodeTouch_v1", "state": { "lastupdated": "2021-03-06T21:25:45.624", "lockstate": "unlocked", }, "swversion": "20201211", "type": "ZHADoorLock", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-xx-0101", } async def test_handler_door_lock(mock_aioresponse, deconz_session, deconz_called_with): """Verify that door lock sensor works.""" locks = deconz_session.sensors.door_lock mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await locks.set_config("0", True) assert deconz_called_with("put", path="/sensors/0/config", json={"lock": True}) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await locks.set_config("0", False) assert deconz_called_with("put", path="/sensors/0/config", json={"lock": False}) async def test_sensor_door_lock(deconz_sensor): """Verify that door lock sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.is_locked is False assert sensor.lock_state == DoorLockLockState.UNLOCKED assert sensor.lock_configuration is False # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 11 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "a43862f76b7fa48b0fbb9107df123b0e" assert sensor.manufacturer == "Onesti Products AS" assert sensor.model_id == "easyCodeTouch_v1" assert sensor.name == "easyCodeTouch_v1" assert sensor.software_version == "20201211" assert sensor.type == "ZHADoorLock" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-xx-0101" 0707010000006D000081A400000000000000000000000164453E3900000B33000000000000000000000000000000000000002600000000deconz-112/tests/sensors/test_fire.py"""Test pydeCONZ fire sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.fire tests/sensors/test_fire.py """ DATA = { "config": { "on": True, "reachable": True, }, "ep": 1, "etag": "2b585d2c016bfd665ba27a8fdad28670", "manufacturername": "LUMI", "modelid": "lumi.sensor_smoke", "name": "sensor_kitchen_smoke", "state": { "fire": False, "lastupdated": "2018-02-20T11:25:02", }, "type": "ZHAFire", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0500", } DATA_DEVELCO = { "config": { "on": True, "battery": 90, "reachable": True, }, "ep": 1, "etag": "abcdef1234567890abcdef1234567890", "manufacturername": "frient A/S", "modelid": "SMSZB-120", "name": "Fire alarm", "state": { "fire": False, "lastupdated": "2021-11-25T08:00:02.003", "lowbattery": False, "test": True, }, "swversion": "20210526 05:57", "type": "ZHAFire", "uniqueid": "00:11:22:33:44:55:66:77-88-9900", } async def test_sensor_fire(deconz_sensor): """Verify that fire sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.fire is False # DeconzSensor assert sensor.battery is None assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.in_test_mode is False assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "2b585d2c016bfd665ba27a8fdad28670" assert sensor.manufacturer == "LUMI" assert sensor.model_id == "lumi.sensor_smoke" assert sensor.name == "sensor_kitchen_smoke" assert sensor.software_version == "" assert sensor.type == "ZHAFire" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0500" async def test_sensor_fire_develco(deconz_sensor): """Verify that develco/frient fire sensor works.""" sensor = await deconz_sensor(DATA_DEVELCO) assert sensor.fire is False # DeconzSensor assert sensor.battery == 90 assert sensor.ep == 1 assert sensor.low_battery is False assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.in_test_mode is True assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "abcdef1234567890abcdef1234567890" assert sensor.manufacturer == "frient A/S" assert sensor.model_id == "SMSZB-120" assert sensor.name == "Fire alarm" assert sensor.software_version == "20210526 05:57" assert sensor.type == "ZHAFire" assert sensor.unique_id == "00:11:22:33:44:55:66:77-88-9900" 0707010000006E000081A400000000000000000000000164453E3900000510000000000000000000000000000000000000002E00000000deconz-112/tests/sensors/test_generic_flag.py"""Test pydeCONZ generic flag sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.generic_flag tests/sensors/test_generic_flag.py """ DATA = { "config": { "on": True, "reachable": True, }, "modelid": "Switch", "name": "Kitchen Switch", "state": { "flag": True, "lastupdated": "2018-07-01T10:40:35", }, "swversion": "1.0.0", "type": "CLIPGenericFlag", "uniqueid": "kitchen-switch", } async def test_sensor_generic_flag(deconz_sensor): """Verify that generic flag sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.flag is True # DeconzSensor assert sensor.battery is None assert sensor.ep is None assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "" assert sensor.manufacturer == "" assert sensor.model_id == "Switch" assert sensor.name == "Kitchen Switch" assert sensor.software_version == "1.0.0" assert sensor.type == "CLIPGenericFlag" assert sensor.unique_id == "kitchen-switch" 0707010000006F000081A400000000000000000000000164453E39000005C8000000000000000000000000000000000000003000000000deconz-112/tests/sensors/test_generic_status.py"""Test pydeCONZ generic status sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.generic_status tests/sensors/test_generic_status.py """ DATA = { "config": { "on": True, "reachable": True, }, "etag": "aacc83bc7d6e4af7e44014e9f776b206", "manufacturername": "Phoscon", "modelid": "PHOSCON_FSM_STATE", "name": "FSM_STATE Motion stair", "state": { "lastupdated": "2019-04-24T00:00:25", "status": 0, }, "swversion": "1.0", "type": "CLIPGenericStatus", "uniqueid": "fsm-state-1520195376277", } async def test_sensor_generic_status(deconz_sensor): """Verify that generic flag sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.status == 0 # DeconzSensor assert sensor.battery is None assert sensor.ep is None assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "aacc83bc7d6e4af7e44014e9f776b206" assert sensor.manufacturer == "Phoscon" assert sensor.model_id == "PHOSCON_FSM_STATE" assert sensor.name == "FSM_STATE Motion stair" assert sensor.software_version == "1.0" assert sensor.type == "CLIPGenericStatus" assert sensor.unique_id == "fsm-state-1520195376277" 07070100000070000081A400000000000000000000000164453E390000079D000000000000000000000000000000000000002A00000000deconz-112/tests/sensors/test_humidity.py"""Test pydeCONZ humidity. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.humidity tests/sensors/test_humidity.py """ DATA = { "config": { "battery": 100, "offset": 0, "on": True, "reachable": True, }, "ep": 1, "etag": "1220e5d026493b6e86207993703a8a71", "manufacturername": "LUMI", "modelid": "lumi.weather", "name": "Mi temperature 1", "state": { "humidity": 3555, "lastupdated": "2019-05-05T14:39:00", }, "swversion": "20161129", "type": "ZHAHumidity", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0405", } async def test_handler_humidity(mock_aioresponse, deconz_session, deconz_called_with): """Verify that humidity sensor works.""" humidity = deconz_session.sensors.humidity mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await humidity.set_config("0", offset=1) assert deconz_called_with("put", path="/sensors/0/config", json={"offset": 1}) async def test_sensor_humidity(deconz_sensor): """Verify that humidity sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.humidity == 3555 assert sensor.offset == 0 assert sensor.scaled_humidity == 35.55 # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "1220e5d026493b6e86207993703a8a71" assert sensor.manufacturer == "LUMI" assert sensor.model_id == "lumi.weather" assert sensor.name == "Mi temperature 1" assert sensor.software_version == "20161129" assert sensor.type == "ZHAHumidity" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0405" 07070100000071000081A400000000000000000000000164453E3900000B7D000000000000000000000000000000000000002D00000000deconz-112/tests/sensors/test_light_level.py"""Test pydeCONZ light level. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.light_level tests/sensors/test_light_level.py """ DATA = { "config": { "alert": "none", "battery": 100, "ledindication": False, "on": True, "pending": [], "reachable": True, "tholddark": 12000, "tholdoffset": 7000, "usertest": False, }, "ep": 2, "etag": "5cfb81765e86aa53ace427cfd52c6d52", "manufacturername": "Philips", "modelid": "SML001", "name": "Motion sensor 4", "state": { "dark": True, "daylight": False, "lastupdated": "2019-05-05T14:37:06", "lightlevel": 6955, "lux": 5, }, "swversion": "6.1.0.18912", "type": "ZHALightLevel", "uniqueid": "00:17:88:01:03:28:8c:9b-02-0400", } async def test_handler_light_level( mock_aioresponse, deconz_session, deconz_called_with ): """Verify that configuring light level sensors works.""" light_level = deconz_session.sensors.light_level mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await light_level.set_config("0", threshold_dark=10, threshold_offset=20) assert deconz_called_with( "put", path="/sensors/0/config", json={"tholddark": 10, "tholdoffset": 20}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await light_level.set_config("0", threshold_dark=1) assert deconz_called_with( "put", path="/sensors/0/config", json={"tholddark": 1}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await light_level.set_config("0", threshold_offset=2) assert deconz_called_with( "put", path="/sensors/0/config", json={"tholdoffset": 2}, ) async def test_sensor_light_level(deconz_sensor): """Verify that light level sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.dark is True assert sensor.daylight is False assert sensor.light_level == 6955 assert sensor.lux == 5 assert sensor.scaled_light_level == 5 assert sensor.threshold_dark == 12000 assert sensor.threshold_offset == 7000 # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 2 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "5cfb81765e86aa53ace427cfd52c6d52" assert sensor.manufacturer == "Philips" assert sensor.model_id == "SML001" assert sensor.name == "Motion sensor 4" assert sensor.software_version == "6.1.0.18912" assert sensor.type == "ZHALightLevel" assert sensor.unique_id == "00:17:88:01:03:28:8c:9b-02-0400" 07070100000072000081A400000000000000000000000164453E39000005F7000000000000000000000000000000000000002A00000000deconz-112/tests/sensors/test_moisture.py"""Test pydeCONZ moisture. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.moisture tests/sensors/test_moisture.py """ DATA = { "config": { "battery": 100, "on": True, "reachable": True, }, "ep": 1, "etag": "814610ebe8c84ea0c87e137ea0a3fee6", "lastseen": "2021-08-15T06:58Z", "manufacturername": "modkam.ru", "modelid": "DIYRuZ_Flower", "name": "Moisture 8", "state": { "lastupdated": "2021-08-15T06:58:52.547", "moisture": 69, }, "swversion": "22/07/2021 10:04", "type": "ZHAMoisture", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0408", } async def test_sensor_moisture(deconz_sensor): """Verify that moisture sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.moisture == 69 # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "814610ebe8c84ea0c87e137ea0a3fee6" assert sensor.manufacturer == "modkam.ru" assert sensor.model_id == "DIYRuZ_Flower" assert sensor.name == "Moisture 8" assert sensor.software_version == "22/07/2021 10:04" assert sensor.type == "ZHAMoisture" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0408" 07070100000073000081A400000000000000000000000164453E39000005D7000000000000000000000000000000000000002C00000000deconz-112/tests/sensors/test_open_close.py"""Test pydeCONZ open close sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.open_close tests/sensors/test_open_close.py """ DATA = { "config": { "battery": 95, "on": True, "reachable": True, "temperature": 3300, }, "ep": 1, "etag": "66cc641d0368110da6882b50090174ac", "manufacturername": "LUMI", "modelid": "lumi.sensor_magnet.aq2", "name": "Back Door", "state": {"lastupdated": "2019-05-05T14:54:32", "open": False}, "swversion": "20161128", "type": "ZHAOpenClose", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0006", } async def test_sensor_open_close(deconz_sensor): """Verify that open/close sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.open is False # DeconzSensor assert sensor.battery == 95 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature == 33 # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "66cc641d0368110da6882b50090174ac" assert sensor.manufacturer == "LUMI" assert sensor.model_id == "lumi.sensor_magnet.aq2" assert sensor.name == "Back Door" assert sensor.software_version == "20161128" assert sensor.type == "ZHAOpenClose" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0006" 07070100000074000081A400000000000000000000000164453E3900000B9F000000000000000000000000000000000000002700000000deconz-112/tests/sensors/test_power.py"""Test pydeCONZ power sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.power tests/sensors/test_power.py """ import pytest DATA = { "config": { "on": True, "reachable": True, }, "ep": 1, "etag": "96e71c7db4685b334d3d0decc3f11868", "manufacturername": "Heiman", "modelid": "SmartPlug", "name": "Power 16", "state": { "current": 34, "lastupdated": "2018-03-12T19:22:13", "power": 64, "voltage": 231, }, "type": "ZHAPower", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0b04", } DATA_ONLY_POWER = { "config": { "on": True, "reachable": True, "temperature": 3400, }, "ep": 2, "etag": "77ab6ddae6dd81469080ad62118d81b6", "lastseen": "2021-07-07T19:30Z", "manufacturername": "LUMI", "modelid": "lumi.plug.maus01", "name": "Power 27", "state": { "lastupdated": "2021-07-07T19:24:59.664", "power": 1, }, "swversion": "05-02-2018", "type": "ZHAPower", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-02-000c", } @pytest.mark.parametrize( "input, expected", [ ( DATA, { "battery": None, "current": 34, "deconz_id": "/sensors/0", "ep": 1, "etag": "96e71c7db4685b334d3d0decc3f11868", "low_battery": None, "manufacturer": "Heiman", "model_id": "SmartPlug", "name": "Power 16", "on": True, "power": 64, "reachable": True, "internal_temperature": None, "software_version": "", "tampered": None, "type": "ZHAPower", "unique_id": "xx:xx:xx:xx:xx:xx:xx:xx-01-0b04", "voltage": 231, }, ), ( DATA_ONLY_POWER, { "battery": None, "current": None, "deconz_id": "/sensors/0", "ep": 2, "etag": "77ab6ddae6dd81469080ad62118d81b6", "low_battery": None, "manufacturer": "LUMI", "model_id": "lumi.plug.maus01", "name": "Power 27", "on": True, "power": 1, "reachable": True, "internal_temperature": 34.0, "software_version": "05-02-2018", "tampered": None, "type": "ZHAPower", "unique_id": "xx:xx:xx:xx:xx:xx:xx:xx-02-000c", "voltage": None, }, ), ], ) async def test_sensor_power(input, expected, deconz_sensor): """Verify that power sensor works.""" sensor = await deconz_sensor(input) for attr, value in expected.items(): assert getattr(sensor, attr) == value 07070100000075000081A400000000000000000000000164453E390000177A000000000000000000000000000000000000002A00000000deconz-112/tests/sensors/test_presence.py"""Test pydeCONZ presence. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.presence tests/sensors/test_presence.py """ from pydeconz.models.sensor.presence import ( PresenceConfigDeviceMode, PresenceConfigTriggerDistance, PresenceStatePresenceEvent, ) DATA = { "config": { "alert": "none", "battery": 100, "delay": 0, "ledindication": False, "on": True, "pending": [], "reachable": True, "sensitivity": 1, "sensitivitymax": 2, "usertest": False, }, "ep": 2, "etag": "5cfb81765e86aa53ace427cfd52c6d52", "manufacturername": "Philips", "modelid": "SML001", "name": "Motion sensor 4", "state": { "lastupdated": "2019-05-05T14:37:06", "presence": False, }, "swversion": "6.1.0.18912", "type": "ZHAPresence", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-02-0406", } DATA_PRESENCE_EVENT = { "config": { "devicemode": "undirected", "on": True, "reachable": True, "sensitivity": 3, "triggerdistance": "medium", }, "etag": "13ff209f9401b317987d42506dd4cd79", "lastannounced": None, "lastseen": "2022-06-28T23:13Z", "manufacturername": "aqara", "modelid": "lumi.motion.ac01", "name": "Aqara FP1", "state": { "lastupdated": "2022-06-28T23:13:38.577", "presence": True, "presenceevent": "leave", }, "swversion": "20210121", "type": "ZHAPresence", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0406", } async def test_handler_presence(mock_aioresponse, deconz_session, deconz_called_with): """Verify that configuring presence sensor works.""" presence = deconz_session.sensors.presence mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await presence.set_config("0", delay=10, duration=20, sensitivity=1) assert deconz_called_with( "put", path="/sensors/0/config", json={"delay": 10, "duration": 20, "sensitivity": 1}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await presence.set_config("0", delay=1) assert deconz_called_with( "put", path="/sensors/0/config", json={"delay": 1}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await presence.set_config("0", duration=2) assert deconz_called_with( "put", path="/sensors/0/config", json={"duration": 2}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await presence.set_config("0", sensitivity=3) assert deconz_called_with( "put", path="/sensors/0/config", json={"sensitivity": 3}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await presence.set_config("0", device_mode=PresenceConfigDeviceMode.LEFT_AND_RIGHT) assert deconz_called_with( "put", path="/sensors/0/config", json={"devicemode": "leftright"}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await presence.set_config("0", reset_presence=True) assert deconz_called_with( "put", path="/sensors/0/config", json={"resetpresence": True}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await presence.set_config("0", trigger_distance=PresenceConfigTriggerDistance.FAR) assert deconz_called_with( "put", path="/sensors/0/config", json={"triggerdistance": "far"}, ) async def test_sensor_presence(deconz_sensor): """Verify that presence sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.dark is None assert sensor.delay == 0 assert sensor.device_mode is None assert sensor.duration is None assert sensor.presence is False assert sensor.presence_event is None assert sensor.sensitivity == 1 assert sensor.max_sensitivity == 2 assert sensor.trigger_distance is None # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 2 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "5cfb81765e86aa53ace427cfd52c6d52" assert sensor.manufacturer == "Philips" assert sensor.model_id == "SML001" assert sensor.name == "Motion sensor 4" assert sensor.software_version == "6.1.0.18912" assert sensor.type == "ZHAPresence" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-02-0406" async def test_sensor_presence_event(deconz_sensor): """Verify that presence event sensor works.""" sensor = await deconz_sensor(DATA_PRESENCE_EVENT) assert sensor.dark is None assert sensor.delay is None assert sensor.device_mode == PresenceConfigDeviceMode.UNDIRECTED assert sensor.duration is None assert sensor.presence is True assert sensor.presence_event == PresenceStatePresenceEvent.LEAVE assert sensor.sensitivity == 3 assert sensor.max_sensitivity is None assert sensor.trigger_distance == PresenceConfigTriggerDistance.MEDIUM # DeconzSensor assert sensor.battery is None assert sensor.ep is None assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "13ff209f9401b317987d42506dd4cd79" assert sensor.manufacturer == "aqara" assert sensor.model_id == "lumi.motion.ac01" assert sensor.name == "Aqara FP1" assert sensor.software_version == "20210121" assert sensor.type == "ZHAPresence" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0406" 07070100000076000081A400000000000000000000000164453E39000005C9000000000000000000000000000000000000002A00000000deconz-112/tests/sensors/test_pressure.py"""Test pydeCONZ pressure sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.pressure tests/sensors/test_pressure.py """ DATA = { "config": { "battery": 100, "on": True, "reachable": True, }, "ep": 1, "etag": "1220e5d026493b6e86207993703a8a71", "manufacturername": "LUMI", "modelid": "lumi.weather", "name": "Mi temperature 1", "state": { "lastupdated": "2019-05-05T14:39:00", "pressure": 1010, }, "swversion": "20161129", "type": "ZHAPressure", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0403", } async def test_sensor_pressure(deconz_sensor): """Verify that pressure sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.pressure == 1010 # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "1220e5d026493b6e86207993703a8a71" assert sensor.manufacturer == "LUMI" assert sensor.model_id == "lumi.weather" assert sensor.name == "Mi temperature 1" assert sensor.software_version == "20161129" assert sensor.type == "ZHAPressure" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0403" 07070100000077000081A400000000000000000000000164453E3900000753000000000000000000000000000000000000003100000000deconz-112/tests/sensors/test_relative_rotary.py"""Test pydeCONZ relative rotary sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.relative_rotary tests/sensors/test_relative_rotary.py """ from pydeconz.models.sensor.relative_rotary import RelativeRotaryEvent DATA = { "config": { "battery": 100, "on": True, "reachable": True, }, "etag": "463728970bdb7d04048fc4373654f45a", "lastannounced": "2022-07-03T13:57:59Z", "lastseen": "2022-07-03T14:02Z", "manufacturername": "Signify Netherlands B.V.", "modelid": "RDM002", "name": "RDM002 44", "state": { "expectedeventduration": 400, "expectedrotation": 75, "lastupdated": "2022-07-03T11:37:49.586", "rotaryevent": 2, }, "swversion": "2.59.19", "type": "ZHARelativeRotary", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-14-fc00", } async def test_sensor_relative_rotary(deconz_sensor): """Verify that relative rotary sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.expected_event_duration == 400 assert sensor.expected_rotation == 75 assert sensor.rotary_event == RelativeRotaryEvent.REPEAT # DeconzSensor assert sensor.battery == 100 assert sensor.ep is None assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "463728970bdb7d04048fc4373654f45a" assert sensor.manufacturer == "Signify Netherlands B.V." assert sensor.model_id == "RDM002" assert sensor.name == "RDM002 44" assert sensor.software_version == "2.59.19" assert sensor.type == "ZHARelativeRotary" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-14-fc00" 07070100000078000081A400000000000000000000000164453E390000229E000000000000000000000000000000000000002800000000deconz-112/tests/sensors/test_switch.py"""Test pydeCONZ switch. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.switch tests/sensors/test_switch.py """ from pydeconz.models.sensor.switch import ( SwitchDeviceMode, SwitchMode, SwitchWindowCoveringType, ) DATA = { "config": { "battery": 90, "group": "201", "on": True, "reachable": True, }, "ep": 2, "etag": "233ae541bbb7ac98c42977753884b8d2", "manufacturername": "Philips", "mode": 1, "modelid": "RWL021", "name": "Dimmer switch 3", "state": { "buttonevent": 1002, "lastupdated": "2019-04-28T20:29:13", }, "swversion": "5.45.1.17846", "type": "ZHASwitch", "uniqueid": "00:17:88:01:02:0e:32:a3-02-fc00", } DATA_CUBE = { "config": { "battery": 90, "on": True, "reachable": True, "temperature": 1100, }, "ep": 3, "etag": "e34fa1c7a19d960e35a1f4d56ac475af", "manufacturername": "LUMI", "mode": 1, "modelid": "lumi.sensor_cube.aqgl01", "name": "Mi Magic Cube", "state": { "buttonevent": 747, "gesture": 7, "lastupdated": "2019-12-12T18:50:40", }, "swversion": "20160704", "type": "ZHASwitch", "uniqueid": "00:15:8d:00:02:8b:3b:24-03-000c", } DATA_HUE_WALL_SWITCH = { "config": { "battery": 100, "devicemode": "dualrocker", "on": True, "pending": [], "reachable": True, }, "ep": 1, "etag": "01173dc5b19bb0a976006eee8d0d3718", "lastseen": "2021-03-12T22:55Z", "manufacturername": "Signify Netherlands B.V.", "mode": 1, "modelid": "RDM001", "name": "RDM001 15", "state": { "buttonevent": 1002, "eventduration": 1, "lastupdated": "2021-03-12T22:21:20.017", }, "swversion": "20210115", "type": "ZHASwitch", "uniqueid": "00:17:88:01:0b:00:05:5d-01-fc00", } DATA_TINT_REMOTE = { "config": { "group": "16388,16389,16390", "on": True, "reachable": True, }, "ep": 1, "etag": "b1336f750d31300afa441a04f2c69b68", "manufacturername": "MLI", "mode": 1, "modelid": "ZBT-Remote-ALL-RGBW", "name": "ZHA Remote 1", "state": { "angle": 10, "buttonevent": 6002, "lastupdated": "2020-09-08T18:58:24.193", "xy": [0.3381, 0.1627], }, "swversion": "2.0", "type": "ZHASwitch", "uniqueid": "00:11:22:33:44:55:66:77-01-1000", } DATA_UBISYS_J1 = { "config": { "mode": "momentary", "on": True, "reachable": False, "windowcoveringtype": 0, }, "ep": 2, "etag": "da5fbb89eca4133b6949537e73b31f77", "lastseen": "2020-11-21T15:47Z", "manufacturername": "ubisys", "mode": 1, "modelid": "J1 (5502)", "name": "J1", "state": { "buttonevent": None, "lastupdated": "none", }, "swversion": "20190129-DE-FB0", "type": "ZHASwitch", "uniqueid": "00:1f:ee:00:00:00:00:09-02-0102", } async def test_handler_switch(mock_aioresponse, deconz_session, deconz_called_with): """Verify that configuring presence sensor works.""" switch = deconz_session.sensors.switch mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await switch.set_config("0", device_mode=SwitchDeviceMode.DUAL_ROCKER) assert deconz_called_with( "put", path="/sensors/0/config", json={"devicemode": "dualrocker"}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await switch.set_config("0", mode=SwitchMode.ROCKER) assert deconz_called_with( "put", path="/sensors/0/config", json={"mode": "rocker"}, ) mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await switch.set_config("0", window_covering_type=SwitchWindowCoveringType.DRAPERY) assert deconz_called_with( "put", path="/sensors/0/config", json={"windowcoveringtype": 4}, ) async def test_sensor_switch(deconz_sensor): """Verify that switch sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.button_event == 1002 assert sensor.gesture is None assert sensor.angle is None assert sensor.xy is None assert sensor.device_mode is None # DeconzSensor assert sensor.battery == 90 assert sensor.ep == 2 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "233ae541bbb7ac98c42977753884b8d2" assert sensor.manufacturer == "Philips" assert sensor.model_id == "RWL021" assert sensor.name == "Dimmer switch 3" assert sensor.software_version == "5.45.1.17846" assert sensor.type == "ZHASwitch" assert sensor.unique_id == "00:17:88:01:02:0e:32:a3-02-fc00" async def test_sensor_switch_sensor_cube(deconz_sensor): """Verify that cube switch sensor works.""" sensor = await deconz_sensor(DATA_CUBE) assert sensor.button_event == 747 assert sensor.gesture == 7 # DeconzSensor assert sensor.battery == 90 assert sensor.ep == 3 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature == 11.0 # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "e34fa1c7a19d960e35a1f4d56ac475af" assert sensor.manufacturer == "LUMI" assert sensor.model_id == "lumi.sensor_cube.aqgl01" assert sensor.name == "Mi Magic Cube" assert sensor.software_version == "20160704" assert sensor.type == "ZHASwitch" assert sensor.unique_id == "00:15:8d:00:02:8b:3b:24-03-000c" async def test_sensor_switch_hue_wall_switch_module(deconz_sensor): """Verify that cube switch sensor works.""" sensor = await deconz_sensor(DATA_HUE_WALL_SWITCH) assert sensor.button_event == 1002 assert sensor.event_duration == 1 assert sensor.device_mode == SwitchDeviceMode.DUAL_ROCKER assert not sensor.angle assert not sensor.gesture assert not sensor.mode assert not sensor.window_covering_type assert not sensor.xy # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert not sensor.low_battery assert sensor.on assert sensor.reachable assert not sensor.tampered assert not sensor.internal_temperature # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "01173dc5b19bb0a976006eee8d0d3718" assert sensor.manufacturer == "Signify Netherlands B.V." assert sensor.model_id == "RDM001" assert sensor.name == "RDM001 15" assert sensor.software_version == "20210115" assert sensor.type == "ZHASwitch" assert sensor.unique_id == "00:17:88:01:0b:00:05:5d-01-fc00" async def test_sensor_switch_tint_remote(deconz_sensor): """Verify that tint remote sensor works.""" sensor = await deconz_sensor(DATA_TINT_REMOTE) assert sensor.button_event == 6002 assert sensor.angle == 10 assert sensor.xy == [0.3381, 0.1627] # DeconzSensor assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "b1336f750d31300afa441a04f2c69b68" assert sensor.manufacturer == "MLI" assert sensor.model_id == "ZBT-Remote-ALL-RGBW" assert sensor.name == "ZHA Remote 1" assert sensor.software_version == "2.0" assert sensor.type == "ZHASwitch" assert sensor.unique_id == "00:11:22:33:44:55:66:77-01-1000" async def test_sensor_switch_ubisys_j1(deconz_sensor): """Verify that tint remote sensor works.""" sensor = await deconz_sensor(DATA_UBISYS_J1) assert sensor.button_event is None assert sensor.angle is None assert sensor.xy is None assert sensor.mode == SwitchMode.MOMENTARY assert sensor.window_covering_type == SwitchWindowCoveringType.ROLLER_SHADE # DeconzSensor assert sensor.ep == 2 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is False assert sensor.tampered is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "da5fbb89eca4133b6949537e73b31f77" assert sensor.manufacturer == "ubisys" assert sensor.model_id == "J1 (5502)" assert sensor.name == "J1" assert sensor.software_version == "20190129-DE-FB0" assert sensor.type == "ZHASwitch" assert sensor.unique_id == "00:1f:ee:00:00:00:00:09-02-0102" 07070100000079000081A400000000000000000000000164453E3900000627000000000000000000000000000000000000002D00000000deconz-112/tests/sensors/test_temperature.py"""Test pydeCONZ temperature sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.temperature tests/sensors/test_temperature.py """ DATA = { "config": { "battery": 100, "offset": 0, "on": True, "reachable": True, }, "ep": 1, "etag": "1220e5d026493b6e86207993703a8a71", "manufacturername": "LUMI", "modelid": "lumi.weather", "name": "Mi temperature 1", "state": { "lastupdated": "2019-05-05T14:39:00", "temperature": 2182, }, "swversion": "20161129", "type": "ZHATemperature", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0402", } async def test_sensor_temperature(deconz_sensor): """Verify that temperature sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.temperature == 2182 assert sensor.scaled_temperature == 21.82 # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "1220e5d026493b6e86207993703a8a71" assert sensor.manufacturer == "LUMI" assert sensor.model_id == "lumi.weather" assert sensor.name == "Mi temperature 1" assert sensor.software_version == "20161129" assert sensor.type == "ZHATemperature" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0402" 0707010000007A000081A400000000000000000000000164453E390000373F000000000000000000000000000000000000002C00000000deconz-112/tests/sensors/test_thermostat.py"""Test pydeCONZ thermostat. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.thermostat tests/sensors/test_thermostat.py """ import pytest from pydeconz.models.sensor.thermostat import ( ThermostatFanMode, ThermostatMode, ThermostatPreset, ThermostatSwingMode, ThermostatTemperatureMeasurement, ) DATA = { "config": { "battery": 59, "displayflipped": None, "heatsetpoint": 2100, "locked": None, "mountingmode": None, "offset": 0, "on": True, "reachable": True, }, "ep": 1, "etag": "6130553ac247174809bae47144ee23f8", "lastseen": "2020-11-29T19:31Z", "manufacturername": "Danfoss", "modelid": "eTRV0100", "name": "Thermostat_stue_sofa", "state": { "errorcode": None, "lastupdated": "2020-11-29T19:28:40.665", "mountingmodeactive": False, "on": True, "temperature": 2102, "valve": 24, "windowopen": "Closed", }, "swversion": "01.02.0008 01.02", "type": "ZHAThermostat", "uniqueid": "14:b4:57:ff:fe:d5:4e:77-01-0201", } DATA_EUROTRONIC = { "config": { "battery": 100, "displayflipped": True, "heatsetpoint": 2100, "locked": False, "mode": "auto", "offset": 0, "on": True, "reachable": True, }, "ep": 1, "etag": "25aac331bc3c4b465cfb2197f6243ea4", "manufacturername": "Eurotronic", "modelid": "SPZB0001", "name": "Living Room Radiator", "state": { "lastupdated": "2019-02-10T22:41:32", "on": False, "temperature": 2149, "valve": 0, }, "swversion": "15181120", "type": "ZHAThermostat", "uniqueid": "00:15:8d:00:01:92:d2:51-01-0201", } DATA_TUYA = { "config": { "battery": 100, "heatsetpoint": 1550, "locked": None, "offset": 0, "on": True, "preset": "auto", "reachable": True, "schedule": {}, "schedule_on": None, "setvalve": True, "windowopen_set": True, }, "ep": 1, "etag": "36850fc8521f7c23606c9304b2e1f7bb", "lastseen": "2020-11-11T21:23Z", "manufacturername": "_TYST11_kfvq6avy", "modelid": "fvq6avy", "name": "fvq6avy", "state": { "lastupdated": "none", "on": None, "temperature": 2290, }, "swversion": "20180727", "type": "ZHAThermostat", "uniqueid": "bc:33:ac:ff:fe:47:a1:95-01-0201", } async def test_handler_thermostat(mock_aioresponse, deconz_session, deconz_called_with): """Verify that configuring thermostat sensor works.""" thermostat = deconz_session.sensors.thermostat mock_aioresponse.put("http://host:80/api/apikey/sensors/0/config") await thermostat.set_config( id="0", cooling_setpoint=1000, enable_schedule=True, external_sensor_temperature=24, external_window_open=True, fan_mode=ThermostatFanMode.AUTO, flip_display=False, heating_setpoint=500, locked=True, mode=ThermostatMode.AUTO, mounting_mode=False, on=True, preset=ThermostatPreset.AUTO, schedule=[], set_valve=True, swing_mode=ThermostatSwingMode.HALF_OPEN, temperature_measurement=ThermostatTemperatureMeasurement.FLOOR_SENSOR, window_open_detection=True, ) assert deconz_called_with( "put", path="/sensors/0/config", json={ "coolsetpoint": 1000, "schedule_on": True, "externalsensortemp": 24, "externalwindowopen": True, "fanmode": "auto", "displayflipped": False, "heatsetpoint": 500, "locked": True, "mode": "auto", "mountingmode": False, "on": True, "preset": "auto", "schedule": [], "setvalve": True, "swingmode": "half open", "temperaturemeasurement": "floor sensor", "windowopen_set": True, }, ) async def test_sensor_danfoss_thermostat(deconz_sensor): """Verify that Danfoss thermostat works. Danfoss thermostat is the simplest kind with only control over temperaturdeconz_sensore. """ sensor = await deconz_sensor(DATA) assert sensor.cooling_setpoint is None assert sensor.display_flipped is None assert sensor.error_code is None assert sensor.external_sensor_temperature is None assert sensor.external_window_open is None assert sensor.fan_mode is None assert sensor.floor_temperature is None assert sensor.heating is None assert sensor.heating_setpoint == 2100 assert sensor.scaled_heating_setpoint == 21.00 assert sensor.locked is None assert sensor.mode is None assert sensor.mounting_mode is None assert sensor.mounting_mode_active is False assert sensor.offset == 0 assert sensor.preset is None assert sensor.state_on assert sensor.swing_mode is None assert sensor.temperature == 2102 assert sensor.scaled_temperature == 21.0 assert sensor.temperature_measurement is None assert sensor.valve == 24 assert sensor.window_open_detection is None # DeconzSensor assert sensor.battery == 59 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "6130553ac247174809bae47144ee23f8" assert sensor.manufacturer == "Danfoss" assert sensor.model_id == "eTRV0100" assert sensor.name == "Thermostat_stue_sofa" assert sensor.software_version == "01.02.0008 01.02" assert sensor.type == "ZHAThermostat" assert sensor.unique_id == "14:b4:57:ff:fe:d5:4e:77-01-0201" async def test_sensor_eurotronic_thermostat(deconz_sensor): """Verify that thermostat sensor works.""" sensor = await deconz_sensor(DATA_EUROTRONIC) assert sensor.cooling_setpoint is None assert sensor.error_code is None assert sensor.fan_mode is None assert sensor.floor_temperature is None assert sensor.heating is None assert sensor.heating_setpoint == 2100 assert sensor.scaled_heating_setpoint == 21.00 assert sensor.locked is False assert sensor.mode == ThermostatMode.AUTO assert sensor.mounting_mode is None assert sensor.mounting_mode_active is None assert sensor.offset == 0 assert sensor.preset is None assert not sensor.state_on assert sensor.swing_mode is None assert sensor.temperature == 2149 assert sensor.scaled_temperature == 21.5 assert sensor.temperature_measurement is None assert sensor.valve == 0 assert sensor.window_open_detection is None # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "25aac331bc3c4b465cfb2197f6243ea4" assert sensor.manufacturer == "Eurotronic" assert sensor.model_id == "SPZB0001" assert sensor.name == "Living Room Radiator" assert sensor.software_version == "15181120" assert sensor.type == "ZHAThermostat" assert sensor.unique_id == "00:15:8d:00:01:92:d2:51-01-0201" async def test_sensor_tuya_thermostat(deconz_sensor): """Verify that Tuya thermostat works.""" sensor = await deconz_sensor(DATA_TUYA) assert sensor.heating_setpoint == 1550 assert sensor.scaled_heating_setpoint == 15.50 assert sensor.locked is None assert sensor.mode is None assert sensor.offset == 0 assert sensor.schedule_enabled is None assert not sensor.state_on assert sensor.temperature == 2290 assert sensor.scaled_temperature == 22.9 assert sensor.valve is None # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "36850fc8521f7c23606c9304b2e1f7bb" assert sensor.manufacturer == "_TYST11_kfvq6avy" assert sensor.model_id == "fvq6avy" assert sensor.name == "fvq6avy" assert sensor.software_version == "20180727" assert sensor.type == "ZHAThermostat" assert sensor.unique_id == "bc:33:ac:ff:fe:47:a1:95-01-0201" # Verify temperature conversion to increase coverage sensor.update( { "config": { "coolsetpoint": 1000, "externalsensortemp": 2000, "heatsetpoint": None, }, "state": { "floortemperature": 4000, }, } ) assert sensor.cooling_setpoint == 1000 assert sensor.scaled_cooling_setpoint == 10 assert sensor.external_sensor_temperature == 2000 assert sensor.scaled_external_sensor_temperature == 20 assert sensor.heating_setpoint is None assert sensor.scaled_heating_setpoint is None assert sensor.floor_temperature == 4000 assert sensor.scaled_floor_temperature == 40 PROPERTY_DATA = [ ( ("config", "displayflipped"), "display_flipped", {True: True, False: False, None: None}, ), ( ("state", "errorcode"), "error_code", {True: True, False: False, None: None}, ), ( ("config", "externalwindowopen"), "external_window_open", {True: True, False: False, None: None}, ), ( ("state", "heating"), "heating", {True: True, False: False, None: None}, ), ( ("config", "locked"), "locked", {True: True, False: False, None: None}, ), ( ("state", "mountingmodeactive"), "mounting_mode_active", {True: True, False: False, None: None}, ), ( ("config", "offset"), "offset", {1: 1, 0: 0, None: None}, ), ( ("config", "schedule_on"), "schedule_enabled", {True: True, False: False, None: None}, ), ( ("state", "valve"), "valve", {1: 1, 0: 0, None: None}, ), ( ("config", "windowopen_set"), "window_open_detection", {True: True, False: False, None: None}, ), ] @pytest.mark.parametrize("path, property, data", PROPERTY_DATA) async def test_thermostat_properties(deconz_sensor, path, property, data): """Verify normal thermostat properties.""" sensor = await deconz_sensor({"config": {}, "state": {}, "type": "ZHAThermostat"}) for input, output in data.items(): sensor.update({path[0]: {path[1]: input}}) assert getattr(sensor, property) == output ENUM_PROPERTY_DATA = [ ( ("config", "fanmode"), "fan_mode", { "auto": ThermostatFanMode.AUTO, "unsupported": ThermostatFanMode.UNKNOWN, None: ThermostatFanMode.UNKNOWN, }, ), ( ("config", "mode"), "mode", { "auto": ThermostatMode.AUTO, "unsupported": ThermostatMode.UNKNOWN, None: ThermostatMode.UNKNOWN, }, ), ( ("config", "preset"), "preset", { "auto": ThermostatPreset.AUTO, "unsupported": ThermostatPreset.UNKNOWN, None: ThermostatPreset.UNKNOWN, }, ), ( ("config", "swingmode"), "swing_mode", { "fully open": ThermostatSwingMode.FULLY_OPEN, "unsupported": ThermostatSwingMode.UNKNOWN, None: ThermostatSwingMode.UNKNOWN, }, ), ( ("config", "temperaturemeasurement"), "temperature_measurement", { "air sensor": ThermostatTemperatureMeasurement.AIR_SENSOR, "unsupported": ThermostatTemperatureMeasurement.UNKNOWN, None: ThermostatTemperatureMeasurement.UNKNOWN, }, ), ] @pytest.mark.parametrize("path, property, data", ENUM_PROPERTY_DATA) async def test_enum_thermostat_properties(deconz_sensor, path, property, data): """Verify enum properties return expected values or None.""" sensor = await deconz_sensor({"config": {}, "state": {}, "type": "ZHAThermostat"}) assert getattr(sensor, property) is None for input, output in data.items(): sensor.update({path[0]: {path[1]: input}}) assert getattr(sensor, property) == output SCALED_PROPERTY_DATA = [ ( ("config", "coolsetpoint"), "cooling_setpoint", {1000: (1000, 10), 0: (0, None), None: (None, None)}, ), ( ("config", "externalsensortemp"), "external_sensor_temperature", {2000: (2000, 20), 0: (0, None), None: (None, None)}, ), ( ("state", "floortemperature"), "floor_temperature", {3000: (3000, 30), 0: (0, None), None: (None, None)}, ), ( ("config", "heatsetpoint"), "heating_setpoint", {4000: (4000, 40), 0: (0, None), None: (None, None)}, ), ( ("state", "temperature"), "temperature", {5000: (5000, 50), 0: (0, 0.0)}, ), ] @pytest.mark.parametrize("path, property, data", SCALED_PROPERTY_DATA) async def test_scaled_thermostat_properties(deconz_sensor, path, property, data): """Verify the scaling properties of thermostat.""" sensor = await deconz_sensor({"config": {}, "state": {}, "type": "ZHAThermostat"}) for input, output in data.items(): sensor.update({path[0]: {path[1]: input}}) assert getattr(sensor, property) == output[0] assert getattr(sensor, f"scaled_{property}") == output[1] 0707010000007B000081A400000000000000000000000164453E3900000640000000000000000000000000000000000000002600000000deconz-112/tests/sensors/test_time.py"""Test pydeCONZ time sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.time tests/sensors/test_time.py """ DATA = { "config": { "battery": 40, "on": True, "reachable": True, }, "ep": 1, "etag": "28e796678d9a24712feef59294343bb6", "lastseen": "2020-11-22T11:26Z", "manufacturername": "Danfoss", "modelid": "eTRV0100", "name": "eTRV Séjour", "state": { "lastset": "2020-11-19T08:07:08Z", "lastupdated": "2020-11-22T10:51:03.444", "localtime": "2020-11-22T10:51:01", "utc": "2020-11-22T10:51:01Z", }, "swversion": "20200429", "type": "ZHATime", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-000a", } async def test_sensor_time(deconz_sensor): """Verify that time sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.last_set == "2020-11-19T08:07:08Z" # DeconzSensor assert sensor.battery == 40 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature is None # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "28e796678d9a24712feef59294343bb6" assert sensor.manufacturer == "Danfoss" assert sensor.model_id == "eTRV0100" assert sensor.name == "eTRV Séjour" assert sensor.software_version == "20200429" assert sensor.type == "ZHATime" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-000a" 0707010000007C000081A400000000000000000000000164453E390000090D000000000000000000000000000000000000002B00000000deconz-112/tests/sensors/test_vibration.py"""Test pydeCONZ vibration sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.vibration tests/sensors/test_vibration.py """ from unittest.mock import Mock DATA = { "config": { "battery": 91, "on": True, "pending": [], "reachable": True, "sensitivity": 21, "sensitivitymax": 21, "temperature": 3200, }, "ep": 1, "etag": "b7599df551944df97b2aa87d160b9c45", "manufacturername": "LUMI", "modelid": "lumi.vibration.aq1", "name": "Vibration 1", "state": { "lastupdated": "2019-03-09T15:53:07", "orientation": [10, 1059, 0], "tiltangle": 83, "vibration": True, "vibrationstrength": 114, }, "swversion": "20180130", "type": "ZHAVibration", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0101", } async def test_sensor_vibration(deconz_sensor): """Verify that vibration sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.orientation == [10, 1059, 0] assert sensor.sensitivity == 21 assert sensor.max_sensitivity == 21 assert sensor.tilt_angle == 83 assert sensor.vibration is True assert sensor.vibration_strength == 114 # DeconzSensor assert sensor.battery == 91 assert sensor.ep == 1 assert sensor.low_battery is None assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is None assert sensor.internal_temperature == 32 # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "b7599df551944df97b2aa87d160b9c45" assert sensor.manufacturer == "LUMI" assert sensor.model_id == "lumi.vibration.aq1" assert sensor.name == "Vibration 1" assert sensor.software_version == "20180130" assert sensor.type == "ZHAVibration" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0101" sensor.register_callback(mock_callback := Mock()) assert sensor._callbacks event = {"state": {"lastupdated": "2019-03-15T10:15:17", "orientation": [0, 84, 6]}} sensor.update(event) mock_callback.assert_called_once() assert sensor.changed_keys == {"state", "lastupdated", "orientation"} sensor.remove_callback(mock_callback) assert not sensor._callbacks 0707010000007D000081A400000000000000000000000164453E3900000603000000000000000000000000000000000000002700000000deconz-112/tests/sensors/test_water.py"""Test pydeCONZ water sensor. pytest --cov-report term-missing --cov=pydeconz.interfaces.sensors --cov=pydeconz.models.sensor.water tests/sensors/test_water.py """ DATA = { "config": { "battery": 100, "on": True, "reachable": True, "temperature": 2500, }, "ep": 1, "etag": "fae893708dfe9b358df59107d944fa1c", "manufacturername": "LUMI", "modelid": "lumi.sensor_wleak.aq1", "name": "water2", "state": { "lastupdated": "2019-01-29T07:13:20", "lowbattery": False, "tampered": False, "water": False, }, "swversion": "20170721", "type": "ZHAWater", "uniqueid": "xx:xx:xx:xx:xx:xx:xx:xx-01-0500", } async def test_sensor_water(deconz_sensor): """Verify that water sensor works.""" sensor = await deconz_sensor(DATA) assert sensor.water is False # DeconzSensor assert sensor.battery == 100 assert sensor.ep == 1 assert sensor.low_battery is False assert sensor.on is True assert sensor.reachable is True assert sensor.tampered is False assert sensor.internal_temperature == 25 # DeconzDevice assert sensor.deconz_id == "/sensors/0" assert sensor.etag == "fae893708dfe9b358df59107d944fa1c" assert sensor.manufacturer == "LUMI" assert sensor.model_id == "lumi.sensor_wleak.aq1" assert sensor.name == "water2" assert sensor.software_version == "20170721" assert sensor.type == "ZHAWater" assert sensor.unique_id == "xx:xx:xx:xx:xx:xx:xx:xx-01-0500" 0707010000007E000081A400000000000000000000000164453E3900001D82000000000000000000000000000000000000002600000000deconz-112/tests/test_alarm_system.py"""Test pydeCONZ alarm systems. pytest --cov-report term-missing --cov=pydeconz.alarm_system tests/test_alarm_system.py """ from pydeconz.models.alarm_system import ( AlarmSystemArmAction, AlarmSystemArmMode, AlarmSystemArmState, AlarmSystemDeviceTrigger, ) async def test_create_alarm_system( mock_aioresponse, deconz_refresh_state, deconz_called_with ): """Verify that alarm system works.""" deconz_session = await deconz_refresh_state( alarm_systems={ "0": { "name": "default", "config": { "armmode": "armed_away", "configured": True, "disarmed_entry_delay": 0, "disarmed_exit_delay": 0, "armed_away_entry_delay": 120, "armed_away_exit_delay": 120, "armed_away_trigger_duration": 120, "armed_stay_entry_delay": 120, "armed_stay_exit_delay": 120, "armed_stay_trigger_duration": 120, "armed_night_entry_delay": 120, "armed_night_exit_delay": 120, "armed_night_trigger_duration": 120, }, "state": {"armstate": "armed_away", "seconds_remaining": 0}, "devices": { "ec:1b:bd:ff:fe:6f:c3:4d-01-0501": {"armmask": "none"}, "00:15:8d:00:02:af:95:f9-01-0101": { "armmask": "AN", "trigger": "state/vibration", }, }, } } ) alarm_systems = deconz_session.alarm_systems assert len(alarm_systems.keys()) == 1 mock_aioresponse.post("http://host:80/api/apikey/alarmsystems") await alarm_systems.create_alarm_system(name="not_default") assert deconz_called_with( "post", path="/alarmsystems", json={"name": "not_default"}, ) mock_aioresponse.put("http://host:80/api/apikey/alarmsystems/0/arm_away") await alarm_systems.arm(id="0", action=AlarmSystemArmAction.AWAY, pin_code="1234") assert deconz_called_with( "put", path="/alarmsystems/0/arm_away", json={"code0": "1234"} ) mock_aioresponse.put("http://host:80/api/apikey/alarmsystems/0/arm_night") await alarm_systems.arm(id="0", action=AlarmSystemArmAction.NIGHT, pin_code="23456") assert deconz_called_with( "put", path="/alarmsystems/0/arm_night", json={"code0": "23456"} ) mock_aioresponse.put("http://host:80/api/apikey/alarmsystems/0/arm_stay") await alarm_systems.arm(id="0", action=AlarmSystemArmAction.STAY, pin_code="345678") assert deconz_called_with( "put", path="/alarmsystems/0/arm_stay", json={"code0": "345678"} ) mock_aioresponse.put("http://host:80/api/apikey/alarmsystems/0/disarm") await alarm_systems.arm( id="0", action=AlarmSystemArmAction.DISARM, pin_code="4567890" ) assert deconz_called_with( "put", path="/alarmsystems/0/disarm", json={"code0": "4567890"} ) mock_aioresponse.put("http://host:80/api/apikey/alarmsystems/0/config") await alarm_systems.set_alarm_system_configuration( id="0", code0="0000", armed_away_entry_delay=0, armed_away_exit_delay=1, armed_away_trigger_duration=2, armed_night_entry_delay=3, armed_night_exit_delay=4, armed_night_trigger_duration=5, armed_stay_entry_delay=6, armed_stay_exit_delay=7, armed_stay_trigger_duration=8, disarmed_entry_delay=9, disarmed_exit_delay=10, ) assert deconz_called_with( "put", path="/alarmsystems/0/config", json={ "code0": "0000", "armed_away_entry_delay": 0, "armed_away_exit_delay": 1, "armed_away_trigger_duration": 2, "armed_night_entry_delay": 3, "armed_night_exit_delay": 4, "armed_night_trigger_duration": 5, "armed_stay_entry_delay": 6, "armed_stay_exit_delay": 7, "armed_stay_trigger_duration": 8, "disarmed_entry_delay": 9, "disarmed_exit_delay": 10, }, ) mock_aioresponse.put("http://host:80/api/apikey/alarmsystems/0/config") await alarm_systems.set_alarm_system_configuration( id="0", code0="4444", ) assert deconz_called_with( "put", path="/alarmsystems/0/config", json={"code0": "4444"}, ) mock_aioresponse.put( "http://host:80/api/apikey/alarmsystems/0/device/00:00:00:00:00:00:00:01-01-0101" ) await alarm_systems.add_device( id="0", unique_id="00:00:00:00:00:00:00:01-01-0101", armed_away=True, armed_night=True, armed_stay=True, trigger=AlarmSystemDeviceTrigger.ON, ) assert deconz_called_with( "put", path="/alarmsystems/0/device/00:00:00:00:00:00:00:01-01-0101", json={"armmask": "ANS", "trigger": "state/on"}, ) mock_aioresponse.put( "http://host:80/api/apikey/alarmsystems/0/device/00:00:00:00:00:00:00:01-01-0101" ) await alarm_systems.add_device( id="0", unique_id="00:00:00:00:00:00:00:01-01-0101", armed_night=True, ) assert deconz_called_with( "put", path="/alarmsystems/0/device/00:00:00:00:00:00:00:01-01-0101", json={"armmask": "N"}, ) mock_aioresponse.put( "http://host:80/api/apikey/alarmsystems/0/device/00:00:00:00:00:00:00:01-01-0101" ) await alarm_systems.add_device( id="0", unique_id="00:00:00:00:00:00:00:01-01-0101", armed_stay=True, is_keypad=True, ) assert deconz_called_with( "put", path="/alarmsystems/0/device/00:00:00:00:00:00:00:01-01-0101", json={}, ) mock_aioresponse.delete( "http://host:80/api/apikey/alarmsystems/0/device/00:00:00:00:00:00:00:01-01-0101" ) await alarm_systems.remove_device( id="0", unique_id="00:00:00:00:00:00:00:01-01-0101", ) assert deconz_called_with( "delete", path="/alarmsystems/0/device/00:00:00:00:00:00:00:01-01-0101", json=None, ) # Test model alarm_system_0 = alarm_systems["0"] assert alarm_system_0.deconz_id == "/alarmsystems/0" assert alarm_system_0.arm_state == AlarmSystemArmState.ARMED_AWAY assert alarm_system_0.seconds_remaining == 0 assert alarm_system_0.pin_configured assert alarm_system_0.arm_mode == AlarmSystemArmMode.ARMED_AWAY assert alarm_system_0.armed_away_entry_delay == 120 assert alarm_system_0.armed_away_exit_delay == 120 assert alarm_system_0.armed_away_trigger_duration == 120 assert alarm_system_0.armed_night_entry_delay == 120 assert alarm_system_0.armed_night_exit_delay == 120 assert alarm_system_0.armed_night_trigger_duration == 120 assert alarm_system_0.armed_stay_entry_delay == 120 assert alarm_system_0.armed_stay_exit_delay == 120 assert alarm_system_0.armed_stay_trigger_duration == 120 assert alarm_system_0.disarmed_entry_delay == 0 assert alarm_system_0.disarmed_exit_delay == 0 assert alarm_system_0.devices == { "ec:1b:bd:ff:fe:6f:c3:4d-01-0501": {"armmask": "none"}, "00:15:8d:00:02:af:95:f9-01-0101": { "armmask": "AN", "trigger": "state/vibration", }, } 0707010000007F000081A400000000000000000000000164453E390000121A000000000000000000000000000000000000001D00000000deconz-112/tests/test_api.py"""Test pydeCONZ api. pytest --cov-report term-missing --cov=pydeconz.api tests/test_api.py """ from unittest.mock import Mock import pytest from pydeconz.interfaces.api_handlers import ID_FILTER_ALL from pydeconz.interfaces.events import EventType async def test_api_items(mock_aioresponse, deconz_refresh_state): """Verify that groups works.""" session = await deconz_refresh_state( lights={"1": {"type": "light"}, "2": {"type": "light"}} ) grouped_apiitems = session.lights assert [*grouped_apiitems.items()] == [ ("1", grouped_apiitems["1"]), ("2", grouped_apiitems["2"]), ] assert [*grouped_apiitems.keys()] == ["1", "2"] assert [*grouped_apiitems.values()] == [ grouped_apiitems["1"], grouped_apiitems["2"], ] apiitems = session.lights.lights assert [*apiitems.items()] == [("1", apiitems["1"]), ("2", apiitems["2"])] assert [*apiitems.keys()] == ["1", "2"] assert [*apiitems.values()] == [apiitems["1"], apiitems["2"]] assert grouped_apiitems["1"] == apiitems["1"] assert apiitems.get("1") == apiitems["1"] assert apiitems.get("3", True) is True with pytest.raises(KeyError): grouped_apiitems["3"] # Subscribe without ID filter unsub_apiitems_all = apiitems.subscribe(apiitems_mock_subscribe_all := Mock()) unsub_apiitems_add = apiitems.subscribe( apiitems_mock_subscribe_add := Mock(), EventType.ADDED ) unsub_apiitems_update = apiitems.subscribe( apiitems_mock_subscribe_update := Mock(), EventType.CHANGED ) assert len(apiitems._subscribers) == 1 assert len(apiitems._subscribers[ID_FILTER_ALL]) == 3 # Subscribe with ID filter unsub_apiitems_1_all = apiitems.subscribe( apiitems_1_mock_subscribe_all := Mock(), id_filter="1" ) unsub_apiitems_1_add = apiitems.subscribe( apiitems_1_mock_subscribe_add := Mock(), EventType.ADDED, id_filter="1" ) unsub_apiitems_1_update = apiitems.subscribe( apiitems_1_mock_subscribe_update := Mock(), EventType.CHANGED, id_filter="1" ) assert len(apiitems._subscribers) == 2 assert len(apiitems._subscribers[ID_FILTER_ALL]) == 3 assert len(apiitems._subscribers["1"]) == 3 item_1 = apiitems["1"] item_1.register_callback(item_1_mock_callback := Mock()) unsub_item_1 = item_1.subscribe(item_1_mock_subscribe := Mock()) mock_aioresponse.get( "http://host:80/api/apikey/lights", payload={"1": {"key1": ""}, "3": {"type": "light"}}, ) await apiitems.update() # Item 1 is updated apiitems_mock_subscribe_all.assert_any_call(EventType.CHANGED, "1") apiitems_mock_subscribe_update.assert_called_with(EventType.CHANGED, "1") apiitems_1_mock_subscribe_all.assert_any_call(EventType.CHANGED, "1") apiitems_1_mock_subscribe_update.assert_called_with(EventType.CHANGED, "1") item_1_mock_callback.assert_called() item_1_mock_subscribe.assert_called() item_1.changed_keys == ("key1") # item 3 is created assert "3" in apiitems apiitems_mock_subscribe_all.assert_called_with(EventType.ADDED, "3") apiitems_mock_subscribe_add.assert_called_with(EventType.ADDED, "3") apiitems_1_mock_subscribe_add.assert_not_called() unsub_item_1() assert len(item_1._subscribers) == 0 unsub_apiitems_all() assert len(apiitems._subscribers[ID_FILTER_ALL]) == 2 unsub_apiitems_add() assert len(apiitems._subscribers[ID_FILTER_ALL]) == 1 unsub_apiitems_update() assert len(apiitems._subscribers[ID_FILTER_ALL]) == 0 unsub_apiitems_1_all() assert len(apiitems._subscribers["1"]) == 2 unsub_apiitems_1_add() assert len(apiitems._subscribers["1"]) == 1 unsub_apiitems_1_update() assert len(apiitems._subscribers["1"]) == 0 # Unsubscribe without ID in subscribers unsub_apiitems_4 = apiitems.subscribe(Mock(), id_filter="4") assert len(apiitems._subscribers["4"]) == 1 del apiitems._subscribers["4"] unsub_apiitems_4() async def test_unsupported_resource_type(deconz_refresh_state): """Verify that creation of APIItems works as expected.""" session = await deconz_refresh_state( alarm_systems={"1": {"type": "unknown_type"}}, groups={"1": {"type": "unknown_type", "scenes": []}}, lights={"1": {"type": "unknown_type"}}, sensors={"1": {"type": "unknown_type"}}, ) assert len(session.alarm_systems.keys()) == 1 assert len(session.groups.keys()) == 1 assert len(session.lights.keys()) == 1 # Legacy support assert len(session.sensors.keys()) == 0 07070100000080000081A400000000000000000000000164453E390000153D000000000000000000000000000000000000002000000000deconz-112/tests/test_config.py"""Test pydeCONZ config. pytest --cov-report term-missing --cov=pydeconz.config tests/test_config.py """ from pydeconz.config import ( ConfigDeviceName, ConfigNTP, ConfigTimeFormat, ConfigUpdateChannel, ConfigZigbeeChannel, ) async def test_create_config( mock_aioresponse, deconz_refresh_state, deconz_called_with ): """Verify that creating a config works.""" deconz_session = await deconz_refresh_state(config=FIXTURE_CONFIG) config = deconz_session.config assert config.api_version == "1.0.4" assert config.bridge_id == "0123456789AB" assert config.device_name == ConfigDeviceName.UNKNOWN assert config.dhcp assert config.firmware_version == "0x26490700" assert config.gateway == "192.168.0.1" assert config.ip_address == "192.168.0.90" assert config.link_button is False assert config.local_time == "2017-11-04T13:01:19" assert config.mac == "00:11:22:33:44:55" assert config.model_id == "deCONZ" assert config.name == "deCONZ-GW" assert config.network_mask == "255.255.255.0" assert config.network_open_duration == 60 assert not config.ntp assert config.pan_id == 50436 assert not config.portal_services assert config.rf_connected assert config.software_update == { "checkforupdate": False, "devicetypes": {"bridge": False, "lights": [], "sensors": []}, "notify": False, "text": "", "updatestate": 0, "url": "", } assert config.software_version == "2.4.82" assert config.time_format == ConfigTimeFormat.FORMAT_24H assert config.time_zone == "Europe/Stockholm" assert not config.utc assert config.uuid == "12345678-90AB-CDEF-1234-1234567890AB" assert config.websocket_notify_all assert config.websocket_port == 443 assert config.whitelist == { "1234567890": { "create date": "2017-11-02T23:13:13", "last use date": "2017-11-04T12:00:03", "name": "deCONZ WebApp", } } assert config.zigbee_channel == ConfigZigbeeChannel.CHANNEL_11 del config.raw["bridgeid"] assert config.bridge_id == "0000000000000000" config.raw["bridgeid"] = "00212EFFFF012345" assert config.bridge_id == "00212E012345" config.raw["ntp"] = "synced" assert config.ntp == ConfigNTP.SYNCED mock_aioresponse.put("http://host:80/api/apikey/config") await config.set_config( discovery=True, group_delay=1000, light_last_seen_interval=100, name="ABC", network_open_duration=10, otau_active=False, permit_join=111, rf_connected=True, time_format=ConfigTimeFormat.FORMAT_24H, time_zone="Europe/Stockholm", unlock=200, update_channel=ConfigUpdateChannel.BETA, utc="2017-11-04T12:01:19", zigbee_channel=ConfigZigbeeChannel.CHANNEL_15, websocket_notify_all=False, ) assert deconz_called_with( "put", path="/config", json={ "discovery": True, "groupdelay": 1000, "lightlastseeninterval": 100, "name": "ABC", "networkopenduration": 10, "otauactive": False, "permitjoin": 111, "rfconnected": True, "timeformat": "24h", "timezone": "Europe/Stockholm", "unlock": 200, "updatechannel": "beta", "utc": "2017-11-04T12:01:19", "zigbeechannel": 15, "websocketnotifyall": False, }, ) FIXTURE_CONFIG = { "UTC": "2017-11-04T12:01:19", "apiversion": "1.0.4", "backup": {"errorcode": 0, "status": "idle"}, "bridgeid": "0123456789AB", "datastoreversion": "60", "devicename": "", "dhcp": True, "disablePermitJoinAutoOff": False, "factorynew": False, "fwversion": "0x26490700", "gateway": "192.168.0.1", "internetservices": {"remoteaccess": "disconnected"}, "ipaddress": "192.168.0.90", "lightlastseeninterval": 60, "linkbutton": False, "localtime": "2017-11-04T13:01:19", "mac": "00:11:22:33:44:55", "modelid": "deCONZ", "name": "deCONZ-GW", "netmask": "255.255.255.0", "networkopenduration": 60, "panid": 50436, "portalconnection": "disconnected", "portalservices": False, "portalstate": { "communication": "disconnected", "incoming": False, "outgoing": False, "signedon": False, }, "proxyaddress": "none", "proxyport": 0, "replacesbridgeid": None, "rfconnected": True, "starterkitid": "", "swupdate": { "checkforupdate": False, "devicetypes": {"bridge": False, "lights": [], "sensors": []}, "notify": False, "text": "", "updatestate": 0, "url": "", }, "swversion": "2.4.82", "timeformat": "24h", "timezone": "Europe/Stockholm", "uuid": "12345678-90AB-CDEF-1234-1234567890AB", "websocketnotifyall": True, "websocketport": 443, "whitelist": { "1234567890": { "create date": "2017-11-02T23:13:13", "last use date": "2017-11-04T12:00:03", "name": "deCONZ WebApp", } }, "wifi": "not-installed", "wifiappw": "", "wifichannel": "1", "wifiip": "192.168.8.1", "wifiname": "Not set", "wifitype": "accesspoint", "zigbeechannel": 11, } 07070100000081000081A400000000000000000000000164453E390000129F000000000000000000000000000000000000002000000000deconz-112/tests/test_events.py"""Test pydeCONZ session class. pytest --cov-report term-missing --cov=pydeconz.interfaces.events tests/test_events.py """ from unittest.mock import Mock import pytest from pydeconz.interfaces.events import EventHandler from pydeconz.models import ResourceGroup from pydeconz.models.event import Event, EventType RAW_EVENT = { "id": "1", "r": ResourceGroup.LIGHT.value, "e": EventType.ADDED.value, } EVENT_HANDLER_DATA = [ (None, None, True), # No filters (EventType.ADDED, None, True), # Filter correct (EventType.CHANGED, None, False), # Filter incorrect ((EventType.ADDED, EventType.CHANGED), None, True), # Filter incorrect (None, ResourceGroup.LIGHT, True), # Filter correct (None, ResourceGroup.SENSOR, False), # Filter incorrect (None, (ResourceGroup.LIGHT, ResourceGroup.SENSOR), True), # Filter incorrect ] @pytest.mark.parametrize("event_filter, resource_filter, expected", EVENT_HANDLER_DATA) async def test_event_handler(event_filter, resource_filter, expected): """Verify event handler behaves according to configured filters.""" event_handler = EventHandler(gateway=Mock()) assert event_handler filters = {} if event_filter: filters["event_filter"] = event_filter if resource_filter: filters["resource_filter"] = resource_filter unsubscribe_callback = event_handler.subscribe(mock_callback := Mock(), **filters) assert len(event_handler._subscribers) == 1 assert unsubscribe_callback event_handler.handler(RAW_EVENT) assert mock_callback.called is expected unsubscribe_callback() assert len(event_handler._subscribers) == 0 EVENT_ADDED_DATA = [ (ResourceGroup.ALARM, EventType.ADDED, "alarmsystem"), (ResourceGroup.GROUP, EventType.ADDED, "group"), (ResourceGroup.LIGHT, EventType.ADDED, "light"), (ResourceGroup.SENSOR, EventType.ADDED, "sensor"), ] @pytest.mark.parametrize("resource, event_type, resource_key", EVENT_ADDED_DATA) async def test_event_added(resource, event_type, resource_key): """Verify added event content.""" data = { "id": "1", "r": resource.value, "e": event_type.value, resource_key: {"k": "v"}, } event = Event.from_dict(data) assert event.id == "1" assert event.group_id == "" assert event.scene_id == "" assert event.resource == resource assert event.type == event_type assert event.added_data == {"k": "v"} assert event.data == data assert event.changed_data == {} EVENT_CHANGED_DATA = [ (ResourceGroup.ALARM, EventType.CHANGED, {"state": {"k": "v"}}), (ResourceGroup.GROUP, EventType.CHANGED, {"name": "g"}), (ResourceGroup.LIGHT, EventType.CHANGED, {"state": {"k": "v"}}), (ResourceGroup.SENSOR, EventType.CHANGED, {"config": {"k": "v"}}), ] @pytest.mark.parametrize("resource, event_type, test_data", EVENT_CHANGED_DATA) async def test_event_changed(resource, event_type, test_data): """Verify changed event content.""" data = { "id": "1", "r": resource.value, "e": event_type.value, **test_data, } event = Event.from_dict(data) assert event.id == "1" assert event.group_id == "" assert event.scene_id == "" assert event.resource == resource assert event.type == event_type assert event.changed_data == test_data assert event.data == data assert event.added_data == {} EVENT_DELETED_DATA = [ (ResourceGroup.ALARM, EventType.DELETED), (ResourceGroup.GROUP, EventType.DELETED), (ResourceGroup.LIGHT, EventType.DELETED), (ResourceGroup.SENSOR, EventType.DELETED), ] @pytest.mark.parametrize("resource, event_type", EVENT_DELETED_DATA) async def test_event_deleted(resource, event_type): """Verify deleted event content.""" data = { "id": "1", "r": resource.value, "e": event_type.value, } event = Event.from_dict(data) assert event.id == "1" assert event.group_id == "" assert event.scene_id == "" assert event.resource == resource assert event.type == event_type assert event.data == data assert event.changed_data == {} assert event.added_data == {} async def test_event_scene_called(): """Verify scene_called event content.""" data = { "gid": "1", "scid": "2", "r": ResourceGroup.SCENE.value, "e": EventType.SCENE_CALLED.value, } event = Event.from_dict(data) assert event.id == "" assert event.group_id == "1" assert event.scene_id == "2" assert event.resource == ResourceGroup.SCENE assert event.type == EventType.SCENE_CALLED assert event.data == data assert event.changed_data == {} assert event.added_data == {} 07070100000082000081A400000000000000000000000164453E3900004275000000000000000000000000000000000000002100000000deconz-112/tests/test_gateway.py"""Test pydeCONZ session class. pytest --cov-report term-missing --cov=pydeconz.gateway tests/test_gateway.py """ from asyncio import gather from unittest.mock import AsyncMock, Mock, patch import aiohttp import pytest from pydeconz import ERRORS, BridgeBusy, RequestError, ResponseError, pydeconzException from pydeconz.models import ResourceGroup from pydeconz.models.alarm_system import AlarmSystemArmState from pydeconz.models.event import EventType from pydeconz.websocket import Signal, State @pytest.fixture def count_subscribers(deconz_session) -> int: """Count the amount of subscribers in all handlers.""" def calculate(): """Count subscribers.""" subscribers = 0 def calc(subscriber_filters) -> int: """Calculate subscriber per filter.""" count = 0 for filter in subscriber_filters.values(): count += len(filter) return count subscribers += calc(deconz_session.alarm_systems._subscribers) subscribers += calc(deconz_session.groups._subscribers) subscribers += calc(deconz_session.scenes._subscribers) for light in deconz_session.lights._handlers: subscribers += calc(light._subscribers) for sensor in deconz_session.sensors._handlers: subscribers += calc(sensor._subscribers) return subscribers yield calculate async def test_websocket_not_setup(deconz_session, mock_wsclient): """Test websocket method is not set up if websocket port is not provided.""" deconz_session.start() assert not deconz_session.websocket mock_wsclient.assert_not_called() async def test_websocket_setup(deconz_session, mock_wsclient): """Test websocket methods work.""" deconz_session.start(websocketport=443) mock_wsclient.assert_called() deconz_session.websocket.start.assert_called() deconz_session.close() deconz_session.websocket.stop.assert_called() async def test_websocket_config_provided_websocket_port( deconz_refresh_state, mock_wsclient ): """Test websocket methods work.""" session = await deconz_refresh_state(config={"websocketport": 8080}) session.start() mock_wsclient.assert_called() session.websocket.start.assert_called() session.close() session.websocket.stop.assert_called() async def test_initial_state(deconz_session, deconz_refresh_state, count_subscribers): """Test refresh_state creates devices as expected.""" assert count_subscribers() == 1 # Scene subscribed to groups unsub = deconz_session.subscribe(session_subscription := Mock()) assert count_subscribers() == 35 await deconz_refresh_state( alarm_systems={"0": {}}, config={"bridgeid": "012345"}, groups={ "g1": { "id": "gid", "scenes": [{"id": "sc1", "name": "scene1"}], "lights": [], } }, lights={"l1": {"type": "light"}}, sensors={"s1": {"type": "ZHAPresence"}}, ) assert session_subscription.call_count == 4 assert deconz_session.config.bridge_id == "012345" assert "0" in deconz_session.alarm_systems assert "g1" in deconz_session.groups assert "l1" in deconz_session.lights assert "g1_sc1" in deconz_session.scenes assert "s1" in deconz_session.sensors assert deconz_session.groups["g1"].id == "gid" assert deconz_session.groups["g1"].deconz_id == "/groups/g1" assert deconz_session.lights["l1"].deconz_id == "/lights/l1" assert deconz_session.scenes["g1_sc1"].deconz_id == "/groups/g1/scenes/sc1" assert deconz_session.sensors["s1"].deconz_id == "/sensors/s1" unsub() assert count_subscribers() == 1 async def test_get_api_key(mock_aioresponse, deconz_session): """Verify that get_api_key method can retrieve an api key.""" api_key = "0123456789abc36" mock_aioresponse.post( "http://host:80/api", payload=[{"success": {"username": api_key}}], ) assert await deconz_session.get_api_key() == api_key async def test_refresh_state(deconz_refresh_state): """Test refresh_state creates devices as expected.""" session = await deconz_refresh_state() assert session.config.bridge_id == "0000000000000000" assert len(session.alarm_systems.values()) == 0 assert len(session.groups.values()) == 0 assert len(session.lights.values()) == 0 assert len(session.sensors.values()) == 0 assert len(session.scenes.values()) == 0 await deconz_refresh_state( alarm_systems={"0": {}}, config={"bridgeid": "012345"}, groups={ "g1": { "id": "gid", "scenes": [{"id": "sc1", "name": "scene1"}], "lights": [], } }, lights={"l1": {"type": "light"}}, sensors={"s1": {"type": "ZHAPresence"}}, ) assert session.config.bridge_id == "012345" assert "0" in session.alarm_systems assert "g1" in session.groups assert "l1" in session.lights assert "g1_sc1" in session.scenes assert "s1" in session.sensors assert session.alarm_systems["0"].deconz_id == "/alarmsystems/0" assert session.groups["g1"].id == "gid" assert session.groups["g1"].deconz_id == "/groups/g1" assert session.lights["l1"].deconz_id == "/lights/l1" assert session.scenes["g1_sc1"].deconz_id == "/groups/g1/scenes/sc1" assert session.sensors["s1"].deconz_id == "/sensors/s1" async def test_request(mock_aioresponse, deconz_session): """Test request method and all its exceptions.""" mock_aioresponse.get( "http://host:80/api/apikey", content_type="application/json", payload={"result": "success"}, ) assert await deconz_session.request("get", "") == {"result": "success"} # Bad content type mock_aioresponse.get( "http://host:80/api/apikey/bad_content_type", content_type="http/text", ) with pytest.raises(ResponseError): await deconz_session.request("get", "/bad_content_type") # Client error with patch.object( deconz_session.session, "request", side_effect=aiohttp.client_exceptions.ClientError, ), pytest.raises(RequestError): await deconz_session.request("get", "/client_error") # Raise on error for error_code, error in ERRORS.items(): mock_aioresponse.get( f"http://host:80/api/apikey/{error_code}", content_type="application/json", payload={ "error": {"type": error_code, "address": "host", "description": ""} }, ) with pytest.raises(error): await deconz_session.request("get", f"/{error_code}") # Raise on error - Unknown error mock_aioresponse.get( "http://host:80/api/apikey/unknown", content_type="application/json", payload=[{"error": {"type": 0, "address": "host", "description": ""}}], ) with pytest.raises(pydeconzException): await deconz_session.request("get", "/unknown") # Generic exception with patch.object( deconz_session.session, "request", side_effect=Exception ), pytest.raises(Exception): await deconz_session.request("get", "") await deconz_session.session.close() async def test_session_handler_on_uninitialized_websocket(deconz_session): """Test session_handler is not called when self.websocket is None.""" # Event handler not called when self.websocket is None with patch.object( deconz_session.events, "handler", return_value=True ) as event_handler: await deconz_session.session_handler(signal=Signal.DATA) event_handler.assert_not_called() async def test_session_handler(deconz_session): """Test session_handler works.""" # Mock websocket deconz_session.websocket = Mock() # Event data with patch.object( deconz_session.events, "handler", return_value=True ) as event_handler: await deconz_session.session_handler(signal=Signal.DATA) event_handler.assert_called() @pytest.mark.parametrize( "state, value", [(State.RUNNING, True), (State.STOPPED, False)] ) async def test_session_handler_state_change( deconz_session, mock_websocket_state_change, state, value ): """Test session_handler works.""" await mock_websocket_state_change(state) deconz_session.connection_status_callback.assert_called_with(value) @pytest.mark.parametrize( "event", [ {"e": "added", "r": "scenes"}, {"e": "deleted", "r": "lights"}, {"e": "scene-called", "r": "scenes"}, ], ) async def test_unsupported_events(deconz_session, event): """Test event_handler handles unsupported events and resources.""" assert not deconz_session.events.handler(event) async def test_incomplete_event(deconz_session): """Test event_handler handles unsupported events and resources.""" with pytest.raises(KeyError): deconz_session.events.handler({"e": "deleted"}) async def test_alarmsystem_events(deconz_session, mock_websocket_event): """Test event_handler works.""" deconz_session.subscribe(session_subscription := Mock()) # Add alarmsystem await mock_websocket_event( event=EventType.ADDED, resource=ResourceGroup.ALARM, id="1", data={ "alarmsystem": { "name": "default", "config": { "armmode": "armed_away", "configured": True, "disarmed_entry_delay": 0, "disarmed_exit_delay": 0, "armed_away_entry_delay": 120, "armed_away_exit_delay": 120, "armed_away_trigger_duration": 120, "armed_stay_entry_delay": 120, "armed_stay_exit_delay": 120, "armed_stay_trigger_duration": 120, "armed_night_entry_delay": 120, "armed_night_exit_delay": 120, "armed_night_trigger_duration": 120, }, "state": { "armstate": "disarmed", "seconds_remaining": 0, }, "devices": {}, } }, ) assert "1" in deconz_session.alarm_systems assert deconz_session.alarm_systems["1"].arm_state == AlarmSystemArmState.DISARMED session_subscription.assert_called_once_with(EventType.ADDED, "1") # Update alarmsystem deconz_session.alarm_systems["1"].register_callback( mock_alarmsystem_callback := Mock() ) await mock_websocket_event( resource=ResourceGroup.ALARM, id="1", data={"state": {"armstate": "armed_away"}}, ) mock_alarmsystem_callback.assert_called() assert deconz_session.alarm_systems["1"].changed_keys == {"state", "armstate"} assert deconz_session.alarm_systems["1"].arm_state == AlarmSystemArmState.ARMED_AWAY async def test_light_events(deconz_session, mock_websocket_event): """Test event_handler works.""" deconz_session.subscribe(session_subscription := Mock()) # Add light await mock_websocket_event( event=EventType.ADDED, resource=ResourceGroup.LIGHT, id="1", unique_id="1", data={ "light": { "type": "On/Off light", "state": { "bri": 1, "reachable": True, }, }, }, ) assert "1" in deconz_session.lights assert deconz_session.lights["1"].brightness == 1 session_subscription.assert_called_once_with(EventType.ADDED, "1") # Update light deconz_session.lights["1"].register_callback(mock_light_callback := Mock()) await mock_websocket_event( resource=ResourceGroup.LIGHT, id="1", unique_id="1", data={"state": {"bri": 2}}, ) mock_light_callback.assert_called() assert deconz_session.lights["1"].changed_keys == {"state", "bri"} assert deconz_session.lights["1"].brightness == 2 async def test_group_events(deconz_session, deconz_refresh_state, mock_websocket_event): """Test event_handler works.""" deconz_session.subscribe(session_subscription := Mock()) await deconz_refresh_state( lights={ "2": { "type": "On/Off light", "state": { "bri": 1, "reachable": True, }, } } ) # Add group await mock_websocket_event( event=EventType.ADDED, resource=ResourceGroup.GROUP, id="1", data={ "group": { "action": {"bri": 1}, "lights": ["2"], "scenes": [], "state": {"any_on": False}, } }, ) assert "1" in deconz_session.groups assert deconz_session.groups["1"].brightness == 1 assert session_subscription.call_count == 2 session_subscription.assert_any_call(EventType.ADDED, "1") session_subscription.assert_any_call(EventType.ADDED, "2") # Update group deconz_session.groups["1"].register_callback(mock_group_callback := Mock()) await mock_websocket_event( resource=ResourceGroup.GROUP, id="1", data={"state": {"any_on": True}}, ) mock_group_callback.assert_called() assert deconz_session.groups["1"].changed_keys == {"state", "any_on"} assert deconz_session.groups["1"].any_on async def test_sensor_events(deconz_session, mock_websocket_event): """Test event_handler works.""" unsub_sensor_mock = deconz_session.sensors.subscribe(sensor_subscription := Mock()) deconz_session.subscribe(session_subscription := Mock()) # Add sensor await mock_websocket_event( event=EventType.ADDED, resource=ResourceGroup.SENSOR, id="1", unique_id="1", data={ "sensor": { "type": "ZHAPresence", "config": { "reachable": True, }, } }, ) assert "1" in deconz_session.sensors assert deconz_session.sensors["1"].reachable session_subscription.assert_called_once_with(EventType.ADDED, "1") sensor_subscription.assert_called_once() unsub_sensor_mock() # Update sensor deconz_session.sensors["1"].register_callback(mock_sensor_callback := Mock()) await mock_websocket_event( resource=ResourceGroup.SENSOR, id="1", unique_id="1", data={ "config": {"reachable": False}, }, ) mock_sensor_callback.assert_called() assert deconz_session.sensors["1"].changed_keys == {"config", "reachable"} session_subscription.assert_called_with(EventType.CHANGED, "1") assert not deconz_session.sensors["1"].reachable sensor_subscription.assert_called_once() @patch("pydeconz.gateway.sleep", new_callable=AsyncMock) async def test_retry_on_bridge_busy(_, deconz_refresh_state): """Verify a max count of 4 bridge busy messages.""" session = await deconz_refresh_state(lights={"1": {"type": "light"}}) request_mock = AsyncMock(side_effect=BridgeBusy) with pytest.raises(BridgeBusy), patch.object(session, "_request", new=request_mock): await session.request_with_retry("put", "field", {"key1": "on"}) assert request_mock.call_count == 3 assert not session._sleep_tasks @patch("pydeconz.gateway.sleep", new_callable=AsyncMock) async def test_request_exception_bridge_busy_pass_on_retry(_, deconz_refresh_state): """Verify retry can return an expected response.""" session = await deconz_refresh_state(lights={"1": {"type": "light"}}) request_mock = AsyncMock(side_effect=(BridgeBusy, {"response": "ok"})) with patch.object(session, "_request", new=request_mock): assert await session.request_with_retry("put", "field", {"key1": "on"}) == { "response": "ok" } assert request_mock.call_count == 2 assert not session._sleep_tasks @patch("pydeconz.gateway.sleep", new_callable=AsyncMock) async def test_reset_retry_with_a_second_request(_, deconz_refresh_state): """Verify an ongoing retry can be reset by a new request.""" session = await deconz_refresh_state(lights={"1": {"type": "light"}}) request_mock = AsyncMock(side_effect=(BridgeBusy, BridgeBusy, {"response": "ok"})) with patch.object(session, "_request", new=request_mock): collected_responses = await gather( session.request_with_retry("put", "field", {"key1": "on"}), session.request_with_retry("put", "field", {"key2": "on"}), ) assert request_mock.call_count == 3 assert not session._sleep_tasks assert collected_responses == [{}, {"response": "ok"}] 07070100000083000081A400000000000000000000000164453E390000235F000000000000000000000000000000000000002000000000deconz-112/tests/test_groups.py"""Test pydeCONZ groups. pytest --cov-report term-missing --cov=pydeconz.group tests/test_groups.py """ from unittest.mock import Mock import pytest from pydeconz.models.light.light import LightAlert, LightColorMode, LightEffect async def test_handler_group(mock_aioresponse, deconz_called_with, deconz_session): """Verify that groups works.""" groups = deconz_session.groups # Set attributes mock_aioresponse.put("http://host:80/api/apikey/groups/0") await groups.set_attributes( id="0", hidden=False, light_sequence=["1", "2"], lights=["3", "4"], multi_device_ids=["5", "6"], name="Group", ) assert deconz_called_with( "put", path="/groups/0", json={ "hidden": False, "lightsequence": ["1", "2"], "lights": ["3", "4"], "multideviceids": ["5", "6"], "name": "Group", }, ) mock_aioresponse.put("http://host:80/api/apikey/groups/0") await groups.set_attributes(id="0", name="Group") assert deconz_called_with( "put", path="/groups/0", json={"name": "Group"}, ) # Set state mock_aioresponse.put("http://host:80/api/apikey/groups/0/action") await groups.set_state( id="0", alert=LightAlert.SHORT, brightness=200, color_loop_speed=10, color_temperature=400, effect=LightEffect.COLOR_LOOP, hue=1000, on=True, on_time=100, saturation=150, toggle=True, transition_time=250, xy=(0.1, 0.1), ) assert deconz_called_with( "put", path="/groups/0/action", json={ "alert": "select", "bri": 200, "colorloopspeed": 10, "ct": 400, "effect": "colorloop", "hue": 1000, "on": True, "ontime": 100, "sat": 150, "toggle": True, "transitiontime": 250, "xy": (0.1, 0.1), }, ) mock_aioresponse.put("http://host:80/api/apikey/groups/0/action") await groups.set_state("0", on=True) assert deconz_called_with( "put", path="/groups/0/action", json={"on": True}, ) async def test_group(deconz_refresh_state): """Verify that groups works.""" deconz_session = await deconz_refresh_state( groups={ "0": { "action": { "bri": 132, "colormode": "hs", "ct": 0, "effect": "none", "hue": 0, "on": False, "sat": 127, "scene": None, "xy": [0, 0], }, "devicemembership": [], "etag": "e31c23b3bd9ece918f23ee17ef430304", "id": "11", "lights": ["14", "15", "12"], "name": "Hall", "scenes": [ { "id": "1", "name": "warmlight", "lightcount": 3, "transitiontime": 10, } ], "state": {"all_on": False, "any_on": True}, "type": "LightGroup", } } ) group = deconz_session.groups["0"] assert group.state is True assert group.all_on is False assert group.any_on is True assert group.device_membership == [] assert group.hidden is None assert group.id == "11" assert group.lights == ["14", "15", "12"] assert group.light_sequence is None assert group.multi_device_ids is None assert group.brightness == 132 assert group.hue == 0 assert group.saturation == 127 assert group.color_temp == 0 assert group.xy == (0, 0) assert group.color_mode == LightColorMode.HS assert group.effect == LightEffect.NONE assert group.reachable is True assert group.deconz_id == "/groups/0" assert group.etag == "e31c23b3bd9ece918f23ee17ef430304" assert group.manufacturer == "" assert group.model_id == "" assert group.name == "Hall" assert group.software_version == "" assert group.type == "LightGroup" assert group.unique_id == "" group.register_callback(mock_callback := Mock()) assert group._callbacks event = {"state": {"all_on": False, "any_on": False}} group.update(event) assert group.all_on is False assert group.any_on is False mock_callback.assert_called_once() assert group.changed_keys == {"state", "all_on", "any_on"} group.remove_callback(mock_callback) assert not group._callbacks group.raw["action"]["xy"] = (65555, 65555) assert group.xy == (1, 1) del group.raw["action"]["xy"] assert group.xy is None ENUM_PROPERTY_DATA = [ ( ("action", "colormode"), "color_mode", { "ct": LightColorMode.CT, "hs": LightColorMode.HS, "xy": LightColorMode.XY, None: LightColorMode.UNKNOWN, }, ), ( ("action", "effect"), "effect", { "colorloop": LightEffect.COLOR_LOOP, "none": LightEffect.NONE, None: LightEffect.UNKNOWN, }, ), ] @pytest.mark.parametrize("path, property, data", ENUM_PROPERTY_DATA) async def test_enum_group_properties(deconz_refresh_state, path, property, data): """Verify enum properties return expected values or None.""" deconz_session = await deconz_refresh_state( groups={"0": {"action": {}, "scenes": []}} ) group = deconz_session.groups["0"] assert getattr(group, property) is None for input, output in data.items(): data = {path[0]: input} if len(path) == 2: data = {path[0]: {path[1]: input}} group.update(data) assert getattr(group, property) == output @pytest.mark.parametrize( "light_state, expected_group_state", [ ( { "bri": 1, "ct": 1, "hue": 1, "sat": 1, "xy": (0.1, 0.1), "colormode": "xy", "reachable": True, }, { "brightness": 1, "ct": 1, "hue": 1, "sat": 1, "xy": (0.1, 0.1), "colormode": LightColorMode.XY, "effect": LightEffect.NONE, }, ), ( { "bri": 1, "ct": 1, "colormode": "ct", "reachable": True, }, { "brightness": 1, "ct": 1, "hue": None, "sat": None, "xy": None, "colormode": LightColorMode.CT, "effect": LightEffect.NONE, }, ), ( { "bri": 1, "reachable": True, }, { "brightness": 1, "ct": None, "hue": None, "sat": None, "xy": None, "colormode": LightColorMode.HS, "effect": LightEffect.NONE, }, ), ], ) async def test_update_color_state( light_state, expected_group_state, deconz_refresh_state, ): """Verify that groups works.""" deconz_session = await deconz_refresh_state( groups={ "0": { "action": { "bri": 132, "colormode": "hs", "ct": 0, "effect": "none", "hue": 0, "on": False, "sat": 127, "scene": None, "xy": [0, 0], }, "devicemembership": [], "etag": "e31c23b3bd9ece918f23ee17ef430304", "id": "11", "lights": ["14", "15", "12"], "name": "Hall", "scenes": [{"id": "1", "name": "warmlight"}], "state": {"all_on": False, "any_on": True}, "type": "LightGroup", } }, lights={"14": {"type": "Dimmable light", "state": light_state}}, ) group = deconz_session.groups["0"] light = deconz_session.lights["14"] group.update_color_state(light, update_all_attributes=True) assert group.brightness == expected_group_state["brightness"] assert group.color_temp == expected_group_state["ct"] assert group.hue == expected_group_state["hue"] assert group.saturation == expected_group_state["sat"] assert group.xy == expected_group_state["xy"] assert group.color_mode == expected_group_state["colormode"] assert group.effect == expected_group_state["effect"] 07070100000084000081A400000000000000000000000164453E3900000414000000000000000000000000000000000000002000000000deconz-112/tests/test_lights.py"""Test pydeCONZ lights. pytest --cov-report term-missing --cov=pydeconz.light tests/test_lights.py """ from tests import lights as light_test_data async def test_create_all_light_types(deconz_refresh_state): """Verify that light types work.""" deconz_session = await deconz_refresh_state( lights={ "0": light_test_data.test_configuration_tool.DATA, "1": light_test_data.test_cover.DATA, "2": light_test_data.test_light.DATA, "3": light_test_data.test_lock.DATA, "4": light_test_data.test_siren.DATA, "5": {"type": "unsupported device"}, }, ) lights = deconz_session.lights assert len(lights._handlers) == 6 assert lights["0"].type == "Configuration tool" assert lights["1"].type == "Window covering device" assert lights["2"].type == "Extended color light" assert lights["3"].type == "Door Lock" assert lights["4"].type == "Warning device" assert lights["5"].type == "unsupported device" # legacy support 07070100000085000081A400000000000000000000000164453E390000107A000000000000000000000000000000000000002000000000deconz-112/tests/test_scenes.py"""Test pydeCONZ scenes. pytest --cov-report term-missing --cov=pydeconz.scene tests/test_scenes.py """ async def test_handler_scene(mock_aioresponse, deconz_called_with, deconz_session): """Verify that groups works.""" scenes = deconz_session.scenes mock_aioresponse.post("http://host:80/api/apikey/groups/0/scenes") await scenes.create_scene(group_id="0", name="Garage") assert deconz_called_with("post", path="/groups/0/scenes", json={"name": "Garage"}) mock_aioresponse.put("http://host:80/api/apikey/groups/0/scenes/1/recall") await scenes.recall(group_id="0", scene_id="1") assert deconz_called_with( "put", path="/groups/0/scenes/1/recall", json={}, ) mock_aioresponse.put("http://host:80/api/apikey/groups/0/scenes/1/store") await scenes.store(group_id="0", scene_id="1") assert deconz_called_with( "put", path="/groups/0/scenes/1/store", json={}, ) mock_aioresponse.put("http://host:80/api/apikey/groups/0/scenes/1") await scenes.set_attributes(group_id="0", scene_id="1", name="new name") assert deconz_called_with( "put", path="/groups/0/scenes/1", json={"name": "new name"}, ) mock_aioresponse.put("http://host:80/api/apikey/groups/0/scenes/1") await scenes.set_attributes(group_id="0", scene_id="1") assert deconz_called_with( "put", path="/groups/0/scenes/1", json={}, ) async def test_scene(deconz_refresh_state): """Verify that groups works.""" deconz_session = await deconz_refresh_state( groups={ "0": { "action": { "bri": 132, "colormode": "hs", "ct": 0, "effect": "none", "hue": 0, "on": False, "sat": 127, "scene": None, "xy": [0, 0], }, "devicemembership": [], "etag": "e31c23b3bd9ece918f23ee17ef430304", "id": "11", "lights": ["14", "15", "12"], "name": "Hall", "scenes": [ { "id": "1", "name": "warmlight", "lightcount": 3, "transitiontime": 10, } ], "state": {"all_on": False, "any_on": True}, "type": "LightGroup", } } ) assert len(deconz_session.scenes.values()) == 1 scene = deconz_session.scenes["0_1"] assert scene.deconz_id == "/groups/0/scenes/1" assert scene.id == "1" assert scene.light_count == 3 assert scene.transition_time == 10 assert scene.name == "warmlight" await deconz_refresh_state( groups={ "0": { "action": { "bri": 132, "colormode": "hs", "ct": 0, "effect": "none", "hue": 0, "on": False, "sat": 127, "scene": None, "xy": [0, 0], }, "devicemembership": [], "etag": "e31c23b3bd9ece918f23ee17ef430304", "id": "11", "lights": ["14", "15", "12"], "name": "Hall", "scenes": [ { "id": "1", "name": "coldlight", "lightcount": 3, "transitiontime": 10, }, {"id": "2", "name": "New scene"}, ], "state": {"all_on": False, "any_on": True}, "type": "LightGroup", } } ) assert len(deconz_session.scenes.values()) == 2 # Update scene assert scene.name == "coldlight" # Add scene scene2 = deconz_session.scenes["0_2"] assert scene2.deconz_id == "/groups/0/scenes/2" assert scene2.id == "2" assert scene2.name == "New scene" 07070100000086000081A400000000000000000000000164453E3900000E32000000000000000000000000000000000000002100000000deconz-112/tests/test_sensors.py"""Test pydeCONZ sensors. pytest --cov-report term-missing --cov=pydeconz.sensor tests/test_sensors.py """ from pydeconz.models import ResourceType from tests import sensors as sensor_test_data async def test_create_all_sensors(deconz_refresh_state): """Verify that creating all sensors work.""" deconz_session = await deconz_refresh_state( sensors={ "0": sensor_test_data.test_air_purifier.DATA, "1": sensor_test_data.test_air_quality.DATA, "2": sensor_test_data.test_alarm.DATA, "3": sensor_test_data.test_ancillary_control.DATA, "4": sensor_test_data.test_battery.DATA, "5": sensor_test_data.test_carbon_monoxide.DATA, "6": sensor_test_data.test_consumption.DATA, "7": sensor_test_data.test_daylight.DATA, "8": sensor_test_data.test_door_lock.DATA, "9": sensor_test_data.test_fire.DATA, "10": sensor_test_data.test_generic_flag.DATA, "11": sensor_test_data.test_generic_status.DATA, "12": sensor_test_data.test_humidity.DATA, "13": sensor_test_data.test_light_level.DATA, "14": sensor_test_data.test_moisture.DATA, "15": sensor_test_data.test_open_close.DATA, "16": sensor_test_data.test_power.DATA, "17": sensor_test_data.test_presence.DATA, "18": sensor_test_data.test_pressure.DATA, "19": sensor_test_data.test_relative_rotary.DATA, "20": sensor_test_data.test_switch.DATA, "21": sensor_test_data.test_temperature.DATA, "22": sensor_test_data.test_thermostat.DATA, "23": sensor_test_data.test_time.DATA, "24": sensor_test_data.test_vibration.DATA, "25": sensor_test_data.test_water.DATA, }, ) sensors = deconz_session.sensors assert len(sensors._handlers) == 26 assert sensors["0"].type == ResourceType.ZHA_AIR_PURIFIER.value assert sensors["1"].type == ResourceType.ZHA_AIR_QUALITY.value assert sensors["2"].type == ResourceType.ZHA_ALARM.value assert sensors["3"].type == ResourceType.ZHA_ANCILLARY_CONTROL.value assert sensors["4"].type == ResourceType.ZHA_BATTERY.value assert sensors["5"].type == ResourceType.ZHA_CARBON_MONOXIDE.value assert sensors["6"].type == ResourceType.ZHA_CONSUMPTION.value assert sensors["7"].type == ResourceType.DAYLIGHT.value assert sensors["8"].type == ResourceType.ZHA_DOOR_LOCK.value assert sensors["9"].type == ResourceType.ZHA_FIRE.value assert sensors["10"].type == ResourceType.CLIP_GENERIC_FLAG.value assert sensors["11"].type == ResourceType.CLIP_GENERIC_STATUS.value assert sensors["12"].type == ResourceType.ZHA_HUMIDITY.value assert sensors["13"].type == ResourceType.ZHA_LIGHT_LEVEL.value assert sensors["14"].type == ResourceType.ZHA_MOISTURE.value assert sensors["15"].type == ResourceType.ZHA_OPEN_CLOSE.value assert sensors["16"].type == ResourceType.ZHA_POWER.value assert sensors["17"].type == ResourceType.ZHA_PRESENCE.value assert sensors["18"].type == ResourceType.ZHA_PRESSURE.value assert sensors["19"].type == ResourceType.ZHA_RELATIVE_ROTARY.value assert sensors["20"].type == ResourceType.ZHA_SWITCH.value assert sensors["21"].type == ResourceType.ZHA_TEMPERATURE.value assert sensors["22"].type == ResourceType.ZHA_THERMOSTAT.value assert sensors["23"].type == ResourceType.ZHA_TIME.value assert sensors["24"].type == ResourceType.ZHA_VIBRATION.value assert sensors["25"].type == ResourceType.ZHA_WATER.value 07070100000087000081A400000000000000000000000164453E3900001085000000000000000000000000000000000000001F00000000deconz-112/tests/test_utils.py"""Test pydeCONZ utilities. pytest --cov-report term-missing --cov=pydeconz.utils tests/test_utils.py """ from unittest.mock import AsyncMock, Mock, patch import aiohttp import pytest from pydeconz import errors, utils API_KEY = "1234567890" IP = "127.0.0.1" PORT = "80" async def test_delete_api_key() -> None: """Test a successful call of delete_api_key.""" session = Mock() with patch("pydeconz.utils.request", new=AsyncMock(return_value=True)): await utils.delete_api_key(session, IP, PORT, API_KEY) async def test_delete_all_keys() -> None: """Test a successful call of delete_all_keys. Delete all keys doesn't care what happens with delete_api_key. """ session = Mock() with patch( "pydeconz.utils.request", new=AsyncMock(return_value={"whitelist": {1: "123", 2: "456"}}), ): await utils.delete_all_keys(session, IP, PORT, API_KEY) async def test_get_bridge_id() -> None: """Test a successful call of get_bridgeid.""" session = Mock() with patch( "pydeconz.utils.request", new=AsyncMock(return_value={"bridgeid": "12345"}), ): response = await utils.get_bridge_id(session, IP, PORT, API_KEY) assert response == "12345" async def test_discovery() -> None: """Test a successful call to discovery.""" session = Mock() with patch( "pydeconz.utils.request", new=AsyncMock( return_value=[ { "id": "123456FFFFABCDEF", "internalipaddress": "host1", "internalport": "port1", "macaddress": "a:b:c", "name": "gateway", }, { "id": "234567BCDEFG", "internalipaddress": "host2", "internalport": "port2", }, ] ), ): response = await utils.discovery(session) assert [ { "id": "123456ABCDEF", "host": "host1", "port": "port1", "mac": "a:b:c", "name": "gateway", }, { "id": "234567BCDEFG", "host": "host2", "port": "port2", "mac": "", "name": "", }, ] == response async def test_discovery_response_empty() -> None: """Test an empty discovery returns an empty list.""" session = Mock() with patch("pydeconz.utils.request", new=AsyncMock(return_value={})): response = await utils.discovery(session) assert not response async def test_request() -> None: """Test a successful call of request.""" response = Mock() response.content_type = "application/json" response.json = AsyncMock(return_value={"json": "response"}) session = AsyncMock(return_value=response) result = await utils.request(session, "url") assert result == {"json": "response"} async def test_request_fails_client_error() -> None: """Test a successful call of request.""" session = AsyncMock(side_effect=aiohttp.ClientError) with pytest.raises(errors.RequestError) as e_info: await utils.request(session, "url") assert str(e_info.value) == "Error requesting data from url: " async def test_request_fails_invalid_content() -> None: """Test a successful call of request.""" response = Mock() response.content_type = "application/binary" session = AsyncMock(return_value=response) with pytest.raises(errors.ResponseError) as e_info: await utils.request(session, "url") assert str(e_info.value) == "Invalid content type: application/binary" async def test_request_fails_raise_error() -> None: """Test a successful call of request.""" response = Mock() response.content_type = "application/json" response.json = AsyncMock( return_value=[ {"error": {"type": 1, "address": "address", "description": "description"}} ] ) session = AsyncMock(return_value=response) with pytest.raises(errors.Unauthorized) as e_info: await utils.request(session, "url") assert str(e_info.value) == "1 address description" 07070100000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000B00000000TRAILER!!!710 blocks
Locations
Projects
Search
Status Monitor
Help
OpenBuildService.org
Documentation
API Documentation
Code of Conduct
Contact
Support
@OBShq
Terms
openSUSE Build Service is sponsored by
The Open Build Service is an
openSUSE project
.
Sign Up
Log In
Places
Places
All Projects
Status Monitor