Metadata-Version: 2.5
Name: aiocomexio
Version: 0.12.0
Summary: Asyncio-native Python client for the Comexio IO-Server
Project-URL: Homepage, https://github.com/kayl-codes/aiocomexio
Project-URL: Issues, https://github.com/kayl-codes/aiocomexio/issues
Author: kayl-codes
License-Expression: MIT
License-File: LICENSE
Keywords: asyncio,building-automation,comexio,home-automation
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Home Automation
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.13
Requires-Dist: aiohttp>=3.11
Requires-Dist: cryptography>=42.0.0
Provides-Extra: dev
Requires-Dist: bandit==1.9.4; extra == 'dev'
Requires-Dist: complexipy==8.1.0; extra == 'dev'
Requires-Dist: mypy==2.3.1; extra == 'dev'
Requires-Dist: pytest-asyncio==1.4.0; extra == 'dev'
Requires-Dist: pytest-cov==7.1.0; extra == 'dev'
Requires-Dist: pytest==9.1.1; extra == 'dev'
Requires-Dist: ruff==0.16.6; extra == 'dev'
Requires-Dist: syrupy==6.1.1; extra == 'dev'
Description-Content-Type: text/markdown

# aiocomexio

Asyncio-native Python client for the [Comexio IO-Server](https://www.comexio.com/), a local
building-automation controller.

No dependency on Home Assistant or any other framework — usable as a standalone library. It will
provide the communication layer of the
[homeassistant-comexio](https://github.com/kayl-codes/homeassistant-comexio) integration.

> **Status: beta.** The code was moved over from the integration step by step and is in use there. It covers:
> admin login, read access (including live values of physical IOs and system info — free
> storage, time server settings, uptime), value writes, the Web-IO
> lifecycle (including the command protocol, HTTP or HTTPS), marker / KNX object management and
> function plan editing — plans, elements, wires, block instance settings, scene blocks and
> Quantum modules with their configs, run / stop, 1-Wire sensors, Doorstation detection with a
> camera snapshot, the settings of an extension (its IP address)
> (`ComexioClient`), admin page scraping (`aiocomexio.scrape`), config parsing
> (`aiocomexio.config`, including Quickbus devices such as Multitouch panels and their IOs), KNX
> DPT tables (`aiocomexio.knx`), Web-IO command builders (`aiocomexio.webio`, with separate
> classes for Quickbus IOs and 1-Wire sensors), function plan rendering
> (with block settings in the hover text) and diffing (`aiocomexio.function_plan`) and the
> block-type reference catalog that resolves Comexio catalog ids per server by stable key
> (`aiocomexio.reference_catalog`). The API may still change.

## Installation

```bash
pip install aiocomexio
```

## Usage

The client works on an `aiohttp.ClientSession` you create and close yourself.
`session_kwargs()` gives it the settings Comexio needs (a cookie jar that keeps the admin
login for an IP-address host, a request timeout; `progress_log_interval=10` adds a "still
waiting" log line for slow requests). Give every client its own session: `login()` clears the
session's cookie jar. Several tasks can share one client while it logs in: `login()` waits for
the requests already in flight (at most 60 s), and requests that come up meanwhile wait until the
login is done, so renewing the session does not cut them off. Two exceptions: a request still
running after those 60 s may fail as not logged in, and requests that waited for a login that
failed or was cancelled may run without an admin session. Marker and KNX writes
(`set_marker_value`, `set_knx_value`) go through Comexio's `/api/` with a separate API user: pass
`api_username` / `api_password` to the client.
`set_io_value` sends only the relay and dimmer outputs (`Q1`, `Q2`, …) and the Dali groups
(`DG1`, …) to `/api/`, which refuses nearly every other IO. It writes the others (the 0-10 V
analog outputs `AO1`, …, and the IOs Comexio writes itself, such as a Multiroom-Server's
`Alarm1` or a Touchpanel's `Bright1`) over the admin session like Comexio's own IO page, so
they need a `login()` first
(`io_written_over_admin_session(identifier)` tells which IOs). Do not pass Quickbus IOs to
`set_io_value`: their identifiers repeat across devices, so they cannot be addressed by name.
Every call either returns data or raises a `ComexioError` subclass
(`ComexioAuthenticationError`, `ComexioConnectionError`, `ComexioResponseError`,
`ComexioDataError`, `ComexioRequestRejectedError` for a write the server refused,
`ComexioCreatedWithoutIdError` for a create the server confirmed but whose id could not be read
back — the object exists, do not create it again). One exception: the value setters raise a
plain `ValueError`, which is not a `ComexioError`, for a non-finite float (`nan`, `inf`) before
sending anything — catch it separately if your values can be non-finite.

An answer is read up to 16 MiB (the largest one measured on an installation with 30 function plans is
865 KiB); a successful answer that is longer, or that the device sends compressed although every request
asks for `Accept-Encoding: identity`, is refused with `ComexioDataError`, never cut. An error-status answer
sent compressed keeps its status (`ComexioResponseError`); its
`body` is then not the page but the note `('gzip'-encoded body not read)`, naming the coding. The one exception is
the credentials POST of `login()`: its answer is not read at all (the admin page that follows decides whether
the login worked), so neither its size nor its coding can fail the login. The request was
sent by then, so for a write a `ComexioDataError` does not prove it failed — as with `ComexioConnectionError`,
look at the object before repeating a create. An answer is decoded with its declared charset, else UTF-8: a
declared charset that is no text encoding (`hex`, `zlib`, ...) or that transforms the page instead of reading it
(`unicode_escape`, `raw_unicode_escape`, `punycode`, `idna`) counts as undeclared, and a
`fallback_charset_resolver` of an injected `aiohttp.ClientSession` is not consulted.

### HTTPS with a pinned certificate

Comexio serves HTTPS with a self-signed certificate, so the client pins its SHA-256 fingerprint
instead of verifying a CA. `fetch_certificate_fingerprint(host)` reads it from a bare TLS
handshake — no HTTP request, no credentials. Pass it as `fingerprint=` and every request goes over
HTTPS; a server presenting any other certificate raises `ComexioCertificateMismatchError` (not a
`ComexioConnectionError`: retrying cannot help, the certificate was replaced or something else
answers) before the request that met it is sent — if that request was a redirect hop, the hops
before it did reach the pinned server. Without `fingerprint` the client uses plain HTTP and
follows redirects only on the Comexio server: on the same scheme, host and port, or on to HTTPS on
port 443 of the same host. A redirect anywhere else (another host, or another port of the same
host) raises `ComexioResponseError` before anything is sent there. A redirect on to HTTPS checks
the certificate against the system's CAs, so Comexio's self-signed one fails there with
`ComexioConnectionError` — pin it instead.

Fetching the fingerprint is trust on first use: whatever answers that one handshake gets pinned.
So fetch it **once, at setup**, store it, and pass the stored value on every later start —
fetching it anew on each start would accept whatever certificate answers. If you want to rule out
an impostor at setup too, compare `format_fingerprint()` with the SHA-256 value Comexio's admin
page lists under its certificate hashes. A pinned client follows redirects only within its HTTPS
origin (same scheme, host and port). For a host name that resolves to several addresses, aiohttp
reports only the last failure, so a mismatch there can arrive as `ComexioConnectionError`; compare
a fresh `fetch_certificate_fingerprint()` with the pin to tell the two apart.

```python
import asyncio

import aiohttp

from aiocomexio import ComexioClient, fetch_certificate_fingerprint, format_fingerprint, session_kwargs
from aiocomexio.config import parse_config

HOST = "192.168.0.20"


async def setup() -> str:
    """Run once: read the fingerprint, check it, store it with the rest of your configuration."""
    fingerprint = await fetch_certificate_fingerprint(HOST)
    print("Comexio certificate:", format_fingerprint(fingerprint))
    return fingerprint


async def main(stored_fingerprint: str) -> None:
    """Every later start: use the stored fingerprint, never a freshly fetched one."""
    async with aiohttp.ClientSession(**session_kwargs()) as session:
        client = ComexioClient(HOST, "admin", "secret", session=session, fingerprint=stored_fingerprint)
        await client.login()
        raw = await client.get_raw_config()
        config = parse_config(raw.variables, io_types=raw.io_types, io_input_types=raw.io_input_types)
        print(len(config["markers"]), "markers on Comexio", raw.comexio_version)


asyncio.run(main(stored_fingerprint="<the value setup() returned>"))
```

### 1-Wire sensors

The three 1-Wire calls all need the admin session (`login()` first):

- `get_onewire_device_count()` counts the 1-Wire devices on the IO home page. `0` means there
  are none, so the status poll can be skipped. The count and the status request are separate
  requests with no shared snapshot: a sensor added between them can already appear in the current
  status, and a sensor added after both is seen by the next poll.
- `get_onewire_sensors()` returns the active configured sensors keyed by address (`OT1`, `OH3`, …);
  inactive entries are left out. An active entry without a usable `Prefix` or `Id` is skipped (and
  logged); if active entries exist but none is usable, the call raises `ComexioDataError`. Two
  active entries with the same address collapse to one: the later one wins (logged). Two entries
  with the same `Id` but different prefixes both stay, and a warning is logged, because a function
  plan names a sensor by its `Id` alone. It costs one config fetch.
- `get_onewire_status()` returns the live state of the sensors Comexio reports, keyed by address.
  It is read from Comexio's 1-Wire monitor, which is not an official interface, so a changed answer
  raises `ComexioDataError`. An item without an address is skipped (and logged), so a sensor can be
  missing from the map; if the answer has items but none has an address, the call raises
  `ComexioDataError`. Two items with the same address collapse to one, the later one wins (logged).
  Compare the keys with `get_onewire_sensors()` if you need every sensor.

Each `OneWireStatus` has `visible` (False: the sensor was not found), `online` (False: read error,
for example a CRC error) and `value` (the reading as a float). `value` is `None` if there is none,
if it is not finite, or if it is not a number at all (that case is logged). `ok` is True only when
the sensor is visible and online. A read error can still carry a value, so check `ok` before
trusting `value`.

```python
import asyncio

import aiohttp

from aiocomexio import ComexioClient, session_kwargs


async def report_1wire_problems(client: ComexioClient) -> None:
    """Print the sensors whose status says not found or read error (not the ones missing from it)."""
    if not await client.get_onewire_device_count():
        return
    for address, status in (await client.get_onewire_status()).items():
        if not status.ok:
            print(address, "read error" if status.visible else "not found")


async def main(stored_fingerprint: str) -> None:
    async with aiohttp.ClientSession(**session_kwargs()) as session:
        client = ComexioClient("192.168.0.20", "admin", "secret", session=session, fingerprint=stored_fingerprint)
        await client.login()
        await report_1wire_problems(client)


asyncio.run(main(stored_fingerprint="<the value setup() returned>"))
```

The status is for spotting broken sensors. To get the values pushed, build Web-IO commands for the
sensors. `config.parse_config` does not read them, so add them under the key `"onewire"` of the
parsed config. Take them from the same config fetch, with
`onewire.parse_onewire_config(raw.variables["FubModules"])`: `get_onewire_sensors()` fetches a config
of its own, which can differ if the 1-Wire setup changes in between. Then `webio.build_webio_commands` (or
`generate_webio_json` with `WebioClass.ONEWIRE`, class `<webio name> [1Wire]`) builds one command per
sensor:

- The command is named `HA 1Wire <address>`.
- It pushes `{"id": "<address>", "value": …, "type": "onewire"}`.
- An analog sensor gets the generic analog range and a digital one gets 0/1.
- A sensor of unknown kind gets no command.

In a function plan, a sensor is element type 12 with its `config_id` as `ref_id`. Wire it to its
command like a marker. To show the sensors by name in `function_plan.render_plan_svg`, pass
`onewire_by_id`: the sensors keyed by `str(config_id)`, each `{"name": …, "address": …, "type":
sensor.kind, "value": …}`. Without it a sensor renders as `1-Wire ref=<config_id>`. A plan
cannot tell apart sensors that share an `Id` under different prefixes (`parse_onewire_config` logs
them): leave such an Id out of the map, and its elements render as `1-Wire ref=<config_id>`
instead of showing one of the sensors.
`render_flow_svg` takes the same map (and `knx_by_id`) for its box labels. A value that is not
finite (`nan`, `inf`) shows as no value.

### Doorstation

A Comexio Doorstation is an extension of the IO-Server with a camera. Its IOs (`PIR1`, the bell
buttons `DI1`–`DI3`, `NameLED1`, …) come through `config.parse_config` like those of any other
extension. The camera calls need the admin session (`login()` first):

- `get_doorstations()` reads the IO home page and returns the Doorstations keyed by extension id.
  Each `Doorstation` has `ext_id`, `camera_id` and `name` (the extension name, `None` if the page
  carries none). An empty map means the page lists extensions but no Doorstation; a page with neither
  the device counters of its extension blocks (the ones `get_onewire_device_count()` reads) nor an
  element with a non-empty `data-camera_id` raises `ComexioDataError`. Elements whose ids are not positive integers are skipped (and logged); if there are
  such elements but none is usable, the call raises `ComexioDataError`. Detection follows the IO home page
  of Comexio OS 11.0.2: the elements of a Doorstation tab carry `data-extid`, `data-extname` and
  `data-camera_id`.
- `get_extension(ext_id)` reads the edit form of one extension (`/admin/extension/edit/?id=<ext_id>&json_return=true`,
  which answers with the HTML of the form, not JSON) and returns an `ExtensionInfo`: `ext_id`, `type_id`
  (`DOORSTATION_TYPE_ID`, 16, for a Doorstation; `is_doorstation` tells), `name` and `serial` (`None` if empty; an
  extension learned offline has no serial), `location` (the raw "IP-Adresse (optional)" field, `None` if the input has
  no value) and `ip`: that field as a usable IPv4 address, `None` for `local`, a host name, IPv6 or anything else
  (`location` tells them apart). A Doorstation with an address of its own is the device that serves the camera
  stream (the exact stream URL is not read from the form). The login rows of the form (user name, password) are
  never read. An id Comexio does not know (it answers with the form for a new
  extension) or a form without the address field raises `ComexioDataError`, the login form
  `ComexioAuthenticationError`, an `ext_id` that is no positive int `TypeError` / `ValueError`.
- `doorstation_stream_url(camera_id)` is the URL of the camera's MJPEG stream on the IO-Server
  (`/proxy/cam/<camera_id>/`, `multipart/x-mixed-replace`, one JPEG per part).
- `get_doorstation_snapshot(camera_id)` returns the current image: the first complete JPEG of that
  stream, read with the admin session. A stream without a frame within 10 seconds, or one that breaks
  off, raises `ComexioConnectionError`; one that ends or passes 4 MB without a frame (a frame must end
  within the first 4 MB; bytes past it are not kept), a malformed frame (also one with an empty scan
  or one that ends before its first scan), or a stream sent compressed (it is asked for unencoded and
  never decompressed, so the 4 MB cap holds) raises `ComexioDataError`; the login form raises
  `ComexioAuthenticationError`.

The Doorstation serves the same stream under its own address, reportedly without a login. If you
use it, keep the Doorstation in a network segment that only trusted hosts can reach.

## Development

```bash
pip install -e ".[dev]"
pre-commit install

ruff check .            # lint
ruff format --check .   # formatting check; drop --check to auto-format
mypy                    # strict type check (src/aiocomexio)
pytest                  # tests
```

## License

[MIT](LICENSE)
