Metadata-Version: 2.4
Name: usbairq
Version: 1.0.0
Summary: Asynchronous library to talk to air-Q devices over their USB serial port.
Author-email: Daniel Lehmann <daniel.lehmann@air-q.com>
License-Expression: Apache-2.0
Project-URL: Homepage, https://github.com/CorantGmbH/usbairq
Project-URL: Bug Tracker, https://github.com/CorantGmbH/usbairq/issues
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Hardware
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aioairq>=0.6.0
Requires-Dist: pyserial>=3.5
Provides-Extra: dev
Requires-Dist: ruff; extra == "dev"
Requires-Dist: pre-commit; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Dynamic: license-file

[![PyPI version](https://img.shields.io/pypi/v/usbairq)](https://pypi.org/project/usbairq/)
[![license](https://img.shields.io/github/license/CorantGmbH/usbairq)](https://github.com/CorantGmbH/usbairq/blob/main/LICENSE)
# PyPI package `usbairq`

Python library for asynchronous data access to local air-Q devices **over USB**,
without any network.

`usbairq` is [`aioairq`](https://github.com/CorantGmbH/aioairq) with a different
transport: `USBAirQ` inherits from `AirQ`, so the API, the encryption and the
returned data are identical — only the bytes travel over the device's serial port
instead of HTTP.

## Requirements

- An air-Q **with a USB port**. Ask the device: `GET /config` reports it as
  `"usb": true`. At the moment this is the air-Q Radon Science.
- A firmware version that serves the serial API.
- On the PC: no driver installation on current Windows, macOS and Linux.

## Usage

```python
import asyncio
from usbairq import USBAirQ

PASSWORD = "airqsetup"


async def main():
    ports = USBAirQ.discover()  # e.g. ["/dev/ttyUSB0"]

    async with await USBAirQ.connect(ports[0], PASSWORD) as airq:
        config = await airq.get_config()
        print(f"Available sensors: {config['sensors']}")

        data = await airq.get_latest_data()
        print(f"Averaged data: {data}")

        await airq.set_device_name("Living room")


asyncio.run(main())
```

`USBAirQ.discover()` lists all connected FTDI bridges — the serial equivalent of
mDNS. The chip is not exclusive to the air-Q, so treat the entries as candidates:
`connect()` verifies the password and raises `InvalidAuth` or `TimeoutError` if
the port belongs to something else.

Every method of [`aioairq.AirQ`](https://github.com/CorantGmbH/aioairq) is
available: `get_config`, `get_latest_data`, `get_log`, `fetch_device_info`,
`blink`, `get_night_mode` / `set_night_mode`, `get_led_theme` / `set_led_theme`,
brightness, `restart`, `shutdown`, …

### How long a call takes

Almost all of it is the device thinking, not the link, so these numbers hold for
the local HTTP API too:

| Call | Time |
| --- | --- |
| `get_latest_data()` | ~40 ms |
| `get_config()`, `fetch_device_info()` | ~260 ms — a large payload to serialise and encrypt |
| `set_device_name()` and other config writes | **~5.5 s** — the device persists to flash |

Config writes return `None`, so a script that awaits one looks stuck for those
five seconds with nothing to show. Print something before the call if that
matters. A device in its first minute after boot is slower across the board,
because sensor warm-up, Wi-Fi and the SD mount are all competing.

### Historical data

The device stores its measurements on the SD card as `year/month/day/timestamp`.
Walk the tree and download a day, exactly as over WLAN:

```python
import asyncio
from usbairq import USBAirQ

PASSWORD = "airqsetup"


async def main():
    async with await USBAirQ.connect(USBAirQ.discover()[0], PASSWORD) as airq:
        for year in sorted(await airq.get_historical_files_list()):
            for month in sorted(await airq.get_historical_files_list(year), key=int):
                for day in sorted(await airq.get_historical_files_list(f"{year}/{month}"), key=int):
                    for stamp in await airq.get_historical_files_list(f"{year}/{month}/{day}"):
                        path = f"{year}/{month}/{day}/{stamp}"
                        records = await airq.get_historical_file(path)
                        print(f"{path}: {len(records)} measurements")


asyncio.run(main())
```

[`examples/download_history.py`](examples/download_history.py) is the runnable
version of this, with a `--csv` option that writes every measurement to one file.

`get_historical_file()` prefers the device's compressed mirror and falls back to
the plain file when it is missing. `compressed=False` forces the plain route. If
the device password was changed after the data was written, pass `recrypt=True`:
the device then re-encrypts its cleartext mirror with the current password.

Measured on an air-Q Radon Science at 921600 baud, for one full 48 KB day file:

| Route | Bytes | Time |
| --- | --- | --- |
| `/file_zlib` (`compressed=True`, the default) | 8320 | **0.35 s** |
| `/file` (`compressed=False`) | 49573 | 1.25 s |
| `/file_recrypt` (`recrypt=True`) | 49573 | 2.19 s |

So the compressed mirror is worth roughly a factor of three and a half, and using
it is the default. Two caveats, both firmware-side: the file currently being written
has no mirror yet, and neither does the **last file of each day** — the device
compresses a file only when it hits its 48 KB limit and a new one is started, so a
day's final file stays uncompressed for good. Both simply take the `/file` fallback.

Throughput is about 38 kB/s on the plain route. From firmware 2.3.0 on that is
device-side work — SD read, per-line encryption, framing — not the link; older
firmware is capped near 10 kB/s by its scheduler tick regardless of the baud rate.
Fetching a whole day over USB now costs about what it costs over WLAN. A download
can still stall for a moment when it collides with the measurement loop writing to
the SD card, since both contend for the same card lock.

### Console speed

The console runs at **921600 baud** from firmware 2.3.0 on, which is the default
here. There is no command to change it: the rate is compiled into the firmware
(`MICROPY_HW_UART_REPL_BAUD`), because the port that carries the API also carries
the boot log, and a device that answers at an unexpected rate is a device nobody
can talk to. For older firmware pass `baudrate=115200` to `connect()`.

### Damaged lines

A serial link has no checksum, so a flipped byte is caught by what sits on top of
it: the device answers `400` when it cannot decrypt the request, `404` or `405`
when the path or method it received is not the one that was sent, and nothing at
all when the `##AQ1` prefix itself was hit. A damaged response breaks the JSON or
the AES padding.

Every request is therefore sent up to `RETRY_ATTEMPTS` times, a broken chunk
stream restarts the whole download, and only the last attempt's failure reaches
the caller. Errors the device means — a `500`, or the `404` from a missing
`/file_zlib` mirror — are passed straight on rather than retried.

### Not available over USB

- **`from_device_id()`** — USB devices are addressed by port, not by mDNS name.

## Protocol

Newline-delimited JSON frames with a magic prefix, in both directions:

```text
##AQ1 {"path": "/data", "method": "GET"}
##AQ1 {"id": "<DeviceID>", "status": 200, "content": "<base64(iv || aes)>"}
```

Lines without the `##AQ1 ` prefix — boot loader output, firmware logs — are
discarded by both sides. `content` is Base64 of a 16-byte IV followed by the
AES-256-CBC ciphertext, keyed with the device password padded to 32 bytes, i.e.
exactly what the local HTTP API returns. Error responses carry a plaintext
`error` and an HTTP-style status instead, surfaced as `usbairq.DeviceError`.

Requests are answered strictly one at a time; `usbairq` serialises concurrent
calls internally. Request frames are capped at 4096 bytes.

A response too large for one frame — the file routes — arrives as a numbered
chunk stream, closed by a frame carrying `eof`:

```text
##AQ1 {"path": "/file", "method": "GET", "request": "<base64(iv || aes)>"}
##AQ1 {"id": "<DeviceID>", "status": 200, "seq": 0, "chunk": "…", "eof": false}
##AQ1 {"id": "<DeviceID>", "status": 200, "seq": 1, "chunk": "",  "eof": true}
```

Chunks are not encrypted a second time: what the file routes serve is already
encrypted on the SD card. A gap in `seq` raises `InvalidAirQResponse`.

## Development

```sh
git clone https://github.com/CorantGmbH/usbairq
cd usbairq

python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

# The unit tests speak the protocol to a fake device and need no hardware
pytest

# With a device attached, the on-device tests are enabled by AIRQ_PORT
AIRQ_PORT=/dev/ttyUSB0 AIRQ_PASS=12345678 pytest
```

On Linux, access to `/dev/ttyUSB0` usually requires membership in the `dialout`
group (`sudo usermod -aG dialout $USER`, then log in again).

This repository uses [pre-commit](https://pre-commit.com/) for linting and
formatting with [Ruff](https://github.com/astral-sh/ruff). Install the git hooks
once with `pre-commit install`.
