Fix five correctness bugs in the Python wrapper

python/src/libkp/__init__.py

@@ -54,6 +54,7 @@ async def main():
     ParseError,
     ProtocolRejectedError,
     SessionError,
+    TimeoutErrorLibKP,
     TooShortError,
     UnknownSlotError,
 )
@@ -214,6 +215,7 @@ async def main():
     "ConnectError",
     "ConnectionClosedError",
     "ProtocolRejectedError",
+    "TimeoutErrorLibKP",
     "CommandError",
     "DisconnectedError",
     "UnknownSlotError",

python/src/libkp/control.py

@@ -23,6 +23,7 @@
 from typing import Any, ClassVar
 
 from . import _generated as gen
+from .errors import UnknownSlotError
 from .nrpn import control_change, program_change
 
 __all__ = [
@@ -212,13 +213,18 @@ class ModuleSlot(str, Enum):
 
     @classmethod
     def parse(cls, name: str | ModuleSlot) -> ModuleSlot:
-        """Resolve a slot name case-insensitively (``"rev"``, ``"Dly"``, …)."""
+        """Resolve a slot name case-insensitively (``"rev"``, ``"Dly"``, …).
+
+        Raises :class:`~libkp.errors.UnknownSlotError` — a
+        :class:`~libkp.errors.LibKPError` — for a name outside the eight slots,
+        matching what :meth:`libkp.model.DeviceModel.set_effect_enabled` raises.
+        """
         if isinstance(name, cls):
             return name
         try:
             return cls(str(name).upper())
-        except ValueError as exc:  # pragma: no cover - defensive
-            raise ValueError(f"unknown effect slot {name!r}") from exc
+        except ValueError as exc:
+            raise UnknownSlotError(str(name)) from exc
 
 
 def slot_enable_cc(slot: str | ModuleSlot) -> int:

python/src/libkp/examples/meters.py

@@ -77,6 +77,11 @@
 PULSE_FLASH_SECS = 0.15
 #: Label column width in the meter rows.
 LABEL_WIDTH = 19
+#: Columns a meter row spends on everything except the bar itself: the two-space
+#: indent, the space after the label, the bar's two brackets, two spaces, the
+#: five-column value, two more spaces, and the 17-column ``range …`` readout.
+#: Added to :data:`LABEL_WIDTH` this is the full non-bar width of a row.
+ROW_CHROME = 2 + 1 + 2 + 2 + 5 + 2 + 17
 
 FULL_SCALE = float(gen.FULL_SCALE)
 PHASE_SPAN = gen.FULL_SCALE + 1
@@ -357,14 +362,16 @@ async def run(args: argparse.Namespace) -> int:
     width = min(max(args.width, 8), 512)
     print(f"Connecting to {ip}...", file=sys.stderr)
 
-    model = await DeviceModel.connect(ip, args.port)
-    view = MeterView(ip=ip, width=width, show_all=args.all)
-    events = model.events()
-
     out = sys.stdout
-    out.write(CLEAR + HIDE_CURSOR)
-    out.flush()
+    model = await DeviceModel.connect(ip, args.port)
+    # Everything past the connect goes through the try, so a failure while
+    # setting up the terminal still closes the socket and the ingest tasks.
     try:
+        view = MeterView(ip=ip, width=width, show_all=args.all)
+        events = model.events()
+
+        out.write(CLEAR + HIDE_CURSOR)
+        out.flush()
         frame_time = 1.0 / max(args.fps, 1.0)
         loop = asyncio.get_running_loop()
         while True:
@@ -431,12 +438,17 @@ def build_parser() -> argparse.ArgumentParser:
 
 
 def _default_width() -> int:
-    """A bar width that fits the terminal, with a sane fallback when piped."""
+    """A bar width that fits the terminal, with a sane fallback when piped.
+
+    A meter row is ``LABEL_WIDTH + ROW_CHROME`` columns wide before the bar, so
+    the bar has to stop one column short of that to leave the right margin
+    clear — otherwise every row wraps and the full-screen frame tears.
+    """
     try:
         columns = os.get_terminal_size().columns
     except OSError:
         columns = 100
-    return min(max(columns - LABEL_WIDTH - 26, 12), 60)
+    return min(max(columns - LABEL_WIDTH - ROW_CHROME - 1, 12), 60)
 
 
 def main(argv: list[str] | None = None) -> int:

python/src/libkp/examples/meters_tui.py

@@ -104,6 +104,20 @@ def _short_label(label: str) -> str:
     return label.replace("_", " ")
 
 
+#: Width of the meters table's label column. Derived from the widest label the
+#: table can ever show — the ``--all`` set carries a ``vN`` prefix, so a fixed
+#: guess that fits the bar rows wraps the raw ones onto a second line.
+LABEL_WIDTH = max(len(_short_label(label)) for label, _ident in ALL_ROWS)
+
+#: Columns a meter row spends on everything except the bar itself: the label
+#: column, the 6-column value, the 17-column ``range …`` readout, the bar's two
+#: brackets, the three single-space gaps the grid puts between the four
+#: columns, and the panel's border plus horizontal padding. One extra column is
+#: left spare so a full-scale bar never reaches the right edge — Rich crops an
+#: over-wide cell with an ellipsis rather than wrapping it.
+METER_ROW_CHROME = LABEL_WIDTH + 6 + 17 + 2 + 3 + 4 + 1
+
+
 # --- the run loop ------------------------------------------------------------
 
 
@@ -290,7 +304,7 @@ def action_toggle_raw(self) -> None:
         # -- panel builders (pure over view + snapshot) --------------------
 
         def _bar_width(self) -> int:
-            return min(max(self.size.width - 34, 12), 72)
+            return min(max(self.size.width - METER_ROW_CHROME, 12), 72)
 
         def _rig(self, app: MetersApp) -> Group:
             state: DeviceState = self.view.snapshot
@@ -397,7 +411,7 @@ def _meters(self, app: MetersApp) -> Table:
             width = self._bar_width()
             rows = ALL_ROWS if view.show_all else BAR_ROWS
             table = Table.grid(padding=(0, 1))
-            table.add_column(justify="left", style="grey50", width=16)
+            table.add_column(justify="left", style="grey50", width=LABEL_WIDTH)
             table.add_column()
             table.add_column(justify="right", width=6)
             table.add_column(justify="left", style="grey37")

python/src/libkp/nrpn.py

@@ -112,10 +112,22 @@ def sysex(
 
     ``F0 00 20 33 <product> <device> <function> <instance=0> <page> <number>
     <values…> F7``
+
+    Every header byte is a 7-bit SysEx data byte. An out-of-range one raises
+    :class:`ValueError` rather than being masked: masking would silently
+    retarget the message at a different address (``page=0x80`` would become
+    page 0, the string/morph page) instead of surfacing the mistake.
     """
+    header = (product, device, function, page, number)
+    if any(not 0 <= byte <= 0x7F for byte in header):
+        names = ("product", "device", "function", "page", "number")
+        bad = ", ".join(
+            f"{n}={v}" for n, v in zip(names, header, strict=True) if not 0 <= v <= 0x7F
+        )
+        raise ValueError(f"SysEx header bytes must be 7-bit (0-127): {bad}")
     out = bytearray([0xF0])
     out.extend(MANUFACTURER_ID)
-    out.extend([product & 0x7F, device & 0x7F, function & 0x7F, 0x00, page & 0x7F, number & 0x7F])
+    out.extend([product, device, function, 0x00, page, number])
     out.extend(bytes(values))
     out.append(0xF7)
     return bytes(out)

python/tests/test_conformance.py

@@ -29,6 +29,17 @@ def test_every_vector_file_is_covered():
     assert present == covered, f"uncovered vector files: {sorted(present - covered)}"
 
 
+def test_no_vector_case_list_is_empty():
+    """An emptied case list would make its ``parametrize`` collect nothing and pass."""
+    empty: list[str] = []
+    for path in sorted(VECTORS_DIR.glob("*.json")):
+        document = vector(path.stem)
+        for key, value in document.items():
+            if isinstance(value, list) and not value:
+                empty.append(f"{path.stem}.{key}")
+    assert empty == [], f"empty vector case lists: {empty}"
+
+
 # ---------------------------------------------------------------------------
 # u14.json
 # ---------------------------------------------------------------------------

python/tests/test_control.py

@@ -5,6 +5,7 @@
 import pytest
 
 from libkp import control as c
+from libkp.errors import UnknownSlotError
 
 
 def test_continuous_controller_bytes():
@@ -108,3 +109,15 @@ def test_default_channel_is_zero():
 
 def test_distinct_control_types_do_not_compare_equal():
     assert c.Gain(64) != c.DelayMix(64)
+
+
+def test_unknown_slot_raises_a_libkp_error():
+    """Every slot-name entry point rejects a bad name the same way."""
+    for call in (
+        lambda: c.slot_enable_cc("nope"),
+        lambda: c.ModuleSlot.parse("nope"),
+        lambda: c.SlotEnable("nope", True).message(0),
+        lambda: c.control_from_op("slot_enable", slot="nope", on=True).message(0),
+    ):
+        with pytest.raises(UnknownSlotError):
+            call()

python/tests/test_meters_example.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import io
+import os
 import re
 import sys
 
@@ -186,6 +187,24 @@ def test_render_shows_the_patch_header_blocks_and_bars():
     assert "last param:" in text
 
 
+@pytest.mark.parametrize("columns", [72, 80, 100, 120, 160])
+def test_default_width_keeps_every_row_inside_the_terminal(monkeypatch, columns):
+    """The default bar width must leave the right margin clear.
+
+    A meter row is ``LABEL_WIDTH + ROW_CHROME`` columns wide before the bar, so
+    an over-generous default makes every row wrap and the full-screen frame
+    tears apart.
+    """
+    monkeypatch.setattr(meters.os, "get_terminal_size", lambda *_a: os.terminal_size((columns, 40)))
+    view = meters.MeterView(ip="127.0.0.1", width=meters._default_width(), show_all=True)
+    view.on_event(Status(status([gen.FULL_SCALE] * gen.METER_COUNT)))
+
+    frame = plain(meters.render(view, DeviceState()))
+    rows = [line for line in frame.split("\n") if "range" in line]
+    assert len(rows) == gen.METER_COUNT
+    assert max(len(row) for row in rows) < columns
+
+
 def test_render_all_shows_every_raw_field():
     view = meters.MeterView(ip="127.0.0.1", width=16, show_all=True)
     text = plain(meters.render(view, DeviceState()))

python/tests/test_meters_tui.py

@@ -0,0 +1,106 @@
+"""The bundled Textual ``meters_tui`` example.
+
+The textual-free helpers are tested unconditionally; the widget-level layout
+check needs the optional ``tui`` extra and skips without it.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from libkp import _generated as gen
+from libkp.examples import meters, meters_tui
+from libkp.state import RealtimeStatus
+
+# ---------------------------------------------------------------------------
+# Pure helpers — no Textual required
+# ---------------------------------------------------------------------------
+
+
+def _view(raw: tuple[int, ...]) -> meters.MeterView:
+    view = meters.MeterView(ip="127.0.0.1", width=48, show_all=False)
+    view.status = RealtimeStatus(raw=raw)
+    return view
+
+
+def test_verdict_is_idle_until_the_strobe_moves():
+    assert meters_tui.verdict_key(_view((0,) * gen.METER_COUNT)) == "idle"
+
+
+def test_verdict_is_unknown_until_enough_samples_land():
+    view = _view((1, 0, 0, 8000, *(0,) * 7))
+    assert meters_tui.verdict_key(view) == "unknown"
+
+
+@pytest.mark.parametrize(
+    "phases,expected",
+    [
+        ([8000, 8005, 8010], "intune"),
+        ([8000, 7000, 6000], "sharp"),
+        ([6000, 7000, 8000], "flat"),
+    ],
+)
+def test_verdicts_follow_the_drift_rate(phases, expected):
+    view = _view((1, 0, 0, phases[-1], *(0,) * 7))
+    now = meters_tui._clock()
+    view.phase_history.extend([(now - 0.2, phases[0]), (now - 0.1, phases[1]), (now, phases[2])])
+    assert meters_tui.verdict_key(view) == expected
+
+
+def test_every_verdict_key_has_a_label():
+    for key in ("idle", "unknown", "intune", "sharp", "flat"):
+        assert key in meters_tui._VERDICTS
+
+
+def test_level_color_matches_the_ansi_thresholds():
+    assert meters_tui.level_color(0.0) == "green"
+    assert meters_tui.level_color(0.7) == "yellow"
+    assert meters_tui.level_color(0.99) == "red"
+
+
+def test_cli_mirrors_the_ansi_example():
+    args = meters_tui.build_parser().parse_args(["--ip", "10.0.0.5", "--all"])
+    assert (args.ip, args.all, args.port) == ("10.0.0.5", True, gen.PORT)
+
+
+# ---------------------------------------------------------------------------
+# Layout — needs the optional `tui` extra
+# ---------------------------------------------------------------------------
+
+
+#: What a ``Panel`` spends before its content: ``border: round`` on both sides
+#: plus the ``padding: 0 1`` declared in ``MetersApp.CSS``.
+PANEL_CHROME = 4
+
+
+@pytest.mark.parametrize("columns", [72, 80, 100, 120, 160])
+@pytest.mark.parametrize("show_all", [False, True])
+def test_meter_bars_are_never_truncated_by_the_panel(monkeypatch, columns, show_all):
+    """The bar has to fit the METERS panel's content width.
+
+    Rich crops an over-wide grid cell with an ellipsis rather than wrapping, so
+    an over-generous bar width silently chops the right-hand end of every meter
+    — including the peak-hold marker.
+    """
+    pytest.importorskip("textual")
+    from rich.console import Console
+    from textual.geometry import Size
+
+    app_class = meters_tui._build_app_class()
+    app = app_class(meters_tui.build_parser().parse_args([]), None)
+    app.view.show_all = show_all
+    app.view.status = RealtimeStatus(raw=(gen.FULL_SCALE,) * gen.METER_COUNT)
+    app.view.peaks = [float(gen.FULL_SCALE)] * gen.METER_COUNT
+    app.view.maxs = [gen.FULL_SCALE] * gen.METER_COUNT
+    app.view.mins = [0] * gen.METER_COUNT
+    monkeypatch.setattr(app_class, "size", property(lambda _self: Size(columns, 40)))
+
+    console = Console(width=columns - PANEL_CHROME, no_color=True)
+    with console.capture() as capture:
+        console.print(app._meters(app))
+    lines = capture.get().rstrip("\n").split("\n")
+
+    rows = gen.METER_COUNT if show_all else len(meters.BAR_ROWS)
+    assert len(lines) == rows, "a wrapped row means the bar overflowed the panel"
+    for line in lines:
+        assert "…" not in line, f"bar truncated at {columns} columns: {line!r}"

python/tests/test_nrpn.py

@@ -134,6 +134,29 @@ def test_parse_extended_string_recovers_address_and_text():
     assert nrpn.parse_extended_string(nrpn.set_single(0x00, 0x7F, 0x00, 0x01, 1)) is None
 
 
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"page": 0x80},
+        {"number": 128},
+        {"page": -1},
+        {"function": 0xFF},
+        {"product": 200},
+        {"device": 0x80},
+    ],
+)
+def test_sysex_rejects_an_out_of_range_header_byte(kwargs):
+    """Masking would silently retarget the message at a different address."""
+    args = {"product": 0x00, "device": 0x7F, "function": 0x01, "page": 0x0A, "number": 0x04}
+    args.update(kwargs)
+    with pytest.raises(ValueError, match="7-bit"):
+        nrpn.sysex(**args)
+
+
+def test_sysex_accepts_the_whole_seven_bit_range():
+    assert nrpn.sysex(0x00, 0x7F, 0x01, 0x7F, 0x7F)[8:10] == bytes([0x7F, 0x7F])
+
+
 def test_parses_a_status_header():
     message = bytes.fromhex("f0002033000002007c4e0000f7")
     header, values = NrpnHeader.parse(message)

python/tests/test_public_api.py

@@ -0,0 +1,22 @@
+"""The package's public surface: everything ``__all__`` promises is importable."""
+
+from __future__ import annotations
+
+import libkp
+from libkp import errors
+
+
+def test_every_exported_name_exists():
+    missing = [name for name in libkp.__all__ if not hasattr(libkp, name)]
+    assert missing == []
+
+
+def test_the_whole_error_family_is_exported():
+    """A caller must be able to name every raisable error from the package root."""
+    missing = [name for name in errors.__all__ if not hasattr(libkp, name)]
+    assert missing == [], f"errors missing from libkp: {missing}"
+
+
+def test_every_error_derives_from_the_base():
+    for name in errors.__all__:
+        assert issubclass(getattr(errors, name), libkp.LibKPError)