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