diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 21404b2..6c65589 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -70,24 +70,6 @@ jobs: - run: ruff format --check . - run: python -m pytest -q - homeassistant: - name: Home Assistant integration lint + test - runs-on: ubuntu-latest - defaults: - run: - working-directory: python/examples/homeassistant - steps: - - uses: actions/checkout@v4 - - uses: actions/setup-python@v5 - with: - python-version: "3.14" - - uses: astral-sh/setup-uv@v5 - - run: uv sync - - run: uv run ruff check . - - run: uv run ruff format --check . - - run: uv run pytest -q - - run: uv run python build.py - swift: name: Swift lint + test runs-on: macos-15 diff --git a/.github/workflows/publish-python.yml b/.github/workflows/publish-python.yml new file mode 100644 index 0000000..28efcd3 --- /dev/null +++ b/.github/workflows/publish-python.yml @@ -0,0 +1,56 @@ +name: Publish libkp to PyPI + +# Tag-driven: `git tag python-v0.1.0 && git push --tags` builds python/ and +# publishes it. Authentication is PyPI Trusted Publishing over OIDC — the +# `id-token: write` permission below is the whole credential, so there is no +# API token in this repository. +on: + push: + tags: ["python-v*"] + workflow_dispatch: + +permissions: {} + +jobs: + build: + name: Build the distributions + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.14" + + - name: The tag and the package agree on the version + if: startsWith(github.ref, 'refs/tags/') + run: | + set -euo pipefail + tag="${GITHUB_REF_NAME#python-v}" + pkg=$(grep -m1 '^version = ' python/pyproject.toml | cut -d'"' -f2) + init=$(grep -m1 '^__version__ = ' python/src/libkp/__init__.py | cut -d'"' -f2) + echo "tag=$tag pyproject=$pkg __version__=$init" + test "$tag" = "$pkg" + test "$tag" = "$init" + + - run: pip install build + - run: python -m build python/ + - uses: actions/upload-artifact@v4 + with: + name: dist + path: python/dist/ + + publish: + name: Publish to PyPI + needs: build + runs-on: ubuntu-latest + environment: + name: pypi + url: https://pypi.org/p/libkp + permissions: + id-token: write + steps: + - uses: actions/download-artifact@v4 + with: + name: dist + path: dist/ + - uses: pypa/gh-action-pypi-publish@release/v1 diff --git a/README.md b/README.md index 51f5a19..7c17f87 100644 --- a/README.md +++ b/README.md @@ -29,9 +29,10 @@ with my hands. | Python | `libkp` | Python 3.11+, standard library only | [`python/`](python/) | | Swift | `LibKP` | Swift 6, macOS 13+, `Network` framework | [`swift/`](swift/) | -None of the three is on a package registry yet; depend on the directory as a -path or git dependency. The shape of the API is the same everywhere — find a -device, connect a model, subscribe to its state, send it things: +Python is on PyPI — `pip install libkp`. Rust and Swift are not on a registry +yet; depend on the directory as a path or git dependency. The shape of the API +is the same everywhere — find a device, connect a model, subscribe to its +state, send it things: ```rust use libkp::model::DeviceModel; @@ -69,6 +70,14 @@ cd swift && swift run meters # or `swift run MetersApp` fo Discovery needs UDP port 5727 to itself, so quit Kemper's Rig Manager (or pass `--ip`) before running it. +### Built on libkp + +- [**kemper-homeassistant**](https://github.com/gotwalt/kemper-homeassistant) — + a Home Assistant integration: the rig, the amp, the cabinet, and whether + anyone is playing, held on one session that never polls the device. It + depends on `libkp` from PyPI, and tests against + [`libkp.testing.FakeDevice`](python/README.md#testing-against-a-fake-profiler). + ## Status Tested against a **Profiler Player on firmware 14.2.1**. Other Profiler models diff --git a/assets/kemper.png b/assets/kemper.png new file mode 100644 index 0000000..1509ffa Binary files /dev/null and b/assets/kemper.png differ diff --git a/python/README.md b/python/README.md index a25d2a8..c7456f6 100644 --- a/python/README.md +++ b/python/README.md @@ -14,7 +14,8 @@ the CC control vocabulary, and an observable async device model. ## Install ```sh -pip install -e '.[dev]' # from python/ +pip install libkp # from PyPI +pip install -e '.[dev]' # or from python/, to work on it ``` Or run straight from the source tree: @@ -153,6 +154,7 @@ with DiscoveryPort.acquire() as port: # raises PortUnavailableError if taken | `libkp.state` | The state tree and the pure `DeviceState.apply_update` fold. | | `libkp.model` | `DeviceModel`, the async store over the stream and the control link. | | `libkp.errors` | The exception family, all deriving from `LibKPError`. | +| `libkp.testing` | `FakeDevice`, an in-process Profiler to test against. | | `libkp._generated` | **Generated, data only** — constants and lookup tables. Do not edit. | `_generated.py` is emitted from [`../spec`](../spec) by @@ -371,6 +373,30 @@ for message in unframer.push(raw_stream_bytes): ... ``` +## Testing against a fake Profiler + +`libkp.testing.FakeDevice` is a Profiler stand-in that speaks the real +transport in-process: the greeting, the protocol-selection handshake, the +preamble, then MIDI3 framing or the CBOR dump. Anything built on libkp can hold +a session against it in its own suite, with nothing below the socket mocked and +no device on the desk. + +```python +from libkp import DeviceModel +from libkp.testing import FakeDevice, answer_requests + +fake = await FakeDevice(responder=answer_requests).start() +model = await DeviceModel.connect("127.0.0.1", port=fake.port) +... +await model.close() +await fake.stop() +``` + +It can also hang up mid-session, hold back the greeting, or refuse connections +for a while — the states a reconnect has to survive. libkp's own async tests +drive it; so does the [Home Assistant +integration](https://github.com/gotwalt/kemper-homeassistant). + ## Tests ```sh @@ -388,7 +414,7 @@ The suite covers: checked for message count, pending bytes, exact messages, decoded status frames, the per-function histogram, and the resulting rig/amp/cab names. - **Unit tests** for each module, and async tests that drive `Session` and - `DeviceModel` against an in-process stand-in device (`tests/fake_device.py`). + `DeviceModel` against an in-process stand-in device (`libkp.testing`, shipped with the package). ## Provenance diff --git a/python/examples/homeassistant/README.md b/python/examples/homeassistant/README.md deleted file mode 100644 index 42a339e..0000000 --- a/python/examples/homeassistant/README.md +++ /dev/null @@ -1,119 +0,0 @@ -# Kemper Profiler — Home Assistant integration - -A custom integration that puts a Kemper Profiler on the local network into -Home Assistant: what rig is loaded, and whether anyone is playing through it. - -It is an **example** of using `libkp` in an application, and it is a real -integration: it holds one MIDI3 session for as long as Home Assistant runs, -takes everything it shows from what the device pushes unrequested, and never -polls the device or reconnects in a loop. - -A Profiler is identified by the **serial number** it advertises, not by its -address. Every setup broadcasts once to ask where that serial is now, so a -device whose DHCP lease moves it to another address is followed automatically: -the entry, the device and all five entities stay exactly as they were, history -included. Discovery finding nothing — the port held by Rig Manager, a quiet -network — is not an error; the last known address is used as it stands. - -``` -Kemper Profiler -├─ sensor._rig Rig name "Crunchy Vox" -├─ sensor._amp Amp name "Vintage Twin" -├─ sensor._cabinet Cabinet name "2x12 Alnico" -├─ binary_sensor._active Playing? on / off -└─ sensor._last_activity Last activity timestamp -``` - -## The entities - -| Entity | What it is | -|---|---| -| `sensor` Rig / Amp / Cabinet | The names the device pushes on a rig change. They follow the front panel, a MIDI controller, Rig Manager — anything that loads a rig. | -| `binary_sensor` Active | On while signal is passing through the rig. See below. | -| `sensor` Last activity | When signal was last heard. While *Active* is on it is when the current session began; when *Active* goes off it is the moment of the last note. | - -Everything else the device says — the effect slots, the tempo, the volumes, -the tuner, the bank preview, both channels' states — is in the integration's -**diagnostics** download rather than in entities. Adding an entity for any of -it is one row in the table in `sensor.py`. - -### Activity detection - -The Profiler pushes a meter frame about twenty times a second. Writing an -entity per frame would put 72,000 states an hour into the recorder to say -"someone is playing", so the meter lane is read by one plain callback that -compares a single 14-bit integer per frame and writes Home Assistant state -only when the answer changes: **two state writes per playing session**, -however long the session runs. - -The level it reads is the **rig output** meter — after the rig's own volume, -before the master/monitor/headphone volumes — so a rig turned down reads -quiet, but practising with the monitors off still reads as playing. - -Two options (Settings → Devices & services → Kemper Profiler → Configure): - -- **Quiet window** — how long the output must stay below the threshold before - *Active* turns off. Default 5 minutes. -- **Level threshold** — how loud counts as playing, as a percentage of full - scale. Default 2%. - -Saving them retunes the running detector; it does **not** reconnect to the -device. - -### Losing the connection - -When the stream ends — the amp switched off, the network dropped — the -integration does not redial the address it was using, because that address is -the part that can change. It reloads the config entry instead, which starts -again at discovery: find the serial, follow it to wherever it is now, connect -once. A Profiler that is simply off fails that setup and Home Assistant retries -on its own widening schedule; a session that ends within a minute of opening -waits half a minute before reloading, so nothing can spin. - -## Install - -Build the bundle and copy it into your Home Assistant configuration -directory, next to `configuration.yaml`: - -```sh -uv run python build.py # dist/custom_components/kemper + a zip -uv run python build.py --install ~/homeassistant -``` - -For a Home Assistant OS or supervised install, take -`dist/kemper-.zip` and unpack it into the configuration directory -with the **Samba share**, **Terminal & SSH**, or **File editor** add-on — its -paths are already `custom_components/kemper/…`. - -Then: - -1. Restart Home Assistant. -2. **Settings → Devices & services → Add integration → "Kemper Profiler"**. -3. The flow broadcasts for Profilers on the LAN and lists what answers. If - nothing answers — Rig Manager holds the discovery port exclusively, and so - does a running `meters` example — choose *Enter a host manually* and give - the Profiler's IP address. - -The bundle vendors `libkp` itself, so the integration has no `pip` -requirements and works on an install with no internet access. - -## Development - -The integration is developed against the library beside it: -`custom_components/kemper/libkp` is a relative symlink to `python/src/libkp`, -and the integration imports it as `from .libkp import …`. The same code -therefore runs against the working tree here and against the vendored copy in -a bundle — `build.py` dereferences the symlink and copies the library in. - -```sh -uv sync # a Python 3.14 environment with Home Assistant 2026.8 -uv run ruff check . -uv run ruff format --check . -uv run pytest -q -uv run python build.py -``` - -The tests are end-to-end over a real loopback socket: they drive libkp's own -`FakeDevice` (`python/tests/fake_device.py`), push the same bytes a Profiler -pushes, and assert on entity states. Nothing below the config entry is mocked -except the discovery broadcast. diff --git a/python/examples/homeassistant/build.py b/python/examples/homeassistant/build.py deleted file mode 100644 index b36cb28..0000000 --- a/python/examples/homeassistant/build.py +++ /dev/null @@ -1,96 +0,0 @@ -#!/usr/bin/env python3 -"""Bundle the integration into something a Home Assistant config directory takes. - -In the source tree ``custom_components/kemper/libkp`` is a symlink to the -library beside it, so the integration runs against the working copy with no -install step and no copy to keep in sync. Home Assistant would not tolerate -that symlink in a config directory, so the bundle **dereferences** it: the -library is copied in as an ordinary package, and the shipped integration -depends on nothing but the standard library. - -Usage:: - - python build.py # dist/custom_components/kemper + the zip - python build.py --install ~/homeassistant # copy into a config directory -""" - -from __future__ import annotations - -import argparse -import json -import shutil -import zipfile -from pathlib import Path - -HERE = Path(__file__).resolve().parent -SOURCE = HERE / "custom_components" / "kemper" -DIST = HERE / "dist" -DOMAIN = "kemper" - -#: Never shipped: bytecode caches, editor droppings, and the integration's own -#: tests, which import Home Assistant's test harness. -IGNORE = shutil.ignore_patterns("__pycache__", "*.py[co]", ".DS_Store", "tests") - - -def version() -> str: - """The version in the manifest — the one Home Assistant shows.""" - return json.loads((SOURCE / "manifest.json").read_text(encoding="utf-8"))["version"] - - -def build() -> Path: - """Write ``dist/custom_components/kemper`` and the zip beside it.""" - staged = DIST / "custom_components" / DOMAIN - if DIST.exists(): - shutil.rmtree(DIST) - staged.parent.mkdir(parents=True) - # symlinks=False is the point: the libkp symlink lands as a real directory. - shutil.copytree(SOURCE, staged, symlinks=False, ignore=IGNORE) - - archive = DIST / f"{DOMAIN}-{version()}.zip" - with zipfile.ZipFile(archive, "w", zipfile.ZIP_DEFLATED) as zf: - for path in sorted(staged.rglob("*")): - if path.is_file(): - zf.write(path, path.relative_to(DIST)) - return archive - - -def install(config_dir: Path) -> Path: - """Replace ``/custom_components/kemper`` with the freshly built copy.""" - staged = DIST / "custom_components" / DOMAIN - target = config_dir / "custom_components" / DOMAIN - target.parent.mkdir(parents=True, exist_ok=True) - if target.exists(): - shutil.rmtree(target) - shutil.copytree(staged, target) - return target - - -def main() -> int: - """Build, and optionally install into a config directory.""" - parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) - parser.add_argument( - "--install", - metavar="HA_CONFIG_DIR", - type=Path, - help="also copy the bundle into this Home Assistant configuration directory", - ) - args = parser.parse_args() - - archive = build() - staged = DIST / "custom_components" / DOMAIN - files = sum(1 for path in staged.rglob("*") if path.is_file()) - print(f"built {staged.relative_to(HERE)} ({files} files)") - print(f" {archive.relative_to(HERE)}") - - if args.install is not None: - config_dir = args.install.expanduser().resolve() - if not config_dir.is_dir(): - parser.error(f"{config_dir} is not a directory") - target = install(config_dir) - print(f"installed into {target}") - print("restart Home Assistant, then add the integration from Devices & services") - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/python/examples/homeassistant/custom_components/kemper/__init__.py b/python/examples/homeassistant/custom_components/kemper/__init__.py deleted file mode 100644 index dcaceb1..0000000 --- a/python/examples/homeassistant/custom_components/kemper/__init__.py +++ /dev/null @@ -1,130 +0,0 @@ -"""The Kemper Profiler integration: one config entry, one device, one session. - -Setting an entry up opens exactly one MIDI3 stream to the Profiler and keeps -it. The device tolerates a session; what it does not tolerate is connection -*churn* (``docs/06``, ``docs/11``), so nothing here dials in a loop. - -**Where the device is** is decided fresh at every setup. The entry's identity -is the serial the Profiler advertises, not its address: an entry that knows a -serial broadcasts once, and if that serial answers from somewhere else the -entry is updated to the new address (and to the name and firmware version, -which change too) before anything is dialed. Discovery finding nothing — the -port held by Rig Manager, a quiet network, a device on another subnet — is not -an error; the stored address is used as it stands. - -**Losing the stream** therefore goes back through the same door instead of -through libkp's own redial: reconnecting to a remembered address would keep -dialing an address the device may have left. The coordinator asks for a reload, -setup rediscovers, and a device that is simply switched off fails with -:class:`ConfigEntryNotReady`, which is Home Assistant's own spaced retry. -""" - -from __future__ import annotations - -import logging - -from homeassistant.config_entries import ConfigEntry -from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT, Platform -from homeassistant.core import HomeAssistant -from homeassistant.exceptions import ConfigEntryNotReady - -from .const import CONF_SERIAL, CONF_SW_VERSION -from .coordinator import KemperConfigEntry, KemperCoordinator -from .discovery import async_find_serial -from .libkp import ConnectOptions, ControlPolicy, DeviceModel, LibKPError -from .libkp.protocol import PORT - -_LOGGER = logging.getLogger(__name__) - -PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR, Platform.SENSOR] - -#: How the integration connects. The CBOR control channel is deliberately off: -#: the only thing it adds over the stream is the morph position, which nothing -#: here surfaces, and it would cost the device a second socket for as long as -#: Home Assistant runs. No reconnect policy either — see the module docstring: -#: coming back is a reload, so that the address is looked up again first. -CONNECT_CONTROL = ControlPolicy.OFF - - -async def async_locate(hass: HomeAssistant, entry: KemperConfigEntry) -> str: - """The address to dial, after asking the network where the serial is. - - Returns the stored host unchanged when the entry predates serial-keying, - when nothing answers, or when the device is where it was; otherwise the - entry is updated in place — the address, and the name and version, which a - firmware update or a rename changes just as quietly. - """ - host: str = entry.data[CONF_HOST] - serial: str | None = entry.data.get(CONF_SERIAL) - if not serial: - return host - - found = await async_find_serial(serial) - if found is None: - return host - - updates = { - key: value - for key, value in ( - (CONF_HOST, found.host), - (CONF_NAME, found.name), - (CONF_SW_VERSION, found.version), - ) - if entry.data.get(key) != value - } - if not updates: - return host - if CONF_HOST in updates: - _LOGGER.info( - "Profiler %s answered from %s instead of %s; following it", - serial, - found.host, - host, - ) - hass.config_entries.async_update_entry(entry, data={**entry.data, **updates}) - return found.host - - -async def async_setup_entry(hass: HomeAssistant, entry: KemperConfigEntry) -> bool: - """Find the Profiler, connect to it, and bring its entities up.""" - host = await async_locate(hass, entry) - options = ConnectOptions(port=entry.data.get(CONF_PORT, PORT), control=CONNECT_CONTROL) - try: - model = await DeviceModel.connect(host, options=options) - except (LibKPError, OSError) as err: - raise ConfigEntryNotReady(f"could not connect to the Profiler at {host}: {err}") from err - - coordinator = KemperCoordinator(hass, entry, model) - entry.runtime_data = coordinator - try: - await coordinator.async_start() - await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) - except Exception: - # Whatever went wrong, the socket does not get to outlive the attempt. - await coordinator.async_shutdown() - raise - - entry.async_on_unload(entry.add_update_listener(async_options_updated)) - return True - - -async def async_unload_entry(hass: HomeAssistant, entry: KemperConfigEntry) -> bool: - """Tear the entities down and hang up on the device.""" - unloaded = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) - if unloaded: - await entry.runtime_data.async_shutdown() - return unloaded - - -async def async_options_updated(hass: HomeAssistant, entry: ConfigEntry) -> None: - """Apply changed options in place — never by reloading the entry. - - A reload would close the session and open another one, which is a real cost - to the device; the two options that exist only steer the activity detector, - and it can be retuned while it runs. The same listener sees the address - updates :func:`async_locate` makes, which need no action at all: they are - already what the running session was dialed with. - """ - coordinator: KemperCoordinator | None = getattr(entry, "runtime_data", None) - if coordinator is not None: - coordinator.apply_options() diff --git a/python/examples/homeassistant/custom_components/kemper/activity.py b/python/examples/homeassistant/custom_components/kemper/activity.py deleted file mode 100644 index 0d7f127..0000000 --- a/python/examples/homeassistant/custom_components/kemper/activity.py +++ /dev/null @@ -1,199 +0,0 @@ -"""The activity detector — the one consumer of the device's fast lane. - -The Profiler pushes a meter frame about twenty times a second, unrequested, -for as long as the stream is open (``docs/07``). That rate is right for a -level meter and wrong for Home Assistant: an entity written per frame would -put 72,000 states an hour into the recorder to say "someone is playing". - -So the meter lane is read here and nowhere else, and it produces exactly two -state writes per playing session however long the session runs: - -- a plain callback on the model's event stream does two integer comparisons - per frame and stores a timestamp — no awaits, no state writes, no work that - scales with how long the note lasts; -- the first crossing of the threshold flips the detector **on** and arms one - timer; -- when that timer fires, the detector settles **off** if the window has gone - by with no crossing, and otherwise re-arms itself for the remainder. The - timer is never re-armed per sample, so a two-hour rehearsal costs the same - one timer a single chord does. - -The level read is ``rig_out_level`` (meter v6), the tap *after* rig volume. -``docs/07`` describes the four candidates: the strobe fields say nothing about -level, ``stack_level`` (v4) ignores rig volume — so a rig deliberately turned -down still reads loud — and ``loudness`` (v9) is a slow RMS that both lags the -first note and tails off after the last. v6 follows playing dynamics -immediately, respects the rig's own volume, and is deliberately blind to the -main/monitor/headphone knobs, so turning the monitors down to practise -quietly does not read as "stopped playing". -""" - -from __future__ import annotations - -from collections.abc import Callable -from datetime import datetime - -from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback -from homeassistant.helpers.event import async_call_later -from homeassistant.util import dt as dt_util - -from .libkp import _generated as gen -from .libkp.model import DeviceModel -from .libkp.state import DeviceEvent, Status - - -def raw_threshold(percent: float) -> int: - """The 14-bit meter value a ``percent``-of-full-scale threshold means.""" - return max(0, min(gen.FULL_SCALE, round(gen.FULL_SCALE * percent / 100.0))) - - -class ActivityDetector: - """Whether sound is currently passing through the rig, and when it last did. - - Owns its own subscription to the model (:meth:`start` / :meth:`stop`) and - tells its listeners only when one of those two answers changes, which is - what the ``active`` binary sensor and the ``last_activity`` sensor write on. - """ - - def __init__( - self, - hass: HomeAssistant, - model: DeviceModel, - *, - window: float, - threshold: float, - ) -> None: - self._hass = hass - self._model = model - #: How long without a crossing ends a session, in seconds. - self._window = window - #: The threshold as the meter lane reports it: a 14-bit integer, so the - #: per-frame test is one comparison and no arithmetic. - self._threshold = raw_threshold(threshold) - self._active = False - self._last_signal: datetime | None = None - self._last_activity: datetime | None = None - self._cancel_timer: CALLBACK_TYPE | None = None - self._listeners: list[Callable[[], None]] = [] - self._attached = False - - # -- what the entities read ------------------------------------------ - - @property - def active(self) -> bool: - """True while the threshold has been crossed inside the window.""" - return self._active - - @property - def window(self) -> float: - """The quiet window currently in force, in seconds.""" - return self._window - - @property - def threshold(self) -> int: - """The level threshold currently in force, as the meter reports it.""" - return self._threshold - - @property - def last_activity(self) -> datetime | None: - """When signal was last seen, as of the most recent transition. - - While :attr:`active` is on this is when the current session began; - when it goes off it becomes the moment of the last crossing — the last - note heard — and stays there until the next session starts. - """ - return self._last_activity - - @callback - def add_listener(self, callback_: Callable[[], None]) -> CALLBACK_TYPE: - """Register a callback for transitions; returns its remover.""" - self._listeners.append(callback_) - - @callback - def remove() -> None: - if callback_ in self._listeners: - self._listeners.remove(callback_) - - return remove - - # -- lifecycle ------------------------------------------------------- - - @callback - def start(self) -> None: - """Begin watching the model's events.""" - if self._attached: - return - self._model.add_event_listener(self._on_event) - self._attached = True - - @callback - def stop(self) -> None: - """Stop watching and disarm the timer. Idempotent.""" - if self._attached: - self._model.remove_event_listener(self._on_event) - self._attached = False - self._disarm() - - @callback - def update_options(self, *, window: float, threshold: float) -> None: - """Apply new options in place — no reconnect, no reload. - - Every socket to the device costs it something (``docs/11``), so - changing a number in the options form must not cost a session. A - shorter window is honoured immediately: the armed timer is re-evaluated - against the new one, which can settle the detector off on the spot. - """ - self._window = window - self._threshold = raw_threshold(threshold) - if self._active: - self._disarm() - self._expire(dt_util.utcnow()) - - # -- the fast lane --------------------------------------------------- - - @callback - def _on_event(self, event: DeviceEvent) -> None: - """Called for every event the model decodes, ~20 Hz of them meters. - - Keep this trivial: it runs inside the model's ingest path. - """ - if not isinstance(event, Status): - return - if event.status.rig_out_level <= self._threshold: - return - self._last_signal = dt_util.utcnow() - if self._active: - return - self._active = True - self._last_activity = self._last_signal - self._arm(self._window) - self._notify() - - @callback - def _arm(self, delay: float) -> None: - self._cancel_timer = async_call_later(self._hass, delay, self._expire) - - @callback - def _disarm(self) -> None: - if self._cancel_timer is not None: - self._cancel_timer() - self._cancel_timer = None - - @callback - def _expire(self, now: datetime) -> None: - """The window may have run out — settle off, or re-arm for the rest.""" - self._cancel_timer = None - if not self._active or self._last_signal is None: - return - idle = (now - self._last_signal).total_seconds() - if idle < self._window: - self._arm(self._window - idle) - return - self._active = False - self._last_activity = self._last_signal - self._notify() - - @callback - def _notify(self) -> None: - for listener in list(self._listeners): - listener() diff --git a/python/examples/homeassistant/custom_components/kemper/binary_sensor.py b/python/examples/homeassistant/custom_components/kemper/binary_sensor.py deleted file mode 100644 index f8275ab..0000000 --- a/python/examples/homeassistant/custom_components/kemper/binary_sensor.py +++ /dev/null @@ -1,43 +0,0 @@ -"""The ``active`` binary sensor: is anything actually coming out of the rig. - -It reads nothing itself — :class:`~.activity.ActivityDetector` owns the meter -lane and tells this entity when the answer changes, which is twice per playing -session. -""" - -from __future__ import annotations - -from homeassistant.components.binary_sensor import BinarySensorEntity -from homeassistant.core import HomeAssistant -from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback - -from .coordinator import KemperConfigEntry, KemperCoordinator -from .entity import KemperEntity - - -async def async_setup_entry( - hass: HomeAssistant, - entry: KemperConfigEntry, - async_add_entities: AddConfigEntryEntitiesCallback, -) -> None: - """Set up the Profiler's binary sensors.""" - async_add_entities([KemperActiveBinarySensor(entry.runtime_data)]) - - -class KemperActiveBinarySensor(KemperEntity, BinarySensorEntity): - """On while the rig has passed signal inside the configured window.""" - - _attr_translation_key = "active" - - def __init__(self, coordinator: KemperCoordinator) -> None: - super().__init__(coordinator, "active") - - async def async_added_to_hass(self) -> None: - """Follow the detector as well as the coordinator.""" - await super().async_added_to_hass() - self.async_on_remove(self.coordinator.activity.add_listener(self.async_write_ha_state)) - - @property - def is_on(self) -> bool: - """Whether the detector currently reads as playing.""" - return self.coordinator.activity.active diff --git a/python/examples/homeassistant/custom_components/kemper/config_flow.py b/python/examples/homeassistant/custom_components/kemper/config_flow.py deleted file mode 100644 index 2a5212d..0000000 --- a/python/examples/homeassistant/custom_components/kemper/config_flow.py +++ /dev/null @@ -1,218 +0,0 @@ -"""The config flow: find the Profiler if we can, ask for it if we cannot. - -Discovery comes first because the Profiler answers a UDP broadcast with its -name, serial and firmware version — everything the device registry wants — -without costing it a TCP session. The port is exclusive (one process at a -time), so Rig Manager or a running meters example holding it is an ordinary -outcome, not an error: the flow falls through to a host/port form. - -A manually entered host is checked with exactly **one** session, opened and -closed, and then asked who it is with a short directed poll — so a hand-added -Profiler is keyed by its serial too, and survives moving to another address. -Only a device that answers no poll at all is keyed by its host. There is no -retry loop anywhere in this file; the device does not tolerate connection -churn (``docs/06``). -""" - -from __future__ import annotations - -from typing import Any - -import voluptuous as vol -from homeassistant.config_entries import ( - ConfigEntry, - ConfigFlow, - ConfigFlowResult, - OptionsFlow, -) -from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_NAME, CONF_PORT -from homeassistant.core import callback -from homeassistant.helpers.selector import ( - NumberSelector, - NumberSelectorConfig, - NumberSelectorMode, - SelectOptionDict, - SelectSelector, - SelectSelectorConfig, - SelectSelectorMode, - TextSelector, -) - -from .const import ( - CONF_ACTIVITY_THRESHOLD, - CONF_ACTIVITY_WINDOW, - CONF_SERIAL, - CONF_SW_VERSION, - DEFAULT_ACTIVITY_THRESHOLD, - DEFAULT_ACTIVITY_WINDOW, - DEFAULT_NAME, - DOMAIN, -) -from .discovery import Found, async_discover, async_identify -from .libkp import ConnectOptions, ControlPolicy, DeviceModel, LibKPError, SyncStrategy -from .libkp.protocol import PORT - -#: The sentinel option that leaves the device list for the manual form. -MANUAL = "manual" - - -async def async_check(host: str, port: int) -> None: - """Prove a host is a Profiler: one session, opened and closed. - - Nothing is requested on it — the point is the handshake, and the burst of - reads belongs to the entry that goes on to hold the session. - """ - model = await DeviceModel.connect( - host, - options=ConnectOptions(port=port, control=ControlPolicy.OFF, sync=SyncStrategy.OFF), - ) - await model.close() - - -class KemperConfigFlow(ConfigFlow, domain=DOMAIN): - """Add one Profiler.""" - - VERSION = 1 - - def __init__(self) -> None: - self._found: list[Found] = [] - - async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: - """Look for Profilers, then pick one or type one in.""" - self._found = await async_discover() - if not self._found: - return await self.async_step_manual() - return await self.async_step_pick() - - async def async_step_pick(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: - """Choose among the Profilers that answered.""" - if user_input is not None: - chosen = user_input[CONF_DEVICE] - if chosen == MANUAL: - return await self.async_step_manual() - found = next(device for device in self._found if device.host == chosen) - await self.async_set_unique_id(found.serial or found.host) - self._abort_if_unique_id_configured(updates={CONF_HOST: found.host}) - return self.async_create_entry( - title=found.name, - data={ - CONF_HOST: found.host, - CONF_PORT: PORT, - CONF_NAME: found.name, - CONF_SERIAL: found.serial, - CONF_SW_VERSION: found.version, - }, - ) - - options = [ - SelectOptionDict(value=device.host, label=f"{device.name} ({device.host})") - for device in self._found - ] - options.append(SelectOptionDict(value=MANUAL, label="Enter a host manually")) - return self.async_show_form( - step_id="pick", - data_schema=vol.Schema( - { - vol.Required(CONF_DEVICE): SelectSelector( - SelectSelectorConfig(options=options, mode=SelectSelectorMode.LIST) - ) - } - ), - ) - - async def async_step_manual(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: - """Type in a host that discovery could not find.""" - errors: dict[str, str] = {} - if user_input is not None: - host = user_input[CONF_HOST].strip() - port = int(user_input[CONF_PORT]) - try: - await async_check(host, port) - except LibKPError, OSError: - errors["base"] = "cannot_connect" - except Exception: # the form must survive anything the stack raises - errors["base"] = "unknown" - else: - # It is a Profiler; now ask it who it is, so the entry is keyed - # by the serial rather than by an address that can change. - found = await async_identify(host) or Found( - host=host, name=DEFAULT_NAME, serial=None, version=None - ) - await self.async_set_unique_id(found.serial or host) - self._abort_if_unique_id_configured(updates={CONF_HOST: host}) - return self.async_create_entry( - title=found.name, - data={ - CONF_HOST: host, - CONF_PORT: port, - CONF_NAME: found.name, - CONF_SERIAL: found.serial, - CONF_SW_VERSION: found.version, - }, - ) - - suggested = user_input or {CONF_PORT: PORT} - return self.async_show_form( - step_id="manual", - data_schema=vol.Schema( - { - vol.Required(CONF_HOST, default=suggested.get(CONF_HOST, "")): TextSelector(), - vol.Required(CONF_PORT, default=suggested.get(CONF_PORT, PORT)): vol.All( - vol.Coerce(int), vol.Range(min=1, max=65535) - ), - } - ), - errors=errors, - ) - - @staticmethod - @callback - def async_get_options_flow(config_entry: ConfigEntry) -> KemperOptionsFlow: - """The two knobs the activity detector has.""" - return KemperOptionsFlow() - - -class KemperOptionsFlow(OptionsFlow): - """How loud, and for how long, counts as playing. - - Saving these does **not** reload the entry: the integration applies them to - the running detector, so tuning them never costs the device a session. - """ - - async def async_step_init(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult: - """Show and store the detector's window and threshold.""" - if user_input is not None: - return self.async_create_entry(data=user_input) - - options = self.config_entry.options - return self.async_show_form( - step_id="init", - data_schema=vol.Schema( - { - vol.Required( - CONF_ACTIVITY_WINDOW, - default=options.get(CONF_ACTIVITY_WINDOW, DEFAULT_ACTIVITY_WINDOW), - ): NumberSelector( - NumberSelectorConfig( - min=1, - max=120, - step=1, - unit_of_measurement="min", - mode=NumberSelectorMode.BOX, - ) - ), - vol.Required( - CONF_ACTIVITY_THRESHOLD, - default=options.get(CONF_ACTIVITY_THRESHOLD, DEFAULT_ACTIVITY_THRESHOLD), - ): NumberSelector( - NumberSelectorConfig( - min=0, - max=100, - step=0.5, - unit_of_measurement="%", - mode=NumberSelectorMode.BOX, - ) - ), - } - ), - ) diff --git a/python/examples/homeassistant/custom_components/kemper/const.py b/python/examples/homeassistant/custom_components/kemper/const.py deleted file mode 100644 index 248d8e6..0000000 --- a/python/examples/homeassistant/custom_components/kemper/const.py +++ /dev/null @@ -1,31 +0,0 @@ -"""Constants shared by the Kemper Profiler integration.""" - -from __future__ import annotations - -DOMAIN = "kemper" - -#: What the device is called before discovery has told us otherwise. -DEFAULT_NAME = "Kemper Profiler" -#: Device-registry identity, fixed for every Profiler. -MANUFACTURER = "Kemper" -MODEL = "Profiler" - -#: Entry data keys beyond ``CONF_HOST`` / ``CONF_PORT`` / ``CONF_NAME``: what -#: discovery told us about the device, kept so the device registry can show it -#: without re-polling. -CONF_SERIAL = "serial" -CONF_SW_VERSION = "sw_version" - -#: How long a broadcast poll listens for replies, in seconds. -DISCOVERY_SECONDS = 3.0 -#: How long a poll aimed at one known address listens: the device is either -#: there and answers at once, or it is not. -DIRECTED_DISCOVERY_SECONDS = 1.5 - -#: Options-flow keys and their defaults. The window is in minutes and the -#: threshold in percent of the meter lane's full scale, because those are the -#: units the person filling in the form thinks in; the detector converts. -CONF_ACTIVITY_WINDOW = "activity_window" -CONF_ACTIVITY_THRESHOLD = "activity_threshold" -DEFAULT_ACTIVITY_WINDOW = 5.0 -DEFAULT_ACTIVITY_THRESHOLD = 2.0 diff --git a/python/examples/homeassistant/custom_components/kemper/coordinator.py b/python/examples/homeassistant/custom_components/kemper/coordinator.py deleted file mode 100644 index 1193b7c..0000000 --- a/python/examples/homeassistant/custom_components/kemper/coordinator.py +++ /dev/null @@ -1,191 +0,0 @@ -"""The push coordinator: one :class:`DeviceModel`, one state tree, one device. - -libkp's model is already a store — it holds the device state and hands out a -fresh snapshot whenever *slow* state changes, coalesced to at most one per -ingested chunk. So there is nothing to poll here and no update interval: the -coordinator is a :class:`DataUpdateCoordinator` whose data arrives from a -background task that does nothing but drain the model's snapshot queue. - -That task is also where a lost stream is noticed. libkp can redial one on its -own, and this integration deliberately does not ask it to: the model would -redial the address it was given, and the whole point of keying an entry by the -Profiler's serial is that the address is the part that changes. So a loss ends -the session and asks Home Assistant to reload the entry, which starts again -from discovery. A session that ends almost as soon as it began is treated as a -device that is not really there and the reload waits -:data:`RELOAD_DELAY_SECONDS`, so nothing can spin setup in a loop. - -The fast lane (meters, beat pulse, tuner deviance) never reaches this class. -It is read only by :class:`~.activity.ActivityDetector`, which turns it into -two state writes per playing session. -""" - -from __future__ import annotations - -import asyncio -import contextlib -import logging - -from homeassistant.config_entries import ConfigEntry -from homeassistant.const import CONF_HOST, CONF_NAME -from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback -from homeassistant.helpers.device_registry import DeviceInfo -from homeassistant.helpers.event import async_call_later -from homeassistant.helpers.update_coordinator import DataUpdateCoordinator -from homeassistant.util import dt as dt_util - -from .activity import ActivityDetector -from .const import ( - CONF_ACTIVITY_THRESHOLD, - CONF_ACTIVITY_WINDOW, - CONF_SW_VERSION, - DEFAULT_ACTIVITY_THRESHOLD, - DEFAULT_ACTIVITY_WINDOW, - DEFAULT_NAME, - DOMAIN, - MANUFACTURER, - MODEL, -) -from .libkp.model import DeviceModel -from .libkp.state import Connection, DeviceState - -_LOGGER = logging.getLogger(__name__) - -#: A session that ends sooner than this after it opened says the device is not -#: really available, whatever its handshake said. -SHORT_SESSION_SECONDS = 60.0 -#: How long such a session waits before the entry is reloaded. A healthy -#: session that ends — the amp switched off after a rehearsal — reloads at once. -RELOAD_DELAY_SECONDS = 30.0 - -#: The entry, typed by what :attr:`ConfigEntry.runtime_data` holds. -type KemperConfigEntry = ConfigEntry[KemperCoordinator] - - -def activity_window(entry: ConfigEntry) -> float: - """The configured quiet window, in seconds (the form asks for minutes).""" - return float(entry.options.get(CONF_ACTIVITY_WINDOW, DEFAULT_ACTIVITY_WINDOW)) * 60.0 - - -def activity_threshold(entry: ConfigEntry) -> float: - """The configured level threshold, in percent of the meter full scale.""" - return float(entry.options.get(CONF_ACTIVITY_THRESHOLD, DEFAULT_ACTIVITY_THRESHOLD)) - - -class KemperCoordinator(DataUpdateCoordinator[DeviceState]): - """Publishes the model's slow-lane snapshots to the entity layer.""" - - config_entry: KemperConfigEntry - - def __init__(self, hass: HomeAssistant, entry: KemperConfigEntry, model: DeviceModel) -> None: - super().__init__( - hass, - _LOGGER, - config_entry=entry, - name=f"{DOMAIN} {entry.data[CONF_HOST]}", - update_interval=None, - ) - self.model = model - self.activity = ActivityDetector( - hass, - model, - window=activity_window(entry), - threshold=activity_threshold(entry), - ) - self._task: asyncio.Task[None] | None = None - self._reload_timer: CALLBACK_TYPE | None = None - self._opened = dt_util.utcnow() - #: Set once the entry is being torn down, so the disconnection the - #: teardown itself causes is not mistaken for the device going away. - self._closing = False - - @property - def reload_pending(self) -> bool: - """Whether a lost stream is waiting to reload the entry.""" - return self._reload_timer is not None - - @property - def device_id(self) -> str: - """The device-registry identifier: the serial when discovery knew it, - else the host, else the entry — stable across restarts either way.""" - entry = self.config_entry - return entry.unique_id or entry.entry_id - - @property - def device_info(self) -> DeviceInfo: - """One device per config entry: the Profiler itself.""" - entry = self.config_entry - return DeviceInfo( - identifiers={(DOMAIN, self.device_id)}, - manufacturer=MANUFACTURER, - model=MODEL, - name=entry.data.get(CONF_NAME) or DEFAULT_NAME, - sw_version=entry.data.get(CONF_SW_VERSION), - ) - - async def async_start(self) -> None: - """Seed the first snapshot, attach the detector, start listening.""" - self._opened = dt_util.utcnow() - self.async_set_updated_data(self.model.state()) - self.activity.start() - self._task = self.config_entry.async_create_background_task( - self.hass, self._listen(), name=f"{DOMAIN} {self.config_entry.data[CONF_HOST]} state" - ) - - async def _listen(self) -> None: - """Drain the model's store; every snapshot is an entity update. - - The loop ends when the device goes away, which is the one thing a - snapshot can say that this class acts on rather than passes along. - """ - queue = self.model.subscribe() - try: - while True: - state = await queue.get() - self.async_set_updated_data(state) - if state.connection is Connection.DISCONNECTED: - self._schedule_reload() - return - finally: - self.model.unsubscribe(queue) - - @callback - def _schedule_reload(self) -> None: - """Ask for a reload, so the way back starts at discovery.""" - if self._closing or self._reload_timer is not None: - return - session = (dt_util.utcnow() - self._opened).total_seconds() - delay = 0.0 if session >= SHORT_SESSION_SECONDS else RELOAD_DELAY_SECONDS - _LOGGER.info( - "Lost the stream to the Profiler after %.0f s; reloading in %.0f s to find it again", - session, - delay, - ) - self._reload_timer = async_call_later(self.hass, delay, self._reload) - - @callback - def _reload(self, _now: object) -> None: - self._reload_timer = None - self.hass.config_entries.async_schedule_reload(self.config_entry.entry_id) - - async def async_shutdown(self) -> None: - """Stop listening and hang up. The device sees one clean disconnect.""" - self._closing = True - if self._reload_timer is not None: - self._reload_timer() - self._reload_timer = None - await super().async_shutdown() - self.activity.stop() - if self._task is not None: - self._task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await self._task - self._task = None - await self.model.close() - - def apply_options(self) -> None: - """Re-read the options the detector uses, without touching the socket.""" - entry = self.config_entry - self.activity.update_options( - window=activity_window(entry), threshold=activity_threshold(entry) - ) diff --git a/python/examples/homeassistant/custom_components/kemper/diagnostics.py b/python/examples/homeassistant/custom_components/kemper/diagnostics.py deleted file mode 100644 index 40ff4e9..0000000 --- a/python/examples/homeassistant/custom_components/kemper/diagnostics.py +++ /dev/null @@ -1,60 +0,0 @@ -"""Diagnostics: the whole state tree, which is the debugging window. - -Only five entities are exposed, but the model knows far more than that — the -effect slots, the tempo, the volumes, the tuner, the bank preview, both -channels' states. Dumping the tree here means a bug report carries everything -the device said without any of it having to become an entity first. -""" - -from __future__ import annotations - -from dataclasses import fields, is_dataclass -from enum import Enum -from typing import Any - -from homeassistant.components.diagnostics import async_redact_data -from homeassistant.const import CONF_HOST -from homeassistant.core import HomeAssistant - -from .const import CONF_SERIAL -from .coordinator import KemperConfigEntry - -TO_REDACT = {CONF_HOST, CONF_SERIAL} - - -def plain(value: Any) -> Any: - """A JSON-friendly copy of a state tree: dataclasses to dicts, enums to - their values, private bookkeeping fields left out.""" - if is_dataclass(value) and not isinstance(value, type): - return { - field.name: plain(getattr(value, field.name)) - for field in fields(value) - if not field.name.startswith("_") - } - if isinstance(value, Enum): - return value.value - if isinstance(value, (list, tuple, set)): - return [plain(item) for item in value] - if isinstance(value, dict): - return {str(key): plain(item) for key, item in value.items()} - return value - - -async def async_get_config_entry_diagnostics( - hass: HomeAssistant, entry: KemperConfigEntry -) -> dict[str, Any]: - """Everything this integration knows about one Profiler.""" - coordinator = entry.runtime_data - detector = coordinator.activity - last_activity = detector.last_activity - return { - "entry": { - "data": async_redact_data(dict(entry.data), TO_REDACT), - "options": dict(entry.options), - }, - "activity": { - "active": detector.active, - "last_activity": None if last_activity is None else last_activity.isoformat(), - }, - "state": plain(coordinator.model.state()), - } diff --git a/python/examples/homeassistant/custom_components/kemper/discovery.py b/python/examples/homeassistant/custom_components/kemper/discovery.py deleted file mode 100644 index 026ec39..0000000 --- a/python/examples/homeassistant/custom_components/kemper/discovery.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Finding a Profiler, and finding it again when its address moves. - -The Profiler answers a UDP broadcast with its name, **serial** and firmware -version, and costs itself nothing to do it — no session, no handshake. That -makes the serial the device's real identity and the IP address merely where it -happens to be today: a DHCP lease expires over a weekend and the same amp comes -back on a different address. - -So discovery is used in two places, and both live here rather than in the -config flow: once when a Profiler is added, and once per setup to ask "where is -serial X now". Every poll is a single 3-second listen — there is no loop, and a -port that another program holds is an ordinary answer of "nothing found". -""" - -from __future__ import annotations - -from collections.abc import Iterable -from dataclasses import dataclass - -from .const import DEFAULT_NAME, DIRECTED_DISCOVERY_SECONDS, DISCOVERY_SECONDS -from .libkp import DiscoveryOptions, DiscoveryPort, LibKPError -from .libkp.discovery import Reply - - -@dataclass(frozen=True, slots=True) -class Found: - """One Profiler that answered a discovery poll.""" - - host: str - name: str - serial: str | None - version: str | None - - @classmethod - def from_reply(cls, reply: Reply) -> Found: - """What a raw reply says about the device that sent it.""" - return cls( - host=reply.ip, - name=reply.name or DEFAULT_NAME, - serial=reply.serial, - version=reply.version, - ) - - -async def async_discover( - *, listen_for: float = DISCOVERY_SECONDS, targets: Iterable[str] | None = None -) -> list[Found]: - """Poll for Profilers. An unavailable port or a failed poll means none. - - ``targets`` adds explicit unicast destinations to the broadcast, which is - how a device on the other side of a router — one no broadcast reaches — can - still be asked to identify itself. - """ - try: - with DiscoveryPort.acquire() as port: - replies = await port.poll( - DiscoveryOptions(listen_for=listen_for, extra_targets=list(targets or ())) - ) - except LibKPError, OSError: - return [] - return [Found.from_reply(reply) for reply in replies] - - -async def async_find_serial(serial: str) -> Found | None: - """Where the Profiler with this serial is now, if it answers at all.""" - for found in await async_discover(): - if found.serial == serial: - return found - return None - - -async def async_identify(host: str) -> Found | None: - """Ask one known address who it is: a short, directed poll. - - Used after a manually entered host has been proved to be a Profiler, so - that a hand-added device is keyed by its serial like a discovered one and - survives moving to another address. - """ - for found in await async_discover(listen_for=DIRECTED_DISCOVERY_SECONDS, targets=[host]): - if found.host == host: - return found - return None diff --git a/python/examples/homeassistant/custom_components/kemper/entity.py b/python/examples/homeassistant/custom_components/kemper/entity.py deleted file mode 100644 index 436a346..0000000 --- a/python/examples/homeassistant/custom_components/kemper/entity.py +++ /dev/null @@ -1,30 +0,0 @@ -"""The base entity: device identity, naming and availability in one place.""" - -from __future__ import annotations - -from homeassistant.helpers.update_coordinator import CoordinatorEntity - -from .coordinator import KemperCoordinator -from .libkp.state import Connection - -#: The connection states in which what the entities show is live. A degraded -#: connection is still a connection: the stream — everything these entities -#: read — is open, and only the optional control channel is missing. -LIVE = (Connection.CONNECTED, Connection.DEGRADED) - - -class KemperEntity(CoordinatorEntity[KemperCoordinator]): - """One reading from one Profiler.""" - - _attr_has_entity_name = True - - def __init__(self, coordinator: KemperCoordinator, key: str) -> None: - super().__init__(coordinator) - self._attr_unique_id = f"{coordinator.device_id}_{key}" - self._attr_device_info = coordinator.device_info - - @property - def available(self) -> bool: - """Available while the stream is up — the tree goes stale without it.""" - state = self.coordinator.data - return super().available and state is not None and state.connection in LIVE diff --git a/python/examples/homeassistant/custom_components/kemper/icons.json b/python/examples/homeassistant/custom_components/kemper/icons.json deleted file mode 100644 index 14944be..0000000 --- a/python/examples/homeassistant/custom_components/kemper/icons.json +++ /dev/null @@ -1,27 +0,0 @@ -{ - "entity": { - "binary_sensor": { - "active": { - "default": "mdi:guitar-electric", - "state": { - "off": "mdi:guitar-electric", - "on": "mdi:music-note" - } - } - }, - "sensor": { - "rig_name": { - "default": "mdi:tune-vertical" - }, - "amp_name": { - "default": "mdi:amplifier" - }, - "cabinet_name": { - "default": "mdi:speaker" - }, - "last_activity": { - "default": "mdi:clock-outline" - } - } - } -} diff --git a/python/examples/homeassistant/custom_components/kemper/libkp b/python/examples/homeassistant/custom_components/kemper/libkp deleted file mode 120000 index 454fbc0..0000000 --- a/python/examples/homeassistant/custom_components/kemper/libkp +++ /dev/null @@ -1 +0,0 @@ -../../../../src/libkp \ No newline at end of file diff --git a/python/examples/homeassistant/custom_components/kemper/manifest.json b/python/examples/homeassistant/custom_components/kemper/manifest.json deleted file mode 100644 index c9c0fc6..0000000 --- a/python/examples/homeassistant/custom_components/kemper/manifest.json +++ /dev/null @@ -1,13 +0,0 @@ -{ - "domain": "kemper", - "name": "Kemper Profiler", - "codeowners": ["@gotwalt"], - "config_flow": true, - "dependencies": [], - "documentation": "https://github.com/gotwalt/libkp/tree/main/python/examples/homeassistant", - "integration_type": "device", - "iot_class": "local_push", - "issue_tracker": "https://github.com/gotwalt/libkp/issues", - "requirements": [], - "version": "0.1.0" -} diff --git a/python/examples/homeassistant/custom_components/kemper/sensor.py b/python/examples/homeassistant/custom_components/kemper/sensor.py deleted file mode 100644 index 5fb2f41..0000000 --- a/python/examples/homeassistant/custom_components/kemper/sensor.py +++ /dev/null @@ -1,102 +0,0 @@ -"""Sensors: what is loaded on the Profiler, and when it last made a sound. - -The three name sensors are a table of :class:`SensorEntityDescription` rows -with a ``value_fn`` over the state tree, so the tempo, the volumes, the morph -position or an effect slot's type is one row each whenever they are wanted. -""" - -from __future__ import annotations - -from collections.abc import Callable -from dataclasses import dataclass -from datetime import datetime - -from homeassistant.components.sensor import ( - SensorDeviceClass, - SensorEntity, - SensorEntityDescription, -) -from homeassistant.core import HomeAssistant -from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback - -from .coordinator import KemperConfigEntry, KemperCoordinator -from .entity import KemperEntity -from .libkp.state import DeviceState - - -@dataclass(frozen=True, kw_only=True) -class KemperSensorEntityDescription(SensorEntityDescription): - """A sensor described by where it reads from in the state tree.""" - - value_fn: Callable[[DeviceState], str | None] - - -SENSORS: tuple[KemperSensorEntityDescription, ...] = ( - KemperSensorEntityDescription( - key="rig_name", - translation_key="rig_name", - value_fn=lambda state: state.rig.name, - ), - KemperSensorEntityDescription( - key="amp_name", - translation_key="amp_name", - value_fn=lambda state: state.amp.name, - ), - KemperSensorEntityDescription( - key="cabinet_name", - translation_key="cabinet_name", - value_fn=lambda state: state.cabinet.name, - ), -) - - -async def async_setup_entry( - hass: HomeAssistant, - entry: KemperConfigEntry, - async_add_entities: AddConfigEntryEntitiesCallback, -) -> None: - """Set up the Profiler's sensors.""" - coordinator = entry.runtime_data - entities: list[SensorEntity] = [ - KemperSensor(coordinator, description) for description in SENSORS - ] - entities.append(KemperLastActivitySensor(coordinator)) - async_add_entities(entities) - - -class KemperSensor(KemperEntity, SensorEntity): - """One value read straight out of the state tree.""" - - entity_description: KemperSensorEntityDescription - - def __init__( - self, coordinator: KemperCoordinator, description: KemperSensorEntityDescription - ) -> None: - super().__init__(coordinator, description.key) - self.entity_description = description - - @property - def native_value(self) -> str | None: - """The described value, or ``None`` while the device has not said.""" - state = self.coordinator.data - return None if state is None else self.entity_description.value_fn(state) - - -class KemperLastActivitySensor(KemperEntity, SensorEntity): - """When signal was last heard, written on the detector's transitions only.""" - - _attr_translation_key = "last_activity" - _attr_device_class = SensorDeviceClass.TIMESTAMP - - def __init__(self, coordinator: KemperCoordinator) -> None: - super().__init__(coordinator, "last_activity") - - async def async_added_to_hass(self) -> None: - """Follow the detector as well as the coordinator.""" - await super().async_added_to_hass() - self.async_on_remove(self.coordinator.activity.add_listener(self.async_write_ha_state)) - - @property - def native_value(self) -> datetime | None: - """The last crossing the detector settled on, or ``None`` before one.""" - return self.coordinator.activity.last_activity diff --git a/python/examples/homeassistant/custom_components/kemper/strings.json b/python/examples/homeassistant/custom_components/kemper/strings.json deleted file mode 100644 index 380af65..0000000 --- a/python/examples/homeassistant/custom_components/kemper/strings.json +++ /dev/null @@ -1,69 +0,0 @@ -{ - "config": { - "step": { - "pick": { - "title": "Choose a Profiler", - "description": "These Profilers answered on the local network.", - "data": { - "device": "Profiler" - } - }, - "manual": { - "title": "Connect to a Profiler", - "description": "Enter the address of the Profiler. Its network port is on the device under System > Network.", - "data": { - "host": "Host", - "port": "Port" - }, - "data_description": { - "host": "The Profiler's IP address or host name.", - "port": "Leave this at 5727 unless the Profiler was told to use another port." - } - } - }, - "error": { - "cannot_connect": "Could not open a session with that Profiler. Check the address, and that the Profiler is switched on and on the network.", - "unknown": "Unexpected error" - }, - "abort": { - "already_configured": "This Profiler is already configured" - } - }, - "options": { - "step": { - "init": { - "title": "Activity detection", - "description": "How the Profiler decides that someone is playing. The output level is read from the rig output meter, after the rig volume and before the master volume.", - "data": { - "activity_window": "Quiet window", - "activity_threshold": "Level threshold" - }, - "data_description": { - "activity_window": "How long the output has to stay below the threshold before activity turns off.", - "activity_threshold": "How loud the rig output has to be to count as playing, as a percentage of full scale." - } - } - } - }, - "entity": { - "binary_sensor": { - "active": { - "name": "Active" - } - }, - "sensor": { - "rig_name": { - "name": "Rig" - }, - "amp_name": { - "name": "Amp" - }, - "cabinet_name": { - "name": "Cabinet" - }, - "last_activity": { - "name": "Last activity" - } - } - } -} diff --git a/python/examples/homeassistant/custom_components/kemper/translations/en.json b/python/examples/homeassistant/custom_components/kemper/translations/en.json deleted file mode 100644 index 380af65..0000000 --- a/python/examples/homeassistant/custom_components/kemper/translations/en.json +++ /dev/null @@ -1,69 +0,0 @@ -{ - "config": { - "step": { - "pick": { - "title": "Choose a Profiler", - "description": "These Profilers answered on the local network.", - "data": { - "device": "Profiler" - } - }, - "manual": { - "title": "Connect to a Profiler", - "description": "Enter the address of the Profiler. Its network port is on the device under System > Network.", - "data": { - "host": "Host", - "port": "Port" - }, - "data_description": { - "host": "The Profiler's IP address or host name.", - "port": "Leave this at 5727 unless the Profiler was told to use another port." - } - } - }, - "error": { - "cannot_connect": "Could not open a session with that Profiler. Check the address, and that the Profiler is switched on and on the network.", - "unknown": "Unexpected error" - }, - "abort": { - "already_configured": "This Profiler is already configured" - } - }, - "options": { - "step": { - "init": { - "title": "Activity detection", - "description": "How the Profiler decides that someone is playing. The output level is read from the rig output meter, after the rig volume and before the master volume.", - "data": { - "activity_window": "Quiet window", - "activity_threshold": "Level threshold" - }, - "data_description": { - "activity_window": "How long the output has to stay below the threshold before activity turns off.", - "activity_threshold": "How loud the rig output has to be to count as playing, as a percentage of full scale." - } - } - } - }, - "entity": { - "binary_sensor": { - "active": { - "name": "Active" - } - }, - "sensor": { - "rig_name": { - "name": "Rig" - }, - "amp_name": { - "name": "Amp" - }, - "cabinet_name": { - "name": "Cabinet" - }, - "last_activity": { - "name": "Last activity" - } - } - } -} diff --git a/python/examples/homeassistant/pyproject.toml b/python/examples/homeassistant/pyproject.toml deleted file mode 100644 index 51216e9..0000000 --- a/python/examples/homeassistant/pyproject.toml +++ /dev/null @@ -1,33 +0,0 @@ -[project] -# A development shell, not a distributable: the integration ships as the -# `custom_components/kemper` directory that build.py bundles, and Home -# Assistant loads it from a config directory rather than from a wheel. This -# file exists so `uv sync` can put Home Assistant's test harness on the path. -name = "kemper-homeassistant" -version = "0.1.0" -description = "Development environment for the Kemper Profiler Home Assistant integration." -requires-python = ">=3.14.2" -dependencies = [ - # Pins homeassistant==2026.8.3 and brings pytest + the custom-component fixtures. - "pytest-homeassistant-custom-component==0.13.357", - "ruff>=0.6", -] - -[tool.uv] -package = false - -[tool.pytest.ini_options] -testpaths = ["tests"] -# "." for `custom_components.kemper`, ../../src for the top-level `libkp` that -# libkp's own test harness imports, ../../tests for that harness (fake_device). -pythonpath = [".", "../../src", "../../tests"] -asyncio_mode = "auto" - -[tool.ruff] -line-length = 100 -# The vendored library is linted by python/'s own configuration; dist/ is build -# output, and the symlink would otherwise be linted twice under two names. -extend-exclude = ["custom_components/kemper/libkp", "dist"] - -[tool.ruff.lint] -select = ["E", "F", "I", "W", "UP"] diff --git a/python/examples/homeassistant/tests/conftest.py b/python/examples/homeassistant/tests/conftest.py deleted file mode 100644 index f383d09..0000000 --- a/python/examples/homeassistant/tests/conftest.py +++ /dev/null @@ -1,129 +0,0 @@ -"""Fixtures: a Home Assistant that loads the integration, against a fake Profiler. - -The device side is libkp's own :class:`fake_device.FakeDevice` — the same -in-process stand-in its test suite drives — so these tests exercise the real -session handshake, the real MIDI3 framing and the real state fold, and mock -nothing below the config entry. -""" - -from __future__ import annotations - -import asyncio -from collections.abc import AsyncIterator -from unittest.mock import patch - -import pytest -from fake_device import FakeDevice, answer_requests -from homeassistant.config_entries import ConfigEntryState -from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT -from homeassistant.core import HomeAssistant -from homeassistant.helpers import entity_registry as er -from pytest_homeassistant_custom_component.common import MockConfigEntry - -from custom_components.kemper.const import CONF_SERIAL, CONF_SW_VERSION, DOMAIN -from custom_components.kemper.libkp.errors import PortUnavailableError -from custom_components.kemper.libkp.protocol import PORT - -#: The serial the fixture entry claims, and so the prefix of every unique id. -SERIAL = "FAKE-SERIAL" -DEVICE_NAME = "Test Profiler" - - -@pytest.fixture(autouse=True) -def auto_enable_custom_integrations(enable_custom_integrations): - """Let Home Assistant see `custom_components/kemper` in every test.""" - return - - -@pytest.fixture(autouse=True) -def no_broadcast(): - """No test polls the real LAN. - - The discovery port reads as held by another program, which is a state the - integration is built to shrug off: every path that discovers falls back to - what the entry already knows. A test that wants replies patches - ``custom_components.kemper.discovery.async_discover`` instead. - """ - with patch( - "custom_components.kemper.discovery.DiscoveryPort.acquire", - side_effect=PortUnavailableError(PORT, OSError("held by the test suite")), - ): - yield - - -@pytest.fixture -async def device(socket_enabled: None) -> AsyncIterator[FakeDevice]: - """A Profiler stand-in that answers the model's opening burst. - - ``socket_enabled`` lifts Home Assistant's test-suite ban on real sockets: - these tests deliberately want one, since a loopback TCP session is exactly - what the integration does in the field. - """ - fake = await FakeDevice(responder=answer_requests).start() - try: - yield fake - finally: - # Hang up first: a server whose handlers are still running never - # finishes closing, and an entry that failed to unload leaves one. - await fake.hangup() - await fake.stop() - - -def make_entry(device: FakeDevice) -> MockConfigEntry: - """A config entry pointing at the fake device's ephemeral port.""" - return MockConfigEntry( - domain=DOMAIN, - title=DEVICE_NAME, - unique_id=SERIAL, - data={ - CONF_HOST: "127.0.0.1", - CONF_PORT: device.port, - CONF_NAME: DEVICE_NAME, - CONF_SERIAL: SERIAL, - CONF_SW_VERSION: "1.2.3", - }, - ) - - -@pytest.fixture -async def entry(hass: HomeAssistant, device: FakeDevice) -> AsyncIterator[MockConfigEntry]: - """A loaded config entry, unloaded again with the test. - - Unloading matters here: it is what closes the session and disarms the - detector's timer, and Home Assistant's test harness fails a test that - leaves either behind. - """ - entry = make_entry(device) - entry.add_to_hass(hass) - assert await hass.config_entries.async_setup(entry.entry_id) - await hass.async_block_till_done() - yield entry - if entry.state is ConfigEntryState.LOADED: - assert await hass.config_entries.async_unload(entry.entry_id) - await hass.async_block_till_done() - - -def entity_id(hass: HomeAssistant, platform: str, key: str) -> str: - """The entity id the registry gave one of the integration's unique ids.""" - found = er.async_get(hass).async_get_entity_id(platform, DOMAIN, f"{SERIAL}_{key}") - assert found is not None, f"no {platform} entity registered for {key}" - return found - - -async def wait_until(predicate, timeout: float = 5.0) -> None: - """Wait until ``predicate`` is true; the wire is asynchronous.""" - async with asyncio.timeout(timeout): - while not predicate(): - await asyncio.sleep(0.01) - - -async def wait_for_state( - hass: HomeAssistant, entity: str, value: str, timeout: float = 5.0 -) -> None: - """Wait until ``entity`` reads ``value``; the wire is asynchronous.""" - async with asyncio.timeout(timeout): - while True: - state = hass.states.get(entity) - if state is not None and state.state == value: - return - await asyncio.sleep(0.01) diff --git a/python/examples/homeassistant/tests/test_binary_sensor.py b/python/examples/homeassistant/tests/test_binary_sensor.py deleted file mode 100644 index 2d6dd9c..0000000 --- a/python/examples/homeassistant/tests/test_binary_sensor.py +++ /dev/null @@ -1,161 +0,0 @@ -"""The activity detector, at the rate the device really pushes meters. - -Every assertion here is about *how many* state writes come out: the meter lane -runs at ~20 Hz, and the whole point of the detector is that Home Assistant -sees two states per playing session and not two per second. -""" - -from __future__ import annotations - -from datetime import timedelta - -from conftest import entity_id, wait_for_state -from fake_device import FakeDevice -from homeassistant.const import EVENT_STATE_CHANGED, STATE_OFF, STATE_ON -from homeassistant.core import Event, HomeAssistant, callback -from homeassistant.util import dt as dt_util -from libkp import _generated as gen -from libkp.nrpn import sysex, u14_split -from pytest_homeassistant_custom_component.common import ( - MockConfigEntry, - async_fire_time_changed, -) - -from custom_components.kemper.const import DEFAULT_ACTIVITY_WINDOW - -#: Rig output level (v6) well above and well below the 2% default threshold. -LOUD = 9000 -QUIET = 100 - - -def meter_message(rig_out_level: int) -> bytes: - """One realtime status frame carrying ``rig_out_level`` in v6.""" - values = [0] * gen.METER_COUNT - values[6] = rig_out_level - payload = bytearray() - for value in values: - payload.extend(u14_split(value)) - return sysex(0x00, 0x00, 0x02, gen.PAGE_REALTIME, gen.METER_BLOCK_NUMBER, bytes(payload)) - - -def count_changes(hass: HomeAssistant, entity: str) -> list[Event]: - """Collect every state change of one entity from now on.""" - seen: list[Event] = [] - - @callback - def record(event: Event) -> None: - if event.data["entity_id"] == entity: - seen.append(event) - - hass.bus.async_listen(EVENT_STATE_CHANGED, record) - return seen - - -async def test_quiet_frames_do_not_count_as_playing( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """A stream with nothing plugged into it still pushes meters.""" - active = entity_id(hass, "binary_sensor", "active") - changes = count_changes(hass, active) - - for _ in range(20): - await device.push(meter_message(QUIET)) - await hass.async_block_till_done() - - assert hass.states.get(active).state == STATE_OFF - assert changes == [] - assert hass.states.get(entity_id(hass, "sensor", "last_activity")).state == "unknown" - - -async def test_a_burst_of_frames_is_one_state_write( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """A hundred frames — five seconds of playing — write the state once.""" - active = entity_id(hass, "binary_sensor", "active") - last_activity = entity_id(hass, "sensor", "last_activity") - changes = count_changes(hass, active) - activity_changes = count_changes(hass, last_activity) - - for _ in range(100): - await device.push(meter_message(LOUD)) - await wait_for_state(hass, active, STATE_ON) - await hass.async_block_till_done() - - assert len(changes) == 1 - assert len(activity_changes) == 1 - assert hass.states.get(last_activity).state != "unknown" - - -async def test_the_window_settles_the_sensor_off( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Silence for the whole window turns it off — and that is the second write.""" - active = entity_id(hass, "binary_sensor", "active") - await device.push(meter_message(LOUD)) - await wait_for_state(hass, active, STATE_ON) - - changes = count_changes(hass, active) - window = timedelta(minutes=DEFAULT_ACTIVITY_WINDOW) - async_fire_time_changed(hass, dt_util.utcnow() + window + timedelta(seconds=1)) - await hass.async_block_till_done() - - assert hass.states.get(active).state == STATE_OFF - assert len(changes) == 1 - - -async def test_playing_again_before_the_window_ends_keeps_it_on( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """A pause shorter than the window re-arms the timer instead of settling.""" - active = entity_id(hass, "binary_sensor", "active") - await device.push(meter_message(LOUD)) - await wait_for_state(hass, active, STATE_ON) - - changes = count_changes(hass, active) - # The timer fires early — the model has heard a note since it was armed. - async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=1)) - await hass.async_block_till_done() - - assert hass.states.get(active).state == STATE_ON - assert changes == [] - - -async def test_a_shorter_window_applies_without_a_reconnect( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Saving the options form retunes the running detector.""" - from custom_components.kemper.const import CONF_ACTIVITY_THRESHOLD, CONF_ACTIVITY_WINDOW - - active = entity_id(hass, "binary_sensor", "active") - await device.push(meter_message(LOUD)) - await wait_for_state(hass, active, STATE_ON) - - hass.config_entries.async_update_entry( - entry, options={CONF_ACTIVITY_WINDOW: 1, CONF_ACTIVITY_THRESHOLD: 2} - ) - await hass.async_block_till_done() - - detector = entry.runtime_data.activity - assert detector.window == 60.0 - async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=61)) - await hass.async_block_till_done() - assert hass.states.get(active).state == STATE_OFF - - -async def test_a_louder_threshold_ignores_quiet_playing( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """The threshold is a percentage of full scale, and it is enforced.""" - from custom_components.kemper.const import CONF_ACTIVITY_THRESHOLD, CONF_ACTIVITY_WINDOW - - hass.config_entries.async_update_entry( - entry, options={CONF_ACTIVITY_WINDOW: 5, CONF_ACTIVITY_THRESHOLD: 90} - ) - await hass.async_block_till_done() - - active = entity_id(hass, "binary_sensor", "active") - for _ in range(20): - await device.push(meter_message(LOUD)) - await hass.async_block_till_done() - - assert hass.states.get(active).state == STATE_OFF diff --git a/python/examples/homeassistant/tests/test_build.py b/python/examples/homeassistant/tests/test_build.py deleted file mode 100644 index f8c2450..0000000 --- a/python/examples/homeassistant/tests/test_build.py +++ /dev/null @@ -1,75 +0,0 @@ -"""The bundler: what leaves the source tree, and in what shape. - -The one thing that has to be true of a bundle is that the ``libkp`` symlink -became a real directory: Home Assistant copies a custom component into its -configuration and would find nothing at the other end of a relative symlink. -""" - -from __future__ import annotations - -import zipfile -from pathlib import Path - -import pytest - -import build - - -@pytest.fixture(autouse=True) -def dist_in_tmp(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: - """Build into the test's own directory, never the working tree's dist/.""" - dist = tmp_path / "dist" - monkeypatch.setattr(build, "DIST", dist) - return dist - - -def test_the_library_is_vendored_as_real_files(tmp_path: Path) -> None: - """No symlink survives the copy, and the generated module comes with it.""" - build.build() - staged = tmp_path / "dist" / "custom_components" / "kemper" - - assert (staged / "manifest.json").is_file() - library = staged / "libkp" - assert library.is_dir() - assert not library.is_symlink() - generated = library / "_generated.py" - assert generated.is_file() - assert not generated.is_symlink() - assert "SPEC_VERSION" in generated.read_text(encoding="utf-8") - - -def test_nothing_that_should_not_ship_ships(tmp_path: Path) -> None: - """Bytecode caches and the integration's tests stay behind.""" - build.build() - staged = tmp_path / "dist" / "custom_components" / "kemper" - - assert not list(staged.rglob("__pycache__")) - assert not list(staged.rglob("*.pyc")) - assert not list(staged.rglob("tests")) - - -def test_the_zip_unpacks_into_a_config_directory(tmp_path: Path) -> None: - """Its paths are relative to the configuration directory, ready to unzip.""" - archive = build.build() - assert archive.name == f"kemper-{build.version()}.zip" - - with zipfile.ZipFile(archive) as zf: - names = zf.namelist() - assert "custom_components/kemper/manifest.json" in names - assert "custom_components/kemper/libkp/_generated.py" in names - assert "custom_components/kemper/translations/en.json" in names - - -def test_install_replaces_an_existing_copy(tmp_path: Path) -> None: - """Installing twice leaves one copy, not a merge of two.""" - build.build() - config_dir = tmp_path / "config" - config_dir.mkdir() - - target = build.install(config_dir) - stale = target / "stale.py" - stale.write_text("# left over from an older version\n", encoding="utf-8") - - build.install(config_dir) - assert (target / "manifest.json").is_file() - assert not stale.exists() diff --git a/python/examples/homeassistant/tests/test_config_flow.py b/python/examples/homeassistant/tests/test_config_flow.py deleted file mode 100644 index b9ba946..0000000 --- a/python/examples/homeassistant/tests/test_config_flow.py +++ /dev/null @@ -1,201 +0,0 @@ -"""The config flow: discovery, the manual fallback, and the options form.""" - -from __future__ import annotations - -from unittest.mock import AsyncMock, patch - -import pytest -from fake_device import FakeDevice -from homeassistant.config_entries import SOURCE_USER -from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_PORT -from homeassistant.core import HomeAssistant -from homeassistant.data_entry_flow import FlowResultType -from pytest_homeassistant_custom_component.common import MockConfigEntry - -from custom_components.kemper.config_flow import MANUAL -from custom_components.kemper.const import ( - CONF_ACTIVITY_THRESHOLD, - CONF_ACTIVITY_WINDOW, - CONF_SERIAL, - CONF_SW_VERSION, - DOMAIN, -) -from custom_components.kemper.discovery import Found - -FOUND = Found(host="10.0.0.5", name="Studio Profiler", serial="SER123", version="10.5.2") - - -@pytest.fixture -def no_setup(): - """Stop a created entry from dialing a device the test does not have.""" - with patch( - "custom_components.kemper.async_setup_entry", AsyncMock(return_value=True) - ) as mocked: - yield mocked - - -async def start(hass: HomeAssistant, found: list[Found]) -> dict: - """Run the user step with a canned discovery result.""" - with patch("custom_components.kemper.config_flow.async_discover", return_value=found): - return await hass.config_entries.flow.async_init(DOMAIN, context={"source": SOURCE_USER}) - - -async def test_a_discovered_profiler_needs_no_session( - hass: HomeAssistant, no_setup: AsyncMock -) -> None: - """Discovery already carries the name, serial and version — take them.""" - result = await start(hass, [FOUND]) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "pick" - - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_DEVICE: FOUND.host} - ) - assert result["type"] is FlowResultType.CREATE_ENTRY - assert result["title"] == "Studio Profiler" - assert result["data"][CONF_HOST] == "10.0.0.5" - assert result["result"].unique_id == "SER123" - - -async def test_a_profiler_is_only_added_once(hass: HomeAssistant, no_setup: AsyncMock) -> None: - """The serial is the identity, so the same device cannot be added twice.""" - existing = MockConfigEntry(domain=DOMAIN, unique_id="SER123", data={CONF_HOST: "10.0.0.9"}) - existing.add_to_hass(hass) - - result = await start(hass, [FOUND]) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_DEVICE: FOUND.host} - ) - assert result["type"] is FlowResultType.ABORT - assert result["reason"] == "already_configured" - # The address is refreshed on the way out: devices move between leases. - assert existing.data[CONF_HOST] == "10.0.0.5" - - -async def test_no_replies_falls_through_to_the_form(hass: HomeAssistant) -> None: - """A held discovery port or a quiet network is not an error.""" - result = await start(hass, []) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "manual" - - -async def test_the_manual_form_is_reachable_from_the_list(hass: HomeAssistant) -> None: - """A Profiler on another subnet never answers the broadcast.""" - result = await start(hass, [FOUND]) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_DEVICE: MANUAL} - ) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "manual" - - -async def test_a_manual_host_is_checked_with_one_session( - hass: HomeAssistant, device: FakeDevice, no_setup: AsyncMock -) -> None: - """The check is a real handshake against the fake device, opened once.""" - result = await start(hass, []) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_HOST: "127.0.0.1", CONF_PORT: device.port} - ) - - assert result["type"] is FlowResultType.CREATE_ENTRY - assert result["data"][CONF_PORT] == device.port - assert result["result"].unique_id == "127.0.0.1" - assert len(device.connections) == 1 - - -async def test_a_host_that_does_not_answer_says_so(hass: HomeAssistant) -> None: - """A refused connection is a form error, not a traceback.""" - result = await start(hass, []) - with patch("custom_components.kemper.config_flow.async_check", side_effect=OSError("refused")): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_HOST: "10.0.0.7", CONF_PORT: 5727} - ) - assert result["type"] is FlowResultType.FORM - assert result["errors"] == {"base": "cannot_connect"} - - -async def test_an_unexpected_failure_says_unknown(hass: HomeAssistant) -> None: - """Anything the stack can raise leaves the form usable.""" - result = await start(hass, []) - with patch("custom_components.kemper.config_flow.async_check", side_effect=ValueError("odd")): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_HOST: "10.0.0.7", CONF_PORT: 5727} - ) - assert result["type"] is FlowResultType.FORM - assert result["errors"] == {"base": "unknown"} - - -async def test_the_options_form_retunes_the_detector( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Saving options reaches the running detector, and only it.""" - result = await hass.config_entries.options.async_init(entry.entry_id) - assert result["type"] is FlowResultType.FORM - assert result["step_id"] == "init" - - result = await hass.config_entries.options.async_configure( - result["flow_id"], {CONF_ACTIVITY_WINDOW: 10, CONF_ACTIVITY_THRESHOLD: 5} - ) - await hass.async_block_till_done() - - assert result["type"] is FlowResultType.CREATE_ENTRY - assert entry.options[CONF_ACTIVITY_WINDOW] == 10 - detector = entry.runtime_data.activity - assert detector.window == 600.0 - assert detector.threshold == 819 - - -async def test_a_manual_host_adopts_its_serial( - hass: HomeAssistant, device: FakeDevice, no_setup: AsyncMock -) -> None: - """A hand-typed address is keyed by the serial the device answers with.""" - identified = Found(host="127.0.0.1", name="Studio Profiler", serial="SER123", version="10.5.3") - result = await start(hass, []) - with patch("custom_components.kemper.discovery.async_discover", return_value=[identified]): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_HOST: "127.0.0.1", CONF_PORT: device.port} - ) - - assert result["type"] is FlowResultType.CREATE_ENTRY - assert result["title"] == "Studio Profiler" - assert result["result"].unique_id == "SER123" - assert result["data"][CONF_SERIAL] == "SER123" - assert result["data"][CONF_SW_VERSION] == "10.5.3" - - -async def test_a_silent_device_is_still_keyed_by_its_host( - hass: HomeAssistant, device: FakeDevice, no_setup: AsyncMock -) -> None: - """With the discovery port held, the host is all the identity there is.""" - result = await start(hass, []) - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_HOST: "127.0.0.1", CONF_PORT: device.port} - ) - - assert result["type"] is FlowResultType.CREATE_ENTRY - assert result["result"].unique_id == "127.0.0.1" - assert result["data"][CONF_SERIAL] is None - - -async def test_re_adding_a_moved_device_by_hand_updates_the_entry( - hass: HomeAssistant, device: FakeDevice, no_setup: AsyncMock -) -> None: - """Typing in the new address of a known Profiler moves it, not clones it.""" - existing = MockConfigEntry( - domain=DOMAIN, unique_id="SER123", data={CONF_HOST: "10.0.0.9", CONF_PORT: 5727} - ) - existing.add_to_hass(hass) - identified = Found(host="127.0.0.1", name="Studio Profiler", serial="SER123", version="10.5.3") - - result = await start(hass, []) - with patch("custom_components.kemper.discovery.async_discover", return_value=[identified]): - result = await hass.config_entries.flow.async_configure( - result["flow_id"], {CONF_HOST: "127.0.0.1", CONF_PORT: device.port} - ) - await hass.async_block_till_done() - - assert result["type"] is FlowResultType.ABORT - assert result["reason"] == "already_configured" - assert existing.data[CONF_HOST] == "127.0.0.1" - assert len(hass.config_entries.async_entries(DOMAIN)) == 1 diff --git a/python/examples/homeassistant/tests/test_diagnostics.py b/python/examples/homeassistant/tests/test_diagnostics.py deleted file mode 100644 index 46b7d00..0000000 --- a/python/examples/homeassistant/tests/test_diagnostics.py +++ /dev/null @@ -1,33 +0,0 @@ -"""The diagnostics download: the whole tree, minus the address of the device.""" - -from __future__ import annotations - -from conftest import entity_id, wait_for_state -from fake_device import FakeDevice -from homeassistant.const import CONF_HOST -from homeassistant.core import HomeAssistant -from libkp import _generated as gen -from libkp.nrpn import PAGE_STRINGS, sysex -from pytest_homeassistant_custom_component.common import MockConfigEntry - -from custom_components.kemper.diagnostics import async_get_config_entry_diagnostics - - -async def test_the_dump_is_json_and_redacted( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Enums come out as their values and the host does not come out at all.""" - await device.push( - sysex(0x00, 0x00, 0x03, PAGE_STRINGS, gen.STRING_RIG_NAME, b"Crunchy Vox\x00") - ) - await wait_for_state(hass, entity_id(hass, "sensor", "rig_name"), "Crunchy Vox") - - diagnostics = await async_get_config_entry_diagnostics(hass, entry) - - assert diagnostics["entry"]["data"][CONF_HOST] == "**REDACTED**" - assert diagnostics["state"]["connection"] == "connected" - assert diagnostics["state"]["rig"]["name"] == "Crunchy Vox" - # The parts that never became entities are here, which is the point. - assert len(diagnostics["state"]["effects"]) == 8 - assert len(diagnostics["state"]["status"]["raw"]) == gen.METER_COUNT - assert diagnostics["activity"]["active"] is False diff --git a/python/examples/homeassistant/tests/test_init.py b/python/examples/homeassistant/tests/test_init.py deleted file mode 100644 index dffc69b..0000000 --- a/python/examples/homeassistant/tests/test_init.py +++ /dev/null @@ -1,204 +0,0 @@ -"""Setting the entry up, tearing it down, and what that costs the device.""" - -from __future__ import annotations - -import asyncio -from datetime import timedelta -from unittest.mock import patch - -from conftest import DEVICE_NAME, SERIAL, entity_id, make_entry, wait_until -from fake_device import FakeDevice -from homeassistant.config_entries import ConfigEntryState -from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT -from homeassistant.core import HomeAssistant -from homeassistant.helpers import device_registry as dr -from homeassistant.helpers import entity_registry as er -from homeassistant.util import dt as dt_util -from pytest_homeassistant_custom_component.common import ( - MockConfigEntry, - async_fire_time_changed, -) - -from custom_components.kemper.const import ( - CONF_ACTIVITY_THRESHOLD, - CONF_ACTIVITY_WINDOW, - CONF_SERIAL, - CONF_SW_VERSION, - DOMAIN, -) -from custom_components.kemper.coordinator import RELOAD_DELAY_SECONDS -from custom_components.kemper.discovery import Found -from custom_components.kemper.libkp.session import PROTOCOL_CBOR_CONTROL, PROTOCOL_MIDI3_STREAM - - -async def test_setup_opens_exactly_one_session( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """One stream, and no control channel: the entities need nothing from it.""" - assert entry.state is ConfigEntryState.LOADED - assert device.connection_count(PROTOCOL_MIDI3_STREAM) == 1 - assert device.connection_count(PROTOCOL_CBOR_CONTROL) == 0 - - -async def test_setup_registers_the_device(hass: HomeAssistant, entry: MockConfigEntry) -> None: - """Discovery's name and version reach the device registry.""" - device_entry = dr.async_get(hass).async_get_device(identifiers={(DOMAIN, SERIAL)}) - assert device_entry is not None - assert device_entry.name == DEVICE_NAME - assert device_entry.manufacturer == "Kemper" - assert device_entry.model == "Profiler" - assert device_entry.sw_version == "1.2.3" - - -async def test_unload_hangs_up( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Unloading closes the socket; the fake sees the hangup.""" - assert await hass.config_entries.async_unload(entry.entry_id) - await hass.async_block_till_done() - - assert entry.state is ConfigEntryState.NOT_LOADED - connection = device.connections[0] - async with asyncio.timeout(5): - await connection.closed.wait() - - -async def test_a_device_that_is_not_there_retries_later( - hass: HomeAssistant, device: FakeDevice -) -> None: - """A refused connection is ``ConfigEntryNotReady``, not a hard failure.""" - entry = make_entry(device) - await device.stop() # nothing is listening on that port any more - entry.add_to_hass(hass) - assert not await hass.config_entries.async_setup(entry.entry_id) - await hass.async_block_till_done() - assert entry.state is ConfigEntryState.SETUP_RETRY - - -async def test_changing_options_does_not_reconnect( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """The detector is retuned in place: same model, same socket, new numbers.""" - coordinator = entry.runtime_data - model = coordinator.model - - hass.config_entries.async_update_entry( - entry, options={CONF_ACTIVITY_WINDOW: 1, CONF_ACTIVITY_THRESHOLD: 50} - ) - await hass.async_block_till_done() - - assert entry.state is ConfigEntryState.LOADED - assert entry.runtime_data.model is model - assert device.connection_count(PROTOCOL_MIDI3_STREAM) == 1 - detector = coordinator.activity - assert detector.window == 60.0 - assert detector.threshold == 8192 - - -async def test_the_configured_port_is_the_one_dialed( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """The entry's port, not libkp's default, decides where the model dials.""" - assert entry.data[CONF_PORT] == device.port - assert device.connections - - -async def test_setup_follows_the_serial_to_a_new_address( - hass: HomeAssistant, device: FakeDevice -) -> None: - """The lease moved: the entry knows the serial, so it finds the device again.""" - entry = MockConfigEntry( - domain=DOMAIN, - title=DEVICE_NAME, - unique_id=SERIAL, - data={ - CONF_HOST: "10.0.0.99", # where it used to be - CONF_PORT: device.port, - CONF_NAME: "Old Name", - CONF_SERIAL: SERIAL, - CONF_SW_VERSION: "1.0.0", - }, - ) - entry.add_to_hass(hass) - moved = Found(host="127.0.0.1", name="Studio Profiler", serial=SERIAL, version="10.5.3") - - with patch("custom_components.kemper.discovery.async_discover", return_value=[moved]): - assert await hass.config_entries.async_setup(entry.entry_id) - await hass.async_block_till_done() - - try: - assert entry.state is ConfigEntryState.LOADED - # The entry followed the device, name and firmware version included. - assert entry.data[CONF_HOST] == "127.0.0.1" - assert entry.data[CONF_NAME] == "Studio Profiler" - assert entry.data[CONF_SW_VERSION] == "10.5.3" - # And it is the same device, with the same entities. - assert hass.states.get(entity_id(hass, "sensor", "rig_name")).state != "unavailable" - device_entry = dr.async_get(hass).async_get_device(identifiers={(DOMAIN, SERIAL)}) - assert device_entry is not None - assert device_entry.name == "Studio Profiler" - assert device_entry.sw_version == "10.5.3" - finally: - assert await hass.config_entries.async_unload(entry.entry_id) - await hass.async_block_till_done() - - -async def test_setup_uses_the_stored_host_when_nothing_answers( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """A held discovery port must not stop a Profiler that has not moved.""" - assert entry.state is ConfigEntryState.LOADED - assert entry.data[CONF_HOST] == "127.0.0.1" - assert device.connection_count(PROTOCOL_MIDI3_STREAM) == 1 - - -async def test_a_lost_stream_reloads_the_entry( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Coming back goes through setup, so it starts by finding the device again.""" - coordinator = entry.runtime_data - await device.hangup() - await wait_until(lambda: coordinator.reload_pending) - - # The session was seconds old, so the reload waits rather than spinning. - async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=RELOAD_DELAY_SECONDS + 1)) - await hass.async_block_till_done() - await wait_until(lambda: entry.state is ConfigEntryState.LOADED) - - assert entry.runtime_data is not coordinator - assert device.connection_count(PROTOCOL_MIDI3_STREAM) == 2 - assert hass.states.get(entity_id(hass, "sensor", "rig_name")).state != "unavailable" - - -async def test_unloading_a_lost_entry_cancels_the_reload( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Nothing dials after the entry is gone, and nothing is left armed.""" - coordinator = entry.runtime_data - await device.hangup() - await wait_until(lambda: coordinator.reload_pending) - - assert await hass.config_entries.async_unload(entry.entry_id) - await hass.async_block_till_done() - - assert not coordinator.reload_pending - async_fire_time_changed(hass, dt_util.utcnow() + timedelta(seconds=RELOAD_DELAY_SECONDS + 1)) - await hass.async_block_till_done() - assert entry.state is ConfigEntryState.NOT_LOADED - assert device.connection_count(PROTOCOL_MIDI3_STREAM) == 1 - - -async def test_every_entity_is_keyed_by_the_serial( - hass: HomeAssistant, entry: MockConfigEntry -) -> None: - """Entity identity survives a move, because it never mentions the address.""" - registry = er.async_get(hass) - entities = er.async_entries_for_config_entry(registry, entry.entry_id) - unique_ids = {item.unique_id for item in entities} - assert unique_ids == { - f"{SERIAL}_rig_name", - f"{SERIAL}_amp_name", - f"{SERIAL}_cabinet_name", - f"{SERIAL}_last_activity", - f"{SERIAL}_active", - } diff --git a/python/examples/homeassistant/tests/test_sensor.py b/python/examples/homeassistant/tests/test_sensor.py deleted file mode 100644 index 9924e84..0000000 --- a/python/examples/homeassistant/tests/test_sensor.py +++ /dev/null @@ -1,62 +0,0 @@ -"""The name sensors, driven by what the device actually pushes. - -The messages come from libkp's own message builders, so what these tests put -on the wire is byte-for-byte what a Profiler puts there when a rig is loaded. -""" - -from __future__ import annotations - -from conftest import entity_id, wait_for_state -from fake_device import FakeDevice -from homeassistant.core import HomeAssistant -from libkp import _generated as gen -from libkp.nrpn import PAGE_STRINGS, sysex -from pytest_homeassistant_custom_component.common import MockConfigEntry - -#: The page-0 string tags a rig change pushes, by their spec names. -RIG_NAME = gen.STRING_RIG_NAME -AMP_NAME = gen.STRING_AMP_NAME -CABINET_NAME = gen.STRING_CABINET_NAME - - -def string_tag(number: int, text: str) -> bytes: - """A ``$03`` String Parameter push, as a rig change sends.""" - return sysex(0x00, 0x00, 0x03, PAGE_STRINGS, number, text.encode("ascii") + b"\x00") - - -async def test_the_names_follow_the_device( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Load a rig on the device and the three sensors say what it is.""" - await device.push(string_tag(RIG_NAME, "Crunchy Vox")) - await device.push(string_tag(AMP_NAME, "Vintage Twin")) - await device.push(string_tag(CABINET_NAME, "2x12 Alnico")) - - await wait_for_state(hass, entity_id(hass, "sensor", "rig_name"), "Crunchy Vox") - await wait_for_state(hass, entity_id(hass, "sensor", "amp_name"), "Vintage Twin") - await wait_for_state(hass, entity_id(hass, "sensor", "cabinet_name"), "2x12 Alnico") - - -async def test_a_second_rig_replaces_the_first( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Nothing is cached across rigs: the last push wins.""" - await device.push(string_tag(RIG_NAME, "Crunchy Vox")) - await wait_for_state(hass, entity_id(hass, "sensor", "rig_name"), "Crunchy Vox") - - await device.push(string_tag(RIG_NAME, "Clean Twin")) - await wait_for_state(hass, entity_id(hass, "sensor", "rig_name"), "Clean Twin") - - -async def test_losing_the_stream_makes_the_sensors_unavailable( - hass: HomeAssistant, device: FakeDevice, entry: MockConfigEntry -) -> None: - """Availability follows the connection, not the last value seen.""" - await device.push(string_tag(RIG_NAME, "Crunchy Vox")) - rig = entity_id(hass, "sensor", "rig_name") - await wait_for_state(hass, rig, "Crunchy Vox") - - await device.hangup() - - await wait_for_state(hass, rig, "unavailable") - await wait_for_state(hass, entity_id(hass, "sensor", "last_activity"), "unavailable") diff --git a/python/src/libkp/__init__.py b/python/src/libkp/__init__.py index 2143a47..48199c7 100644 --- a/python/src/libkp/__init__.py +++ b/python/src/libkp/__init__.py @@ -16,6 +16,9 @@ - :mod:`libkp.model` — :class:`~libkp.model.DeviceModel`, the async store over the stream and the control link. +Beside the layers, :mod:`libkp.testing` holds :class:`~libkp.testing.FakeDevice`, +an in-process Profiler for driving all of the above without a device. + Constants and lookup tables come from :mod:`libkp._generated`, which is emitted from the shared spec; the protocol logic is hand-written here and held to the shared conformance vectors. diff --git a/python/tests/fake_device.py b/python/src/libkp/testing.py similarity index 97% rename from python/tests/fake_device.py rename to python/src/libkp/testing.py index d047393..dad3f3d 100644 --- a/python/tests/fake_device.py +++ b/python/src/libkp/testing.py @@ -1,4 +1,9 @@ -"""An in-process stand-in for a Profiler, for exercising the async layers. +"""An in-process stand-in for a Profiler, for tests that want a real session. + +Shipped with the package so that anything built on libkp -- a Home Assistant +integration, a controller, a script -- can drive the real transport in its own +test suite without a device on the desk and without mocking the layers under +test. libkp's own async tests use it as it stands. It speaks just enough of the transport to drive :class:`libkp.session.Session`, :class:`libkp.model.DeviceModel` and the CBOR tooling: any number of concurrent @@ -34,9 +39,9 @@ import asyncio from collections.abc import Callable, Iterable -from libkp import _generated as gen -from libkp import cbor, midi3, nrpn -from libkp.session import ( +from . import _generated as gen +from . import cbor, midi3, nrpn +from .session import ( PROTOCOL_CBOR_CONTROL, PROTOCOL_MIDI3_STREAM, PROTOCOL_RESERVED, diff --git a/python/tests/test_cbor.py b/python/tests/test_cbor.py index ca37e48..768d779 100644 --- a/python/tests/test_cbor.py +++ b/python/tests/test_cbor.py @@ -5,12 +5,11 @@ import asyncio -from fake_device import DEFAULT_DUMP, FakeDevice, wait_for - from libkp import _generated as gen from libkp import cbor from libkp.session import PROTOCOL_CBOR_CONTROL from libkp.state import Num, Text +from libkp.testing import DEFAULT_DUMP, FakeDevice, wait_for def test_encodes_with_minimal_length_heads(): diff --git a/python/tests/test_meters_example.py b/python/tests/test_meters_example.py index b415309..7f63a48 100644 --- a/python/tests/test_meters_example.py +++ b/python/tests/test_meters_example.py @@ -10,11 +10,11 @@ import meters import pytest -from fake_device import FakeDevice from libkp import _generated as gen from libkp.nrpn import PAGE_STRINGS, set_single, sysex, u14_split from libkp.state import BeatPulse, DeviceState, ParamChanged, RealtimeStatus, Status +from libkp.testing import FakeDevice ANSI = re.compile(r"\x1b\[[0-9;?]*[a-zA-Z]") diff --git a/python/tests/test_model.py b/python/tests/test_model.py index 7f6f362..a4ec405 100644 --- a/python/tests/test_model.py +++ b/python/tests/test_model.py @@ -12,7 +12,6 @@ import asyncio import pytest -from fake_device import FakeDevice, answer_requests, ext_param, wait_for from libkp import _generated as gen from libkp import cbor @@ -57,6 +56,7 @@ SyncCompleted, TempoBpm, ) +from libkp.testing import FakeDevice, answer_requests, ext_param, wait_for RIG_NAME = sysex(0x00, 0x00, 0x03, PAGE_STRINGS, 1, b"Test Rig\x00") REV_TYPE = set_single(0x00, 0x00, 0x3D, 0, 179) diff --git a/python/tests/test_session.py b/python/tests/test_session.py index ddd613d..9d729ba 100644 --- a/python/tests/test_session.py +++ b/python/tests/test_session.py @@ -12,7 +12,6 @@ import asyncio import pytest -from fake_device import FakeDevice from libkp.errors import ConnectError, ProtocolRejectedError, TimeoutErrorLibKP from libkp.session import ( @@ -24,6 +23,7 @@ Session, parse_protocol_list, ) +from libkp.testing import FakeDevice IDLE = 0.2