Source code for scamp.transcriber

"""Module containing the :class:`Transcriber` class which records the playback of a group of
:class:`~scamp.instruments.ScampInstrument` objects to create a :class:`~scamp.performance.Performance`"""

#  ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++  #
#  This file is part of SCAMP (Suite for Computer-Assisted Music in Python)                      #
#  Copyright © 2020 Marc Evanstein <marc@marcevanstein.com>.                                     #
#                                                                                                #
#  This program is free software: you can redistribute it and/or modify it under the terms of    #
#  the GNU General Public License as published by the Free Software Foundation, either version   #
#  3 of the License, or (at your option) any later version.                                      #
#                                                                                                #
#  This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;     #
#  without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.     #
#  See the GNU General Public License for more details.                                          #
#                                                                                                #
#  You should have received a copy of the GNU General Public License along with this program.    #
#  If not, see <http://www.gnu.org/licenses/>.                                                   #
#  ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++  #

from __future__ import annotations
from .performance import Performance
from expenvelope import Envelope
from clockblocks import Clock, TempoEnvelope
from .instruments import ScampInstrument
from typing import Sequence


[docs]class Transcriber: """ Class responsible for transcribing notes played by instruments into a :class:`~scamp.performance.Performance`. It is possible to run multiple transcriptions simultaneously, for instance starting at different times, recording different instruments, or recording relative to different clocks. """ def __init__(self): self._transcriptions_in_progress = [] @property def transcriptions_in_progress(self) -> tuple[Performance]: """Tuple of all current transcriptions.""" return tuple(self._transcriptions_in_progress)
[docs] def is_transcribing(self) -> bool: """Checks if any transcriptions are in progress.""" return len(self._transcriptions_in_progress) > 0
[docs] def start_transcribing(self, instrument_or_instruments: ScampInstrument | Sequence[ScampInstrument], clock: Clock, units: str = "beats") -> Performance: """ Starts transcribing new performance on the given clock, consisting of the given instrument :param instrument_or_instruments: the instruments we notate in this Performance :param clock: which clock all timings are relative to :param units: one of ["beats", "time"]. Do we use the beats of the clock or the time? :return: the Performance that this transcription writes to, which will be updated as notes are played and acts as a handle when calling stop_transcribing. """ assert units in ("beats", "time") if not hasattr(instrument_or_instruments, "__len__"): instrument_or_instruments = [instrument_or_instruments] if len(instrument_or_instruments) == 0: raise ValueError("No instruments specified for transcription!") performance = Performance() for instrument in instrument_or_instruments: performance.new_part(instrument) if self not in instrument._transcribers_to_notify: instrument._transcribers_to_notify.append(self) clock.rouse_and_hold() # ensures that the call to clock.beat() is up-to-date, e.g. when session runs as server self._transcriptions_in_progress.append( (performance, clock, clock.beat(), units) ) clock.release_from_suspension() # releases return performance
[docs] def register_note(self, instrument: ScampInstrument, note_info: dict) -> None: """ Called when an instrument wants to register that it finished a note, records note in all transcriptions :param instrument: the ScampInstrument that played the note :param note_info: the note info dictionary on that note, containing time stamps, parameter changes, etc. """ assert note_info["end_time_stamp"] is not None, "Cannot register unfinished note!" param_change_segments = note_info["parameter_change_segments"] if note_info["start_time_stamp"].time_in_master == note_info["end_time_stamp"].time_in_master: return # loop through all the transcriptions in progress for performance, clock, clock_start_beat, units in self._transcriptions_in_progress: # figure out the start_beat and length relative to this transcription's clock and start beat start_beat_in_clock = Transcriber._resolve_time_stamp(note_info["start_time_stamp"], clock, units) end_beat_in_clock = Transcriber._resolve_time_stamp(note_info["end_time_stamp"], clock, units) note_start_beat = start_beat_in_clock - clock_start_beat note_length = end_beat_in_clock - start_beat_in_clock # handle split points (if applicable) by creating a note length sections tuple note_length_sections = None if len(note_info["split_points"]) > 0: note_length_sections = [] last_split = note_start_beat for split_point in note_info["split_points"]: split_point_beat = Transcriber._resolve_time_stamp(split_point, clock, units) note_length_sections.append(split_point_beat - last_split) last_split = split_point_beat if end_beat_in_clock > last_split: note_length_sections.append(end_beat_in_clock - last_split) note_length_sections = tuple(note_length_sections) # get curves for all the parameters extra_parameters = {} for param in note_info["parameter_start_values"]: if param in param_change_segments and len(param_change_segments[param]) > 0: levels = [note_info["parameter_start_values"][param]] # keep track of this in case of gaps between segments beat_of_last_level_recorded = start_beat_in_clock durations = [] curve_shapes = [] for param_change_segment in param_change_segments[param]: # no need to transcribe a param_change_segment that was aborted immediately if param_change_segment.duration == 0 and \ param_change_segment.end_level == param_change_segment.start_level: continue param_start_beat_in_clock = Transcriber._resolve_time_stamp( param_change_segment.start_time_stamp, clock, units) param_end_beat_in_clock = Transcriber._resolve_time_stamp( param_change_segment.end_time_stamp, clock, units) # if there's a gap between the last level we recorded and this segment, we need to fill it with # a flat segment that holds the last level recorded if param_start_beat_in_clock > beat_of_last_level_recorded: durations.append(param_start_beat_in_clock - beat_of_last_level_recorded) levels.append(levels[-1]) curve_shapes.append(0) durations.append(param_end_beat_in_clock - param_start_beat_in_clock) levels.append(param_change_segment.end_level) curve_shapes.append(param_change_segment.curve_shape) beat_of_last_level_recorded = param_end_beat_in_clock # again, if we end the curve early, then we need to add a flat filler segment if beat_of_last_level_recorded < note_start_beat + note_length: durations.append(note_start_beat + note_length - beat_of_last_level_recorded) levels.append(levels[-1]) curve_shapes.append(0) # assign to specific variables for pitch and volume, otherwise put in a dictionary of extra params if param == "pitch": # note that if the length of levels is 1, then there's been no meaningful animation # so just act like it's not animated. This probably shouldn't really come up. (It was # coming up before with zero-length notes, but now those are just skipped anyway.) if len(levels) == 1: pitch = levels[0] else: pitch = Envelope(levels, durations, curve_shapes) elif param == "volume": if len(levels) == 1: volume = levels[0] else: volume = Envelope(levels, durations, curve_shapes) else: if len(levels) == 1: extra_parameters[param] = levels[0] else: extra_parameters[param] = Envelope(levels, durations, curve_shapes) else: # assign to specific variables for pitch and volume, otherwise put in a dictionary of extra params if param == "pitch": pitch = note_info["parameter_start_values"]["pitch"] elif param == "volume": volume = note_info["parameter_start_values"]["volume"] else: extra_parameters[param] = note_info["parameter_start_values"][param] for instrument_part in performance.get_parts_by_instrument(instrument): # it'd be kind of weird for more than one part to have the same instrument, but if they did, # I suppose that each part should transcribe the note instrument_part.new_note( note_start_beat, note_length_sections if note_length_sections is not None else note_length, pitch, volume, note_info["properties"] )
@staticmethod def _resolve_time_stamp(time_stamp, clock, units): assert units in ("beats", "time") return time_stamp.beat_in_clock(clock) if units == "beats" else time_stamp.time_in_clock(clock)
[docs] def stop_transcribing(self, which_performance=None, tempo_envelope_tolerance=0.001) -> Performance: """ Stops transcribing a Performance and returns it. Defaults to the oldest started performance, unless otherwise specified. :param which_performance: which performance to stop transcribing; defaults to oldest started :param tempo_envelope_tolerance: error tolerance when extracting the absolute tempo envelope for the Performance :return: the created Performance """ transcription = None if which_performance is None: if len(self._transcriptions_in_progress) == 0: raise ValueError("Cannot stop transcribing performance, as none has been started!") transcription = self._transcriptions_in_progress.pop(0) else: for i, transcription in enumerate(self._transcriptions_in_progress): if transcription[0] == which_performance: transcription = self._transcriptions_in_progress.pop(i) break if transcription is None: raise ValueError("Cannot stop transcribing given performance, as it was never started!") transcribed_performance, transcription_clock, transcription_start_beat, units = transcription if units == "beats": transcribed_performance.tempo_envelope = transcription_clock.extract_absolute_tempo_envelope( transcription_start_beat, tolerance=tempo_envelope_tolerance ) elif transcription_clock.is_master(): # transcribing based on master time, so there can't be any tempo changes; just use a blank TempoEnvelope transcribed_performance.tempo_envelope = TempoEnvelope() else: transcribed_performance.tempo_envelope = transcription_clock.parent.extract_absolute_tempo_envelope( transcription_start_beat, tolerance=tempo_envelope_tolerance ) return transcribed_performance