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 # ---------------------------------------------------------------------------