subsequence.display

  1"""Live terminal dashboard for composition playback.
  2
  3Provides a persistent status line showing the current bar, section, chord, BPM,
  4and key.  Optionally renders an ASCII grid visualisation of all running patterns
  5above the status line, showing which steps have notes and at what velocity.
  6
  7Log messages scroll above the dashboard without disruption.
  8
  9Enable it with a single call before ``play()``:
 10
 11```python
 12composition.display()          # status line only
 13composition.display(grid=True) # status line + pattern grid
 14composition.play()
 15```
 16
 17The status line updates every beat and looks like::
 18
 19	125.00 BPM  Key: E  Bar: 17.1  [chorus 1/8]  Chord: Em7
 20
 21The grid (when enabled) updates every bar and looks like::
 22
 23	kick_2          |█ · · · █ · · · █ · · · █ · · ·|
 24	snare_1         |· · · · ▓ · · · · · · · ▓ · · ·|
 25	bass            |▓ · · ▓ · · ▓ · ▓ · · · ▓ · · ·|
 26"""
 27
 28import logging
 29import shutil
 30import sys
 31import threading
 32import typing
 33
 34import subsequence.constants
 35
 36if typing.TYPE_CHECKING:
 37	from subsequence.composition import Composition
 38
 39
 40_NOTE_NAMES = ["C", "C#", "D", "D#", "E", "F", "F#", "G", "G#", "A", "A#", "B"]
 41
 42_MAX_GRID_COLUMNS = 32
 43_LABEL_WIDTH = 16
 44_MIN_TERMINAL_WIDTH = 40
 45_SUSTAIN = -1
 46
 47
 48class GridDisplay:
 49
 50	"""Multi-line ASCII grid visualisation of running pattern steps.
 51
 52	Renders one block per pattern showing which grid steps have notes and at
 53	what velocity.  Drum patterns (those with a ``drum_note_map``) show one
 54	row per drum sound; pitched patterns show a single summary row.
 55
 56	Not used directly — instantiated by ``Display`` when ``grid=True``.
 57	"""
 58
 59	def __init__ (self, composition: "Composition", scale: float = 1.0) -> None:
 60
 61		"""Store composition reference for reading pattern state.
 62
 63		Parameters:
 64			composition: The ``Composition`` instance to read running
 65				patterns from.
 66			scale: Horizontal zoom factor.  Snapped to the nearest
 67				integer ``cols_per_step`` so that all on-grid markers
 68				are uniformly spaced.  Default ``1.0`` = one visual
 69				column per grid step (current behaviour).
 70		"""
 71
 72		self._composition = composition
 73		self._scale = scale
 74		self._lines: typing.List[str] = []
 75
 76	@property
 77	def line_count (self) -> int:
 78
 79		"""Number of terminal lines the grid currently occupies."""
 80
 81		return len(self._lines)
 82
 83	# ------------------------------------------------------------------
 84	# Static helpers
 85	# ------------------------------------------------------------------
 86
 87	@staticmethod
 88	def _velocity_char (velocity: typing.Union[int, float]) -> str:
 89
 90		"""Map a MIDI velocity (0-127) to a single ANSI block character.
 91
 92		Returns:
 93			``">"`` for sustain (note still sounding),
 94			``"░"`` for velocity > 0 to < 31.75 (0 - < 25%),
 95			``"▒"`` for velocity >= 31.75 to < 63.5 (25% to < 50%),
 96			``"▓"`` for velocity >= 63.5 to < 95.25 (50% to < 75%),
 97			``"█"`` for velocity >= 95.25 (75% to 100%).
 98		"""
 99
100		if velocity == _SUSTAIN:
101			return ">"
102		if velocity == 0:
103			return "·"
104		
105		pct = velocity / 127.0
106		if pct < 0.25:
107			return "░"
108		if pct < 0.50:
109			return "▒"
110		if pct < 0.75:
111			return "▓"
112		return "█"
113
114	@staticmethod
115	def _cell_char (velocity: typing.Union[int, float], is_on_grid: bool) -> str:
116
117		"""Return the display character for a grid cell.
118
119		Non-zero velocities (attacks and sustain) always show their
120		velocity glyph.  Empty on-grid positions show ``"·"``, empty
121		between-grid positions show ``" "`` (space).
122		"""
123
124		if velocity != 0:
125			return GridDisplay._velocity_char(velocity)
126		return "·" if is_on_grid else " "
127
128	@staticmethod
129	def _midi_note_name (pitch: int) -> str:
130
131		"""Convert a MIDI note number to a human-readable name.
132
133		Examples: 60 → ``"C4"``, 42 → ``"F#2"``, 36 → ``"C2"``.
134		"""
135
136		octave = (pitch // 12) - 1
137		note = _NOTE_NAMES[pitch % 12]
138		return f"{note}{octave}"
139
140	# ------------------------------------------------------------------
141	# Grid building
142	# ------------------------------------------------------------------
143
144	def build (self) -> None:
145
146		"""Rebuild grid lines from the current state of all running patterns."""
147
148		lines: typing.List[str] = []
149		term_width = shutil.get_terminal_size(fallback=(80, 24)).columns
150
151		if term_width < _MIN_TERMINAL_WIDTH:
152			self._lines = []
153			return
154
155		for name, pattern in self._composition.running_patterns.items():
156
157			grid_size = min(getattr(pattern, "_default_grid", 16), _MAX_GRID_COLUMNS)
158			muted = getattr(pattern, "_muted", False)
159			drum_map: typing.Optional[typing.Dict[str, int]] = getattr(pattern, "_drum_note_map", None)
160
161			# Snap to integer cols_per_step for uniform marker spacing.
162			cols_per_step = max(1, round(self._scale))
163			visual_cols = grid_size * cols_per_step
164			display_cols = self._fit_columns(visual_cols, term_width)
165
166			# On-grid columns: exact multiples of cols_per_step.
167			on_grid = frozenset(
168				i * cols_per_step
169				for i in range(grid_size)
170				if i * cols_per_step < display_cols
171			)
172
173			if muted:
174				lines.extend(self._render_muted(name, display_cols, on_grid))
175			elif drum_map:
176				lines.extend(self._render_drum_pattern(name, pattern, drum_map, visual_cols, display_cols, on_grid))
177			else:
178				lines.extend(self._render_pitched_pattern(name, pattern, visual_cols, display_cols, on_grid))
179
180		self._lines = lines
181
182	# ------------------------------------------------------------------
183	# Rendering helpers
184	# ------------------------------------------------------------------
185
186	def _render_muted (self, name: str, display_cols: int, on_grid: typing.FrozenSet[int]) -> typing.List[str]:
187
188		"""Render a muted pattern as a single row of dashes."""
189
190		cells = " ".join("-" if col in on_grid else " " for col in range(display_cols))
191		label = f"({name})"[:_LABEL_WIDTH].ljust(_LABEL_WIDTH)
192		return [f"{label}|{cells}|"]
193
194	def _render_drum_pattern (
195		self,
196		name: str,
197		pattern: typing.Any,
198		drum_map: typing.Dict[str, int],
199		visual_cols: int,
200		display_cols: int,
201		on_grid: typing.FrozenSet[int],
202	) -> typing.List[str]:
203
204		"""Render a drum pattern with one row per distinct drum sound."""
205
206		lines: typing.List[str] = []
207
208		# Build reverse map: {midi_note: drum_name}.
209		reverse_map: typing.Dict[int, str] = {}
210		for drum_name, midi_note in drum_map.items():
211			if midi_note not in reverse_map:
212				reverse_map[midi_note] = drum_name
213
214		# Discover which pitches are present in the pattern.
215		velocity_grid = self._build_velocity_grid(pattern, visual_cols, display_cols)
216
217		if not velocity_grid:
218			return lines
219
220		# Sort rows by MIDI pitch (lowest first — kick before hi-hat).
221
222		for pitch in sorted(velocity_grid):
223			label_text = reverse_map.get(pitch, self._midi_note_name(pitch))
224			label = label_text[:_LABEL_WIDTH].ljust(_LABEL_WIDTH)
225			cells = " ".join(
226				self._cell_char(v, col in on_grid)
227				for col, v in enumerate(velocity_grid[pitch][:display_cols])
228			)
229			lines.append(f"{label}|{cells}|")
230
231		return lines
232
233	def _render_pitched_pattern (
234		self,
235		name: str,
236		pattern: typing.Any,
237		visual_cols: int,
238		display_cols: int,
239		on_grid: typing.FrozenSet[int],
240	) -> typing.List[str]:
241
242		"""Render a pitched pattern as a single summary row."""
243
244		# Collapse all pitches into a single row using max velocity per slot.
245		velocity_grid = self._build_velocity_grid(pattern, visual_cols, display_cols)
246
247		summary = [0] * display_cols
248		for pitch_velocities in velocity_grid.values():
249			for i, vel in enumerate(pitch_velocities[:display_cols]):
250				if vel > summary[i] or (vel == _SUSTAIN and summary[i] == 0):
251					summary[i] = vel
252
253		label = name[:_LABEL_WIDTH].ljust(_LABEL_WIDTH)
254		cells = " ".join(
255			self._cell_char(v, col in on_grid)
256			for col, v in enumerate(summary)
257		)
258		return [f"{label}|{cells}|"]
259
260	# ------------------------------------------------------------------
261	# Internal helpers
262	# ------------------------------------------------------------------
263
264	def _build_velocity_grid (
265		self,
266		pattern: typing.Any,
267		grid_size: int,
268		display_cols: int,
269	) -> typing.Dict[int, typing.List[int]]:
270
271		"""Scan pattern steps and build a {pitch: [velocity_per_slot]} dict.
272
273		Each pitch gets a list of length *display_cols*.  At each grid slot
274		the highest velocity from any note at that position is stored.
275		"""
276
277		total_pulses = int(pattern.length * subsequence.constants.MIDI_QUARTER_NOTE)
278
279		if total_pulses <= 0 or grid_size <= 0:
280			return {}
281
282		pulses_per_slot = total_pulses / grid_size
283
284		velocity_grid: typing.Dict[int, typing.List[int]] = {}
285
286		for pulse, step in pattern.steps.items():
287			# Map pulse → grid slot.
288			slot = int(pulse / pulses_per_slot)
289
290			if slot < 0 or slot >= display_cols:
291				continue
292
293			for note in step.notes:
294				if note.pitch not in velocity_grid:
295					velocity_grid[note.pitch] = [0] * display_cols
296
297				if note.velocity > velocity_grid[note.pitch][slot]:
298					velocity_grid[note.pitch][slot] = note.velocity
299
300				# Fill sustain markers for slots where the note is
301				# still sounding.  Short notes (drums, staccato) never
302				# enter this loop.
303				end_pulse = pulse + note.duration
304				for s in range(slot + 1, display_cols):
305					if s * pulses_per_slot >= end_pulse:
306						break
307					if velocity_grid[note.pitch][s] == 0:
308						velocity_grid[note.pitch][s] = _SUSTAIN
309
310		return velocity_grid
311
312	@staticmethod
313	def _fit_columns (grid_size: int, term_width: int) -> int:
314
315		"""Determine how many grid columns fit in the terminal.
316
317		Each column occupies 2 characters (char + space), plus the label
318		prefix and pipe delimiters.
319		"""
320
321		# label (LABEL_WIDTH) + "|" + cells + "|"
322		# Each cell is "X " (2 chars) but last cell has no trailing space
323		# inside the pipes: "X . X ." is grid_size * 2 - 1 chars.
324		overhead = _LABEL_WIDTH + 2  # label + pipes
325		available = term_width - overhead
326
327		if available <= 0:
328			return 0
329
330		# Each column needs 2 chars (char + space), except the last needs 1.
331		max_cols = (available + 1) // 2
332
333		return min(grid_size, max_cols)
334
335
336class DisplayLogHandler (logging.Handler):
337
338	"""Logging handler that clears and redraws the status line around log output.
339
340	Installed by ``Display.start()`` and removed by ``Display.stop()``. Ensures
341	log messages do not overwrite or corrupt the persistent status line.
342	"""
343
344	def __init__ (self, display: "Display") -> None:
345
346		"""Store reference to the display for clear/redraw calls."""
347
348		super().__init__()
349		self._display = display
350
351	def emit (self, record: logging.LogRecord) -> None:
352
353		"""Clear the status line, write the log message, then redraw."""
354
355		try:
356			# emit() runs on whatever thread logged; hold the display's render
357			# lock across the whole clear → write → redraw sequence so a
358			# concurrent event-loop draw() cannot interleave ANSI sequences.
359			with self._display._render_lock:
360
361				self._display.clear_line()
362
363				msg = self.format(record)
364				sys.stderr.write(msg + "\n")
365				sys.stderr.flush()
366
367				self._display.draw()
368
369		except Exception:
370			self.handleError(record)
371
372
373class Display:
374
375	"""Live-updating terminal dashboard showing composition state.
376
377	Reads bar, section, chord, BPM, and key from the ``Composition`` and renders
378	a persistent region to stderr.  When ``grid=True`` an ASCII pattern grid is
379	rendered above the status line.  A custom ``DisplayLogHandler`` ensures log
380	messages scroll cleanly above the dashboard.
381
382	Example:
383		```python
384		composition.display(grid=True)
385		composition.play()
386		```
387	"""
388
389	def __init__ (self, composition: "Composition", grid: bool = False, grid_scale: float = 1.0) -> None:
390
391		"""Store composition reference for reading playback state.
392
393		Parameters:
394			composition: The ``Composition`` instance to read state from.
395			grid: When True, render an ASCII grid of running patterns
396				above the status line.
397			grid_scale: Horizontal zoom factor for the grid (default
398				``1.0``).  Snapped internally to the nearest integer
399				``cols_per_step`` for uniform marker spacing.
400		"""
401
402		self._composition = composition
403		self._active: bool = False
404		self._handler: typing.Optional[DisplayLogHandler] = None
405		self._saved_handlers: typing.List[logging.Handler] = []
406		self._last_line: str = ""
407		self._last_bar: typing.Optional[int] = None
408		self._cached_section: typing.Any = None
409		self._grid: typing.Optional[GridDisplay] = GridDisplay(composition, scale=grid_scale) if grid else None
410		self._last_grid_bar: typing.Optional[int] = None
411		self._drawn_line_count: int = 0
412		# Serialises terminal writes: update()/draw() run on the event loop,
413		# but DisplayLogHandler.emit() runs on whatever thread logged (e.g.
414		# the web UI's HTTP worker) — unsynchronised, an emit mid-draw would
415		# interleave ANSI sequences and garble the dashboard.  RLock because
416		# emit() holds it across its clear_line() → write → draw() sequence.
417		self._render_lock = threading.RLock()
418
419	def start (self) -> None:
420
421		"""Install the log handler and activate the display.
422
423		Saves existing root logger handlers and replaces them with a
424		``DisplayLogHandler`` that clears/redraws the status line around
425		each log message. Original handlers are restored by ``stop()``.
426		"""
427
428		if self._active:
429			return
430
431		self._active = True
432
433		root_logger = logging.getLogger()
434
435		# Save existing handlers so we can restore them on stop.
436		self._saved_handlers = list(root_logger.handlers)
437
438		# Build the replacement handler, inheriting the formatter from the
439		# first existing handler (if any) for consistent log formatting.
440		self._handler = DisplayLogHandler(self)
441
442		if self._saved_handlers and self._saved_handlers[0].formatter:
443			self._handler.setFormatter(self._saved_handlers[0].formatter)
444		else:
445			self._handler.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s"))
446
447		root_logger.handlers.clear()
448		root_logger.addHandler(self._handler)
449
450	def stop (self) -> None:
451
452		"""Clear the status line and restore original log handlers."""
453
454		if not self._active:
455			return
456
457		self.clear_line()
458		self._active = False
459
460		root_logger = logging.getLogger()
461		root_logger.handlers.clear()
462
463		for handler in self._saved_handlers:
464			root_logger.addHandler(handler)
465
466		self._saved_handlers = []
467		self._handler = None
468
469	def update (self, _: int = 0) -> None:
470
471		"""Rebuild and redraw the dashboard; called on ``"bar"`` and ``"beat"`` events.
472
473		The integer argument (bar or beat number) is ignored — state is read directly from
474		the composition.
475
476		Note: "bar" and "beat" events are emitted as ``asyncio.create_task`` at the start
477		of each pulse, but the tasks only execute *after* ``_advance_pulse()`` completes
478		(which includes sending MIDI via ``_process_pulse()``). The display therefore
479		always trails the audio slightly — this is inherent to the architecture and cannot
480		be avoided without restructuring the sequencer loop.
481		"""
482
483		if not self._active:
484			return
485
486		self._last_line = self._format_status()
487
488		# Rebuild grid data only when the bar counter changes.
489		if self._grid is not None:
490			current_bar = self._composition.sequencer.current_bar
491			if current_bar != self._last_grid_bar:
492				self._last_grid_bar = current_bar
493				self._grid.build()
494
495		self.draw()
496
497	def draw (self) -> None:
498
499		"""Write the current dashboard to the terminal."""
500
501		if not self._active or not self._last_line:
502			return
503
504		with self._render_lock:
505
506			grid_lines = self._grid._lines if self._grid is not None else []
507			total = len(grid_lines) + 1  # grid lines + status line
508
509			if grid_lines:
510				total += 1  # separator line
511
512			# Move cursor up to overwrite the previously drawn region.
513			# Cursor sits on the last line (status) with no trailing newline,
514			# so we move up (total - 1) to reach the first line.
515			if self._drawn_line_count > 1:
516				sys.stderr.write(f"\033[{self._drawn_line_count - 1}A")
517
518			if grid_lines:
519				sep = "-" * len(grid_lines[0])
520				sys.stderr.write(f"\r\033[K{sep}\n")
521				for line in grid_lines:
522					sys.stderr.write(f"\r\033[K{line}\n")
523
524			# Status line (no trailing newline — cursor stays on this line).
525			sys.stderr.write(f"\r\033[K{self._last_line}")
526			# Clear to end of screen in case the grid shrank since the last draw
527			# (e.g. a pattern disappeared), which would otherwise leave the old
528			# status line stranded one line below the new one.
529			sys.stderr.write("\033[J")
530			sys.stderr.flush()
531
532			self._drawn_line_count = total
533
534	def clear_line (self) -> None:
535
536		"""Erase the entire dashboard region from the terminal."""
537
538		if not self._active:
539			return
540
541		with self._render_lock:
542
543			if self._drawn_line_count > 1:
544				# Cursor is on the last line (no trailing newline).
545				# Move up (total - 1) to reach the first line.
546				sys.stderr.write(f"\033[{self._drawn_line_count - 1}A")
547
548				# Clear each line.
549				for _ in range(self._drawn_line_count):
550					sys.stderr.write("\r\033[K\n")
551
552				# Move cursor back up to the starting position.
553				sys.stderr.write(f"\033[{self._drawn_line_count}A")
554			else:
555				sys.stderr.write("\r\033[K")
556
557			sys.stderr.flush()
558			self._drawn_line_count = 0
559
560	def _format_status (self) -> str:
561
562		"""Build the status string from current composition state."""
563
564		parts: typing.List[str] = []
565		comp = self._composition
566
567		parts.append(f"{comp.sequencer.current_bpm:.2f} BPM")
568
569		if comp.key:
570			parts.append(f"Key: {comp.key}")
571
572		bar  = max(0, comp.sequencer.current_bar)  + 1
573		beat = max(0, comp.sequencer.current_beat) + 1
574		parts.append(f"Bar: {bar}.{beat}")
575
576		# Section info (only when form is configured).
577		# Cache refreshes only when the bar counter changes, keeping
578		# the section display in sync with the bar display even though
579		# the form state advances one beat early (due to lookahead).
580		if comp.form_state is not None:
581			current_bar = comp.sequencer.current_bar
582
583			if current_bar != self._last_bar:
584				self._last_bar = current_bar
585				self._cached_section = comp.form_state.get_section_info()
586
587			section = self._cached_section
588
589			if section:
590				section_str = f"[{section.name} {section.bar + 1}/{section.bars}"
591				if section.next_section:
592					section_str += f" \u2192 {section.next_section}"
593				section_str += "]"
594				parts.append(section_str)
595			else:
596				parts.append("[form finished]")
597
598		# Current chord (only when harmony is configured).  Reads the harmony
599		# window at the playhead, so it tracks sub-bar harmonic rhythm and
600		# covers progression-only mode (no engine) — the display refreshes
601		# per beat, which bounds how stale it can be.
602		chord = comp.current_chord()
603		if chord is not None:
604			parts.append(f"Chord: {chord.name()}")
605
606		# Conductor signals (when any are registered).  builder_bar is the
607		# lookahead bar - deliberately the same time base the pattern builders
608		# read, so the status line shows the values shaping what you are about
609		# to hear (the section line above shows playing-bar time instead).
610		conductor = comp.conductor
611		if conductor.signal_names:
612			beat = comp.builder_bar * comp.sequencer.time_signature[0]
613			for name in conductor.signal_names:
614				value = conductor.get(name, beat)
615				parts.append(f"{name.title()}: {value:.2f}")
616
617		return "  ".join(parts)