From 44cc4ca303ea106e1806982cc5ef69575cb247af Mon Sep 17 00:00:00 2001 From: Aaron Gotwalt Date: Thu, 27 Aug 2026 08:15:40 -0700 Subject: [PATCH] Add a Home Assistant integration, and move the Python examples beside the library MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Python examples lived inside the package (`libkp.examples`), which is where neither the Rust (`rust/examples/`) nor the Swift (separate executable targets) examples live, and which would have shipped a Home Assistant integration in the wheel. They now sit in `python/examples/`: `uv run examples/meters.py`, `uv run --extra tui examples/meters_tui.py`. The console scripts are gone with them; the tests import the examples off a second pytest path. `python/examples/homeassistant/` is a custom integration (`custom_components/ kemper`) over the model, and the third front-end on the library after the two terminal views. It exposes only the slow lane — the rig, amp and cabinet names — plus one derived reading: an `active` binary sensor with a `last_activity` timestamp, computed from the meter stream inside the integration so Home Assistant never sees the 20 Hz frames. The detector is a per-frame integer compare and one lazily armed timer: two state writes per playing session however long it runs, with the quiet window and the level threshold as options that apply without reconnecting. Identity is the serial from discovery, so the device and its entities survive a DHCP lease change: setup asks the network where the serial is before dialing and follows it, a lost stream reloads the entry so the way back starts at discovery, and a manually entered host is identified with a directed poll so it is serial-keyed too. The CBOR control channel is left off — nothing here surfaces the morph, and it would cost the device a second socket for as long as Home Assistant runs. libkp is pure standard library, so it is vendored: the integration directory holds `libkp` as a symlink to the library and `build.py` dereferences it into `dist/custom_components/kemper/` (and a zip), which is the whole install. Tests run against libkp's `FakeDevice` under `pytest-homeassistant-custom-component`; CI gets a `homeassistant` job. Verified against a Profiler Player on 14.2.1, including the case where the stored address was stale and the serial answered from a new one. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01KA85wBck75RTmQZ7SioUHV --- .github/workflows/ci.yml | 18 ++ README.md | 2 +- docs/01-overview.md | 2 +- docs/07-realtime-status.md | 2 +- python/README.md | 37 ++- python/examples/README.md | 39 ++++ python/examples/homeassistant/README.md | 119 ++++++++++ python/examples/homeassistant/build.py | 96 ++++++++ .../custom_components/kemper/__init__.py | 130 +++++++++++ .../custom_components/kemper/activity.py | 199 ++++++++++++++++ .../custom_components/kemper/binary_sensor.py | 43 ++++ .../custom_components/kemper/config_flow.py | 218 ++++++++++++++++++ .../custom_components/kemper/const.py | 31 +++ .../custom_components/kemper/coordinator.py | 191 +++++++++++++++ .../custom_components/kemper/diagnostics.py | 60 +++++ .../custom_components/kemper/discovery.py | 82 +++++++ .../custom_components/kemper/entity.py | 30 +++ .../custom_components/kemper/icons.json | 27 +++ .../custom_components/kemper/libkp | 1 + .../custom_components/kemper/manifest.json | 13 ++ .../custom_components/kemper/sensor.py | 102 ++++++++ .../custom_components/kemper/strings.json | 69 ++++++ .../kemper/translations/en.json | 69 ++++++ python/examples/homeassistant/pyproject.toml | 33 +++ .../examples/homeassistant/tests/conftest.py | 129 +++++++++++ .../homeassistant/tests/test_binary_sensor.py | 161 +++++++++++++ .../homeassistant/tests/test_build.py | 75 ++++++ .../homeassistant/tests/test_config_flow.py | 201 ++++++++++++++++ .../homeassistant/tests/test_diagnostics.py | 33 +++ .../examples/homeassistant/tests/test_init.py | 204 ++++++++++++++++ .../homeassistant/tests/test_sensor.py | 62 +++++ python/{src/libkp => }/examples/meters.py | 16 +- python/{src/libkp => }/examples/meters_tui.py | 25 +- python/pyproject.toml | 6 +- python/src/libkp/examples/__init__.py | 7 - python/tests/test_meters_example.py | 2 +- python/tests/test_meters_tui.py | 3 +- 37 files changed, 2477 insertions(+), 60 deletions(-) create mode 100644 python/examples/README.md create mode 100644 python/examples/homeassistant/README.md create mode 100644 python/examples/homeassistant/build.py create mode 100644 python/examples/homeassistant/custom_components/kemper/__init__.py create mode 100644 python/examples/homeassistant/custom_components/kemper/activity.py create mode 100644 python/examples/homeassistant/custom_components/kemper/binary_sensor.py create mode 100644 python/examples/homeassistant/custom_components/kemper/config_flow.py create mode 100644 python/examples/homeassistant/custom_components/kemper/const.py create mode 100644 python/examples/homeassistant/custom_components/kemper/coordinator.py create mode 100644 python/examples/homeassistant/custom_components/kemper/diagnostics.py create mode 100644 python/examples/homeassistant/custom_components/kemper/discovery.py create mode 100644 python/examples/homeassistant/custom_components/kemper/entity.py create mode 100644 python/examples/homeassistant/custom_components/kemper/icons.json create mode 120000 python/examples/homeassistant/custom_components/kemper/libkp create mode 100644 python/examples/homeassistant/custom_components/kemper/manifest.json create mode 100644 python/examples/homeassistant/custom_components/kemper/sensor.py create mode 100644 python/examples/homeassistant/custom_components/kemper/strings.json create mode 100644 python/examples/homeassistant/custom_components/kemper/translations/en.json create mode 100644 python/examples/homeassistant/pyproject.toml create mode 100644 python/examples/homeassistant/tests/conftest.py create mode 100644 python/examples/homeassistant/tests/test_binary_sensor.py create mode 100644 python/examples/homeassistant/tests/test_build.py create mode 100644 python/examples/homeassistant/tests/test_config_flow.py create mode 100644 python/examples/homeassistant/tests/test_diagnostics.py create mode 100644 python/examples/homeassistant/tests/test_init.py create mode 100644 python/examples/homeassistant/tests/test_sensor.py rename python/{src/libkp => }/examples/meters.py (98%) rename python/{src/libkp => }/examples/meters_tui.py (97%) delete mode 100644 python/src/libkp/examples/__init__.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6c65589..21404b2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -70,6 +70,24 @@ 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/README.md b/README.md index 04d3e3e..51f5a19 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,7 @@ of the current rig, its effect blocks, the tuner strobe, and the output meters: ```sh cd rust && cargo run --example meters # add `-- --ip 192.168.1.50` to skip discovery -cd python && python -m libkp.examples.meters # or `pip install -e .` then `libkp-meters` +cd python && uv run examples/meters.py # or `pip install -e .` then `python examples/meters.py` cd swift && swift run meters # or `swift run MetersApp` for the macOS app ``` diff --git a/docs/01-overview.md b/docs/01-overview.md index 488cad6..7a4ea9f 100644 --- a/docs/01-overview.md +++ b/docs/01-overview.md @@ -151,7 +151,7 @@ Even with no requests outstanding, an open MIDI3 stream delivers: comment, amp, cabinet, every effect-slot type and state, and the rig settings. The `meters` example in each language (`rust/examples`, -`python/src/libkp/examples`, `swift/Sources/meters`) does exactly this and +`python/examples`, `swift/Sources/meters`) does exactly this and nothing more: connect, subscribe, and render what arrives. ## The model diff --git a/docs/07-realtime-status.md b/docs/07-realtime-status.md index 5450b79..a47f352 100644 --- a/docs/07-realtime-status.md +++ b/docs/07-realtime-status.md @@ -189,7 +189,7 @@ The note being tracked is reported separately at page `$7D`, number 84 ## The `meters` example Each implementation ships the same example — `rust/examples`, -`python/src/libkp/examples`, `swift/Sources/meters` — and it exists to make this +`python/examples`, `swift/Sources/meters` — and it exists to make this document concrete. It opens a session, ingests the stream, and renders a live terminal view: diff --git a/python/README.md b/python/README.md index 3531ae5..a25d2a8 100644 --- a/python/README.md +++ b/python/README.md @@ -20,15 +20,14 @@ pip install -e '.[dev]' # from python/ Or run straight from the source tree: ```sh -PYTHONPATH=src python -m libkp.examples.meters --help +PYTHONPATH=src python examples/meters.py --help ``` ## The live meters example ```sh -python -m libkp.examples.meters # discover a device, then render -python -m libkp.examples.meters --ip 192.168.1.50 --all --width 48 -libkp-meters --help # same thing, installed as a script +python examples/meters.py # discover a device, then render +python examples/meters.py --ip 192.168.1.50 --all --width 48 ``` A full-screen ANSI view that updates straight off the stream: @@ -63,8 +62,8 @@ Ctrl-C restores the cursor and exits. gauges, an effect-block grid, a live tuner strobe) built on [Textual](https://textual.textualize.io/). It needs the optional `tui` extra (Textual); the library itself stays dependency-free. Run it with `uv` (below), -or `pip install -e '.[tui]'` then `libkp-meters-tui`. Same core flags as the -ANSI example; press `q` to quit and `a` to toggle the raw fields. +or `pip install -e '.[tui]'` then `python examples/meters_tui.py`. Same core +flags as the ANSI example; press `q` to quit and `a` to toggle the raw fields. ## Running the examples with uv @@ -73,33 +72,25 @@ virtualenv — it resolves the package (and any extras) on the fly. From `python ```sh # Zero-dependency ANSI meters -uv run libkp-meters --help -uv run libkp-meters --ip 10.0.0.1 --all +uv run examples/meters.py --help +uv run examples/meters.py --ip 10.0.0.1 --all # Textual TUI — the `--extra tui` pulls in Textual just for this run -uv run --extra tui libkp-meters-tui -uv run --extra tui libkp-meters-tui --ip 10.0.0.1 -``` - -The console scripts above come from `pyproject.toml`; the equivalent module form -works too: - -```sh -uv run python -m libkp.examples.meters -uv run --extra tui python -m libkp.examples.meters_tui +uv run --extra tui examples/meters_tui.py +uv run --extra tui examples/meters_tui.py --ip 10.0.0.1 ``` Pin the interpreter (anything 3.11+) when you want a specific one, and materialize the environment once for repeated runs or editor tooling: ```sh -uv run --python 3.14 --extra tui libkp-meters-tui # choose the interpreter -uv sync --extra tui # create .venv with Textual -uv run libkp-meters-tui # then reuse it +uv run --python 3.14 --extra tui examples/meters_tui.py # choose the interpreter +uv sync --extra tui # create .venv with Textual +uv run examples/meters_tui.py # then reuse it ``` -Only the Textual example needs the `tui` extra; `uv run libkp-meters` needs -nothing beyond the standard library. +Only the Textual example needs the `tui` extra; `uv run examples/meters.py` +needs nothing beyond the standard library. ## Quick start diff --git a/python/examples/README.md b/python/examples/README.md new file mode 100644 index 0000000..b46eefa --- /dev/null +++ b/python/examples/README.md @@ -0,0 +1,39 @@ +# libkp Python examples + +Runnable front-ends over the `libkp` library, kept beside it rather than inside +it (mirroring `../../rust/examples` and Swift's separate executable targets) so +neither ships in the installed package. + +- **`meters.py`** — a zero-dependency, full-screen ANSI view of the current + rig, amp/cab, the eight effect blocks, the tuner strobe, and the realtime + level meters. +- **`meters_tui.py`** — the same view as a richer [Textual](https://textual.textualize.io/) + TUI (rounded panels, colored gauges). Needs the optional `tui` extra. +- **`homeassistant/`** — a Home Assistant custom integration: the loaded rig, + amp and cabinet as sensors, plus an `active` binary sensor that turns the + ~20 Hz meter lane into two state writes per playing session. Its own uv + project, with its own README, tests and bundler. + +## Running them + +From `python/`, with [uv](https://docs.astral.sh/uv/) (no manual virtualenv +needed — it resolves `libkp` and any extras on the fly): + +```sh +uv run examples/meters.py --help +uv run examples/meters.py --ip 192.168.1.50 --all --width 48 + +uv run --extra tui examples/meters_tui.py --help +uv run --extra tui examples/meters_tui.py --ip 192.168.1.50 +``` + +Or, after `pip install -e .` (add `'.[tui]'` for the Textual example): + +```sh +python examples/meters.py --help +python examples/meters_tui.py --help +``` + +Both discover a device over UDP broadcast when `--ip` is omitted; quit +Kemper's Rig Manager first if it holds the discovery port. Ctrl-C quits either +one and restores the terminal. diff --git a/python/examples/homeassistant/README.md b/python/examples/homeassistant/README.md new file mode 100644 index 0000000..42a339e --- /dev/null +++ b/python/examples/homeassistant/README.md @@ -0,0 +1,119 @@ +# 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 new file mode 100644 index 0000000..b36cb28 --- /dev/null +++ b/python/examples/homeassistant/build.py @@ -0,0 +1,96 @@ +#!/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 new file mode 100644 index 0000000..dcaceb1 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/__init__.py @@ -0,0 +1,130 @@ +"""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 new file mode 100644 index 0000000..0d7f127 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/activity.py @@ -0,0 +1,199 @@ +"""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 new file mode 100644 index 0000000..f8275ab --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/binary_sensor.py @@ -0,0 +1,43 @@ +"""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 new file mode 100644 index 0000000..2a5212d --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/config_flow.py @@ -0,0 +1,218 @@ +"""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 new file mode 100644 index 0000000..248d8e6 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/const.py @@ -0,0 +1,31 @@ +"""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 new file mode 100644 index 0000000..1193b7c --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/coordinator.py @@ -0,0 +1,191 @@ +"""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 new file mode 100644 index 0000000..40ff4e9 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/diagnostics.py @@ -0,0 +1,60 @@ +"""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 new file mode 100644 index 0000000..026ec39 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/discovery.py @@ -0,0 +1,82 @@ +"""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 new file mode 100644 index 0000000..436a346 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/entity.py @@ -0,0 +1,30 @@ +"""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 new file mode 100644 index 0000000..14944be --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/icons.json @@ -0,0 +1,27 @@ +{ + "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 new file mode 120000 index 0000000..454fbc0 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/libkp @@ -0,0 +1 @@ +../../../../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 new file mode 100644 index 0000000..c9c0fc6 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/manifest.json @@ -0,0 +1,13 @@ +{ + "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 new file mode 100644 index 0000000..5fb2f41 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/sensor.py @@ -0,0 +1,102 @@ +"""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 new file mode 100644 index 0000000..380af65 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/strings.json @@ -0,0 +1,69 @@ +{ + "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 new file mode 100644 index 0000000..380af65 --- /dev/null +++ b/python/examples/homeassistant/custom_components/kemper/translations/en.json @@ -0,0 +1,69 @@ +{ + "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 new file mode 100644 index 0000000..51216e9 --- /dev/null +++ b/python/examples/homeassistant/pyproject.toml @@ -0,0 +1,33 @@ +[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 new file mode 100644 index 0000000..f383d09 --- /dev/null +++ b/python/examples/homeassistant/tests/conftest.py @@ -0,0 +1,129 @@ +"""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 new file mode 100644 index 0000000..2d6dd9c --- /dev/null +++ b/python/examples/homeassistant/tests/test_binary_sensor.py @@ -0,0 +1,161 @@ +"""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 new file mode 100644 index 0000000..f8c2450 --- /dev/null +++ b/python/examples/homeassistant/tests/test_build.py @@ -0,0 +1,75 @@ +"""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 new file mode 100644 index 0000000..b9ba946 --- /dev/null +++ b/python/examples/homeassistant/tests/test_config_flow.py @@ -0,0 +1,201 @@ +"""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 new file mode 100644 index 0000000..46b7d00 --- /dev/null +++ b/python/examples/homeassistant/tests/test_diagnostics.py @@ -0,0 +1,33 @@ +"""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 new file mode 100644 index 0000000..dffc69b --- /dev/null +++ b/python/examples/homeassistant/tests/test_init.py @@ -0,0 +1,204 @@ +"""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 new file mode 100644 index 0000000..9924e84 --- /dev/null +++ b/python/examples/homeassistant/tests/test_sensor.py @@ -0,0 +1,62 @@ +"""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/examples/meters.py b/python/examples/meters.py similarity index 98% rename from python/src/libkp/examples/meters.py rename to python/examples/meters.py index be3ed70..aeaa910 100644 --- a/python/src/libkp/examples/meters.py +++ b/python/examples/meters.py @@ -12,7 +12,7 @@ - level bars for stack, rig output and loudness, with peak-hold, - a tempo pulse indicator and the last parameter seen. -Run it with ``python -m libkp.examples.meters`` (Ctrl-C quits and restores the +Run it with ``uv run examples/meters.py`` (Ctrl-C quits and restores the terminal). The realtime field identities it renders come from observed experimentation and are described by the shared spec. """ @@ -28,12 +28,12 @@ from collections import deque from dataclasses import dataclass, field -from .. import _generated as gen -from .. import params -from ..discovery import find_first -from ..errors import LibKPError -from ..model import DeviceModel -from ..state import ( +from libkp import _generated as gen +from libkp import params +from libkp.discovery import find_first +from libkp.errors import LibKPError +from libkp.model import DeviceModel +from libkp.state import ( BeatPulse, DeviceEvent, DeviceState, @@ -409,7 +409,7 @@ async def run(args: argparse.Namespace) -> int: def build_parser() -> argparse.ArgumentParser: """The command-line interface.""" parser = argparse.ArgumentParser( - prog="python -m libkp.examples.meters", + prog="examples/meters.py", description=( "Live terminal view of a Kemper Profiler: rig, amp/cab, the eight " "effect blocks, the tuner strobe, and the realtime level meters." diff --git a/python/src/libkp/examples/meters_tui.py b/python/examples/meters_tui.py similarity index 97% rename from python/src/libkp/examples/meters_tui.py rename to python/examples/meters_tui.py index f6cfd4e..9ee1657 100644 --- a/python/src/libkp/examples/meters_tui.py +++ b/python/examples/meters_tui.py @@ -1,7 +1,7 @@ """A polished **Textual** live view of the Profiler's current patch and stream. This is the rich, widget-based counterpart to the zero-dependency -:mod:`libkp.examples.meters` ANSI view. It connects (discovering the device if +:mod:`meters` ANSI view. It connects (discovering the device if no ``--ip`` is given) with the model's default options -- the stream, the control link that carries the morph, and the read-only request burst -- then drives a full-screen Textual UI straight off the async @@ -21,12 +21,12 @@ :func:`main` so importing or collecting the package never requires it. Install it with ``pip install 'libkp[tui]'`` and run:: - python -m libkp.examples.meters_tui --ip 10.0.0.1 - python -m libkp.examples.meters_tui --all --fps 60 + uv run --extra tui examples/meters_tui.py --ip 10.0.0.1 + uv run --extra tui examples/meters_tui.py --all --fps 60 The reusable view logic — the wrap-aware strobe drift-rate verdict, the peak-hold decay, the tempo-pulse handling and the field labels — is shared with -:mod:`libkp.examples.meters`. The realtime field identities come from observed +:mod:`meters`. The realtime field identities come from observed experimentation and are described by the shared spec. Press ``q`` / ``Ctrl-C`` to quit and ``a`` to toggle the full set of raw meter fields. """ @@ -38,12 +38,7 @@ import sys import time -from .. import _generated as gen -from ..discovery import find_first -from ..errors import LibKPError -from ..model import DeviceModel -from ..state import Connection, DeviceState -from .meters import ( +from meters import ( ALL_ROWS, BAR_ROWS, FPS, @@ -54,6 +49,12 @@ note_name, ) +from libkp import _generated as gen +from libkp.discovery import find_first +from libkp.errors import LibKPError +from libkp.model import DeviceModel +from libkp.state import Connection, DeviceState + # --- shared, textual-free view helpers --------------------------------------- #: Accent color for the panel frames and titles (indigo-400). @@ -464,9 +465,9 @@ def _footer(self, app: MetersApp) -> Group: def build_parser() -> argparse.ArgumentParser: - """The command-line interface (mirrors :mod:`libkp.examples.meters`).""" + """The command-line interface (mirrors :mod:`meters`).""" parser = argparse.ArgumentParser( - prog="python -m libkp.examples.meters_tui", + prog="examples/meters_tui.py", description=( "Polished Textual live view of a Kemper Profiler: rig, amp/cab, the " "eight effect blocks, the tuner strobe, and the realtime meters." diff --git a/python/pyproject.toml b/python/pyproject.toml index 0bde614..c71d7eb 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -17,13 +17,9 @@ dependencies = [] dev = ["pytest>=8", "ruff>=0.6"] tui = ["textual>=0.60"] -[project.scripts] -libkp-meters = "libkp.examples.meters:main" -libkp-meters-tui = "libkp.examples.meters_tui:main" - [tool.hatch.build.targets.wheel] packages = ["src/libkp"] [tool.pytest.ini_options] testpaths = ["tests"] -pythonpath = ["src"] +pythonpath = ["src", "examples"] diff --git a/python/src/libkp/examples/__init__.py b/python/src/libkp/examples/__init__.py deleted file mode 100644 index d245cc4..0000000 --- a/python/src/libkp/examples/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -"""Runnable examples built on the :mod:`libkp` public API. - -- :mod:`libkp.examples.meters` — a live full-screen terminal view of the current - patch, effect blocks, tuner strobe, and level meters. - -Run one with ``python -m libkp.examples.meters --help``. -""" diff --git a/python/tests/test_meters_example.py b/python/tests/test_meters_example.py index a12d113..b415309 100644 --- a/python/tests/test_meters_example.py +++ b/python/tests/test_meters_example.py @@ -8,11 +8,11 @@ import re import sys +import meters import pytest from fake_device import FakeDevice from libkp import _generated as gen -from libkp.examples import meters from libkp.nrpn import PAGE_STRINGS, set_single, sysex, u14_split from libkp.state import BeatPulse, DeviceState, ParamChanged, RealtimeStatus, Status diff --git a/python/tests/test_meters_tui.py b/python/tests/test_meters_tui.py index 0aed026..a3793f2 100644 --- a/python/tests/test_meters_tui.py +++ b/python/tests/test_meters_tui.py @@ -6,10 +6,11 @@ from __future__ import annotations +import meters +import meters_tui import pytest from libkp import _generated as gen -from libkp.examples import meters, meters_tui from libkp.state import RealtimeStatus # ---------------------------------------------------------------------------