From 1c4c64f753f9d47549f2e7676eb0f4cdc4753fbd Mon Sep 17 00:00:00 2001 From: NotoriousArnav Date: Tue, 17 Feb 2026 16:34:12 +0530 Subject: [PATCH 1/2] Add step sequencer with audio engine fixes - New standalone step sequencer TUI (simplesampler-seq) with 16-step grid, cursor navigation, pattern switching, count-in, and metronome - Fix audio crackling: configurable blocksize (1024 for sequencer), _Voice class with __slots__ for faster callback, voice cap (64), optimized drain loop - Add [/] keys for prev/next pattern navigation with wrap-around, solving access to 10+ patterns beyond the 1-9 shortcuts - Master config (ss_config.toml) with metronome and sequencer settings - Pattern save/load, bar-boundary pattern switching, BPM control - Non-blocking UI callbacks to keep engine thread off the GIL --- .gitignore | 1 + pyproject.toml | 1 + src/seq.py | 11 + src/simplesampler/audio/playback.py | 61 +- src/simplesampler/schemas/ss_config.py | 59 ++ src/simplesampler/sequencer/__init__.py | 0 src/simplesampler/sequencer/app.py | 719 ++++++++++++++++++++++++ src/simplesampler/sequencer/engine.py | 201 +++++++ src/simplesampler/sequencer/schema.py | 104 ++++ 9 files changed, 1137 insertions(+), 20 deletions(-) create mode 100644 src/seq.py create mode 100644 src/simplesampler/schemas/ss_config.py create mode 100644 src/simplesampler/sequencer/__init__.py create mode 100644 src/simplesampler/sequencer/app.py create mode 100644 src/simplesampler/sequencer/engine.py create mode 100644 src/simplesampler/sequencer/schema.py diff --git a/.gitignore b/.gitignore index 60df7b4..650621c 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,4 @@ wheels/ .venv *_bank.json +*_patterns.json diff --git a/pyproject.toml b/pyproject.toml index 91073b7..30bfe39 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,7 @@ dependencies = [ [project.scripts] simplesampler = "simplesampler.main:main" +simplesampler-seq = "simplesampler.sequencer.app:main" [build-system] requires = ["hatchling"] diff --git a/src/seq.py b/src/seq.py new file mode 100644 index 0000000..be6f576 --- /dev/null +++ b/src/seq.py @@ -0,0 +1,11 @@ +"""Dev entry point for the sequencer. Run with: uv run src/seq.py bank.json""" + +import sys +import os + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "simplesampler", "..")) + +from simplesampler.sequencer.app import main + +if __name__ == "__main__": + main() diff --git a/src/simplesampler/audio/playback.py b/src/simplesampler/audio/playback.py index 8918032..9da3080 100644 --- a/src/simplesampler/audio/playback.py +++ b/src/simplesampler/audio/playback.py @@ -2,25 +2,44 @@ import wave import numpy as np from collections import deque -from typing import List, Dict import os import sys +class _Voice: + """Lightweight voice object for the audio callback hot path. + + Uses __slots__ for fast attribute access — dict key hashing is + measurably slower when called thousands of times per second. + """ + + __slots__ = ("data", "idx") + + def __init__(self, data: np.ndarray): + self.data = data + self.idx = 0 + + class AudioPlayer: RATE = 44100 CHANNELS = 2 - BLOCKSIZE = 256 # ~5.8ms at 44100 Hz + MAX_VOICES = 64 # Drop oldest voices beyond this limit + + # Absolute ceiling: ~33ms at 44100 Hz. Keeps latency bounded + # even if the caller passes a huge value. + _MAX_BLOCKSIZE = 1456 + + def __init__(self, blocksize: int = 256): + self.blocksize = min(blocksize, self._MAX_BLOCKSIZE) - def __init__(self): # Lock-free pending queue: play_data() appends here, # callback drains into its own local list each cycle. - self._pending: deque = deque() - self._voices: List[Dict] = [] + self._pending: deque[_Voice] = deque() + self._voices: list[_Voice] = [] self.stream = sd.OutputStream( samplerate=self.RATE, - blocksize=self.BLOCKSIZE, + blocksize=self.blocksize, channels=self.CHANNELS, dtype="float32", latency="low", @@ -36,7 +55,7 @@ def play_data(self, data: np.ndarray): if data is None or len(data) == 0: return # deque.append is atomic in CPython — no lock needed - self._pending.append({"data": data, "idx": 0}) + self._pending.append(_Voice(data)) def play_wave_file(self, file_path: str): """Loads and plays a wav file immediately.""" @@ -56,33 +75,35 @@ def _callback(self, outdata: np.ndarray, frames: int, time, status): print(f"Audio status: {status}", file=sys.stderr) # Drain pending voices into our local list (lock-free reads) - while True: - try: - voice = self._pending.popleft() - self._voices.append(voice) - except IndexError: - break + pending = self._pending + voices = self._voices + while pending: + voices.append(pending.popleft()) + + # Enforce voice cap — drop oldest voices first + if len(voices) > self.MAX_VOICES: + del voices[: len(voices) - self.MAX_VOICES] # Zero the output buffer outdata[:] = 0.0 # Mix active voices - i = len(self._voices) - 1 + i = len(voices) - 1 while i >= 0: - voice = self._voices[i] - data = voice["data"] - idx = voice["idx"] + voice = voices[i] + data = voice.data + idx = voice.idx remaining = len(data) - idx to_read = min(frames, remaining) if to_read > 0: outdata[:to_read] += data[idx : idx + to_read] - voice["idx"] += to_read + voice.idx += to_read # Remove finished voices - if voice["idx"] >= len(data): - self._voices.pop(i) + if voice.idx >= len(data): + voices.pop(i) i -= 1 diff --git a/src/simplesampler/schemas/ss_config.py b/src/simplesampler/schemas/ss_config.py new file mode 100644 index 0000000..b2e9507 --- /dev/null +++ b/src/simplesampler/schemas/ss_config.py @@ -0,0 +1,59 @@ +""" +Master configuration for SimpleSampler (ss_config.toml). + +Search order: + 1. $XDG_CONFIG_HOME/simplesampler/ss_config.toml + 2. ./ss_config.toml (working directory) + +If not found, all defaults apply silently. +""" + +import os +import tomllib +from pydantic import BaseModel +from typing import Tuple + + +class MetronomeConfig(BaseModel): + enabled: bool = True + sound: str = "" # Path to WAV — empty means generated sine click + volume: float = 0.7 # 0.0 – 1.0 + accent_beat_1: bool = True # Louder click on beat 1 + + +class SequencerConfig(BaseModel): + default_bpm: int = 120 + steps_per_beat: int = 4 + time_signature: Tuple[int, int] = (4, 4) + pattern_count: int = 4 # Default number of empty patterns to create + + +class SSConfig(BaseModel): + metronome: MetronomeConfig = MetronomeConfig() + sequencer: SequencerConfig = SequencerConfig() + + +def load_config(override_path: str | None = None) -> SSConfig: + """ + Load ss_config.toml from the override path, XDG config dir, or cwd. + Returns defaults if no file is found. + """ + paths: list[str] = [] + + if override_path: + paths.append(override_path) + + # XDG_CONFIG_HOME (default ~/.config) + xdg = os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config")) + paths.append(os.path.join(xdg, "simplesampler", "ss_config.toml")) + + # Current working directory + paths.append(os.path.join(os.getcwd(), "ss_config.toml")) + + for path in paths: + if os.path.isfile(path): + with open(path, "rb") as f: + data = tomllib.load(f) + return SSConfig.model_validate(data) + + return SSConfig() diff --git a/src/simplesampler/sequencer/__init__.py b/src/simplesampler/sequencer/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/simplesampler/sequencer/app.py b/src/simplesampler/sequencer/app.py new file mode 100644 index 0000000..b48a7c8 --- /dev/null +++ b/src/simplesampler/sequencer/app.py @@ -0,0 +1,719 @@ +""" +SimpleSampler Step Sequencer TUI. + +Textual app with a step grid (rows = pads from bank, columns = steps), +cursor navigation, pattern switching, and playback with count-in. +""" + +import argparse +import asyncio +import json +import os +import sys +import numpy as np + +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.widgets import Label, Static +from textual.containers import Horizontal, ScrollableContainer + +from simplesampler.audio.playback import AudioPlayer +from simplesampler.schemas.config import Bank +from simplesampler.schemas.ss_config import SSConfig, load_config +from simplesampler.sequencer.schema import SequenceFile +from simplesampler.sequencer.engine import SequencerEngine + +# 10 MB sample cache limit +MAX_PRELOAD_BYTES = 10 * 1024 * 1024 + +# Characters for step display +STEP_ON = "\u25a0" # ■ +STEP_OFF = "\u00b7" # · +CURSOR = "\u25a3" # ▣ + + +class StepCell(Static): + """A single cell in the step grid.""" + + def __init__(self, pad_id: int, step: int, **kwargs): + super().__init__(STEP_OFF, **kwargs) + self.pad_id = pad_id + self.step = step + self.active = False + + def toggle(self) -> bool: + self.active = not self.active + self._update_display() + return self.active + + def set_active(self, active: bool): + self.active = active + self._update_display() + + def _update_display(self): + self.update(STEP_ON if self.active else STEP_OFF) + + +class SequencerApp(App): + CSS = """ + #status-bar { + dock: top; + height: 1; + padding: 0 1; + background: $primary; + color: $text; + text-style: bold; + } + #grid-container { + height: 1fr; + padding: 0 1; + } + .pad-row { + height: 1; + layout: horizontal; + } + .pad-label { + width: 14; + height: 1; + padding: 0 1 0 0; + } + .step-cell { + width: 3; + height: 1; + text-align: center; + content-align: center middle; + } + .step-cell.--cursor { + background: $accent; + text-style: bold; + } + .step-cell.--playhead { + background: $warning 40%; + } + .step-cell.--cursor.--playhead { + background: $accent; + text-style: bold; + } + .step-cell.--active { + color: $text; + } + .step-cell.--inactive { + color: $text-muted; + } + .step-header { + width: 3; + height: 1; + text-align: center; + content-align: center middle; + text-style: dim; + } + .header-pad-label { + width: 14; + height: 1; + } + #header-row { + height: 1; + layout: horizontal; + padding: 0 1; + } + #help-bar { + dock: bottom; + height: 2; + padding: 0 1; + background: $surface; + color: $text-muted; + } + """ + + BINDINGS = [ + Binding("p", "toggle_play", "Play/Stop", show=True), + Binding("space,enter", "toggle_step", "Toggle Step", show=True), + Binding("equal,plus", "bpm_up", "BPM +5", show=False), + Binding("minus", "bpm_down", "BPM -5", show=False), + Binding("n", "add_pattern", "New Pattern", show=True), + Binding("d", "delete_pattern", "Del Pattern", show=False), + Binding("m", "toggle_metronome", "Metronome", show=True), + Binding("s", "save_patterns", "Save", show=True), + Binding("l", "load_patterns", "Load", show=False), + Binding("q", "quit", "Quit", show=True), + Binding("up", "cursor_up", "Up", show=False), + Binding("down", "cursor_down", "Down", show=False), + Binding("left", "cursor_left", "Left", show=False), + Binding("right", "cursor_right", "Right", show=False), + Binding("1", "switch_pattern_1", "Pattern 1", show=False), + Binding("2", "switch_pattern_2", "Pattern 2", show=False), + Binding("3", "switch_pattern_3", "Pattern 3", show=False), + Binding("4", "switch_pattern_4", "Pattern 4", show=False), + Binding("5", "switch_pattern_5", "Pattern 5", show=False), + Binding("6", "switch_pattern_6", "Pattern 6", show=False), + Binding("7", "switch_pattern_7", "Pattern 7", show=False), + Binding("8", "switch_pattern_8", "Pattern 8", show=False), + Binding("9", "switch_pattern_9", "Pattern 9", show=False), + Binding("left_square_bracket", "prev_pattern", "Prev Pattern", show=False), + Binding("right_square_bracket", "next_pattern", "Next Pattern", show=False), + ] + + def __init__( + self, + bank_path: str, + pattern_path: str | None = None, + config_path: str | None = None, + **kwargs, + ): + super().__init__(**kwargs) + self.bank_path = bank_path + self.pattern_path = pattern_path + + # Load config + self.config: SSConfig = load_config(config_path) + seq_cfg = self.config.sequencer + + # Load bank + self.bank = self._load_bank() + self.pad_ids: list[int] = [p.id for p in self.bank.pads] + self.pad_names: dict[int, str] = {p.id: p.name for p in self.bank.pads} + self.pad_colors: dict[int, str] = {p.id: p.color for p in self.bank.pads} + + # Load or create sequence + if pattern_path and os.path.isfile(pattern_path): + self.sequence = SequenceFile.load(pattern_path) + else: + self.sequence = SequenceFile.create_default( + bpm=seq_cfg.default_bpm, + time_signature=seq_cfg.time_signature, + steps_per_beat=seq_cfg.steps_per_beat, + pattern_count=seq_cfg.pattern_count, + ) + + # Audio — larger blocksize than the sampler (1024 ≈ 23ms vs 256 ≈ 5.8ms). + # The sequencer plays pre-programmed patterns so the extra latency is + # imperceptible, but the bigger buffer gives far more GIL headroom and + # eliminates the output-underflow crackling. + self.audio = AudioPlayer(blocksize=1024) + self.sample_cache: dict[int, np.ndarray] = {} + self._preload_samples() + + # Load metronome click WAV if configured + metro_click = None + if self.config.metronome.sound and os.path.isfile(self.config.metronome.sound): + metro_click = self.audio.load_wav(self.config.metronome.sound) + + # Engine + self.engine = SequencerEngine( + audio=self.audio, + sequence=self.sequence, + sample_cache=self.sample_cache, + metronome_cfg=self.config.metronome, + metronome_click=metro_click, + on_step=self._on_step_callback, + on_count_in_beat=self._on_count_in_callback, + on_playback_start=self._on_playback_start_callback, + ) + + # Cursor position and previous position for targeted updates + self._cursor_row = 0 # Index into self.pad_ids + self._cursor_col = 0 # Step index + self._prev_cursor_row = 0 + self._prev_cursor_col = 0 + + # Playback head (for UI highlight) + self._playhead: int = -1 + self._prev_playhead: int = -1 + + # Pending pattern display + self._pending_pattern: int | None = None + + # Cell lookup cache — populated on_mount to avoid query_one per tick + self._cells: dict[tuple[int, int], StepCell] = {} # (pad_id, step) -> cell + self._status_label: Label | None = None # cached status bar widget + + def _load_bank(self) -> Bank: + try: + with open(self.bank_path, "r") as f: + data = json.load(f) + return Bank(**data) + except Exception as e: + print(f"Error loading bank: {e}", file=sys.stderr) + sys.exit(1) + + def _preload_samples(self): + print("Preloading samples...", file=sys.stderr) + total_size = 0 + for pad in self.bank.pads: + if pad.sample_path and os.path.exists(pad.sample_path): + try: + size = os.path.getsize(pad.sample_path) + if total_size + size <= MAX_PRELOAD_BYTES: + self.sample_cache[pad.id] = self.audio.load_wav(pad.sample_path) + total_size += size + print( + f" Loaded: {pad.name} ({size / 1024:.1f} KB)", + file=sys.stderr, + ) + else: + print(f" Skipped: {pad.name} (cache full)", file=sys.stderr) + except Exception as e: + print(f" Error: {pad.sample_path}: {e}", file=sys.stderr) + print( + f"Preload complete: {total_size / 1024 / 1024:.2f} MB cached", + file=sys.stderr, + ) + + # --- Compose UI --- + + def compose(self) -> ComposeResult: + total_steps = self.sequence.total_steps + pat_name = self._current_pattern_name() + pat_count = len(self.sequence.patterns) + sig = self.sequence.time_signature + + yield Label( + self._status_text(), + id="status-bar", + ) + + # Header row with step numbers + with Horizontal(id="header-row"): + yield Static("", classes="header-pad-label") + for s in range(total_steps): + yield Static(str(s + 1), classes="step-header") + + # Grid rows + with ScrollableContainer(id="grid-container"): + for row_idx, pad_id in enumerate(self.pad_ids): + with Horizontal(classes="pad-row"): + name = self.pad_names.get(pad_id, f"Pad {pad_id}") + # Truncate long names + if len(name) > 12: + name = name[:11] + "\u2026" + yield Static(name, classes="pad-label") + for s in range(total_steps): + cell = StepCell( + pad_id, + s, + classes="step-cell --inactive", + id=f"cell-{pad_id}-{s}", + ) + yield cell + + yield Static( + "[Space] Toggle [P] Play/Stop [+/-] BPM " + "[1-9] Pattern [\\[/\\]] Prev/Next [N] New [D] Del [M] Metro [S] Save [Q] Quit", + id="help-bar", + ) + + def on_mount(self) -> None: + """Load pattern data into grid, build cell cache, set initial cursor.""" + # Cache the status bar label + self._status_label = self.query_one("#status-bar", Label) + # Build cell lookup cache once — eliminates query_one during playback + total_steps = self.sequence.total_steps + for pad_id in self.pad_ids: + for s in range(total_steps): + try: + cell = self.query_one(f"#cell-{pad_id}-{s}", StepCell) + self._cells[(pad_id, s)] = cell + except Exception: + pass + self._sync_grid_from_sequence() + self._update_cursor() + + # --- Status bar --- + + def _status_text(self) -> str: + bpm = self.sequence.bpm + sig = self.sequence.time_signature + pat_name = self._current_pattern_name() + pat_idx = self.sequence.active_pattern + 1 + pat_total = len(self.sequence.patterns) + metro = "ON" if self.config.metronome.enabled else "OFF" + playing = "" + if self.engine.playing: + playing = f" \u25b6 Playing [step {self._playhead + 1}/{self.sequence.total_steps}]" + elif self._playhead >= 0: + playing = " Count-in..." + else: + playing = " \u25a0 Stopped" + + pending = "" + if self._pending_pattern is not None: + pend_name = self.sequence.patterns[self._pending_pattern].name + pending = f" \u2192 {pend_name}" + + return ( + f"BPM: {bpm} | {sig[0]}/{sig[1]} | " + f"Pattern: {pat_name}{pending} [{pat_idx}/{pat_total}] | " + f"Metro: {metro}{playing}" + ) + + def _refresh_status(self): + if self._status_label is not None: + self._status_label.update(self._status_text()) + + def _current_pattern_name(self) -> str: + idx = self.sequence.active_pattern + if 0 <= idx < len(self.sequence.patterns): + return self.sequence.patterns[idx].name + return "?" + + # --- Cursor management --- + + def _update_cursor(self): + """Update cursor highlight — only touches the old and new cursor cells.""" + # Remove cursor from old position + old_pad = ( + self.pad_ids[self._prev_cursor_row] + if self._prev_cursor_row < len(self.pad_ids) + else None + ) + if old_pad is not None: + old_cell = self._cells.get((old_pad, self._prev_cursor_col)) + if old_cell is not None: + old_cell.set_class(False, "--cursor") + + # Add cursor to new position + new_pad = ( + self.pad_ids[self._cursor_row] + if self._cursor_row < len(self.pad_ids) + else None + ) + if new_pad is not None: + new_cell = self._cells.get((new_pad, self._cursor_col)) + if new_cell is not None: + new_cell.set_class(True, "--cursor") + + # Track for next update + self._prev_cursor_row = self._cursor_row + self._prev_cursor_col = self._cursor_col + + def _move_playhead(self, new_step: int): + """Move the playhead highlight from prev column to new column. + + Only touches cells in the two affected columns — O(num_pads) not O(num_pads * num_steps). + """ + old = self._prev_playhead + # Remove --playhead from old column + if old >= 0: + for pad_id in self.pad_ids: + cell = self._cells.get((pad_id, old)) + if cell is not None: + cell.set_class(False, "--playhead") + + # Add --playhead to new column + if new_step >= 0: + for pad_id in self.pad_ids: + cell = self._cells.get((pad_id, new_step)) + if cell is not None: + cell.set_class(True, "--playhead") + + self._prev_playhead = new_step + + def _clear_playhead(self): + """Remove playhead highlight from the previous column.""" + if self._prev_playhead >= 0: + for pad_id in self.pad_ids: + cell = self._cells.get((pad_id, self._prev_playhead)) + if cell is not None: + cell.set_class(False, "--playhead") + self._prev_playhead = -1 + + def action_cursor_up(self): + if self._cursor_row > 0: + self._cursor_row -= 1 + self._update_cursor() + + def action_cursor_down(self): + if self._cursor_row < len(self.pad_ids) - 1: + self._cursor_row += 1 + self._update_cursor() + + def action_cursor_left(self): + if self._cursor_col > 0: + self._cursor_col -= 1 + self._update_cursor() + + def action_cursor_right(self): + if self._cursor_col < self.sequence.total_steps - 1: + self._cursor_col += 1 + self._update_cursor() + + # --- Step toggling --- + + def action_toggle_step(self): + if not self.pad_ids: + return + pad_id = self.pad_ids[self._cursor_row] + step = self._cursor_col + pattern = self.sequence.patterns[self.sequence.active_pattern] + pad_key = str(pad_id) + + # Ensure step list exists + total = self.sequence.total_steps + if pad_key not in pattern.steps: + pattern.steps[pad_key] = [0] * total + + # Toggle + current = pattern.steps[pad_key][step] + pattern.steps[pad_key][step] = 0 if current else 1 + + # Update cell via cache + cell = self._cells.get((pad_id, step)) + if cell is not None: + cell.set_active(pattern.steps[pad_key][step] == 1) + cell.set_class(cell.active, "--active") + cell.set_class(not cell.active, "--inactive") + + # --- Sync grid from sequence data --- + + def _sync_grid_from_sequence(self): + """Update all grid cells to reflect the current pattern's step data.""" + pattern = self.sequence.patterns[self.sequence.active_pattern] + total_steps = self.sequence.total_steps + + for pad_id in self.pad_ids: + pad_key = str(pad_id) + steps = pattern.steps.get(pad_key, []) + for s in range(total_steps): + cell = self._cells.get((pad_id, s)) + if cell is not None: + active = s < len(steps) and steps[s] == 1 + cell.set_active(active) + cell.set_class(active, "--active") + cell.set_class(not active, "--inactive") + + # --- Playback --- + + def action_toggle_play(self): + if self.engine.playing or self._playhead >= 0: + self.engine.stop() + self._playhead = -1 + self._pending_pattern = None + self._clear_playhead() + self._refresh_status() + else: + self._playhead = -2 # Sentinel: count-in started + self._refresh_status() + self.engine.start() + + def _on_step_callback(self, step: int): + """Called from engine thread on each step. + + Posts UI work to the event loop without blocking the engine + thread — critical for keeping audio timing tight. + """ + self._playhead = step + self._post_to_main(self._tick_ui, step) + + def _tick_ui(self, step: int): + """Run on the main thread — move playhead and refresh status bar.""" + self._move_playhead(step) + self._refresh_status() + + def _on_count_in_callback(self, beat: int): + """Called from engine thread during count-in.""" + self._playhead = -2 + self._post_to_main(self._refresh_status) + + def _on_playback_start_callback(self): + self._post_to_main(self._refresh_status) + + def _post_to_main(self, callback, *args): + """Fire-and-forget: schedule callback on Textual's event loop. + + Unlike call_from_thread, this does NOT block the calling thread. + Uses run_coroutine_threadsafe with Textual's app context so + widget mutations (set_class, update) trigger proper repaints. + """ + loop = self._loop + if loop is None or loop.is_closed(): + return + + async def _run(): + with self._context(): + callback(*args) + + asyncio.run_coroutine_threadsafe(_run(), loop=loop) + + # --- BPM --- + + def action_bpm_up(self): + self.sequence.bpm = min(300, self.sequence.bpm + 5) + self._refresh_status() + + def action_bpm_down(self): + self.sequence.bpm = max(20, self.sequence.bpm - 5) + self._refresh_status() + + # --- Pattern switching --- + + def _switch_pattern(self, index: int): + """Switch to pattern by 0-based index. During playback, queues for bar boundary.""" + if index < 0 or index >= len(self.sequence.patterns): + return + if self.engine.playing: + self._pending_pattern = index + self.engine.queue_pattern_switch(index) + self._refresh_status() + else: + self.sequence.active_pattern = index + self._pending_pattern = None + self._sync_grid_from_sequence() + self._update_cursor() + self._refresh_status() + + def action_switch_pattern_1(self): + self._switch_pattern(0) + + def action_switch_pattern_2(self): + self._switch_pattern(1) + + def action_switch_pattern_3(self): + self._switch_pattern(2) + + def action_switch_pattern_4(self): + self._switch_pattern(3) + + def action_switch_pattern_5(self): + self._switch_pattern(4) + + def action_switch_pattern_6(self): + self._switch_pattern(5) + + def action_switch_pattern_7(self): + self._switch_pattern(6) + + def action_switch_pattern_8(self): + self._switch_pattern(7) + + def action_switch_pattern_9(self): + self._switch_pattern(8) + + def action_prev_pattern(self): + """Switch to previous pattern, wrapping around to the last.""" + count = len(self.sequence.patterns) + if count <= 1: + return + current = ( + self.sequence.active_pattern + if not self.engine.playing + else ( + self._pending_pattern + if self._pending_pattern is not None + else self.sequence.active_pattern + ) + ) + new_idx = (current - 1) % count + self._switch_pattern(new_idx) + + def action_next_pattern(self): + """Switch to next pattern, wrapping around to the first.""" + count = len(self.sequence.patterns) + if count <= 1: + return + current = ( + self.sequence.active_pattern + if not self.engine.playing + else ( + self._pending_pattern + if self._pending_pattern is not None + else self.sequence.active_pattern + ) + ) + new_idx = (current + 1) % count + self._switch_pattern(new_idx) + + # --- Add / delete patterns --- + + def action_add_pattern(self): + from simplesampler.sequencer.schema import Pattern, _pattern_names + + count = len(self.sequence.patterns) + name = _pattern_names(count + 1)[-1] + self.sequence.patterns.append(Pattern(name=name)) + self._refresh_status() + + def action_delete_pattern(self): + if len(self.sequence.patterns) <= 1: + return # Can't delete the last pattern + idx = self.sequence.active_pattern + self.sequence.patterns.pop(idx) + if self.sequence.active_pattern >= len(self.sequence.patterns): + self.sequence.active_pattern = len(self.sequence.patterns) - 1 + self._sync_grid_from_sequence() + self._update_cursor() + self._refresh_status() + + # --- Metronome --- + + def action_toggle_metronome(self): + self.config.metronome.enabled = not self.config.metronome.enabled + self._refresh_status() + + # --- Save / Load --- + + def action_save_patterns(self): + path = self.pattern_path or self._default_pattern_path() + try: + self.sequence.save(path) + self.pattern_path = path + self.notify(f"Saved: {path}", severity="information") + except Exception as e: + self.notify(f"Save failed: {e}", severity="error") + + def action_load_patterns(self): + path = self.pattern_path or self._default_pattern_path() + if not os.path.isfile(path): + self.notify(f"File not found: {path}", severity="warning") + return + try: + self.sequence = SequenceFile.load(path) + self.engine.sequence = self.sequence + self._sync_grid_from_sequence() + self._update_cursor() + self._refresh_status() + self.notify(f"Loaded: {path}", severity="information") + except Exception as e: + self.notify(f"Load failed: {e}", severity="error") + + def _default_pattern_path(self) -> str: + base = os.path.splitext(self.bank_path)[0] + return f"{base}_patterns.json" + + # --- Cleanup --- + + async def action_quit(self): + self.engine.stop() + self.audio.cleanup() + self.exit() + + def on_unmount(self): + self.engine.stop() + self.audio.cleanup() + + +def main(): + parser = argparse.ArgumentParser( + description="SimpleSampler Step Sequencer", + prog="simplesampler-seq", + ) + parser.add_argument("bank", help="Path to bank JSON file") + parser.add_argument( + "-p", "--pattern", help="Path to pattern JSON file", default=None + ) + parser.add_argument("-c", "--config", help="Path to ss_config.toml", default=None) + args = parser.parse_args() + + if not os.path.isfile(args.bank): + print(f"Bank file not found: {args.bank}", file=sys.stderr) + sys.exit(1) + + app = SequencerApp( + bank_path=args.bank, + pattern_path=args.pattern, + config_path=args.config, + ) + app.run() + + +if __name__ == "__main__": + main() diff --git a/src/simplesampler/sequencer/engine.py b/src/simplesampler/sequencer/engine.py new file mode 100644 index 0000000..b03eeed --- /dev/null +++ b/src/simplesampler/sequencer/engine.py @@ -0,0 +1,201 @@ +""" +Sequencer playback engine. + +Daemon thread that advances through steps at BPM tempo, fires samples +via AudioPlayer, handles count-in, metronome, and bar-boundary pattern switching. +""" + +import threading +import time +import math +import numpy as np +from typing import Callable + +from simplesampler.audio.playback import AudioPlayer +from simplesampler.sequencer.schema import SequenceFile +from simplesampler.schemas.ss_config import MetronomeConfig + + +def generate_click( + frequency: float = 1000.0, + duration: float = 0.02, + volume: float = 0.7, + rate: int = 44100, +) -> np.ndarray: + """Generate a short sine-wave click for the metronome.""" + n_samples = int(rate * duration) + t = np.linspace(0, duration, n_samples, dtype=np.float32) + # Sine with fast exponential decay envelope + envelope = np.exp(-t * 40.0).astype(np.float32) + mono = (np.sin(2.0 * math.pi * frequency * t) * envelope * volume).astype( + np.float32 + ) + return np.column_stack((mono, mono)) + + +class SequencerEngine: + """Step sequencer playback engine with count-in and metronome.""" + + def __init__( + self, + audio: AudioPlayer, + sequence: SequenceFile, + sample_cache: dict[int, np.ndarray], + metronome_cfg: MetronomeConfig, + metronome_click: np.ndarray | None = None, + on_step: Callable[[int], None] | None = None, + on_count_in_beat: Callable[[int], None] | None = None, + on_playback_start: Callable[[], None] | None = None, + ): + self.audio = audio + self.sequence = sequence + self.sample_cache = sample_cache + self.metronome_cfg = metronome_cfg + self.on_step = on_step # Called with current step index + self.on_count_in_beat = on_count_in_beat # Called with beat number (1-4) + self.on_playback_start = on_playback_start + + # Metronome sounds + if metronome_click is not None: + self._click_normal = metronome_click + self._click_accent = metronome_click # Same if user-provided + else: + vol = metronome_cfg.volume + self._click_normal = generate_click(880.0, 0.02, vol) + self._click_accent = generate_click(1760.0, 0.02, min(vol * 1.3, 1.0)) + + self._playing = False + self._thread: threading.Thread | None = None + self._stop_event = threading.Event() + self._current_step = 0 + + # Pattern switching: queued index applied at bar boundary + self._pending_pattern: int | None = None + + @property + def playing(self) -> bool: + return self._playing + + @property + def current_step(self) -> int: + return self._current_step + + def queue_pattern_switch(self, index: int): + """Queue a pattern switch — applied at end of current bar.""" + if 0 <= index < len(self.sequence.patterns): + self._pending_pattern = index + + def start(self): + """Start playback with count-in.""" + if self._playing: + return + # Wait for previous daemon thread to finish if still alive + if self._thread is not None and self._thread.is_alive(): + self._thread.join(timeout=1.0) + self._stop_event.clear() + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + + def stop(self): + """Stop playback. + + Sets the stop event and returns immediately. The daemon thread + will exit on its own — no join() to avoid deadlocking with + call_from_thread in the step callback. + """ + self._stop_event.set() + self._playing = False + self._current_step = 0 + + def _step_interval(self) -> float: + """Seconds per step at current BPM.""" + return 60.0 / self.sequence.bpm / self.sequence.steps_per_beat + + def _run(self): + """Main playback loop — count-in then step through pattern.""" + interval = self._step_interval() + beats_per_bar = self.sequence.time_signature[0] + + # --- Count-in: play metronome for one full bar of beats --- + if self.metronome_cfg.enabled: + beat_interval = 60.0 / self.sequence.bpm + next_time = time.perf_counter() + for beat in range(1, beats_per_bar + 1): + if self._stop_event.is_set(): + return + # Play click + if self.metronome_cfg.accent_beat_1 and beat == 1: + self.audio.play_data(self._click_accent) + else: + self.audio.play_data(self._click_normal) + if self.on_count_in_beat: + self.on_count_in_beat(beat) + next_time += beat_interval + sleep_dur = next_time - time.perf_counter() + if sleep_dur > 0: + if self._stop_event.wait(timeout=sleep_dur): + return + else: + # Silent count-in — still wait one bar duration + bar_duration = (60.0 / self.sequence.bpm) * beats_per_bar + if self._stop_event.wait(timeout=bar_duration): + return + + # --- Playback loop --- + self._playing = True + self._current_step = 0 + if self.on_playback_start: + self.on_playback_start() + + total_steps = self.sequence.total_steps + next_time = time.perf_counter() + + while not self._stop_event.is_set(): + # Recalculate interval in case BPM changed + interval = self._step_interval() + + # Bar boundary: apply pending pattern switch + if self._current_step == 0 and self._pending_pattern is not None: + self.sequence.active_pattern = self._pending_pattern + self._pending_pattern = None + + # Get current pattern + pat_idx = self.sequence.active_pattern + if 0 <= pat_idx < len(self.sequence.patterns): + pattern = self.sequence.patterns[pat_idx] + else: + pattern = None + + # Fire samples for this step + if pattern is not None: + for pad_id_str, steps in pattern.steps.items(): + if self._current_step < len(steps) and steps[self._current_step]: + pad_id = int(pad_id_str) + if pad_id in self.sample_cache: + self.audio.play_data(self.sample_cache[pad_id]) + + # Metronome on beat boundaries + if self.metronome_cfg.enabled: + if self._current_step % self.sequence.steps_per_beat == 0: + beat_num = self._current_step // self.sequence.steps_per_beat + 1 + if self.metronome_cfg.accent_beat_1 and beat_num == 1: + self.audio.play_data(self._click_accent) + else: + self.audio.play_data(self._click_normal) + + # Notify UI + if self.on_step: + self.on_step(self._current_step) + + # Advance step + self._current_step = (self._current_step + 1) % total_steps + + # Drift-compensated sleep + next_time += interval + sleep_dur = next_time - time.perf_counter() + if sleep_dur > 0: + if self._stop_event.wait(timeout=sleep_dur): + break + + self._playing = False + self._current_step = 0 diff --git a/src/simplesampler/sequencer/schema.py b/src/simplesampler/sequencer/schema.py new file mode 100644 index 0000000..f9a79f1 --- /dev/null +++ b/src/simplesampler/sequencer/schema.py @@ -0,0 +1,104 @@ +""" +Pydantic models for sequencer pattern files. + +Pattern JSON structure: +{ + "bpm": 120, + "time_signature": [4, 4], + "steps_per_beat": 4, + "active_pattern": 0, + "patterns": [ + { + "name": "A", + "steps": { + "0": [1, 0, 0, 0, 1, 0, 0, 0, ...], + "2": [0, 0, 0, 0, 1, 0, 0, 0, ...] + } + } + ] +} + +Keys under "steps" are pad IDs (as strings — JSON limitation). +Only pads with at least one active step need entries. +""" + +import json +from pydantic import BaseModel +from typing import Dict, List, Tuple + + +class Pattern(BaseModel): + name: str + steps: Dict[str, List[int]] = {} # pad_id (str) -> list of 0/1 + + +class SequenceFile(BaseModel): + bpm: int = 120 + time_signature: Tuple[int, int] = (4, 4) + steps_per_beat: int = 4 + active_pattern: int = 0 + patterns: List[Pattern] = [] + + @property + def total_steps(self) -> int: + """Number of steps in one bar.""" + beats_per_bar = self.time_signature[0] + return beats_per_bar * self.steps_per_beat + + def ensure_step_lengths(self): + """Ensure all step lists match total_steps, padding or trimming.""" + n = self.total_steps + for pattern in self.patterns: + for pad_id, steps in pattern.steps.items(): + if len(steps) < n: + steps.extend([0] * (n - len(steps))) + elif len(steps) > n: + pattern.steps[pad_id] = steps[:n] + + def save(self, path: str): + """Save sequence to JSON file.""" + with open(path, "w") as f: + json.dump(self.model_dump(), f, indent=2) + + @classmethod + def load(cls, path: str) -> "SequenceFile": + """Load sequence from JSON file.""" + with open(path, "r") as f: + data = json.load(f) + seq = cls.model_validate(data) + seq.ensure_step_lengths() + return seq + + @classmethod + def create_default( + cls, + bpm: int, + time_signature: Tuple[int, int], + steps_per_beat: int, + pattern_count: int, + ) -> "SequenceFile": + """Create a new sequence file with empty patterns.""" + names = _pattern_names(pattern_count) + patterns = [Pattern(name=n) for n in names] + seq = cls( + bpm=bpm, + time_signature=time_signature, + steps_per_beat=steps_per_beat, + patterns=patterns, + ) + return seq + + +def _pattern_names(count: int) -> list[str]: + """Generate pattern names: A, B, C, ... Z, AA, AB, ...""" + names = [] + for i in range(count): + name = "" + n = i + while True: + name = chr(ord("A") + n % 26) + name + n = n // 26 - 1 + if n < 0: + break + names.append(name) + return names From 29efcea8f976dc9f92fc30a70e7e194acb41f819 Mon Sep 17 00:00:00 2001 From: NotoriousArnav Date: Tue, 17 Feb 2026 17:01:50 +0530 Subject: [PATCH 2/2] Fix thread-safety races in sequencer engine and app - Add _starting guard flag to engine.start() preventing double thread spawn during count-in; update playing property to reflect count-in state - Wrap _run() in try/finally via _run_inner() for guaranteed flag cleanup - Stop engine before sequence swap in action_load_patterns to prevent data race with daemon thread - Clear pending pattern refs before list mutation in action_delete_pattern - Bounds-check _pending_pattern in _status_text to handle stale index - Handle malformed TOML/invalid config in ss_config.py load_config() - Remove unused variables in compose() --- src/simplesampler/schemas/ss_config.py | 13 +++++++---- src/simplesampler/sequencer/app.py | 31 +++++++++++++++++++++----- src/simplesampler/sequencer/engine.py | 20 ++++++++++++----- 3 files changed, 49 insertions(+), 15 deletions(-) diff --git a/src/simplesampler/schemas/ss_config.py b/src/simplesampler/schemas/ss_config.py index b2e9507..e0d3d5d 100644 --- a/src/simplesampler/schemas/ss_config.py +++ b/src/simplesampler/schemas/ss_config.py @@ -9,8 +9,9 @@ """ import os +import sys import tomllib -from pydantic import BaseModel +from pydantic import BaseModel, ValidationError from typing import Tuple @@ -52,8 +53,12 @@ def load_config(override_path: str | None = None) -> SSConfig: for path in paths: if os.path.isfile(path): - with open(path, "rb") as f: - data = tomllib.load(f) - return SSConfig.model_validate(data) + try: + with open(path, "rb") as f: + data = tomllib.load(f) + return SSConfig.model_validate(data) + except (tomllib.TOMLDecodeError, ValidationError) as e: + print(f"Warning: ignoring bad config {path}: {e}", file=sys.stderr) + continue return SSConfig() diff --git a/src/simplesampler/sequencer/app.py b/src/simplesampler/sequencer/app.py index b48a7c8..f9b73b8 100644 --- a/src/simplesampler/sequencer/app.py +++ b/src/simplesampler/sequencer/app.py @@ -263,9 +263,6 @@ def _preload_samples(self): def compose(self) -> ComposeResult: total_steps = self.sequence.total_steps - pat_name = self._current_pattern_name() - pat_count = len(self.sequence.patterns) - sig = self.sequence.time_signature yield Label( self._status_text(), @@ -337,8 +334,11 @@ def _status_text(self) -> str: pending = "" if self._pending_pattern is not None: - pend_name = self.sequence.patterns[self._pending_pattern].name - pending = f" \u2192 {pend_name}" + if 0 <= self._pending_pattern < len(self.sequence.patterns): + pend_name = self.sequence.patterns[self._pending_pattern].name + pending = f" \u2192 {pend_name}" + else: + self._pending_pattern = None return ( f"BPM: {bpm} | {sig[0]}/{sig[1]} | " @@ -636,6 +636,11 @@ def action_delete_pattern(self): if len(self.sequence.patterns) <= 1: return # Can't delete the last pattern idx = self.sequence.active_pattern + # Clear pending pattern references on both app and engine BEFORE + # mutating the list — avoids the engine daemon thread applying a + # stale index between the pop and the cleanup. + self._pending_pattern = None + self.engine._pending_pattern = None self.sequence.patterns.pop(idx) if self.sequence.active_pattern >= len(self.sequence.patterns): self.sequence.active_pattern = len(self.sequence.patterns) - 1 @@ -665,13 +670,27 @@ def action_load_patterns(self): if not os.path.isfile(path): self.notify(f"File not found: {path}", severity="warning") return + # Stop engine before swapping the sequence to avoid racing the + # daemon thread which reads self.sequence on every step tick. + was_playing = self.engine.playing + if was_playing: + self.engine.stop() + self._playhead = -1 + self._pending_pattern = None + self._clear_playhead() try: self.sequence = SequenceFile.load(path) self.engine.sequence = self.sequence + self.engine._pending_pattern = None self._sync_grid_from_sequence() self._update_cursor() self._refresh_status() - self.notify(f"Loaded: {path}", severity="information") + if was_playing: + self.notify( + f"Loaded: {path} (playback stopped)", severity="information" + ) + else: + self.notify(f"Loaded: {path}", severity="information") except Exception as e: self.notify(f"Load failed: {e}", severity="error") diff --git a/src/simplesampler/sequencer/engine.py b/src/simplesampler/sequencer/engine.py index b03eeed..9f79ea3 100644 --- a/src/simplesampler/sequencer/engine.py +++ b/src/simplesampler/sequencer/engine.py @@ -65,6 +65,7 @@ def __init__( self._click_accent = generate_click(1760.0, 0.02, min(vol * 1.3, 1.0)) self._playing = False + self._starting = False # True while count-in is in progress self._thread: threading.Thread | None = None self._stop_event = threading.Event() self._current_step = 0 @@ -74,7 +75,7 @@ def __init__( @property def playing(self) -> bool: - return self._playing + return self._playing or self._starting @property def current_step(self) -> int: @@ -87,11 +88,12 @@ def queue_pattern_switch(self, index: int): def start(self): """Start playback with count-in.""" - if self._playing: + if self._playing or self._starting: return # Wait for previous daemon thread to finish if still alive if self._thread is not None and self._thread.is_alive(): self._thread.join(timeout=1.0) + self._starting = True self._stop_event.clear() self._thread = threading.Thread(target=self._run, daemon=True) self._thread.start() @@ -105,6 +107,7 @@ def stop(self): """ self._stop_event.set() self._playing = False + self._starting = False self._current_step = 0 def _step_interval(self) -> float: @@ -113,6 +116,15 @@ def _step_interval(self) -> float: def _run(self): """Main playback loop — count-in then step through pattern.""" + try: + self._run_inner() + finally: + self._playing = False + self._starting = False + self._current_step = 0 + + def _run_inner(self): + """Actual playback logic, wrapped by _run for cleanup.""" interval = self._step_interval() beats_per_bar = self.sequence.time_signature[0] @@ -143,6 +155,7 @@ def _run(self): # --- Playback loop --- self._playing = True + self._starting = False self._current_step = 0 if self.on_playback_start: self.on_playback_start() @@ -196,6 +209,3 @@ def _run(self): if sleep_dur > 0: if self._stop_event.wait(timeout=sleep_dur): break - - self._playing = False - self._current_step = 0