From b4608d8590b4d62fec64c4cad1248ba886d3e8c9 Mon Sep 17 00:00:00 2001 From: Kelly Fox Date: Tue, 7 Jul 2026 19:17:47 -0500 Subject: [PATCH] feat(midi): add MIDI CC control surface for live performance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a process-wide midi_control.py service (scene jumps, style-cycle, transport, live effect/generator param sweeps) driven by MIDI, so a performer can control a running playlist in real time from a controller. Every action routes through the cheapest correct mechanism for its latency class — Playlist Events for discrete actions, a new request_jump() primitive generalizing skip_event for scene jumps, and direct unlocked setattr() via a new LIVE_PARAMS registry on FrameEffect/GenerativeSource for continuous CC sweeps — with no coalescing, since this module never writes to the DMA socket directly. Ships a sensible default [midi_control].cc_map out of the box (16-pad scene-jump bank + transport + an example knob bank) and targets ensemble systems by MIDI channel. Verified end-to-end on real Ultimate 64 hardware with a KeyLab mkII 49: skip/cycle_style/toggle_pause/jump and a live knob-driven param sweep all confirmed working. Also fixes a doctor.py bug found along the way: print_report's category_order omitted any category not already known, so a Diagnostic in an uncovered category (like the new midi_control one) would silently vanish from the printed report while still counting toward the ok/warn/error totals. --- CLAUDE.md | 5 + c64cast.schema.json | 276 ++++++++++++++++++++ c64cast/cli.py | 28 ++ c64cast/config.py | 156 +++++++++++- c64cast/doctor.py | 43 +++- c64cast/effects.py | 8 + c64cast/generators.py | 12 + c64cast/introspect.py | 6 + c64cast/midi_control.py | 313 +++++++++++++++++++++++ c64cast/playlist.py | 62 +++++ config/c64cast.example.toml | 36 +++ docs/architecture.md | 20 ++ tests/test_doctor.py | 33 +++ tests/test_example_toml_drift.py | 1 + tests/test_generative.py | 48 ++++ tests/test_midi_control.py | 421 +++++++++++++++++++++++++++++++ tests/test_playlist.py | 205 +++++++++++++++ 17 files changed, 1670 insertions(+), 3 deletions(-) create mode 100644 c64cast/midi_control.py create mode 100644 tests/test_midi_control.py diff --git a/CLAUDE.md b/CLAUDE.md index d6e2457..b674e97 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -147,6 +147,11 @@ c64cast/ ├── framebuffer.py Software VIC mirror used by preview + recording ├── preview.py Pygame preview window + cv2.VideoWriter recorder ├── control_plane.py FastAPI HTTP control plane (pause/resume/skip/reload) +├── midi_control.py Process-wide MIDI control surface for live performance: +│ scene jumps/style-cycle/transport via Playlist Events + +│ live effect/generator param sweeps via LIVE_PARAMS; +│ own MIDI port (separate from MidiScene), MIDI channel +│ selects the target ensemble system ├── c64.py Centralized C64 hardware constants (VIC/SID/CIA/KERNAL) ├── interstitial.py InterstitialScene: centered "UP NEXT" text + parallax bg ├── backgrounds.py Parallax background styles for the interstitial diff --git a/c64cast.schema.json b/c64cast.schema.json index bea3032..31e0015 100644 --- a/c64cast.schema.json +++ b/c64cast.schema.json @@ -719,6 +719,282 @@ } } }, + "midi_control": { + "type": "object", + "description": "MIDI CC control surface for live performance: scene jumps, style cycling, transport, live effect params (extra).", + "additionalProperties": false, + "properties": { + "enabled": { + "type": "boolean", + "description": "Run the MIDI control listener; requires the 'midi' extra.", + "default": false + }, + "port": { + "type": [ + "string", + "null" + ], + "description": "MIDI input port name (substring match, case-insensitive). None = first available port." + }, + "broadcast_channel": { + "type": "integer", + "description": "1-based MIDI channel that targets every system at once in ensemble mode. Other channels 1..N target the Nth system in ensemble order. Ignored in single-system mode (the one playlist is always the target).", + "default": 16 + }, + "jump_transition": { + "type": "string", + "description": "How a 'jump' action changes scenes: 'cut' (instant, no interstitial \u2014 the live-performance default) or 'interstitial' (routes through the normal UP-NEXT card).", + "enum": [ + "cut", + "interstitial" + ], + "default": "cut" + }, + "cc_map": { + "type": "array", + "description": "MIDI-message -> action mappings ([[midi_control.cc_map]] tables); see --describe section:midi_control. Set to [] to disable the shipped defaults, or override/extend individual entries. Each entry: type ('cc'|'note'|'pc'), number (0-127), action ('pause'|'resume'|'toggle_pause'|'skip'|'cycle_style'|'jump'|'param'); 'jump' also needs an int scene; 'param' also needs a string target ('effect.' or 'source.', matching a LIVE_PARAMS entry on the current scene's effect/generator).", + "default": [ + { + "type": "note", + "number": 36, + "action": "skip" + }, + { + "type": "note", + "number": 37, + "action": "cycle_style" + }, + { + "type": "note", + "number": 38, + "action": "toggle_pause" + }, + { + "type": "note", + "number": 39, + "action": "jump", + "scene": 0 + }, + { + "type": "note", + "number": 40, + "action": "jump", + "scene": 0 + }, + { + "type": "note", + "number": 41, + "action": "jump", + "scene": 1 + }, + { + "type": "note", + "number": 42, + "action": "jump", + "scene": 2 + }, + { + "type": "note", + "number": 43, + "action": "jump", + "scene": 3 + }, + { + "type": "note", + "number": 44, + "action": "jump", + "scene": 4 + }, + { + "type": "note", + "number": 45, + "action": "jump", + "scene": 5 + }, + { + "type": "note", + "number": 46, + "action": "jump", + "scene": 6 + }, + { + "type": "note", + "number": 47, + "action": "jump", + "scene": 7 + }, + { + "type": "note", + "number": 48, + "action": "jump", + "scene": 8 + }, + { + "type": "note", + "number": 49, + "action": "jump", + "scene": 9 + }, + { + "type": "note", + "number": 50, + "action": "jump", + "scene": 10 + }, + { + "type": "note", + "number": 51, + "action": "jump", + "scene": 11 + }, + { + "type": "note", + "number": 52, + "action": "jump", + "scene": 12 + }, + { + "type": "note", + "number": 53, + "action": "jump", + "scene": 13 + }, + { + "type": "note", + "number": 54, + "action": "jump", + "scene": 14 + }, + { + "type": "note", + "number": 55, + "action": "jump", + "scene": 15 + }, + { + "type": "pc", + "number": 0, + "action": "jump", + "scene": 0 + }, + { + "type": "pc", + "number": 1, + "action": "jump", + "scene": 1 + }, + { + "type": "pc", + "number": 2, + "action": "jump", + "scene": 2 + }, + { + "type": "pc", + "number": 3, + "action": "jump", + "scene": 3 + }, + { + "type": "pc", + "number": 4, + "action": "jump", + "scene": 4 + }, + { + "type": "pc", + "number": 5, + "action": "jump", + "scene": 5 + }, + { + "type": "pc", + "number": 6, + "action": "jump", + "scene": 6 + }, + { + "type": "pc", + "number": 7, + "action": "jump", + "scene": 7 + }, + { + "type": "pc", + "number": 8, + "action": "jump", + "scene": 8 + }, + { + "type": "pc", + "number": 9, + "action": "jump", + "scene": 9 + }, + { + "type": "pc", + "number": 10, + "action": "jump", + "scene": 10 + }, + { + "type": "pc", + "number": 11, + "action": "jump", + "scene": 11 + }, + { + "type": "pc", + "number": 12, + "action": "jump", + "scene": 12 + }, + { + "type": "pc", + "number": 13, + "action": "jump", + "scene": 13 + }, + { + "type": "pc", + "number": 14, + "action": "jump", + "scene": 14 + }, + { + "type": "pc", + "number": 15, + "action": "jump", + "scene": 15 + }, + { + "type": "cc", + "number": 13, + "action": "param", + "target": "effect.decay" + }, + { + "type": "cc", + "number": 14, + "action": "param", + "target": "source.speed" + }, + { + "type": "cc", + "number": 15, + "action": "param", + "target": "source.scale" + }, + { + "type": "cc", + "number": 16, + "action": "param", + "target": "source.scroll_speed" + } + ] + } + } + }, "menu": { "type": "object", "description": "On-C64 SPACE-key menu for live scene tweaks.", diff --git a/c64cast/cli.py b/c64cast/cli.py index 44f8a32..fed25a0 100644 --- a/c64cast/cli.py +++ b/c64cast/cli.py @@ -1003,6 +1003,7 @@ def _resolve_configs(args: argparse.Namespace) -> tuple[cfgmod.LoadResult, list[ paths=[None], is_ensemble=False, master_control=cfg.control, + master_midi_control=cfg.midi_control, ) return loaded, [cfg] @@ -1122,6 +1123,7 @@ def main(argv=None) -> int: paths=loaded.paths, is_ensemble=loaded.is_ensemble, master_control=loaded.master_control, + master_midi_control=loaded.master_midi_control, ) diagnostics = validate_load_result(merged, probe_u64=not cfgs[0].debug.skip_probe) return print_report(diagnostics) @@ -1217,6 +1219,7 @@ def main(argv=None) -> int: ) control_server = None + midi_control_listener = None # SIGTERM → graceful shutdown. SIGINT continues to raise KeyboardInterrupt # via the default handler so the user can still Ctrl+C interactively. @@ -1303,8 +1306,33 @@ def _on_sighup(_signum, _frame): except RuntimeError as e: log.error("control plane disabled: %s", e) + # Optional MIDI control surface for live performance. One listener + # for the whole ensemble (like [control]); MIDI channel selects the + # target system. See midi_control.py's module docstring for the + # latency rationale. + midi_cfg = loaded.master_midi_control if loaded.is_ensemble else cfgs[0].midi_control + if midi_cfg.enabled: + try: + cfgmod.validate_midi_control_cfg(midi_cfg) + from .midi_control import build_midi_control_listener + + midi_control_listener = build_midi_control_listener( + playlists={st.name: st.playlist for st in stacks}, + cfg=midi_cfg, + ) + midi_control_listener.start() + except (cfgmod.ConfigError, RuntimeError, ValueError) as e: + log.error("MIDI control disabled: %s", e) + _run_playlists(stacks, stop_event) finally: + # Stop input surfaces before tearing down what they act on — same + # ordering the keyboard/vision controllers already follow. + if midi_control_listener is not None: + try: + midi_control_listener.stop() + except Exception: + log.exception("MIDI control shutdown failed") if control_server is not None: try: control_server.stop() diff --git a/c64cast/config.py b/c64cast/config.py index e85cf37..e93d212 100644 --- a/c64cast/config.py +++ b/c64cast/config.py @@ -1602,6 +1602,99 @@ class ControlPlaneCfg: ) +_MIDI_CC_TYPE_CHOICES = ("cc", "note", "pc") +_MIDI_ACTION_CHOICES = ( + "pause", + "resume", + "toggle_pause", + "skip", + "cycle_style", + "jump", + "param", +) + +# Shipped out of the box so MIDI control works with no config edits, per a +# typical 16-pad-grid + knob-bank live controller (Launch Control XL / APC +# style). See midi_control.py's module docstring for the full mapping +# rationale; kept here (not imported from midi_control.py) so config stays +# import-light, same rationale as the DAC_CURVE_CHOICES-style constants above. +_DEFAULT_MIDI_CC_MAP: tuple[dict[str, Any], ...] = ( + {"type": "note", "number": 36, "action": "skip"}, + {"type": "note", "number": 37, "action": "cycle_style"}, + {"type": "note", "number": 38, "action": "toggle_pause"}, + {"type": "note", "number": 39, "action": "jump", "scene": 0}, # "home"/panic + # Scene-jump bank: notes 40-55 -> scenes 0-15 (a 16-pad grid row/block), + # and the same bank via Program Change for foot-controller performers. + *({"type": "note", "number": 40 + i, "action": "jump", "scene": i} for i in range(16)), + *({"type": "pc", "number": i, "action": "jump", "scene": i} for i in range(16)), + # Knob bank: deliberately clear of MidiScene's CC1/7/71-75 synth-control + # range, in case a shared controller feeds both via a virtual MIDI Thru. + # A CC mapped to a scene whose current effect/source doesn't declare that + # LIVE_PARAM is a silent no-op — safe to leave mapped across any playlist. + {"type": "cc", "number": 13, "action": "param", "target": "effect.decay"}, + {"type": "cc", "number": 14, "action": "param", "target": "source.speed"}, + {"type": "cc", "number": 15, "action": "param", "target": "source.scale"}, + {"type": "cc", "number": 16, "action": "param", "target": "source.scroll_speed"}, +) + + +@dataclass +class MidiControlCfg: + """Process-wide MIDI control surface for live performance: scene jumps, + style cycling, transport, and live effect/generator parameter sweeps + from a MIDI controller. Off by default; requires the `midi` extra. + + Opens its OWN mido.open_input() — a separate port from any MidiScene's, + even if both read the same physical controller via OS-level MIDI + routing (mido ports are exclusive opens). One listener governs the + whole ensemble (mirrors [control]): MIDI channel selects which system a + message targets, so a performer retargets with a controller-side + channel switch instead of a config/menu round trip.""" + + enabled: bool = field( + default=False, + metadata={"help": "Run the MIDI control listener; requires the 'midi' extra."}, + ) + port: str | None = field( + default=None, + metadata={ + "help": "MIDI input port name (substring match, case-insensitive). " + "None = first available port." + }, + ) + broadcast_channel: int = field( + default=16, + metadata={ + "help": "1-based MIDI channel that targets every system at once in " + "ensemble mode. Other channels 1..N target the Nth system in " + "ensemble order. Ignored in single-system mode (the one playlist " + "is always the target)." + }, + ) + jump_transition: str = field( + default="cut", + metadata={ + "help": "How a 'jump' action changes scenes: 'cut' (instant, no " + "interstitial — the live-performance default) or 'interstitial' " + "(routes through the normal UP-NEXT card).", + "choices": ("cut", "interstitial"), + }, + ) + cc_map: list[dict[str, Any]] = field( + default_factory=lambda: [dict(d) for d in _DEFAULT_MIDI_CC_MAP], + metadata={ + "help": "MIDI-message -> action mappings ([[midi_control.cc_map]] " + "tables); see --describe section:midi_control. Set to [] to disable " + "the shipped defaults, or override/extend individual entries. Each " + "entry: type ('cc'|'note'|'pc'), number (0-127), action " + "('pause'|'resume'|'toggle_pause'|'skip'|'cycle_style'|'jump'|" + "'param'); 'jump' also needs an int scene; 'param' also needs a " + "string target ('effect.' or 'source.', matching a " + "LIVE_PARAMS entry on the current scene's effect/generator)." + }, + ) + + @dataclass class MenuCfg: """On-C64 menu. When enabled, SPACE on the C64 keyboard opens an on-screen @@ -1665,6 +1758,7 @@ class Config: color: ColorCfg = field(default_factory=ColorCfg) dsp: DSPCfg = field(default_factory=DSPCfg) control: ControlPlaneCfg = field(default_factory=ControlPlaneCfg) + midi_control: MidiControlCfg = field(default_factory=MidiControlCfg) menu: MenuCfg = field(default_factory=MenuCfg) # Set only on the master Config produced by load_master(). Per-system # Configs in the returned list always have ensemble = None. @@ -1838,6 +1932,7 @@ def load(path: str | None) -> Config: ("recording", cfg.recording), ("dsp", cfg.dsp), ("control", cfg.control), + ("midi_control", cfg.midi_control), ("menu", cfg.menu), ): if section in data: @@ -2004,13 +2099,16 @@ class LoadResult: In single-system mode: `cfgs = [the_one_config]`, `names = ["system"]`, `paths = [args.config or None]`, `is_ensemble = False`. `master_control` holds the master TOML's [control] section (in - single-system mode this is just the loaded config's [control]).""" + single-system mode this is just the loaded config's [control]). + `master_midi_control` is the [midi_control] analog — also process-wide, + not per-system-cascaded (see _CASCADE_SECTIONS).""" cfgs: list[Config] names: list[str] paths: list[str | None] is_ensemble: bool master_control: ControlPlaneCfg + master_midi_control: MidiControlCfg def load_master(path: str | None) -> LoadResult: @@ -2032,6 +2130,7 @@ def load_master(path: str | None) -> LoadResult: paths=[None], is_ensemble=False, master_control=cfg.control, + master_midi_control=cfg.midi_control, ) path = DEFAULT_CONFIG_PATH @@ -2053,6 +2152,7 @@ def load_master(path: str | None) -> LoadResult: paths=[path], is_ensemble=False, master_control=cfg.control, + master_midi_control=cfg.midi_control, ) log.info("loading ensemble master %s", path) @@ -2076,6 +2176,7 @@ def load_master(path: str | None) -> LoadResult: ("preview", defaults.preview), ("recording", defaults.recording), ("control", defaults.control), + ("midi_control", defaults.midi_control), ("menu", defaults.menu), ): if section in raw: @@ -2118,6 +2219,7 @@ def load_master(path: str | None) -> LoadResult: paths=paths, is_ensemble=True, master_control=defaults.control, + master_midi_control=defaults.midi_control, ) @@ -2948,6 +3050,58 @@ def validate_dac_bitmap_tempo_cfg(cfg: Config) -> None: ) +def validate_midi_control_cfg(midi_cfg: MidiControlCfg) -> None: + """Guard [midi_control]: jump_transition choice, broadcast_channel + range, and every cc_map entry's shape. Takes the already-resolved + MidiControlCfg (loaded.master_midi_control in ensemble mode, else + cfgs[0].midi_control — see cli.py) rather than a whole Config, since + [midi_control] is process-wide like [control], not per-system-cascaded. + No-op when disabled.""" + if not midi_cfg.enabled: + return + if midi_cfg.jump_transition not in ("cut", "interstitial"): + raise ConfigError( + "[midi_control].jump_transition must be 'cut' or 'interstitial', " + f"got {midi_cfg.jump_transition!r}" + ) + if not 1 <= midi_cfg.broadcast_channel <= 16: + raise ConfigError( + f"[midi_control].broadcast_channel must be 1..16, got {midi_cfg.broadcast_channel}" + ) + for i, entry in enumerate(midi_cfg.cc_map): + if not isinstance(entry, dict): + raise ConfigError(f"[midi_control].cc_map[{i}] must be a table, got {entry!r}") + kind = entry.get("type") + if kind not in _MIDI_CC_TYPE_CHOICES: + raise ConfigError( + f"[midi_control].cc_map[{i}].type must be one of " + f"{', '.join(_MIDI_CC_TYPE_CHOICES)}, got {kind!r}" + ) + number = entry.get("number") + if not isinstance(number, int) or not 0 <= number <= 127: + raise ConfigError(f"[midi_control].cc_map[{i}].number must be 0..127, got {number!r}") + action = entry.get("action") + if action not in _MIDI_ACTION_CHOICES: + raise ConfigError( + f"[midi_control].cc_map[{i}].action must be one of " + f"{', '.join(_MIDI_ACTION_CHOICES)}, got {action!r}" + ) + if action == "jump" and not isinstance(entry.get("scene"), int): + raise ConfigError(f"[midi_control].cc_map[{i}] action 'jump' needs an int 'scene'") + if action == "param": + target = entry.get("target") + if ( + not isinstance(target, str) + or "." not in target + or target.split(".", 1)[0] not in ("effect", "source") + ): + raise ConfigError( + f"[midi_control].cc_map[{i}] action 'param' needs a string 'target' " + "of the form 'effect.' or 'source.', got " + f"{target!r}" + ) + + def validate_scene_cfg(s: SceneCfg, cfg: Config, *, audio_enabled: bool) -> None: """Pre-construction validation for a SceneCfg. diff --git a/c64cast/doctor.py b/c64cast/doctor.py index f2a3943..2da414a 100644 --- a/c64cast/doctor.py +++ b/c64cast/doctor.py @@ -27,6 +27,7 @@ LoadResult, validate_dac_bitmap_tempo_cfg, validate_dac_curve_cfg, + validate_midi_control_cfg, validate_scene_cfg, ) from .orchestrator import OrchestratorError @@ -53,7 +54,7 @@ class Diagnostic: ("preview", "pygame", "[preview] enabled local window"), ("control", "fastapi", "[control] enabled HTTP plane"), ("obs", "obsws_python", "obs_status overlay"), - ("midi", "mido", "midi scenes"), + ("midi", "mido", "midi scenes; [midi_control] live control"), ("logging", "rich", "colored log output"), ("vision", "mediapipe", "[vision] enabled gesture control"), ("tr", "serial", "TeensyROM serial backend"), @@ -94,6 +95,7 @@ def validate_load_result(loaded: LoadResult, *, probe_u64: bool = True) -> list[ out.extend(_validate_audio_nmi_rate(loaded)) out.extend(_validate_dac_curve(loaded)) out.extend(_validate_dac_bitmap_tempo(loaded)) + out.extend(_validate_midi_control(loaded)) if loaded.is_ensemble: out.extend(_validate_cross_system_orchestration(loaded)) out.extend(_probe_extras()) @@ -349,6 +351,35 @@ def _validate_dac_bitmap_tempo(loaded: LoadResult) -> list[Diagnostic]: return out +def _validate_midi_control(loaded: LoadResult) -> list[Diagnostic]: + """Flag a malformed [midi_control] section. Process-wide (like + [control]), so this validates loaded.master_midi_control once rather + than looping per system. Offline — delegates to + config.validate_midi_control_cfg.""" + try: + validate_midi_control_cfg(loaded.master_midi_control) + except ConfigError as e: + return [ + Diagnostic( + level="error", + category="midi_control", + subject="midi_control", + message=str(e), + hint="See [midi_control] in the config reference / --describe section:midi_control.", + ) + ] + if loaded.master_midi_control.enabled: + return [ + Diagnostic( + level="ok", + category="midi_control", + subject="midi_control", + message=f"{len(loaded.master_midi_control.cc_map)} cc_map entries configured.", + ) + ] + return [] + + def _validate_cross_system_orchestration(loaded: LoadResult) -> list[Diagnostic]: """Each `orchestrate=true` scene must have a same-name follower in every other system. If not, the Playlist falls back to building the @@ -1330,7 +1361,15 @@ def print_report(diagnostics: list[Diagnostic], file: IO[str] | None = None) -> by_category.setdefault(d.category, []).append(d) # Stable ordering by category, then error > warn > ok within each. - category_order = ["environment", "scene", "audio", "orchestrator", "extras", "connectivity"] + category_order = [ + "environment", + "scene", + "audio", + "midi_control", + "orchestrator", + "extras", + "connectivity", + ] for cat in category_order: rows = by_category.get(cat) if not rows: diff --git a/c64cast/effects.py b/c64cast/effects.py index 67a29d9..ab8e84b 100644 --- a/c64cast/effects.py +++ b/c64cast/effects.py @@ -69,6 +69,12 @@ class FrameEffect: name = "base" + # Live-tunable params: name -> (min, max) for a CC-style [0, 1] sweep. + # midi_control.py scales into this range and setattr()s directly — + # only declare independent single-numeric fields here (a plain + # setattr is GIL-atomic; a value split across two fields wouldn't be). + LIVE_PARAMS: dict[str, tuple[float, float]] = {} + def apply( self, frame: np.ndarray, t: float, modulation: MusicModulation | None = None ) -> np.ndarray: @@ -96,6 +102,8 @@ class TrailsEffect(FrameEffect): _LEVEL_DECAY = 0.06 # extra decay from sustained loudness _MAX_DECAY = 0.97 # hard ceiling — must stay < 1 or the tail never fades + LIVE_PARAMS = {"decay": (0.0, 0.96)} # stays under _MAX_DECAY's reactive headroom + def __init__(self, decay: float = 0.85): self.decay = float(decay) self._prev: np.ndarray | None = None diff --git a/c64cast/generators.py b/c64cast/generators.py index 2559467..401809f 100644 --- a/c64cast/generators.py +++ b/c64cast/generators.py @@ -76,6 +76,12 @@ class GenerativeSource(BaseFrameSource): name = "base" + # Live-tunable params: name -> (min, max) for a CC-style [0, 1] sweep. + # midi_control.py scales into this range and setattr()s directly — + # only declare independent single-numeric fields here (a plain + # setattr is GIL-atomic; a value split across two fields wouldn't be). + LIVE_PARAMS: dict[str, tuple[float, float]] = {} + # Reactive-modulation mapping constants (used only on the music-reactive # render path; the unmodulated path never touches them). Tuned on real HW # (Cam Link A/B vs the static path) so the reaction is unmistakable after @@ -127,6 +133,8 @@ class PlasmaSource(GenerativeSource): """Classic sine-sum plasma whose hue cycles over time. The spatial field is precomputed once; per-frame work is one modulo + HSV→BGR convert.""" + LIVE_PARAMS = {"speed": (0.0, 2.0), "scale": (0.1, 4.0)} + def __init__( self, *, @@ -164,6 +172,8 @@ class TunnelSource(GenerativeSource): """Infinite-zoom tunnel: hue is driven by per-pixel depth (1/radius) and angle, scrolled over time. Depth + angle fields are precomputed once.""" + LIVE_PARAMS = {"speed": (0.0, 2.0)} + def __init__(self, *, width: int = GEN_WIDTH, height: int = GEN_HEIGHT, speed: float = 0.5): super().__init__(width=width, height=height) self.speed = float(speed) @@ -226,6 +236,8 @@ class FireSource(GenerativeSource): _LEVEL_HEIGHT = 0.85 # extra heat gain at full level (taller, hotter flames) _ONSET_FLARE = 0.80 # extra heat gain on a full-strength transient + LIVE_PARAMS = {"scroll_speed": (0.0, 4.0)} + def __init__( self, *, width: int = GEN_WIDTH, height: int = GEN_HEIGHT, scroll_speed: float = 1.1 ): diff --git a/c64cast/introspect.py b/c64cast/introspect.py index 1c9c835..aeab09e 100644 --- a/c64cast/introspect.py +++ b/c64cast/introspect.py @@ -133,6 +133,12 @@ def __repr__(self) -> str: "Host-side audio DSP before the 4-bit DAC: compressor/limiter, expander (replaces the hard gate), pre-emphasis, and mic AGC.", ), ("control", cfgmod.ControlPlaneCfg, "HTTP control plane (extra)."), + ( + "midi_control", + cfgmod.MidiControlCfg, + "MIDI CC control surface for live performance: scene jumps, style " + "cycling, transport, live effect params (extra).", + ), ("menu", cfgmod.MenuCfg, "On-C64 SPACE-key menu for live scene tweaks."), ) diff --git a/c64cast/midi_control.py b/c64cast/midi_control.py new file mode 100644 index 0000000..ee17d74 --- /dev/null +++ b/c64cast/midi_control.py @@ -0,0 +1,313 @@ +"""Process-wide MIDI control surface for live performance: scene jumps, +style cycling, transport, and live effect/generator parameter sweeps from +a MIDI controller — turns a playlist run into something a performer can +drive in real time. + +Unlike :mod:`midi_scene` (a *Scene* that plays the SID directly and is only +live while that scene is on screen), this is a standalone service that runs +for the whole process, mirroring :mod:`control_plane`'s "one server for the +whole ensemble" shape. It opens its OWN ``mido.open_input()`` — mido ports +are exclusive opens, so this is always a second port, never shared with a +running :class:`~c64cast.midi_scene.MidiScene` even on the same physical +controller (route the controller to two virtual MIDI ports, or use OS-level +MIDI Thru, if you want one physical device to feed both). + +Every action bottoms out in one of two cheap, already-existing mechanisms: + +- Discrete actions (pause/resume/skip/cycle_style/jump) set a + :class:`~c64cast.playlist.Playlist` ``threading.Event`` — the same + mechanism :mod:`control_plane` and :mod:`keyboard` already use. Picked up + at the next clean frame boundary (one frame period, worst case). +- Continuous parameter sweeps (a CC mapped to an effect/generator's + ``LIVE_PARAMS`` entry) are a direct, unlocked ``setattr()`` onto the + running scene's ``effect``/``source`` — no Event, no frame-boundary wait, + picked up on the render loop's very next read of that attribute. This is + the cheapest path in the system. + +Because neither path touches the DMA socket from this module's reader +thread (unlike :class:`~c64cast.midi_scene.MidiScene`, which writes SID +registers directly), there is nothing to coalesce: every message is +dispatched immediately. The 1ms poll interval mirrors ``MidiScene._reader`` +for the same reason it was chosen there — it keeps latency tight — but +without that class's ``_CONTROL_FLUSH_INTERVAL_S`` throttle, since there's +no DMA burst risk here to guard against. + +Ensemble targeting is by MIDI channel: channel *N* (1-based) addresses the +Nth system in ensemble order, and a reserved broadcast channel (default 16) +addresses every system at once. A performer retargets by switching their +controller's transmit channel — zero app-side round trip, unlike a menu. +Single-system mode ignores channel entirely. + +Out of scope (see CLAUDE.md's [midi_control] note / docs/architecture.md): +anything that would need a scene rebuild (display-mode switches, scene-type +changes) — those cost real network/DMA setup time and are categorically +wrong for a live-hit control. Only Playlist-level Events and +``LIVE_PARAMS``-declared single-numeric-attribute writes are exposed. + +Requires the `midi` extra (``pip install c64cast[midi]``). +""" + +from __future__ import annotations + +import logging +import threading +import time +from collections.abc import Mapping +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from .config import MidiControlCfg + from .playlist import Playlist + +log = logging.getLogger(__name__) + +# Typed as Any so Pyright doesn't flag every mido.XXX as accessing attributes +# of None — the MIDI_AVAILABLE flag is the runtime guard. Mirrors +# midi_scene.py's import-guard pattern exactly. +try: + import mido as _mido + + mido: Any = _mido + MIDI_AVAILABLE = True +except ImportError: + mido = None + MIDI_AVAILABLE = False + +_CC_TYPES = ("cc", "note", "pc") +_ACTIONS = ("pause", "resume", "toggle_pause", "skip", "cycle_style", "jump", "param") + +# 1ms poll — mirrors MidiScene._reader's interval ("keeps note latency +# tight"). No coalescing here (see module docstring): every message this +# module receives is a plain in-process Event/attribute write, not a DMA +# write, so there's no burst risk to throttle against. +_POLL_INTERVAL_S = 0.001 + + +@dataclass(frozen=True) +class _CCMapping: + kind: str # "cc" | "note" | "pc" + number: int # 0-127 + action: str # one of _ACTIONS + scene: int | None = None # for "jump" + target: str | None = None # for "param": "effect." | "source." + + +def _parse_cc_map(raw: list[dict[str, Any]]) -> dict[tuple[str, int], _CCMapping]: + """Parse cc_map dicts into a (kind, number)-keyed lookup table. Raises + ValueError on a malformed entry — mirrors config.validate_midi_control_cfg's + checks, kept independent so this module is testable without config.py's + ConfigError (the same "config stays import-light" rationale in reverse: + midi_control.py doesn't need to import config's private choice tuples). + A later entry with the same (kind, number) overwrites an earlier one, same + as any TOML list — matches user-override-wins for the shipped defaults.""" + out: dict[tuple[str, int], _CCMapping] = {} + for i, entry in enumerate(raw): + if not isinstance(entry, dict): + raise ValueError(f"cc_map[{i}] must be a table, got {entry!r}") + kind = entry.get("type") + if kind not in _CC_TYPES: + raise ValueError(f"cc_map[{i}].type must be one of {_CC_TYPES}, got {kind!r}") + number = entry.get("number") + if not isinstance(number, int) or not 0 <= number <= 127: + raise ValueError(f"cc_map[{i}].number must be 0..127, got {number!r}") + action = entry.get("action") + if action not in _ACTIONS: + raise ValueError(f"cc_map[{i}].action must be one of {_ACTIONS}, got {action!r}") + scene = entry.get("scene") + if action == "jump" and not isinstance(scene, int): + raise ValueError(f"cc_map[{i}] action 'jump' needs an int 'scene'") + target = entry.get("target") + if action == "param" and (not isinstance(target, str) or "." not in target): + raise ValueError( + f"cc_map[{i}] action 'param' needs a string 'target' " + "('effect.' or 'source.')" + ) + out[(kind, number)] = _CCMapping( + kind=kind, number=number, action=action, scene=scene, target=target + ) + return out + + +class MidiControlListener: + """Opens its own MIDI input port and dispatches mapped messages to one + or more :class:`~c64cast.playlist.Playlist` instances. Construct via + :func:`build_midi_control_listener`; call :meth:`start` / :meth:`stop` + to run the reader thread for the life of the process.""" + + def __init__( + self, + playlists: Mapping[str, Playlist], + cc_map: list[dict[str, Any]], + *, + port: str | None = None, + broadcast_channel: int = 16, + jump_transition: str = "cut", + ) -> None: + if not playlists: + raise ValueError("midi_control needs at least one playlist") + self._playlists = dict(playlists) + self._all = list(self._playlists.values()) + self.port_name = port + self.broadcast_channel = broadcast_channel + self.jump_transition = jump_transition + self._mapping = _parse_cc_map(cc_map) + self._midi_port: Any = None + self._stop = threading.Event() + self._reader_thread: threading.Thread | None = None + self._warned_channels: set[int] = set() + + # ---- MIDI plumbing -------------------------------------------------- + def _open_port(self) -> None: + assert mido is not None + if self.port_name in (None, "", "default"): + names = mido.get_input_names() + if not names: + raise RuntimeError("midi_control: no MIDI input ports available") + self._midi_port = mido.open_input(names[0]) + log.info("midi_control: opened MIDI port %r", names[0]) + return + names = mido.get_input_names() + match = next((n for n in names if self.port_name.lower() in n.lower()), None) + if match is None: + raise RuntimeError( + f"midi_control: no MIDI input port matches {self.port_name!r}; available: {names}" + ) + self._midi_port = mido.open_input(match) + log.info("midi_control: opened MIDI port %r", match) + + def start(self) -> None: + if mido is None: + raise RuntimeError("midi_control requires mido: pip install c64cast[midi]") + self._open_port() + self._stop.clear() + self._reader_thread = threading.Thread( + target=self._reader, daemon=True, name="midi-control-reader" + ) + self._reader_thread.start() + log.info( + "midi_control: listening (%d mapping(s), %d target(s))", + len(self._mapping), + len(self._all), + ) + + def stop(self) -> None: + self._stop.set() + if self._reader_thread is not None: + self._reader_thread.join(timeout=1.0) + self._reader_thread = None + if self._midi_port is not None: + try: + self._midi_port.close() + except Exception: + log.debug("midi_control: port close failed", exc_info=True) + self._midi_port = None + + def _reader(self) -> None: + port = self._midi_port + if port is None: + return + try: + while not self._stop.is_set(): + for msg in port.iter_pending(): + try: + self._dispatch(msg) + except Exception: + log.exception("midi_control: dispatch failed for %r", msg) + time.sleep(_POLL_INTERVAL_S) + except Exception: + log.exception("midi_control reader crashed") + + # ---- dispatch --------------------------------------------------------- + def _targets(self, msg: Any) -> list[Playlist]: + """channel == broadcast_channel-1 -> every playlist; other channel N + (0-based) -> the Nth playlist in ensemble order if in range, else no + target (logged once at debug, not per-message — a performer's + controller idly sending on unrelated channels shouldn't spam logs). + Single-playlist mode ignores channel entirely.""" + if len(self._all) <= 1: + return self._all + channel = getattr(msg, "channel", 0) + if channel == self.broadcast_channel - 1: + return self._all + if 0 <= channel < len(self._all): + return [self._all[channel]] + if channel not in self._warned_channels: + self._warned_channels.add(channel) + log.debug( + "midi_control: channel %d has no target system (channels 1..%d " + "address a system; %d is the broadcast channel)", + channel + 1, + len(self._all), + self.broadcast_channel, + ) + return [] + + def _dispatch(self, msg: Any) -> None: + if msg.type == "note_on" and msg.velocity > 0: + kind, number, value = "note", msg.note, msg.velocity + elif msg.type == "control_change": + kind, number, value = "cc", msg.control, msg.value + elif msg.type == "program_change": + kind, number, value = "pc", msg.program, msg.program + else: + return # note_off, note_on velocity==0, pitchwheel, etc. — not mapped + mapping = self._mapping.get((kind, number)) + if mapping is None: + return + for pl in self._targets(msg): + try: + self._apply(pl, mapping, value) + except Exception: + log.exception( + "midi_control: action %r failed on system %r", mapping.action, pl.name + ) + + def _apply(self, pl: Playlist, mapping: _CCMapping, value: int) -> None: + action = mapping.action + if action == "pause": + pl.pause_event.set() + elif action == "resume": + pl.resume_event.set() + elif action == "toggle_pause": + (pl.resume_event if pl.pause_event.is_set() else pl.pause_event).set() + elif action == "skip": + pl.skip_event.set() + elif action == "cycle_style": + pl.cycle_event.set() + elif action == "jump": + if mapping.scene is not None and 0 <= mapping.scene < len(pl.scenes): + pl.request_jump(mapping.scene, skip_interstitial=self.jump_transition == "cut") + elif action == "param": + self._apply_param(pl, mapping.target, value) + + def _apply_param(self, pl: Playlist, target: str | None, value_0_127: int) -> None: + if target is None or pl.current is None: + return + holder_attr, _, name = target.partition(".") + holder = getattr(pl.current, holder_attr, None) + if holder is None: + return + live_params = getattr(type(holder), "LIVE_PARAMS", {}) + if name not in live_params: + return # target has no such LIVE_PARAM — a documented, silent no-op + lo, hi = live_params[name] + setattr(holder, name, lo + (value_0_127 / 127.0) * (hi - lo)) + + +def build_midi_control_listener( + playlists: Mapping[str, Playlist], cfg: MidiControlCfg +) -> MidiControlListener: + """The one entry point cli.py calls (mirrors + control_plane.start_control_server's shape, minus the auto-start — call + .start() on the result). Raises RuntimeError when the `midi` extra isn't + installed, same pattern MidiScene.__init__ already uses.""" + if not MIDI_AVAILABLE: + raise RuntimeError("midi_control requires mido: pip install c64cast[midi]") + return MidiControlListener( + playlists, + cfg.cc_map, + port=cfg.port, + broadcast_channel=cfg.broadcast_channel, + jump_transition=cfg.jump_transition, + ) diff --git a/c64cast/playlist.py b/c64cast/playlist.py index 149ec51..9fdfbd6 100644 --- a/c64cast/playlist.py +++ b/c64cast/playlist.py @@ -118,6 +118,17 @@ def __init__( self.resume_event = threading.Event() self.skip_event = threading.Event() self.cycle_event = threading.Event() + # Jump-to-index request (e.g. from midi_control.py). Reuses + # skip_event to force the current scene done at the next clean + # frame boundary; _advance() then consumes _jump_target instead of + # advancing to index+1. Locked because it's written from whatever + # thread requests the jump and read from the run-loop thread. + # Last-write-wins: a burst of requests before the run loop drains + # collapses to the final target, which is correct for a performer + # mashing pads. + self._jump_target: int | None = None + self._jump_skip_interstitial = True + self._jump_lock = threading.Lock() # On-C64 menu plumbing. menu_event is toggled by the poller on SPACE # (open when running / close when open); menu_active is held set by the # run loop while the MenuOverlay is injected, which flips the poller into @@ -179,6 +190,32 @@ def request_reload( self._pending_interstitial = new_interstitial self.reload_event.set() + def request_jump(self, index: int, *, skip_interstitial: bool = True) -> None: + """Cut to scenes[index] at the next clean frame boundary (reuses + skip_event to force the current scene done — see _advance's + end-of-scene branch, which consumes _jump_target in place of the + usual index+1). No-op in single-scene mode (nowhere to jump to). + + skip_interstitial=True (the default, and what a live-performance + control surface should always pass) bypasses the "UP NEXT" card + entirely for a hard cut, still gated on the ensemble audio claim + for the target scene like single-scene looping is. False routes + through the normal transitioning/interstitial path instead, for + callers that still want the card. + + Known limitation: a jump requested while self.current is None + (the brief startup / finished-without-loop window) is dropped — + re-send once the playlist has a current scene.""" + if self.single_scene: + self.log.debug("jump to %d ignored — single-scene mode", index) + return + if not 0 <= index < len(self.scenes): + raise ValueError(f"jump index {index} out of range (0..{len(self.scenes) - 1})") + with self._jump_lock: + self._jump_target = index + self._jump_skip_interstitial = skip_interstitial + self.skip_event.set() + def _apply_reload(self) -> None: """Swap in the queued scenes + interstitial factory. The current scene is torn down so its overlays release threads/network state @@ -433,6 +470,31 @@ def _advance(self) -> None: elif not self.transitioning and self.current.is_done: self._fade_out(self.current) self._safe_teardown(self.current) + with self._jump_lock: + jump_target = self._jump_target + jump_skip_interstitial = self._jump_skip_interstitial + self._jump_target = None + if jump_target is not None: + self.index = jump_target + if jump_skip_interstitial: + scene = self.scenes[self.index] + if not self._wait_for_audio_claim(scene): + self.current = None + return + self.log.info( + "scene %d/%d → %r (jump)", self.index + 1, len(self.scenes), scene.name + ) + self.current = scene + self._safe_setup(self.current) + self.transitioning = False + return + resolved = self._resolve_next_index() + if resolved is None: + self.current = None + return + self.index = resolved + self._enter_interstitial() + return next_index = self.index + 1 if next_index >= len(self.scenes): if not self.loop: diff --git a/config/c64cast.example.toml b/config/c64cast.example.toml index 21d75a5..6f2538f 100644 --- a/config/c64cast.example.toml +++ b/config/c64cast.example.toml @@ -577,6 +577,42 @@ enabled = false # requires the `control` extra host = "127.0.0.1" port = 8765 +[midi_control] +# Process-wide MIDI control surface for live performance: scene jumps, +# style cycling, transport, and live effect/generator parameter sweeps +# from a MIDI controller. Off by default; requires the `midi` extra. Opens +# its OWN MIDI input port — separate from any [[scenes]] type = "midi" +# port, even on the same physical controller. One listener governs the +# whole ensemble: MIDI channel selects which system a message targets. +enabled = false +port = "" # substring match; "" = first available port +broadcast_channel = 16 # this channel targets every ensemble system at once; + # other channels target the Nth system in ensemble order +jump_transition = "cut" # "cut" (instant, no interstitial) | "interstitial" + +# Ships with a sensible default cc_map (transport on notes 36-39, a 16-scene +# jump bank on notes 40-55 + Program Change 0-15, a knob bank on CC13-16) if +# left unset — see `--describe section:midi_control` for the full default. +# Setting cc_map here (as below) REPLACES it wholesale, not merges — copy the +# default from --describe first if you just want to add to it. +[[midi_control.cc_map]] +type = "note" +number = 36 +action = "skip" + +[[midi_control.cc_map]] +type = "note" +number = 40 +action = "jump" +scene = 0 + +[[midi_control.cc_map]] +type = "cc" +number = 13 +action = "param" +target = "effect.decay" # "effect." | "source.", matching a + # LIVE_PARAMS entry on the current scene's effect/generator + [menu] # On-C64 menu: SPACE opens an on-screen panel of context-sensitive knobs for # the current scene (palette mode, PETSCII style, …) with a live preview; diff --git a/docs/architecture.md b/docs/architecture.md index 3eb7250..e45fe26 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -307,3 +307,23 @@ When `[control] enabled = true` and the `control` extra is installed (`pip insta * `POST /reload` → re-reads the config from disk and rebuilds the scene list at the next interstitial boundary The same three events are fed by the keyboard poller, so HTTP and the C64 keyboard are equivalent control surfaces. + +### `midi_control.py` — process-wide MIDI control surface (optional, live performance) + +When `[midi_control] enabled = true` and the `midi` extra is installed, `MidiControlListener` opens **its own** `mido.open_input()` and runs a 1ms-poll reader thread for the whole process — a *service*, not a Scene, unlike `midi_scene.py`'s `MidiScene` (only live while that scene is on screen). Mido ports are exclusive opens, so this is always a second port: it never shares a running `MidiScene`'s port even on the same physical controller (route the controller to two virtual ports, or OS-level MIDI Thru, to feed both from one device). + +**Why no coalescing, unlike MidiScene.** `MidiScene._reader` coalesces continuous CCs to `_CONTROL_FLUSH_INTERVAL_S = 1/60` because it writes SID registers over the DMA socket and bursting it is unsafe. `midi_control.py` never touches the DMA socket from its reader thread — every action it takes is either a `threading.Event.set()` on a `Playlist` or a plain in-process `setattr()` — so there is nothing to throttle: every message dispatches the instant it arrives. This is the deliberate, load-bearing difference between the two modules; don't "fix" it by adding a flush interval. + +**Two dispatch paths, chosen for latency:** + +* **Discrete actions** (`pause`/`resume`/`toggle_pause`/`skip`/`cycle_style`/`jump`) reuse the existing `Playlist` Event-flag pattern — the same mechanism `control_plane.py` and `keyboard.py` use — picked up at the next clean frame boundary (one frame period, worst case; the run loop's pacing wait isn't interrupted early by these Events, same floor the keyboard/control-plane already accept). +* **`jump`** is `Playlist.request_jump(index, skip_interstitial=...)` — a new primitive generalizing `skip_event`'s "force `is_done=True` at the next boundary" pattern to an arbitrary target index instead of always `index+1`. `_advance()`'s end-of-scene branch consumes a locked `_jump_target` (last-write-wins: a burst of pad hits collapses to the final target) in place of the usual `index+1` computation. `jump_transition = "cut"` (the default, and what a live-hit control should always want) skips `_enter_interstitial()` entirely and goes straight to `_safe_setup` via the same `_wait_for_audio_claim` gate single-scene looping uses — a jump into an ensemble audio-gated scene **blocks on the real gate**, it does not silently retarget elsewhere (an explicit jump must not be second-guessed). `jump_transition = "interstitial"` routes through the normal "UP NEXT" card instead, for non-performance callers. +* **`param`** (continuous CC → an effect/generator knob) skips the Event/frame-boundary machinery entirely: `setattr(holder, name, scaled_value)` directly on `scene.effect`/`scene.source`, an unlocked cross-thread write already-safe by the same precedent `MidiScene.adsr`/`.pulse_width` etc. use (GIL-atomic single-attribute writes, read by the render loop's normal next pass). This is the cheapest path in the system — no Event, no `_advance()` indirection, picked up on literally the next frame render. + +**`LIVE_PARAMS` registry.** `FrameEffect`/`GenerativeSource` each carry a class-level `LIVE_PARAMS: dict[str, tuple[float, float]]` (name → (min, max)), empty by default. A CC's 0-127 value is linearly scaled into the declared range and written with a plain `setattr` — no per-class wiring beyond the one-line class attribute. Only declare **independent single-numeric fields** here: a `setattr` is GIL-atomic for one attribute, not for two that must change together. Currently declared: `TrailsEffect.decay`, `PlasmaSource.{speed,scale}`, `TunnelSource.speed`, `FireSource.scroll_speed`; `PulseEffect`/`RgbShiftEffect` are purely modulation-driven (no persistent knob) and inherit the empty base — a CC mapped to them is a documented, silent no-op. To add a new live-tunable param to an effect/generator, add one entry to its `LIVE_PARAMS` dict; nothing else needs to change. + +**Ensemble targeting is by MIDI channel**, not a config-time system list: channel *N* (1-based) addresses the Nth system in ensemble order (the same order `playlists.keys()` / `stacks` iterate in), and a reserved `broadcast_channel` (default 16) addresses every system at once. A performer retargets by switching their controller's transmit channel — zero app-side round trip, unlike a menu or HTTP call. Single-system mode ignores channel entirely. Mirrors `control_plane.py`'s "one service for the whole ensemble" shape (`build_midi_control_listener(playlists: Mapping[str, Playlist], cfg)`), not a per-system controller. + +**Config** (`[midi_control]`, `MidiControlCfg`) is process-wide like `[control]`: not in `_CASCADE_SECTIONS`, tracked via `LoadResult.master_midi_control` (parsed from the master TOML in ensemble mode, or the single config in single-system mode) rather than per-system `cfg.midi_control`. `cc_map` is a free-form `list[dict[str, Any]]`, following the `[[scenes.overlays]]` convention — validated by the consuming code (`config.validate_midi_control_cfg` / `midi_control._parse_cc_map`, deliberately duplicated so `midi_control.py` stays testable without importing `config.ConfigError`) rather than encoded rigidly in the dataclass. Ships a default map out of the box (notes 36-39 transport, notes 40-55 + Program Change 0-15 as a 16-scene jump bank, CC13-16 as an example knob bank) so the feature works with no config edits — see `--describe section:midi_control` for the exact table. + +**Explicitly out of scope:** anything that would need a scene rebuild — display-mode switches, scene-type changes — is not exposed. Those cost real network/DMA setup time (`_build_display_mode`, teardown+setup) and are categorically wrong for a live-hit control; this mirrors the existing "live" vs "rebuild" axis in `config._APPLY_CHOICES` for the on-C64 menu. Only Playlist-level Events and `LIVE_PARAMS`-declared single-numeric-attribute writes are ever touched. diff --git a/tests/test_doctor.py b/tests/test_doctor.py index 3ada7c7..cf36468 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -923,6 +923,39 @@ def test_exit_code_one_when_any_error(self): self.assertEqual(doctor.print_report(diags, file=buf), 1) self.assertIn("[ERR ]", buf.getvalue()) + def test_midi_control_category_is_rendered(self): + # Regression: category_order previously omitted "midi_control", so + # a midi_control Diagnostic was silently dropped from the printed + # report (still counted in the ok/warn/error totals, but invisible + # to the reader — the worst kind of drift, since nothing failed). + diags = [doctor.Diagnostic("ok", "midi_control", "midi_control", "11 entries")] + buf = io.StringIO() + doctor.print_report(diags, file=buf) + self.assertIn("MIDI_CONTROL", buf.getvalue()) + self.assertIn("11 entries", buf.getvalue()) + + def test_every_category_used_in_source_is_in_category_order(self): + # General drift guard: every category="..." literal doctor.py + # actually constructs a Diagnostic with must be covered by + # print_report's category_order, or it silently vanishes from the + # printed report (same failure class as the midi_control omission + # above — this test would have caught it). + import inspect + import re + + source = inspect.getsource(doctor) + used = set(re.findall(r'category="([a-z_]+)"', source)) + self.assertTrue(used, "regex found no categories — pattern drifted from doctor.py's style") + # Extract print_report's category_order literal the same way, so + # this test doesn't need to import a private name. + report_source = inspect.getsource(doctor.print_report) + order_match = re.search(r"category_order = \[(.*?)\]", report_source, re.DOTALL) + assert order_match is not None + covered = set(re.findall(r'"([a-z_]+)"', order_match.group(1))) + self.assertEqual( + used - covered, set(), "categories missing from print_report's category_order" + ) + class EnvironmentProbeTest(unittest.TestCase): """The env probe is the dev-environment guard: it catches the desynced diff --git a/tests/test_example_toml_drift.py b/tests/test_example_toml_drift.py index d721162..eeb3f62 100644 --- a/tests/test_example_toml_drift.py +++ b/tests/test_example_toml_drift.py @@ -45,6 +45,7 @@ "color": cfgmod.ColorCfg, "dsp": cfgmod.DSPCfg, "control": cfgmod.ControlPlaneCfg, + "midi_control": cfgmod.MidiControlCfg, "menu": cfgmod.MenuCfg, } diff --git a/tests/test_generative.py b/tests/test_generative.py index 93bd979..acc1b6d 100644 --- a/tests/test_generative.py +++ b/tests/test_generative.py @@ -29,6 +29,33 @@ def test_registry_nonempty_and_named(self): self.assertIn("tunnel", names) self.assertIn("fire", names) + def test_live_params_declared_with_valid_ranges(self): + # midi_control.py scales a CC into each declared (min, max) range and + # setattr()s it directly — a malformed range would silently corrupt + # a live-performance param sweep, so pin the shape here. + expected = { + "plasma": {"speed", "scale"}, + "tunnel": {"speed"}, + "fire": {"scroll_speed"}, + } + for name in generator_names(): + g = build_generator(name) + live_params = g.LIVE_PARAMS + self.assertEqual(set(live_params), expected.get(name, set()), name) + for param, (lo, hi) in live_params.items(): + self.assertLess(lo, hi, f"{name}.{param}") + self.assertTrue(hasattr(g, param), f"{name}.{param} not a real attribute") + + def test_live_params_settable_via_generic_setattr(self): + # The exact mechanism midi_control.py uses: setattr(obj, name, val) + # with no per-class wiring. Constructed directly (not via the + # registry) so the concrete type declares `speed` for pyright. + g = generators.PlasmaSource() + lo, hi = g.LIVE_PARAMS["speed"] + mid = lo + 0.5 * (hi - lo) + setattr(g, "speed", mid) # noqa: B010 — exercises the dynamic-name path deliberately + self.assertAlmostEqual(g.speed, mid) + def test_plasma_frame_shape_and_determinism(self): g = build_generator("plasma") f0 = g.render(0.0) @@ -106,6 +133,27 @@ def test_beat_phase_advances_hue(self): class EffectTest(unittest.TestCase): + def test_live_params_declared_with_valid_ranges(self): + # midi_control.py scales a CC into each declared (min, max) range and + # setattr()s it directly — pin the shape for every registered + # effect, including pulse/rgb_shift (which inherit the empty base + # default: no persistent knob worth exposing, a mapped CC no-ops). + expected = {"trails": {"decay"}, "pulse": set(), "rgb_shift": set()} + for name in expected: + eff = build_effect(name) + live_params = eff.LIVE_PARAMS + self.assertEqual(set(live_params), expected[name], name) + for param, (lo, hi) in live_params.items(): + self.assertLess(lo, hi, f"{name}.{param}") + self.assertTrue(hasattr(eff, param), f"{name}.{param} not a real attribute") + + def test_live_params_settable_via_generic_setattr(self): + eff = TrailsEffect() + lo, hi = eff.LIVE_PARAMS["decay"] + mid = lo + 0.5 * (hi - lo) + setattr(eff, "decay", mid) # noqa: B010 — exercises the dynamic-name path deliberately + self.assertAlmostEqual(eff.decay, mid) + def test_trails_first_frame_passthrough_then_blends(self): eff = build_effect("trails") a = np.zeros((4, 4, 3), np.uint8) diff --git a/tests/test_midi_control.py b/tests/test_midi_control.py new file mode 100644 index 0000000..aa9e402 --- /dev/null +++ b/tests/test_midi_control.py @@ -0,0 +1,421 @@ +"""Host-side unit tests for midi_control.py. + +Mirrors test_midi_scene.py's approach: real mido.Message objects drive +dispatch, _open_port is exercised against a patched `mido` module (no real +MIDI hardware touched), and Playlist targets are MagicMock's satisfying only +what MidiControlListener reads (pause/resume/skip/cycle_event, scenes, +request_jump, current) — mirroring test_control_plane.py's _fake_playlist +helper. No FakeAPI/DMA fake needed anywhere here: this module never touches +hardware, only in-process Playlist Events and plain attribute writes. +""" + +from __future__ import annotations + +import threading +import time +import unittest +from typing import Any +from unittest import mock + +try: + import mido as _mido + + mido: Any = _mido + HAVE_MIDI = True +except ImportError: + mido = None + HAVE_MIDI = False + +from c64cast import midi_control +from c64cast.midi_control import MidiControlListener, _parse_cc_map + + +def _fake_playlist(name: str, *, scene_count: int = 16) -> Any: + """A MagicMock'd Playlist satisfying what MidiControlListener reads: + .name, .scenes (list), .pause/resume/skip/cycle_event (real Events), + .request_jump(), .current (a scene stand-in with .effect/.source).""" + pl = mock.MagicMock(name=f"playlist-{name}") + pl.name = name + pl.scenes = [mock.MagicMock() for _ in range(scene_count)] + pl.pause_event = threading.Event() + pl.resume_event = threading.Event() + pl.skip_event = threading.Event() + pl.cycle_event = threading.Event() + pl.current = mock.MagicMock() + return pl + + +class _FakePort: + """Minimal mido-input stand-in: never yields a message, closes cleanly.""" + + def __init__(self) -> None: + self.closed = False + + def iter_pending(self): + return iter(()) + + def close(self) -> None: + self.closed = True + + +class _ScriptedPort: + """Yields one queued batch of messages on the first iter_pending() call, + then nothing — mirrors test_midi_scene.py's _ScriptedPort.""" + + def __init__(self, batch) -> None: + self._batch = batch + self._done = False + self.closed = False + + def iter_pending(self): + if self._done: + return iter(()) + self._done = True + return iter(self._batch) + + def close(self) -> None: + self.closed = True + + +class ParseCCMapTests(unittest.TestCase): + def test_valid_entries_parse(self): + m = _parse_cc_map( + [ + {"type": "note", "number": 36, "action": "skip"}, + {"type": "pc", "number": 0, "action": "jump", "scene": 2}, + {"type": "cc", "number": 13, "action": "param", "target": "effect.decay"}, + ] + ) + self.assertEqual(m[("note", 36)].action, "skip") + self.assertEqual(m[("pc", 0)].scene, 2) + self.assertEqual(m[("cc", 13)].target, "effect.decay") + + def test_later_entry_overwrites_earlier_same_key(self): + m = _parse_cc_map( + [ + {"type": "note", "number": 36, "action": "skip"}, + {"type": "note", "number": 36, "action": "cycle_style"}, + ] + ) + self.assertEqual(m[("note", 36)].action, "cycle_style") + + def test_bad_type_rejected(self): + with self.assertRaises(ValueError): + _parse_cc_map([{"type": "bogus", "number": 1, "action": "skip"}]) + + def test_bad_number_rejected(self): + with self.assertRaises(ValueError): + _parse_cc_map([{"type": "cc", "number": 999, "action": "skip"}]) + + def test_bad_action_rejected(self): + with self.assertRaises(ValueError): + _parse_cc_map([{"type": "cc", "number": 1, "action": "bogus"}]) + + def test_jump_without_scene_rejected(self): + with self.assertRaises(ValueError): + _parse_cc_map([{"type": "note", "number": 1, "action": "jump"}]) + + def test_param_without_target_rejected(self): + with self.assertRaises(ValueError): + _parse_cc_map([{"type": "cc", "number": 1, "action": "param"}]) + + def test_non_dict_entry_rejected(self): + with self.assertRaises(ValueError): + _parse_cc_map(["not a dict"]) # type: ignore[list-item] + + +@unittest.skipUnless(HAVE_MIDI, "mido not installed (midi extra)") +class _MidiControlTestCase(unittest.TestCase): + pass + + +class DispatchActionTests(_MidiControlTestCase): + """Direct _dispatch() calls, no thread — mirrors test_midi_scene.py's + ControlChangeTests / HandleMsgDispatchTests style.""" + + def _listener(self, cc_map, **kwargs) -> tuple[MidiControlListener, Any]: + pl = _fake_playlist("system") + listener = MidiControlListener({"system": pl}, cc_map, **kwargs) + return listener, pl + + def test_note_on_dispatches_skip(self): + listener, pl = self._listener([{"type": "note", "number": 36, "action": "skip"}]) + listener._dispatch(mido.Message("note_on", note=36, velocity=100)) + self.assertTrue(pl.skip_event.is_set()) + + def test_note_on_dispatches_cycle_style(self): + listener, pl = self._listener([{"type": "note", "number": 37, "action": "cycle_style"}]) + listener._dispatch(mido.Message("note_on", note=37, velocity=100)) + self.assertTrue(pl.cycle_event.is_set()) + + def test_toggle_pause_sets_pause_when_not_paused(self): + listener, pl = self._listener([{"type": "note", "number": 38, "action": "toggle_pause"}]) + listener._dispatch(mido.Message("note_on", note=38, velocity=100)) + self.assertTrue(pl.pause_event.is_set()) + self.assertFalse(pl.resume_event.is_set()) + + def test_toggle_pause_sets_resume_when_paused(self): + listener, pl = self._listener([{"type": "note", "number": 38, "action": "toggle_pause"}]) + pl.pause_event.set() + listener._dispatch(mido.Message("note_on", note=38, velocity=100)) + self.assertTrue(pl.resume_event.is_set()) + + def test_note_on_dispatches_jump(self): + listener, pl = self._listener( + [{"type": "note", "number": 40, "action": "jump", "scene": 3}] + ) + listener._dispatch(mido.Message("note_on", note=40, velocity=100)) + pl.request_jump.assert_called_once_with(3, skip_interstitial=True) + + def test_jump_transition_interstitial_passes_skip_interstitial_false(self): + listener, pl = self._listener( + [{"type": "note", "number": 40, "action": "jump", "scene": 3}], + jump_transition="interstitial", + ) + listener._dispatch(mido.Message("note_on", note=40, velocity=100)) + pl.request_jump.assert_called_once_with(3, skip_interstitial=False) + + def test_jump_out_of_range_scene_is_noop(self): + listener, pl = self._listener( + [{"type": "note", "number": 40, "action": "jump", "scene": 99}] + ) + listener._dispatch(mido.Message("note_on", note=40, velocity=100)) + pl.request_jump.assert_not_called() + + def test_program_change_dispatches_jump(self): + listener, pl = self._listener([{"type": "pc", "number": 5, "action": "jump", "scene": 5}]) + listener._dispatch(mido.Message("program_change", program=5)) + pl.request_jump.assert_called_once_with(5, skip_interstitial=True) + + def test_note_off_never_dispatches(self): + listener, pl = self._listener([{"type": "note", "number": 36, "action": "skip"}]) + listener._dispatch(mido.Message("note_off", note=36)) + self.assertFalse(pl.skip_event.is_set()) + + def test_note_on_velocity_zero_never_dispatches(self): + listener, pl = self._listener([{"type": "note", "number": 36, "action": "skip"}]) + listener._dispatch(mido.Message("note_on", note=36, velocity=0)) + self.assertFalse(pl.skip_event.is_set()) + + def test_unmapped_message_is_noop(self): + listener, pl = self._listener([{"type": "note", "number": 36, "action": "skip"}]) + listener._dispatch(mido.Message("note_on", note=99, velocity=100)) + self.assertFalse(pl.skip_event.is_set()) + + +class ParamActionTests(_MidiControlTestCase): + def _listener_with_effect(self, decay_range=(0.0, 0.96)): + pl = _fake_playlist("system") + effect = mock.MagicMock() + type(effect).LIVE_PARAMS = {"decay": decay_range} + pl.current.effect = effect + listener = MidiControlListener( + {"system": pl}, + [{"type": "cc", "number": 13, "action": "param", "target": "effect.decay"}], + ) + return listener, pl, effect + + def test_cc_scales_and_sets_live_param(self): + listener, pl, effect = self._listener_with_effect() + listener._dispatch(mido.Message("control_change", control=13, value=127)) + self.assertAlmostEqual(effect.decay, 0.96, places=4) + + def test_cc_zero_maps_to_range_minimum(self): + listener, pl, effect = self._listener_with_effect() + listener._dispatch(mido.Message("control_change", control=13, value=0)) + self.assertAlmostEqual(effect.decay, 0.0, places=4) + + def test_cc_midpoint_scales_linearly(self): + listener, pl, effect = self._listener_with_effect(decay_range=(0.0, 1.0)) + listener._dispatch(mido.Message("control_change", control=13, value=64)) + self.assertAlmostEqual(effect.decay, 64 / 127.0, places=4) + + def test_target_without_live_param_entry_is_noop(self): + # effect declares LIVE_PARAMS but not the mapped name. + pl = _fake_playlist("system") + effect = mock.MagicMock() + type(effect).LIVE_PARAMS = {"decay": (0.0, 0.96)} + pl.current.effect = effect + listener = MidiControlListener( + {"system": pl}, + [{"type": "cc", "number": 13, "action": "param", "target": "effect.nonexistent"}], + ) + before = effect.mock_calls + listener._dispatch(mido.Message("control_change", control=13, value=64)) + # No new attribute assertion recorded beyond what MagicMock already had. + self.assertEqual(effect.mock_calls, before) + + def test_holder_missing_is_noop(self): + pl = _fake_playlist("system") + pl.current = mock.MagicMock(spec=[]) # no .effect/.source attrs at all + listener = MidiControlListener( + {"system": pl}, + [{"type": "cc", "number": 13, "action": "param", "target": "effect.decay"}], + ) + # Should not raise. + listener._dispatch(mido.Message("control_change", control=13, value=64)) + + def test_no_current_scene_is_noop(self): + pl = _fake_playlist("system") + pl.current = None + listener = MidiControlListener( + {"system": pl}, + [{"type": "cc", "number": 13, "action": "param", "target": "effect.decay"}], + ) + listener._dispatch(mido.Message("control_change", control=13, value=64)) + + +class ChannelTargetingTests(_MidiControlTestCase): + def test_single_system_ignores_channel(self): + pl = _fake_playlist("only") + listener = MidiControlListener( + {"only": pl}, [{"type": "note", "number": 36, "action": "skip"}] + ) + listener._dispatch(mido.Message("note_on", note=36, velocity=100, channel=7)) + self.assertTrue(pl.skip_event.is_set()) + + def _ensemble(self, n=3): + pls = {f"sys{i}": _fake_playlist(f"sys{i}") for i in range(n)} + listener = MidiControlListener( + pls, + [{"type": "note", "number": 36, "action": "skip"}], + broadcast_channel=16, + ) + return listener, pls + + def test_channel_targets_nth_system(self): + listener, pls = self._ensemble() + listener._dispatch(mido.Message("note_on", note=36, velocity=100, channel=1)) + self.assertFalse(pls["sys0"].skip_event.is_set()) + self.assertTrue(pls["sys1"].skip_event.is_set()) + self.assertFalse(pls["sys2"].skip_event.is_set()) + + def test_broadcast_channel_targets_all(self): + listener, pls = self._ensemble() + listener._dispatch(mido.Message("note_on", note=36, velocity=100, channel=15)) + for pl in pls.values(): + self.assertTrue(pl.skip_event.is_set()) + + def test_out_of_range_channel_targets_nothing(self): + listener, pls = self._ensemble() + listener._dispatch(mido.Message("note_on", note=36, velocity=100, channel=10)) + for pl in pls.values(): + self.assertFalse(pl.skip_event.is_set()) + + +class CrashGuardTests(_MidiControlTestCase): + def test_one_target_raising_does_not_block_others(self): + good = _fake_playlist("good") + bad = _fake_playlist("bad") + bad.skip_event = mock.MagicMock() + bad.skip_event.set.side_effect = RuntimeError("boom") + listener = MidiControlListener( + {"bad": bad, "good": good}, + [{"type": "note", "number": 36, "action": "skip"}], + broadcast_channel=16, + ) + with self.assertLogs("c64cast.midi_control", level="ERROR"): + listener._dispatch(mido.Message("note_on", note=36, velocity=100, channel=15)) + self.assertTrue(good.skip_event.is_set()) + + def test_reader_thread_survives_dispatch_exception(self): + pl = _fake_playlist("system") + listener = MidiControlListener( + {"system": pl}, [{"type": "note", "number": 36, "action": "skip"}] + ) + with ( + mock.patch.object(listener, "_dispatch", side_effect=RuntimeError("boom")), + self.assertLogs("c64cast.midi_control", level="ERROR"), + ): + listener._midi_port = _ScriptedPort([mido.Message("note_on", note=36, velocity=100)]) + listener._stop.clear() + t = threading.Thread(target=listener._reader, daemon=True) + t.start() + time.sleep(0.05) + self.assertTrue(t.is_alive()) + listener._stop.set() + t.join(timeout=1.0) + self.assertFalse(t.is_alive()) + + +class PortSelectionTests(_MidiControlTestCase): + """_open_port name resolution, exercised with mido patched so no real + MIDI hardware is touched — mirrors test_midi_scene.py's + PortSelectionTests.""" + + def _patch_mido(self, names, opened): + fake = mock.MagicMock() + fake.get_input_names.return_value = names + fake.open_input.side_effect = lambda n: opened.append(n) or _FakePort() + return mock.patch.object(midi_control, "mido", fake) + + def _listener(self, port=None): + pl = _fake_playlist("system") + return MidiControlListener({"system": pl}, [], port=port) + + def test_empty_port_picks_first(self): + listener = self._listener(port="") + opened: list[str] = [] + with self._patch_mido(["Port A", "Port B"], opened): + listener._open_port() + self.assertEqual(opened, ["Port A"]) + + def test_no_ports_raises(self): + listener = self._listener(port="") + with self._patch_mido([], []): + with self.assertRaises(RuntimeError): + listener._open_port() + + def test_substring_match(self): + listener = self._listener(port="launch") + opened: list[str] = [] + with self._patch_mido(["IAC Bus 1", "Launch Control XL"], opened): + listener._open_port() + self.assertEqual(opened, ["Launch Control XL"]) + + def test_no_match_raises(self): + listener = self._listener(port="nonexistent") + with self._patch_mido(["IAC Bus 1"], []): + with self.assertRaises(RuntimeError): + listener._open_port() + + +class LifecycleTests(_MidiControlTestCase): + def test_start_stop_lifecycle(self): + pl = _fake_playlist("system") + listener = MidiControlListener( + {"system": pl}, [{"type": "note", "number": 36, "action": "skip"}] + ) + with mock.patch.object( + listener, + "_open_port", + side_effect=lambda: setattr(listener, "_midi_port", _FakePort()), + ): + listener.start() + try: + reader = listener._reader_thread + assert reader is not None + self.assertTrue(reader.is_alive()) + finally: + port = listener._midi_port + listener.stop() + self.assertTrue(port.closed) + self.assertIsNone(listener._reader_thread) + + def test_empty_playlists_rejected(self): + with self.assertRaises(ValueError): + MidiControlListener({}, []) + + +class BuildListenerTests(unittest.TestCase): + def test_raises_when_midi_unavailable(self): + pl = _fake_playlist("system") + fake_cfg = mock.MagicMock() + fake_cfg.cc_map = [] + with mock.patch.object(midi_control, "MIDI_AVAILABLE", False): + with self.assertRaises(RuntimeError): + midi_control.build_midi_control_listener({"system": pl}, fake_cfg) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_playlist.py b/tests/test_playlist.py index 319dd05..a6dd0e7 100644 --- a/tests/test_playlist.py +++ b/tests/test_playlist.py @@ -494,6 +494,211 @@ def fire_skip(): self.assertGreaterEqual(s.teardown_count, 1, "skip should tear down the current scene") self.assertGreater(counter["n"], 1, "skip should land on a new interstitial") + # --- request_jump ------------------------------------------------------- + + def test_request_jump_single_scene_is_noop(self): + pl = Playlist( + [FakeScene("A", frames_until_done=10_000_000)], + FakeApi(), + target_fps=200.0, + heartbeat_interval=0.0, + stop_event=threading.Event(), + interstitial_factory=_transition_factory()[0], + ) + pl.request_jump(0) + self.assertIsNone(pl._jump_target) + self.assertFalse(pl.skip_event.is_set()) + + def test_request_jump_out_of_range_raises(self): + pl = Playlist( + [FakeScene("A", frames_until_done=1), FakeScene("B", frames_until_done=1)], + FakeApi(), + target_fps=200.0, + heartbeat_interval=0.0, + stop_event=threading.Event(), + interstitial_factory=_transition_factory()[0], + ) + with self.assertRaises(ValueError): + pl.request_jump(5) + + def test_request_jump_last_write_wins(self): + pl = Playlist( + [FakeScene(n, frames_until_done=1) for n in "ABC"], + FakeApi(), + target_fps=200.0, + heartbeat_interval=0.0, + stop_event=threading.Event(), + interstitial_factory=_transition_factory()[0], + ) + pl.request_jump(1) + pl.request_jump(2) + self.assertEqual(pl._jump_target, 2) + + def test_request_jump_lands_on_target_index(self): + # A long-running scene at index 0; jump straight to index 2 (C) — + # the walk-forward index+1 path must never land there on its own + # within the short run window. + scenes = [ + FakeScene("A", frames_until_done=10_000_000), + FakeScene("B", frames_until_done=1), + FakeScene("C", frames_until_done=10_000_000), + ] + api = FakeApi() + stop = threading.Event() + factory, counter = _transition_factory() + pl = Playlist( + scenes, + api, + target_fps=200.0, + heartbeat_interval=0.0, + stop_event=stop, + interstitial_factory=factory, + ) + + def fire_jump(): + time.sleep(0.05) + pl.request_jump(2) + time.sleep(0.2) + stop.set() + + threading.Thread(target=fire_jump, daemon=True).start() + pl.run() + self.assertGreaterEqual(scenes[2].setup_count, 1, "jump should land on scene C") + self.assertEqual(scenes[1].setup_count, 0, "jump must skip scene B entirely") + + def test_request_jump_skip_interstitial_bypasses_the_card(self): + # Both scenes run "forever" (target_fps=200 over a 0.25s window + # can't reach 10M frames) so the only scene transition possible in + # this window is the jump itself — a stray natural completion of B + # can't sneak in an unrelated interstitial and confound the count. + scenes = [ + FakeScene("A", frames_until_done=10_000_000), + FakeScene("B", frames_until_done=10_000_000), + ] + api = FakeApi() + stop = threading.Event() + factory, counter = _transition_factory() + pl = Playlist( + scenes, + api, + target_fps=200.0, + heartbeat_interval=0.0, + stop_event=stop, + interstitial_factory=factory, + ) + + result: dict[str, int | None] = {"baseline": None} + + def fire_jump(): + # Wait past the playlist's own startup interstitial (every + # playlist enters scene 0 via one "UP NEXT" card) so the + # baseline below reflects a settled, running scene A — + # otherwise a race against that first card would make the + # baseline nondeterministic. Assertions run in the main thread + # after pl.run() returns (an AssertionError raised here, in a + # background thread, wouldn't fail the test). + deadline = time.time() + 1.0 + while scenes[0].setup_count < 1 and time.time() < deadline: + time.sleep(0.005) + result["baseline"] = counter["n"] + pl.request_jump(1, skip_interstitial=True) + time.sleep(0.2) + stop.set() + + threading.Thread(target=fire_jump, daemon=True).start() + pl.run() + baseline = result["baseline"] + assert baseline is not None, "startup interstitial never completed" + self.assertEqual(counter["n"], baseline, "a cut jump must never build an interstitial") + self.assertGreaterEqual(scenes[1].setup_count, 1) + + def test_request_jump_interstitial_transition_uses_the_card(self): + scenes = [ + FakeScene("A", frames_until_done=10_000_000), + FakeScene("B", frames_until_done=10_000_000), + ] + api = FakeApi() + stop = threading.Event() + factory, counter = _transition_factory() + pl = Playlist( + scenes, + api, + target_fps=200.0, + heartbeat_interval=0.0, + stop_event=stop, + interstitial_factory=factory, + ) + + result: dict[str, int | None] = {"baseline": None} + + def fire_jump(): + deadline = time.time() + 1.0 + while scenes[0].setup_count < 1 and time.time() < deadline: + time.sleep(0.005) + result["baseline"] = counter["n"] + pl.request_jump(1, skip_interstitial=False) + time.sleep(0.2) + stop.set() + + threading.Thread(target=fire_jump, daemon=True).start() + pl.run() + baseline = result["baseline"] + assert baseline is not None, "startup interstitial never completed" + self.assertGreater( + counter["n"], baseline, "an interstitial-routed jump must build the card" + ) + self.assertGreaterEqual(scenes[1].setup_count, 1, "must still land on the target scene") + + def test_jump_to_audio_gated_scene_waits_on_the_same_gate_as_looping(self): + # A jump target that competes for the ensemble audio lock must + # block via _wait_for_audio_claim (the same gate single-scene + # looping uses) rather than silently falling through to + # _resolve_next_index's skip-past-gated-scenes behavior. Proven + # here by pre-setting stop_event so the wait exits immediately + # with current=None, instead of landing on the gated scene. + from unittest.mock import MagicMock + + from c64cast.ensemble import Ensemble, SystemStack + + def _stack(name): + return SystemStack( + name=name, + cfg=MagicMock(), + api=MagicMock(), + audio=None, + source=None, + playlist=MagicMock(), + key_poller=MagicMock(), + framebuffer=None, + preview_window=None, + recorder=None, + ) + + class AudioGatedScene(FakeScene): + def competes_for_audio_lock(self): + return True + + a = FakeScene("A", frames_until_done=1) + b = AudioGatedScene("B", frames_until_done=1) + stop_event = threading.Event() + pl = Playlist( + [a, b], + FakeApi(), + target_fps=200.0, + heartbeat_interval=0.0, + stop_event=stop_event, + interstitial_factory=_transition_factory()[0], + ) + ens = Ensemble(stacks=[_stack("sys"), _stack("other")], stop_event=stop_event) + ens.try_claim_audio("other") # slot held elsewhere + pl.ensemble = ens + pl.current = a + a.is_done = True + stop_event.set() # wait loop inside _wait_for_audio_claim exits immediately + pl.request_jump(1) + pl._advance() + self.assertIsNone(pl.current) + # --- busy-deferral (overlay.is_busy() defers scene teardown) --------- def test_busy_overlay_defers_scene_teardown(self):