From 59ca523a4606b85898d96016a6ccb10652df4814 Mon Sep 17 00:00:00 2001 From: javydekoning Date: Tue, 11 Nov 2025 20:43:17 +0100 Subject: [PATCH 1/2] feat: minor improvements and readme update --- .gitignore | 4 ++- README.md | 26 +++++++-------- docs/example_payloads/roomPayload.json | 0 main.py | 44 ++++++++++++++++++++++++++ run_tests.sh | 9 +++++- xcomfort/devices.py | 9 ++---- xcomfort/room.py | 3 ++ 7 files changed, 74 insertions(+), 21 deletions(-) create mode 100644 docs/example_payloads/roomPayload.json create mode 100644 main.py diff --git a/.gitignore b/.gitignore index 16a0ee1..0a6bb54 100644 --- a/.gitignore +++ b/.gitignore @@ -130,4 +130,6 @@ dmypy.json test.py -.github/linters/pyproject.toml \ No newline at end of file +.github/linters/pyproject.toml + +uv.lock \ No newline at end of file diff --git a/README.md b/README.md index 712849b..104f2d7 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,14 @@ uv pip install -e . ## Usage +Easy mode: + +```sh +uv run main.py --ip x.x.x.x --auth-key "ABCD1234WXYZ" +``` + +Roll your own; + ```python import asyncio from xcomfort import Bridge @@ -46,7 +54,6 @@ async def main(): await runTask asyncio.run(main()) - ``` ## Development @@ -56,24 +63,17 @@ asyncio.run(main()) You can run the tests using uvx without any local dependency management: ```bash -# Run tests with uvx (no local installation needed) -uvx --with aiohttp --with rx --with pycryptodome --with pytest-asyncio pytest tests/ -v - -# Or use the convenience script ./run_tests.sh - -# Or install dev dependencies and run tests locally -uv pip install -e ".[dev]" -pytest ``` +To run Github workflows locally + +Install [act](https://nektosact.com/installation/index.html) and run flows locally using `act`. + ### Dependencies The project includes the following dependencies: + - `aiohttp` - For async HTTP client functionality - `rx` - For reactive programming - `pycryptodome` - For cryptographic operations - -## To run Github workflows locally - -Install [act](https://nektosact.com/installation/index.html) and run flows locally using `act`. diff --git a/docs/example_payloads/roomPayload.json b/docs/example_payloads/roomPayload.json new file mode 100644 index 0000000..e69de29 diff --git a/main.py b/main.py new file mode 100644 index 0000000..1a061aa --- /dev/null +++ b/main.py @@ -0,0 +1,44 @@ +import argparse +import asyncio +import logging +from xcomfort import Bridge + +def observe_device(device): + device.state.subscribe(lambda state: print(f"Device state [{device.device_id}] '{device.name}': {state}")) + +async def main(ip: str, auth_key: str): + bridge = Bridge(ip, auth_key) + + runTask = asyncio.create_task(bridge.run()) + + devices = await bridge.get_devices() + + for device in devices.values(): + observe_device(device) + + # Wait 50 seconds. Try flipping the light switch manually while you wait + await asyncio.sleep(50) + + # Turn off all the lights. + # for device in devices.values(): + # await device.switch(False) + # + # await asyncio.sleep(5) + + await bridge.close() + await runTask + +if __name__ == "__main__": + # Configure debug logging + logging.basicConfig( + level=logging.DEBUG, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' + ) + + parser = argparse.ArgumentParser(description="Test xComfort Bridge connection") + parser.add_argument("--ip", required=True, help="IP address of the xComfort Bridge") + parser.add_argument("--auth-key", required=True, help="Authentication key for the xComfort Bridge") + + args = parser.parse_args() + + asyncio.run(main(args.ip, args.auth_key)) \ No newline at end of file diff --git a/run_tests.sh b/run_tests.sh index ce69fee..9fc29b5 100755 --- a/run_tests.sh +++ b/run_tests.sh @@ -1,3 +1,10 @@ #!/bin/bash -# Script to run tests with uvx without local dependency management +# Script to run tests and linting with uvx without local dependency management + +set -e # Exit on first error + +curl -L -o ./.github/linters/pyproject.toml https://raw.githubusercontent.com/home-assistant/core/refs/heads/dev/pyproject.toml + +uvx ruff check --config .github/linters/.ruff.toml xcomfort/ + uvx --with aiohttp --with rx --with pycryptodome --with pytest-asyncio pytest tests/ -v diff --git a/xcomfort/devices.py b/xcomfort/devices.py index 2ea2d6d..1fc1c83 100644 --- a/xcomfort/devices.py +++ b/xcomfort/devices.py @@ -199,6 +199,7 @@ def __init__(self, bridge, device_id, name, comp_id): def handle_state(self, payload): """Handle RcTouch state updates.""" + _LOGGER.debug("RcTouch %s: Received payload: %s", self.name, payload) temperature = None humidity = None if "info" in payload: @@ -350,12 +351,8 @@ def __init__(self, bridge, device_id, name, comp_id, payload): self.is_on = bool(payload["curstate"]) # Subscribe to component state updates if this is a multisensor - comp = bridge._comps.get(comp_id) # noqa: SLF001 - if comp is not None and comp.comp_type in ( - ComponentTypes.PUSH_BUTTON_MULTI_SENSOR_1_CHANNEL, - ComponentTypes.PUSH_BUTTON_MULTI_SENSOR_2_CHANNEL, - ComponentTypes.PUSH_BUTTON_MULTI_SENSOR_4_CHANNEL, - ): + if self.has_sensors: + comp = self.bridge._comps.get(self.comp_id) # noqa: SLF001 comp.state.subscribe(lambda _: self._on_component_update()) # Find and subscribe to companion sensor device self._find_and_subscribe_sensor_device() diff --git a/xcomfort/room.py b/xcomfort/room.py index f6415d6..312647a 100644 --- a/xcomfort/room.py +++ b/xcomfort/room.py @@ -78,6 +78,7 @@ def __init__(self, bridge, room_id, name: str): def handle_state(self, payload): """Handle room state updates.""" + _LOGGER.debug("Room %s: Received payload: %s", self.name, payload) old_state = self.state.value if old_state is not None: @@ -89,6 +90,7 @@ def handle_state(self, payload): humidity = payload.get("humidity", None) power = payload.get("power", 0.0) + mode = None if "currentMode" in payload: # When handling from _SET_ALL_DATA mode = RctMode(payload.get("currentMode", None)) if "mode" in payload: # When handling from _SET_STATE_INFO @@ -101,6 +103,7 @@ def handle_state(self, payload): self.modesetpoints[RctMode(mode_data["mode"])] = float(mode_data["value"]) _LOGGER.debug("Room %s: Loaded mode setpoints: %s", self.name, self.modesetpoints) + currentstate = None if "state" in payload: currentstate = RctState(payload.get("state", None)) From 77a65bf4879f205eee2e27c255880593bcb8f6bd Mon Sep 17 00:00:00 2001 From: javydekoning Date: Tue, 11 Nov 2025 20:46:31 +0100 Subject: [PATCH 2/2] feat: minor improvements and readme update --- main.py | 27 ++++++++++++++++++++------- run_tests.sh | 2 +- 2 files changed, 21 insertions(+), 8 deletions(-) diff --git a/main.py b/main.py index 1a061aa..f6fbf1a 100644 --- a/main.py +++ b/main.py @@ -1,12 +1,25 @@ +"""Example script demonstrating xComfort Bridge usage.""" + import argparse import asyncio import logging + from xcomfort import Bridge +_LOGGER = logging.getLogger(__name__) + + def observe_device(device): - device.state.subscribe(lambda state: print(f"Device state [{device.device_id}] '{device.name}': {state}")) + """Subscribe to device state changes and log them.""" + device.state.subscribe( + lambda state: _LOGGER.info( + "Device state [%s] '%s': %s", device.device_id, device.name, state + ) + ) + async def main(ip: str, auth_key: str): + """Run the main example demonstrating bridge connection and device observation.""" bridge = Bridge(ip, auth_key) runTask = asyncio.create_task(bridge.run()) @@ -15,9 +28,9 @@ async def main(ip: str, auth_key: str): for device in devices.values(): observe_device(device) - + # Wait 50 seconds. Try flipping the light switch manually while you wait - await asyncio.sleep(50) + await asyncio.sleep(50) # Turn off all the lights. # for device in devices.values(): @@ -34,11 +47,11 @@ async def main(ip: str, auth_key: str): level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) - + parser = argparse.ArgumentParser(description="Test xComfort Bridge connection") parser.add_argument("--ip", required=True, help="IP address of the xComfort Bridge") parser.add_argument("--auth-key", required=True, help="Authentication key for the xComfort Bridge") - + args = parser.parse_args() - - asyncio.run(main(args.ip, args.auth_key)) \ No newline at end of file + + asyncio.run(main(args.ip, args.auth_key)) diff --git a/run_tests.sh b/run_tests.sh index 9fc29b5..3267f3d 100755 --- a/run_tests.sh +++ b/run_tests.sh @@ -5,6 +5,6 @@ set -e # Exit on first error curl -L -o ./.github/linters/pyproject.toml https://raw.githubusercontent.com/home-assistant/core/refs/heads/dev/pyproject.toml +uvx ruff check --config .github/linters/.ruff.toml main.py uvx ruff check --config .github/linters/.ruff.toml xcomfort/ - uvx --with aiohttp --with rx --with pycryptodome --with pytest-asyncio pytest tests/ -v