subsequence.sequencer

The Sequencer — clock, scheduling, and MIDI delivery.

Owns the pulse clock, the event heap, and the output ports, turning scheduled patterns and callbacks into timed MIDI messages. Composition drives one internally; you rarely construct it directly.

Module Contents

class subsequence.sequencer.BpmTransition[source]

State for a gradual BPM transition.

class subsequence.sequencer.MidiEvent[source]

Represents a MIDI event scheduled at a specific pulse.

sequence is a FIFO tie-breaker: when two events share a pulse, the one pushed first dispatches first. Without it, heapq ordering of same-pulse events is undefined, which would break NRPN/RPN sequences (CC 99 → 98 → 6 → 38) and risk reordering bank-select-before-program-change. The Sequencer assigns sequence via its monotonic _event_counter at push time (see Sequencer._push_event); direct constructions in tests can leave it at the default.

classmethod from_mido(pulse: int, msg: mido.Message | mido.MetaMessage, device: int = 0) MidiEvent[source]

Convert a mido.Message to a MidiEvent.

to_mido() mido.Message | mido.MetaMessage | None[source]

Convert this event to a mido.Message, or None if it’s an internal type (like OSC).

class subsequence.sequencer.PatternLike[source]

Bases: Protocol

Protocol for pattern objects that can be scheduled.

on_reschedule() None[source]

Hook called immediately before the pattern is rescheduled.

class subsequence.sequencer.ScheduledCallback[source]

Tracks a repeating callback and its scheduling metadata.

class subsequence.sequencer.ScheduledCallbackSequence[source]

Tracks a self-rescheduling callback whose firing interval varies per hop.

The variable-interval counterpart to ScheduledCallback — built for clocks that walk irregular spans (the harmonic clock under a bound progression). Each fire targets a boundary pulse; the callback’s return value sets the distance to the next boundary.

callback[source]

Called with the boundary pulse it is preparing. Returns the number of beats to the next boundary, or None to stop firing (the sequence is dropped from the queue). May be sync or async.

boundary_pulse[source]

The pulse this fire prepares (the musical boundary, not the fire time).

lookahead_pulses[source]

How far before each boundary the callback fires.

next_fire_pulse[source]

When the next fire is due.

class subsequence.sequencer.ScheduledPattern[source]

Tracks a repeating pattern and its scheduling metadata.

class subsequence.sequencer.Sequencer(output_device_name: str | None = None, initial_bpm: float = 125, time_signature: Tuple[int, int] = (4, 4), input_device_name: str | None = None, clock_follow: bool = False, clock_output: bool = False, record: bool = False, record_filename: str | None = None, spin_wait: bool = True, _jitter_log: List[float] | None = None)[source]

The engine that drives Subsequence timing and MIDI output.

The Sequencer maintains a stable clock (internal or external), handles the scheduling of MIDI events, and triggers pattern rebuilds.

Initialize the sequencer with MIDI devices and initial BPM.

Parameters:
  • output_device_name – MIDI output device name. When omitted, auto-discovers available devices - uses the only device if one is found, or prompts the user to choose if multiple are available.

  • initial_bpm – Tempo in BPM (ignored when clock_follow is True)

  • input_device_name – Optional MIDI input device name for clock/transport

  • clock_follow – When True, follow external MIDI clock instead of internal clock

  • clock_output – When True, send MIDI timing clock (0xF8), start (0xFA), and stop (0xFC) messages so connected hardware can sync to Subsequence’s tempo. Mutually exclusive with clock_follow (ignored when both are set, to prevent feedback loops).

  • record – When True, record all MIDI events to a file.

  • record_filename – Optional filename for the recording (defaults to timestamp).

  • spin_wait – When True (default), use a hybrid sleep+spin strategy for the final sub-millisecond of each pulse interval. This significantly reduces clock jitter at the cost of ~1–5% extra CPU. Set to False to use pure asyncio.sleep() (lower CPU, higher jitter).

  • _jitter_log – Optional list to append per-pulse jitter values (seconds) to during playback. Intended for the clock jitter benchmark — not for general use.

add_input_device(name: str, port: Any) int[source]

Register an additional input device. Returns the device index.

add_output_device(name: str, port: Any, latency_ms: float = 0.0) int[source]

Register an additional output device. Returns the device index.

latency_ms is the device’s physical output latency for compensation (see set_device_latency()).

disable_spin_wait() None[source]

Disable the hybrid sleep+spin timing strategy.

By default the sequencer busy-waits for the final sub-millisecond of each pulse interval to minimise clock jitter. Call this to revert to pure asyncio.sleep() — lower CPU usage at the cost of higher jitter (typically ±0.5–2 ms on Linux vs ~3–4 μs with spin-wait enabled (see the README clock-accuracy benchmark)).

Can also be set at construction time: Sequencer(spin_wait=False).

on_event(event_name: str, callback: Callable[Ellipsis, Any]) None[source]

Register a callback for a named event.

async panic() None[source]

Send a MIDI panic message to all channels.

async play() None[source]

Convenience method to start playback and wait for completion.

save_recording() None[source]

Save the recorded session to a MIDI file.

async schedule_callback_repeating(callback: Callable[[int], Any], interval_beats: float, start_pulse: int = 0, reschedule_lookahead: float = 1) None[source]

Schedule a repeating callback on a beat interval.

async schedule_callback_sequence(callback: Callable[[int], float | None | Coroutine[Any, Any, float | None]], start_pulse: int = 0, reschedule_lookahead: float = 1) None[source]

Schedule a self-rescheduling callback with a variable firing interval.

Where schedule_callback_repeating() fires on a fixed beat interval, this primitive lets the callback decide each hop: it is called lookahead before every boundary pulse, receives that boundary pulse, and returns the number of beats to the next boundary — or None to stop. Built for clocks that walk irregular spans, e.g. the harmonic clock under a bound progression’s harmonic rhythm.

The first fire targets start_pulse as its boundary and is due lookahead before it (immediately, when that is already past — the same backshift idiom as the repeating scheduler).

Parameters:
  • callbackfn(boundary_pulse) -> beats_to_next_boundary | None. May be sync or async.

  • start_pulse – The first boundary pulse.

  • reschedule_lookahead – How many beats before each boundary the callback fires.

async schedule_pattern(pattern: PatternLike, start_pulse: int) None[source]

Schedules a pattern’s notes and CC events into the sequencer’s event queue.

If pattern.mirrors is non-empty, every note, CC, pitch bend, program change, SysEx, NRPN/RPN burst, and drone event is duplicated onto each mirror destination — see the “MIDI mirroring” section of the README.

Bandwidth note: each mirror destination multiplies the per-pattern event count. A dense pattern with two mirrors emits 3× the original. For DIN-MIDI hardware on a saturated bus this can matter; over USB or IAC it is rarely a concern.

Tuning + mirrors: CcEvent.channel and CcEvent.device overrides (used by polyphonic microtonal tuning to rotate notes onto separate channels) apply to the primary destination only. Mirror destinations always use their own pinned (device, channel) — i.e. a polyphonic- tuning pattern mirrored to another synth will collapse all channel rotations onto the mirror’s single channel, losing per-note bend isolation on that destination. Apply tuning per-pattern if both ends need it.

async schedule_pattern_repeating(pattern: PatternLike, start_pulse: int) None[source]

Schedule a pattern and register it for rescheduling each cycle.

set_bpm(bpm: float) None[source]

Instantly change the tempo.

Note: If clock_follow is enabled and the sequencer is running, this method will be ignored as the tempo is slaved to the external source. When Ableton Link is active, the new BPM is proposed to the Link network instead of being applied locally — the network-authoritative tempo is then picked up on the next pulse.

set_device_latency(device: subsequence.midi_utils.DeviceId, latency_ms: float) None[source]

Set an output device’s physical latency (ms) for delay compensation.

Latency is normalised engine-wide: the slowest device plays at its logical time and every faster device’s output is deferred by max_latency its_latency so all devices sound together. See _dispatch_with_compensation().

Parameters:
  • device – Output device id (int index, name str, or None for device 0).

  • latency_ms – Non-negative physical output latency in milliseconds. Raises ValueError if negative or the device is unknown.

set_target_bpm(target_bpm: float, bars: int, shape: str | subsequence.easing.EasingFn = 'linear') None[source]

Smoothly transition to a new tempo over a fixed number of bars.

Parameters:
  • target_bpm – The BPM to ramp toward.

  • bars – Duration of the transition in bars.

  • shape – Easing curve — a name string (e.g. "ease_in_out") or any callable that maps [0, 1] → [0, 1]. Defaults to "linear". "ease_in_out" or "s_curve" are recommended for natural- sounding tempo changes. See subsequence.easing.

Note

When Ableton Link is active the shared network tempo is authoritative, so a local ramp cannot be honoured — this call is ignored. Use set_bpm() to propose a new tempo to the Link session instead.

async start() None[source]

Start the sequencer playback in a separate asyncio task.

When an input device is configured, the MIDI input port is opened here (after the event loop is running) so that call_soon_threadsafe works. When clock_output is True, a MIDI Start (0xFA) message is sent before the first clock tick so connected hardware begins from the top.

async stop() None[source]

Stop the sequencer playback and cleanup resources.

property midi_in: Any | None[source]

Return the primary input port (device 0), or None.

property midi_out: Any | None[source]

Return the primary output port (device 0), or None.