"""
Module containing user-facing playback classes: :class:`Ensemble`, :class:`ScampInstrument`, and :class:`NoteHandle`/
:class:`ChordHandle`
The underlying implementation of playback is done by :class:`~scamp.playback_implementations.PlaybackImplementation` and
all of its subclasses, which are found in playback_implementations.py.
"""
# ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ #
# 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
import itertools
from ._soundfont_host import get_best_preset_match_for_name, print_soundfont_presets
from ._midi import get_available_midi_output_devices, print_available_midi_output_devices
from .utilities import SavesToJSON, NoteProperty
from .spelling import SpellingPolicy
from .note_properties import NoteProperties
from .playback_implementations import PlaybackImplementation, SoundfontPlaybackImplementation, \
MIDIStreamPlaybackImplementation, OSCPlaybackImplementation
from .settings import engraving_settings, playback_settings
from ._animation import get_animation_driver
from clockblocks import wait, current_clock, Clock, ClockKilledError, DeadClockError, TimeStamp, Moment
from clockblocks.utilities import meaningfully_less_than, meaningfully_greater_than
from expenvelope import EnvelopeSegment
import logging
import warnings
from threading import Lock
from typing import Sequence, TypeAlias
from numbers import Real
from expenvelope import Envelope
from copy import deepcopy
import atexit
[docs]
class Ensemble(SavesToJSON):
"""
Host for multiple :class:`ScampInstrument` objects, keeping shared resources, and shared default settings.
A :class:`~scamp.session.Session` is, among other things, an Ensemble.
:param default_audio_driver: value to initialize default_audio_driver instance variable to
:param default_soundfont: value to initialize default_soundfont instance variable to
:param default_spelling_policy: a :class:`~scamp.spelling.SpellingPolicy` (or a string or tuple interpretable as
such) to use for all instruments in this ensemble, overriding scamp defaults.
:param instruments: list of instruments to populate this ensemble with. NOTE: generally it is not a good idea
to initialize an Ensemble with this argument, but better to use the new_part methods after the fact. This is
because instrument playback implementations look to share ensemble resources when they are created, and this
is not possible if they are not already part of an ensemble.
:ivar default_audio_driver: the audio driver instruments in this ensemble will default to. If "default", then
this defers to the scamp global playback_settings default.
:ivar default_soundfont: the soundfont that instruments in this ensemble will default to. If "default", then
this defers to the scamp global playback_settings default.
:ivar instruments: List of all of the ScampInstruments within the Ensemble.
"""
def __init__(self, default_soundfont: str = "default", default_audio_driver: str = "default",
default_spelling_policy: SpellingPolicy | str | tuple = None,
instruments: Sequence[ScampInstrument] = None):
self.default_soundfont = default_soundfont
self.default_audio_driver = default_audio_driver
self._default_spelling_policy = SpellingPolicy.interpret(default_spelling_policy) \
if default_spelling_policy is not None else None
self._instruments = list(instruments) if instruments is not None else []
for instrument in self._instruments:
instrument.set_ensemble(self)
@property
def instruments(self):
"""
Returns a tuple of the instruments currently in this Ensemble.
"""
return tuple(self._instruments)
[docs]
def add_instrument(self, instrument: ScampInstrument) -> ScampInstrument:
"""
Adds an instance of :class:`ScampInstrument` to this Ensemble. Generally, creating of and instrument and adding
it to an ensemble are done simultaneously via one of the "new_instrument" methods.
:param instrument: instrument to add to this ensemble
:return: self
"""
if not hasattr(instrument, "name") or instrument.name is None:
instrument.name = "Track " + str(len(self._instruments) + 1)
self._instruments.append(instrument)
instrument.set_ensemble(self)
return instrument
[docs]
def pop_instrument(self, index):
"""
Pops the instrument at the given index, severing its ties to the ensemble.
:param index: which instrument
"""
inst = self._instruments.pop(index)
inst.ensemble = None
inst.name_count = 0
return inst
[docs]
def new_silent_part(self, name: str = None, default_spelling_policy: SpellingPolicy | str | tuple = None,
clef_preference="from_name", start_note_fixed: bool = True) -> ScampInstrument:
"""
Creates and returns a new ScampInstrument for this Ensemble with no PlaybackImplementations.
:param name: name of the new part
:param default_spelling_policy: the :attr:`~ScampInstrument.default_spelling_policy` for the new part
:param clef_preference: the :attr:`~ScampInstrument.clef_preference` for the new part
:param start_note_fixed: the :attr:`~ScampInstrument.start_note_fixed` default for the new part. Only
relevant if adding a MIDI playback implementation.
:return: the newly created ScampInstrument
"""
return self.add_instrument(ScampInstrument(name, self, default_spelling_policy=default_spelling_policy,
clef_preference=clef_preference, start_note_fixed=start_note_fixed))
@staticmethod
def _resolve_preset_from_name(name, soundfont):
# if preset is auto, try to find a match in the soundfont
if name is None:
preset = (0, 0)
else:
preset_match, match_score = get_best_preset_match_for_name(name, which_soundfont=soundfont)
if match_score > 1.0:
preset = preset_match.bank, preset_match.preset
print("Using preset {} for {}".format(preset_match.name, name))
else:
logging.warning("Could not find preset matching {}. "
"Falling back to preset 0 (probably piano).".format(name))
preset = (0, 0)
return preset
[docs]
def new_part(self, name: str = None, preset="auto", soundfont: str = "default", num_channels: int = 8,
audio_driver: str = "default", max_pitch_bend: int = "default",
start_note_fixed: bool = True, default_spelling_policy: SpellingPolicy | str | tuple = None,
clef_preference="from_name", volume_cc_num: int = 11,
note_on_and_off_only: bool = False) -> ScampInstrument:
"""
Creates and returns a new ScampInstrument for this Ensemble that uses a SoundfontPlaybackImplementation. Unless
otherwise specified, the default soundfont for this Ensemble/Session will be used, and we will search for the
preset that best matches the name given.
:param name: name used for this instrument in score, etc.
:param preset: if an int, assumes bank #0; can also be a tuple of form (bank, preset). If "auto", searches
for a preset of the appropriate name.
:param soundfont: the name of the soundfont to use for fluidsynth playback
:param num_channels: maximum of midi channels available to this midi part. It's wise to use more when doing
microtonal playback, since pitch bends are applied per channel.
:param audio_driver: which audio driver to use for this instrument (defaults to ensemble default)
:param max_pitch_bend: max pitch bend to use for this instrument
:param start_note_fixed: default :attr:`~ScampInstrument.start_note_fixed` for notes started with
:func:`~ScampInstrument.start_note` on this part.
:param default_spelling_policy: the :attr:`~ScampInstrument.default_spelling_policy` for the new part
:param clef_preference: the :attr:`~ScampInstrument.clef_preference` for the new part
:param volume_cc_num: The cc value used for volume changes. Defaults to 11 (expression).
:param note_on_and_off_only: Deprecated; use `start_note_fixed`.
:return: the newly created ScampInstrument
"""
start_note_fixed = _resolve_deprecated_note_on_and_off_only(start_note_fixed, note_on_and_off_only)
# Resolve soundfont and audio driver to ensemble defaults if necessary (these may well be the string
# "default", in which case it gets resolved to the playback_settings default)
soundfont = self.default_soundfont if soundfont == "default" else soundfont
audio_driver = self.default_audio_driver if audio_driver == "default" else audio_driver
# if preset is auto, try to find a match in the soundfont
if preset == "auto":
preset = Ensemble._resolve_preset_from_name(name, soundfont)
elif isinstance(preset, int):
preset = (0, preset)
name = "Track " + str(len(self._instruments) + 1) if name is None else name
instrument = self.new_silent_part(name, default_spelling_policy=default_spelling_policy,
clef_preference=clef_preference, start_note_fixed=start_note_fixed)
instrument.add_soundfont_playback(preset=preset, soundfont=soundfont, num_channels=num_channels,
audio_driver=audio_driver, max_pitch_bend=max_pitch_bend,
volume_cc_num=volume_cc_num)
return instrument
[docs]
def new_midi_part(self, name: str = None, midi_output_device: int | str = None,
num_channels: int = 8, midi_output_name: str = None, max_pitch_bend: int = "default",
start_note_fixed: bool = True, default_spelling_policy: SpellingPolicy | str | tuple = None,
clef_preference="from_name", start_channel: int = 0,
volume_cc_num: int = 11, note_on_and_off_only: bool = False) -> ScampInstrument:
"""
Creates and returns a new ScampInstrument for this Ensemble that uses a MIDIStreamPlaybackImplementation.
This means that when notes are played by this instrument, midi messages are sent out to the given device.
:param name: name used for this instrument in score, etc. for a preset of the appropriate name.
:param midi_output_device: name or number of the device used to output midi. Call
get_available_midi_output_devices to check what's available.
:param num_channels: maximum of midi channels available to this midi part. It's wise to use more when doing
microtonal playback, since pitch bends are applied per channel.
:param midi_output_name: name of this part
:param max_pitch_bend: max pitch bend to use for this instrument
:param start_note_fixed: default :attr:`~ScampInstrument.start_note_fixed` for notes started with
:func:`~ScampInstrument.start_note` on this part.
:param default_spelling_policy: the :attr:`~ScampInstrument.default_spelling_policy` for the new part
:param clef_preference: the :attr:`~ScampInstrument.clef_preference` for the new part
:param start_channel: the first channel to use. For instance, if start_channel is 4, and num_channels is 5,
we will use channels (4, 5, 6, 7, 8). NOTE: channel counting in SCAMP starts from 0, so this may show
up as channels 5-9 in your MIDI software.
:param volume_cc_num: The cc value used for volume changes. Defaults to 11 (expression).
:param note_on_and_off_only: Deprecated; use `start_note_fixed`.
:return: the newly created ScampInstrument
"""
start_note_fixed = _resolve_deprecated_note_on_and_off_only(start_note_fixed, note_on_and_off_only)
name = "Track " + str(len(self._instruments) + 1) if name is None else name
midi_output_device = name if midi_output_device is None else midi_output_device
instrument = self.new_silent_part(name, default_spelling_policy=default_spelling_policy,
clef_preference=clef_preference, start_note_fixed=start_note_fixed)
instrument.add_streaming_midi_playback(midi_output_device=midi_output_device, num_channels=num_channels,
midi_output_name=midi_output_name, max_pitch_bend=max_pitch_bend,
start_channel=start_channel, volume_cc_num=volume_cc_num)
return instrument
[docs]
def new_osc_part(self, name: str = None, port: int = None, ip_address: str = "127.0.0.1",
message_prefix: str = None, osc_message_addresses: dict = "default",
default_spelling_policy: SpellingPolicy | str | tuple = None, clef_preference="from_name",
start_note_fixed: bool = True) -> ScampInstrument:
"""
Creates and returns a new ScampInstrument for this Ensemble that uses a OSCPlaybackImplementation. This means
that when notes are played by this instrument, osc messages are sent out to the specified address
:param name: name used for this instrument in score, etc. for a preset of the appropriate name.
:param port: port osc messages are sent to
:param ip_address: ip_address osc messages are sent to
:param message_prefix: prefix used for this instrument in osc messages
:param osc_message_addresses: dictionary defining the address used for each type of playback message. defaults
to using "start_note", "end_note", "change_pitch", "change_volume", "change_parameter". The default can
be changed in playback settings.
:param default_spelling_policy: the :attr:`~ScampInstrument.default_spelling_policy` for the new part
:param clef_preference: the :attr:`~ScampInstrument.clef_preference` for the new part
:param start_note_fixed: default :attr:`~ScampInstrument.start_note_fixed` for notes started with
:func:`~ScampInstrument.start_note` on this part.
:return: the newly created ScampInstrument
"""
name = "Track " + str(len(self._instruments) + 1) if name is None else name
instrument = self.new_silent_part(name, default_spelling_policy=default_spelling_policy,
clef_preference=clef_preference, start_note_fixed=start_note_fixed)
instrument.add_osc_playback(port=port, ip_address=ip_address, message_prefix=message_prefix,
osc_message_addresses=osc_message_addresses)
return instrument
def _get_part_name_count(self, name):
return sum(i.name == name for i in self._instruments)
[docs]
def get_instrument_by_name(self, name: str, which: int = 0):
"""
Returns the instrument of the given name.
:param name: name of the instrument to return
:param which: If there are multiple with the same name, this parameter specifies the one returned. (If none
match the number given by which, the first name match is returned)
"""
# if there are multiple instruments of the same name, which determines which one is chosen
imperfect_match = None
for instrument in self._instruments:
if name == instrument.name:
if which == instrument.name_count:
return instrument
else:
imperfect_match = instrument if imperfect_match is None else imperfect_match
return imperfect_match
[docs]
def print_default_soundfont_presets(self) -> None:
"""
Prints a list of presets available with the default soundfont.
"""
print_soundfont_presets(self.default_soundfont)
[docs]
@staticmethod
def get_available_midi_output_devices() -> enumerate:
"""
Returns an enumeration of available ports and devices for midi output.
"""
return get_available_midi_output_devices()
[docs]
@staticmethod
def print_available_midi_output_devices() -> None:
"""
Prints a list of available ports and devices for midi output.
"""
print_available_midi_output_devices()
@property
def default_spelling_policy(self) -> SpellingPolicy:
"""
Default spelling policy used for transcriptions made with this Ensemble.
"""
return self._default_spelling_policy
@default_spelling_policy.setter
def default_spelling_policy(self, value: SpellingPolicy | str | tuple):
self._default_spelling_policy = SpellingPolicy.interpret(value) if value is not None else None
def _to_dict(self):
return {
"default_soundfont": self.default_soundfont,
"default_audio_driver": self.default_audio_driver,
"default_spelling_policy": self.default_spelling_policy,
"instruments": self._instruments
}
@classmethod
def _from_dict(cls, json_dict):
return cls(**json_dict)
def __str__(self):
return "Ensemble(instruments=[{}])".format(", ".join(str(i) for i in self._instruments))
def __repr__(self):
return "Ensemble({})".format(", ".join("{}={}".format(k, repr(v)) for k, v in self._to_dict().items()))
NotePropertiesCompatible: TypeAlias = str | dict | NoteProperty | NoteProperties | Sequence['NotePropertiesCompatible']
PitchCompatible: TypeAlias = float | Envelope | Sequence[float] | Sequence[Sequence[float]] | None
VolumeCompatible: TypeAlias = float | Envelope | Sequence[float] | Sequence[Sequence[float]]
DurationCompatible: TypeAlias = float | tuple[float, ...]
def _resolve_deprecated_note_on_and_off_only(start_note_fixed: bool, note_on_and_off_only: bool) -> bool:
"""Fold the deprecated ``note_on_and_off_only`` constructor flag into ``start_note_fixed``."""
if note_on_and_off_only:
warnings.warn("'note_on_and_off_only' is deprecated: fully fixed notes are now the default for start_note "
"when no argument is an Envelope. Use 'start_note_fixed' (or start_note's 'fixed') instead.",
DeprecationWarning, stacklevel=3)
return True
return start_note_fixed
_RAISE_VOLUME_WARNING = (
"Cannot raise volume above {:.2f}. By default start_note uses the start volume as the note velocity, with "
"expression at 100% and only able to fall. Pass an explicit velocity to create headroom; volume then drives "
"expression instead."
)
[docs]
class ScampInstrument(SavesToJSON):
"""
Instrument class that does the playing of the notes. Generally this will be created through one of the
"new_part" methods of the Session or Ensemble class.
:param name: name of this instrument (e.g. when printed in a score)
:param ensemble: Ensemble to which this instrument will belong.
:param default_spelling_policy: sets :attr:`ScampInstrument.default_spelling_policy`
:param clef_preference: sets :attr:`ScampInstrument.clef_preference`
:param playback_implementations: PlaybackImplementation(s) used to actually playback notes
:ivar name: name of this instrument (e.g. when printed in a score)
:ivar name_count: when there are multiple instruments of the same name within an Ensemble, this variable assigns
each a unique index (starting with 0), to distinguish them
:ivar ensemble: Ensemble to which this instrument will belong.
:ivar playback_implementations: list of PlaybackImplementation(s) used to actually playback notes
"""
_note_id_generator = itertools.count()
def __init__(self, name: str = None, ensemble: Ensemble = None,
default_spelling_policy: SpellingPolicy | str | tuple = None,
clef_preference="from_name", playback_implementations: Sequence[PlaybackImplementation] = None,
start_note_fixed: bool = True):
super().__init__()
self.name = "" if name is None else name
self._clef_preference = None
self.clef_preference = clef_preference
self._transcribers_to_notify = []
self._note_info_by_id = {}
self.playback_implementations = [] if playback_implementations is None else playback_implementations
# What a start_note call's fixed="auto" resolves to (unless an Envelope argument or explicit velocity
# requires freedom). True means notes are fully fixed by default.
self.start_note_fixed = start_note_fixed
# A policy for spelling notes used as the default for this instrument. Overrides any broader defaults.
# (Has a getter and setter method allowing constructor strings to be passed.)
self.default_spelling_policy = default_spelling_policy
# this lock stops multiple threads from simultaneously accessing the self._note_info_by_id
self._note_info_lock = Lock()
# how far down the sustain pedal was last pressed, used as the default for pedal_change
self._last_pedal_press_amount = 1.0
#: used when exporting to json to see if this is the top level object being exported, or part of an ensemble
self._export_as_stand_alone = False
self.ensemble = None
self.name_count = 0
if ensemble is not None:
self.set_ensemble(ensemble)
atexit.register(self.end_all_notes)
[docs]
def set_ensemble(self, ensemble: Ensemble) -> None:
"""
Sets the ensemble that this instrument belongs to. Generally this happens automatically.
:param ensemble: the :class:`Ensemble` that this instrument should belong to.
"""
if self.ensemble == ensemble:
# already set to this ensemble
return
self.ensemble = ensemble
# used to help distinguish between identically named instruments in the same ensemble
self.name_count = ensemble._get_part_name_count(self.name)
[docs]
def play_note(self, pitch: PitchCompatible, volume: VolumeCompatible, length: DurationCompatible,
properties: NotePropertiesCompatible = None, blocking: bool = True, clock: Clock = None,
velocity: float = None, silent: bool = False, transcribe: bool = True) -> None:
"""
Play a note on this instrument, with the given pitch, volume and length.
:param pitch: either a number, an Envelope, or a list used to create an Envelope. MIDI pitch values are used,
with 60 representing middle C. However, microtones are allowed; for instance, a pitch of 64.7 produces an
F4 30 cents flat. A pitch of `None` simply translates to a rest.
:param volume: either a number, an Envelope, or a list used to create an Envelope. Volume is scaled from 0 to 1,
with 0 representing silence and 1 representing max volume.
:param length: either a number (of beats), or a tuple representing a set of tied segments
:param properties: Catch-all for a wide range of other playback and notation details that we may want to convey
about a note. See :ref:`The Note Properties Argument`
:param blocking: if True, don't return until the note is done playing; if False, return immediately
:param clock: which clock to use. If None, capture the clock from context.
:param velocity: for MIDI-based playback implementations (or other implementations that take velocity), fixes
the note-on velocity (0 to 1), turning volume into expression cc. By default (`None`) velocity
follows volume.
:param silent: if True, note is not played back, but is still transcribed when a
:class:`~scamp.transcriber.Transcriber` is active. (Generally ignored by end user.)
:param transcribe: if False, note is not transcribed even when a :class:`~scamp.transcriber.Transcriber` is
active. (Generally ignored by end user.)
"""
clock, blocking = self._resolve_clock(clock, blocking)
# A convenience: passing "None" to the pitch just causes a wait call
if pitch is None:
if blocking:
clock.wait(sum(length) if hasattr(length, '__len__') else length)
return
if not (hasattr(length, "__len__") and all(x > 0 for x in length) or length > 0):
raise ValueError("Note length must be positive.")
properties = NoteProperties.interpret(properties)
self._resolve_spelling_policies(properties)
if hasattr(pitch, "__len__"):
pitch = Envelope.from_list(pitch)
pitch.parsed_from_list = True
if hasattr(volume, "__len__"):
volume = Envelope.from_list(volume)
volume.parsed_from_list = True
ScampInstrument._normalize_envelopes(pitch, volume, length, properties)
adjusted_pitch, adjusted_volume, adjusted_length, did_an_adjustment = \
properties.apply_playback_adjustments(pitch, volume, length)
# A playback adjustment makes the heard and notated versions differ, so they play as two separate
# notes: an audible, adjusted one which is not transcribed, and a silent, unmodified one which is.
# With no adjustment the same note is both heard and notated.
# When fast-forwarding, playback adjustments are ignored, and the transcribed note is silent.
# Each forked note is described so that, if the clock that forked it ends before it's done, the resulting
# warning names the note rather than an internal clock.
# the adjusted note, only when we did an adjustment and aren't fast-forwarding
if did_an_adjustment and not clock.is_fast_forwarding():
playback_clock = clock.fork(
self._do_play_note, name="DO_PLAY_NOTE",
args=(adjusted_pitch, adjusted_volume, adjusted_length, properties),
kwargs={"velocity": velocity, "transcribe": False, "silent": silent}
)
playback_clock.description = f"a playback adjustment on {self.name!r}"
# the transcribed, unmodified note, silenced when paired with an adjusted note or when fast-forwarding
transcribe_silent = did_an_adjustment or clock.is_fast_forwarding() or silent
if blocking:
self._do_play_note(pitch, volume, length, properties, velocity=velocity,
silent=transcribe_silent, transcribe=transcribe)
else:
transcription_clock = clock.fork(
self._do_play_note, name="DO_PLAY_NOTE",
args=(pitch, volume, length, properties),
kwargs={"velocity": velocity, "silent": transcribe_silent, "transcribe": transcribe}
)
transcription_clock.description = f"a note on {self.name!r}"
def _resolve_spelling_policies(self, properties: NoteProperties):
"""
Resolves the spelling policies for the NoteProperties, based on instrument or ensemble defaults, if applicable
"""
# resolve the spelling policy based on defaults (local first, then more global)
if len(properties.spelling_policies) == 0:
# if the note doesn't say how to be spelled, check the instrument
if self.default_spelling_policy is not None:
properties.spelling_policies = [self.default_spelling_policy]
# if the instrument doesn't have a default spelling policy check the host (probably a Session)
elif self.ensemble is not None and self.ensemble.default_spelling_policy is not None:
properties.spelling_policies = [self.ensemble.default_spelling_policy]
# if the host doesn't have a default, then fall back to engraving_settings
else:
properties.spelling_policies = [engraving_settings.default_spelling_policy]
def _resolve_clock(self, clock, blocking):
"""
Resolves the clock argument, as well as the blocking state, given to several functions.
"""
if clock is not None:
# if the clock is given explicitly, go with that
return clock, blocking
elif current_clock() is not None:
# otherwise, try to get the clock active on the current thread
return current_clock(), blocking
elif isinstance(self.ensemble, Clock):
# If there is none, but the ensemble is a clock (meaning it's a Session probably), use that.
# Note that, in this case, we shouldn't block, because it causes issues with multiple threads
# calling wait on the same clock at the same time. This would happen when the Session is run as a server.
return self.ensemble, False
else:
return Clock(), blocking
@staticmethod
def _normalize_envelopes(pitch, volume, length, properties):
# length can either be a single number of beats or a list/tuple or segments to be split
# sum_length will represent the total number of beats in either case
sum_length = sum(length) if hasattr(length, "__len__") else length
# normalize envelopes to the duration of the note if the setting say to do so
if isinstance(pitch, Envelope) and (playback_settings.resize_parameter_envelopes == "always" or
playback_settings.resize_parameter_envelopes == "lists" and
hasattr(pitch, "parsed_from_list")):
pitch.normalize_to_duration(sum_length)
if isinstance(volume, Envelope) and (playback_settings.resize_parameter_envelopes == "always" or
playback_settings.resize_parameter_envelopes == "lists" and
hasattr(volume, "parsed_from_list")):
volume.normalize_to_duration(sum_length)
for param, value in properties.extra_playback_parameters.items():
if isinstance(value, Envelope) and (playback_settings.resize_parameter_envelopes == "always" or
playback_settings.resize_parameter_envelopes == "lists" and
hasattr(value, "parsed_from_list")):
value.normalize_to_duration(sum_length)
def _do_play_note(self, pitch, volume, length, properties, velocity=None, silent=False, transcribe=True):
"""
This runs the actual thread that plays the note, and is scheduled when play_note is called.
If playback adjustments were made, then we schedule the altered version of _do_play_note to play back, but with
"transcribe" set to false, and we schedule an unaltered version of _do_play_note to run silently, but with
"transcribe" set to true. This way the transcription is not affected by performance adjustments.
:param pitch: either a number, an Envelope
:param volume: either a number, an Envelope
:param length: either a number (of beats), or a tuple representing a set of tied segments
:param properties: a NoteProperties dictionary
:param velocity: an explicit note-on velocity, or None to follow volume (see start_note)
:param silent: if True, don't actually do any of the playback; just go through the motions for transcribing it
:param transcribe: if False, don't notify Transcribers at the end of the note
"""
clock = current_clock()
# We know the whole shape of the note up front: a plain scalar note can be fully fixed, but any Envelope
# (or an explicit velocity) requires the note to stay free to animate/use channel-wide cc messages.
if velocity is not None or isinstance(volume, Envelope) or isinstance(pitch, Envelope) \
or any(isinstance(param_val, Envelope) for param_val in properties.extra_playback_parameters.values()):
fixed = False
else:
fixed = True
# start the note. (Note that this will also start the animation of pitch, volume,
# and any other parameters if they are envelopes.)
note_flags = []
if silent:
note_flags.append("silent")
if not transcribe:
note_flags.append("no_transcribe")
note_handle = self.start_note(pitch, volume, properties, clock=clock, fixed=fixed, velocity=velocity,
flags=note_flags)
try:
if hasattr(length, "__len__"):
for length_segment in length:
clock.wait(length_segment)
note_handle.split()
else:
clock.wait(length)
note_handle.end()
except (ClockKilledError, DeadClockError) as e:
note_handle.end()
raise e
[docs]
def play_chord(self, pitches: Sequence[PitchCompatible], volume: VolumeCompatible, length: DurationCompatible,
properties: NotePropertiesCompatible = None, blocking: bool = True, clock: Clock = None,
velocity: float = None, silent: bool = False, transcribe: bool = True) -> None:
"""
Play a chord with the given pitches, volume, and length. Essentially, this is a convenience method that
bundles together several calls to "play_note" and takes a list of pitches rather than a single pitch
:param pitches: a list of pitches for the notes of this chord
:param volume: see :func:`play_note`
:param length: see :func:`play_note`
:param properties: see :ref:`The Note Properties Argument`
:param blocking: see description for "play_note"
:param clock: see description for "play_note"
:param velocity: see :func:`play_note`
:param silent: see description for "play_note"
:param transcribe: see description for "play_note"
"""
clock, blocking = self._resolve_clock(clock, blocking)
# A convenience: passing "None" to the pitch just causes a wait call
if pitches is None:
if blocking:
clock.wait(sum(length) if hasattr(length, '__len__') else length)
return
if not hasattr(pitches, "__len__"):
raise ValueError("'pitches' must be a list of pitches.")
properties = NoteProperties.interpret(properties)
self._resolve_spelling_policies(properties)
for i, pitch in enumerate(pitches):
# play the pitches individually, duplicating the properties dictionary for each.
# (note that individual noteheads and spellings need to be separated out)
properties_copy = properties.duplicate()
if len(properties.noteheads) > 1:
# if we've been given multiple noteheads, assign them note by note
# (if given too few, just repeat the last notehead)
properties_copy.noteheads = [properties_copy.noteheads[i]
if i < len(properties_copy.noteheads)
else properties_copy.noteheads[-1]]
if len(properties.spelling_policies) > 1:
# same with spelling policies
properties_copy.spelling_policies = [properties_copy.spelling_policies[i]
if i < len(properties_copy.spelling_policies)
else properties_copy.spelling_policies[-1]]
self.play_note(pitch, volume, length, properties=properties_copy,
blocking=(i == len(pitches) - 1) if blocking else False,
clock=clock, velocity=velocity, silent=silent, transcribe=transcribe)
[docs]
def start_note(self, pitch: PitchCompatible, volume: VolumeCompatible, properties: NotePropertiesCompatible = None,
clock: Clock = None, fixed: bool | str = "auto", velocity: float = None,
flags: Sequence[str] = None) -> NoteHandle:
"""
Start a note with the given pitch, volume, and properties
:param pitch: the pitch / starting pitch of the note
:param volume: the volume / starting volume of the note
:param properties: see :ref:`The Note Properties Argument`
:param clock: the clock on which to run any animation of pitch, volume, etc. If None, captures the clock from
context.
:param fixed: whether the note is locked at note-on. `True` maps volume directly to velocity and allows the
note to share a midi channel with other notes, but forbids any later changes. `False` gives the note its
own channel and allows pitch, volume, and other parameters to change. A note with an
:class:`~expenvelope.envelope.Envelope` argument or an explicit `velocity` must animate, so it is always
unfixed, ignoring this argument. The default `"auto"` falls back to this instrument's
:attr:`start_note_fixed`.
:param velocity: an explicit note-on velocity (0 to 1), decoupling the attack from volume. By default
(`None`) velocity follows the start volume, so expression starts at 100% and volume can only be lowered.
Given a value, the note attacks at that velocity while the volume argument drives expression directly,
over its full range.
:param flags: list of strings that act as flags for how the note should be processed. Should probably be
ignored by a normal user.
:return: a NoteHandle with which to later manipulate the note
"""
clock, _ = self._resolve_clock(clock, None)
# standardize properties if necessary, turn pitch and volume into lists if necessary
properties = NoteProperties.interpret(properties)
self._resolve_spelling_policies(properties)
pitch = Envelope.from_list(pitch) if hasattr(pitch, "__len__") else pitch
volume = Envelope.from_list(volume) if hasattr(volume, "__len__") else volume
# get the starting values for all the parameters to pass to the playback implementations
start_pitch = pitch.start_level() if isinstance(pitch, Envelope) else pitch
start_volume = volume.start_level() if isinstance(volume, Envelope) else volume
other_param_start_values = properties.get_extra_parameter_start_values()
# Resolve the fixed request ("auto" -> the instrument default), then force the note free if an Envelope
# argument or an explicit velocity requires it to animate.
explicitly_fixed = fixed is True
if fixed == "auto":
fixed = self.start_note_fixed
elif fixed is not True and fixed is not False:
raise ValueError("Invalid 'fixed' value {!r}; use True, False, or 'auto'.".format(fixed))
must_animate = velocity is not None or isinstance(volume, Envelope) or isinstance(pitch, Envelope) \
or any(isinstance(v, Envelope) for v in properties.extra_playback_parameters.values())
if must_animate:
if explicitly_fixed:
logging.warning("Ignoring fixed=True: a note with an Envelope argument or an explicit velocity "
"must be free to animate.")
fixed = False
# The note-on velocity and the divisor mapping volume onto expression. When velocity follows volume they
# are equal (to the start volume, or an envelope's peak), so expression starts at 100% and can only fall.
# An explicit velocity instead leaves expression its full range, with volume sent raw.
if velocity is None:
note_velocity = volume.max_level() if isinstance(volume, Envelope) else start_volume
volume_ceiling = note_velocity
else:
note_velocity = velocity
volume_ceiling = 1
flags = [] if flags is None else list(flags)
if fixed and "fixed" not in flags:
flags.append("fixed")
with self._note_info_lock:
# generate a new id for this note, and set up all of its info
note_id = next(ScampInstrument._note_id_generator)
self._note_info_by_id[note_id] = {
"clock": clock,
"start_time_stamp": TimeStamp.now(clock),
"end_time_stamp": None,
"split_points": [],
"parameter_start_values": dict(other_param_start_values, pitch=start_pitch, volume=start_volume),
"parameter_values": dict(other_param_start_values, pitch=start_pitch, volume=start_volume),
"parameter_change_segments": {},
"segments_list_lock": Lock(),
"note_info_lock": self._note_info_lock,
"properties": properties,
"velocity": note_velocity,
"volume_ceiling": volume_ceiling,
"explicit_velocity": velocity,
"fixed": fixed,
"flags": flags
}
if clock.is_fast_forwarding() and "silent" not in self._note_info_by_id[note_id]["flags"]:
self._note_info_by_id[note_id]["flags"].append("silent")
if "silent" not in self._note_info_by_id[note_id]["flags"]:
# otherwise, call all the playback implementation!
for playback_implementation in self.playback_implementations:
playback_implementation.start_note(
note_id, start_pitch, start_volume,
properties, self._note_info_by_id[note_id]
)
# we now exit the lock, since otherwise the following calls will not be able to happen
# create a handle for this note
handle = NoteHandle(note_id, self)
# start all the note animation for pitch, volume, and any extra parameters
# note that, if the note is silent, then start_note has added the silent flag to the note_info dict
# this will cause unsynchronized animation threads not to fire
if isinstance(pitch, Envelope):
handle.change_pitch(pitch.levels[1:], pitch.durations, pitch.curve_shapes, clock)
if isinstance(volume, Envelope):
handle.change_volume(volume.levels[1:], volume.durations, volume.curve_shapes, clock)
for param, value in properties.extra_playback_parameters.items():
if isinstance(value, Envelope):
handle.change_parameter(param, value.levels[1:], value.durations, value.curve_shapes, clock)
return handle
[docs]
def start_chord(self, pitches: Sequence[PitchCompatible], volume: VolumeCompatible,
properties: NotePropertiesCompatible = None, clock: Clock = None, fixed: bool | str = "auto",
velocity: float = None, flags: Sequence[str] = None) -> ChordHandle:
"""
Simple utility for starting chords without starting each note individually.
:param pitches: a list of pitches
:param volume: see :func:`start_note`
:param properties: see :ref:`The Note Properties Argument`
:param clock: see start_note
:param fixed: see start_note
:param velocity: see start_note
:param flags: see start_note
:return: a ChordHandle, which is used to manipulate the chord thereafter. Pitch change calls on the ChordHandle
are based on the first note of the chord; all other notes are shifted in parallel
"""
assert hasattr(pitches, "__len__")
properties = NoteProperties.interpret(properties)
self._resolve_spelling_policies(properties)
# we should either be given a number of noteheads equal to the number of pitches or just one notehead for all
assert len(properties.noteheads) == len(pitches) or len(properties.noteheads) == 1, \
"Wrong number of noteheads for chord."
note_handles = []
pitches = [Envelope.from_list(pitch) if hasattr(pitch, "__len__") else pitch for pitch in pitches]
first_pitch_start_level = pitches[0].start_level() if isinstance(pitches[0], Envelope) else pitches[0]
intervals = [(pitch.start_level() if isinstance(pitch, Envelope) else pitch) - first_pitch_start_level
for pitch in pitches]
for i, pitch in enumerate(pitches):
# for all but the last pitch, play it without blocking, so we can start all the others
# also copy the properties dictionary, and pick out the correct notehead if we've been given several
properties_copy = deepcopy(properties)
if len(properties.noteheads) > 1:
properties_copy.noteheads = [properties_copy.noteheads[i]]
note_handles.append(self.start_note(pitch, volume, properties=properties_copy, clock=clock,
fixed=fixed, velocity=velocity, flags=flags))
return ChordHandle(note_handles, intervals)
[docs]
def change_note_parameter(self, note_id: int | NoteHandle, param_name: str,
target_value_or_values: float | Sequence[float],
transition_length_or_lengths: float | Sequence[float] = 0,
transition_curve_shape_or_shapes: float | Sequence[float] = 0,
clock: Clock = None) -> None:
"""
Changes the value of parameter of note playback over a given time; can also take a sequence of targets and times
:param note_id: which note to affect (an id or a NoteHandle)
:param param_name: name of the parameter to affect. "pitch" and "volume" are special cases
:param target_value_or_values: target value (or list of values) for the parameter
:param transition_length_or_lengths: transition time(s) in beats to the target value(s)
:param transition_curve_shape_or_shapes: curve shape(s) for the transition(s)
:param clock: which clock all of this happens on; by default, reuses the clock that the note started on.
"""
with self._note_info_lock:
note_id = note_id.note_id if isinstance(note_id, NoteHandle) else note_id
note_info = self._note_info_by_id[note_id]
if clock is None:
clock = note_info["clock"]
assert isinstance(clock, Clock), "Invalid clock argument."
if note_info.get("fixed"):
logging.warning("Can't change the {} of a fixed note; start it with fixed=False to animate it."
.format(param_name))
return
if param_name == "volume" and note_info["volume_ceiling"] < 1:
# Expression can never exceed the note-on velocity, so volume can't rise above it. A
# follows-volume note starts at its ceiling (expression 100%); an explicit velocity leaves room.
ceiling = note_info["volume_ceiling"]
if hasattr(target_value_or_values, "__len__"):
if any(v > ceiling for v in target_value_or_values):
logging.warning(_RAISE_VOLUME_WARNING.format(ceiling))
target_value_or_values = [min(v, ceiling) for v in target_value_or_values]
elif target_value_or_values > ceiling:
logging.warning(_RAISE_VOLUME_WARNING.format(ceiling))
target_value_or_values = ceiling
# A sampled value is recorded and, unless the note is silent, pushed to playback. Pitch and
# volume have dedicated playback methods; any other parameter goes through change_note_parameter.
silent = "silent" in note_info["flags"]
if param_name == "pitch":
def apply_to_playback_implementation(impl, value): impl.change_note_pitch(note_id, value)
elif param_name == "volume":
def apply_to_playback_implementation(impl, value): impl.change_note_volume(note_id, value)
else:
def apply_to_playback_implementation(impl, value):
impl.change_note_parameter(note_id, param_name, value)
def parameter_change_function(value):
if not silent:
for playback_implementation in self.playback_implementations:
apply_to_playback_implementation(playback_implementation, value)
note_info["parameter_values"][param_name] = value
assert param_name in note_info["parameter_values"], \
"Cannot change parameter {}, as it was undefined at note start.".format(param_name)
# reuse existing segments list if we've animated this parameter before, otherwise mint one
segments_list = note_info["parameter_change_segments"].setdefault(param_name, [])
# if there was a previous segment changing this same parameter, and it's not done yet, we should abort it
if len(segments_list) > 0:
segments_list[-1].abort_if_running()
# All of a clock family's animations are sampled together by a single demand-driven tick, so
# concurrent animations coincide instead of each waking the scheduler on its own schedule.
driver = get_animation_driver(clock)
if hasattr(target_value_or_values, "__len__"):
# assume linear segments unless otherwise specified
transition_curve_shape_or_shapes = [0] * len(target_value_or_values) if \
transition_curve_shape_or_shapes == 0 else transition_curve_shape_or_shapes
assert hasattr(transition_length_or_lengths, "__len__") and \
hasattr(transition_curve_shape_or_shapes, "__len__")
assert len(target_value_or_values) == len(transition_length_or_lengths) == \
len(transition_curve_shape_or_shapes), \
"List of target values must be accompanied by a equal length list of transition lengths and shapes."
# Build the chained segments up front (each starts where the previous lands), then start
# them one after another: each segment's on_finish starts the next segment, so aborting
# the running one (e.g. when the note ends) leaves the rest of the sequence unstarted.
segments = []
segment_start_value = note_info["parameter_values"][param_name]
for target, length, shape in zip(target_value_or_values, transition_length_or_lengths,
transition_curve_shape_or_shapes):
segments.append(_ParameterChangeSegment(
parameter_change_function, segment_start_value, target, length, shape, clock, driver))
segment_start_value = target
def start_segment(i):
with note_info["segments_list_lock"]:
segments_list.append(segments[i])
on_finish = (lambda: start_segment(i + 1)) if i + 1 < len(segments) else None
segments[i].begin(silent=silent, on_finish=on_finish)
start_segment(0)
else:
parameter_change_segment = _ParameterChangeSegment(
parameter_change_function, note_info["parameter_values"][param_name], target_value_or_values,
transition_length_or_lengths, transition_curve_shape_or_shapes, clock, driver)
with note_info["segments_list_lock"]:
segments_list.append(parameter_change_segment)
parameter_change_segment.begin(silent=silent)
[docs]
def change_note_pitch(self, note_id: int | NoteHandle, target_value_or_values: float | Sequence[float],
transition_length_or_lengths: float | Sequence[float] = 0,
transition_curve_shape_or_shapes: float | Sequence[float] = 0,
clock: Clock = None) -> None:
"""
Change the pitch of an already started note; can also take a sequence of targets and times.
:param note_id: which note to affect (an id or a NoteHandle)
:param target_value_or_values: target value (or list of values) for the parameter
:param transition_length_or_lengths: transition time(s) in beats to the target value(s)
:param transition_curve_shape_or_shapes: curve shape(s) for the transition(s)
:param clock: which clock all of this happens on; by default, reuses the clock that the note started on.
"""
self.change_note_parameter(note_id, "pitch", target_value_or_values, transition_length_or_lengths,
transition_curve_shape_or_shapes, clock)
[docs]
def change_note_volume(self, note_id: int | NoteHandle, target_value_or_values: float | Sequence[float],
transition_length_or_lengths: float | Sequence[float] = 0,
transition_curve_shape_or_shapes: float | Sequence[float] = 0,
clock: Clock = None) -> None:
"""
Change the volume of an already started note; can also take a sequence of targets and times
:param note_id: which note to affect (an id or a NoteHandle)
:param target_value_or_values: target value (or list thereof) for the parameter
:param transition_length_or_lengths: transition time(s) in beats to the target value(s)
:param transition_curve_shape_or_shapes: curve shape(s) for the transition(s)
:param clock: which clock all of this happens on; "from_note" simply reuses the clock that the note started on.
"""
self.change_note_parameter(note_id, "volume", target_value_or_values, transition_length_or_lengths,
transition_curve_shape_or_shapes, clock)
[docs]
def split_note(self, note_id: int | NoteHandle) -> None:
"""
Adds a split point in a note, causing it later to be rendered as tied pieces.
:param note_id: Which note or NoteHandle to split
"""
with self._note_info_lock:
note_id = note_id.note_id if isinstance(note_id, NoteHandle) else note_id
note_info = self._note_info_by_id[note_id]
note_info["split_points"].append(TimeStamp.now(note_info["clock"]))
[docs]
def end_note(self, note_id: int | NoteHandle = None) -> None:
"""
Ends the note with the given note id. If none is specified, ends oldest note started.
Note that this only applies to notes started in an open-ended way with :func:`start_note`, notes created using
:func:`play_note` have their lifecycle controlled automatically.
:param note_id: either the id itself or a NoteHandle with that id. Default of None ends the oldest note
"""
with self._note_info_lock:
# in case we're passed a NoteHandle instead of an actual id number, get the number from the handle
note_id = note_id.note_id if isinstance(note_id, NoteHandle) else note_id
if note_id is not None:
# as specific note_id has been given, so it had better belong to a currently playing note!
if note_id not in self._note_info_by_id:
logging.warning("Tried to end a note that was never started (or already ended)!")
return
elif len(self._note_info_by_id) > 0:
# no specific id was given, so end the oldest note
# (note that ids just count up, so the lowest active id is the oldest)
note_id = min(self._note_info_by_id.keys())
else:
logging.warning("Tried to end a note that was never started (or already ended)!")
return
note_info = self._note_info_by_id[note_id]
# resolve the clock to use
clock = note_info["clock"]
# end any segments that are still changing
for param_name in note_info["parameter_change_segments"]:
if len(note_info["parameter_change_segments"][param_name]) > 0:
note_info["parameter_change_segments"][param_name][-1].abort_if_running()
# transcribe the note, if applicable
note_info["end_time_stamp"] = TimeStamp.now(clock)
if "no_transcribe" not in note_info["flags"]:
for transcriber in self._transcribers_to_notify:
transcriber.register_note(self, note_info)
# do the sonic implementation of ending the note, as long as it's not silent
if "silent" not in note_info["flags"]:
for playback_implementation in self. playback_implementations:
playback_implementation.end_note(note_id)
# remove from active notes and delete the note info
del self._note_info_by_id[note_id]
[docs]
def end_all_notes(self) -> None:
"""
Ends all notes currently playing
"""
while len(self._note_info_by_id) > 0:
self.end_note()
[docs]
def num_notes_playing(self) -> int:
"""
Returns the number of notes currently playing.
"""
return len(self._note_info_by_id)
"""
---------------------------------------- Adding and removing playback ----------------------------------------
"""
[docs]
def add_soundfont_playback(self, preset: str | int | tuple[int, int] = "auto", soundfont: str = "default",
num_channels: int = 8, audio_driver: str = "default", max_pitch_bend: int = "default",
note_on_and_off_only: bool = False,
volume_cc_num: int = 11) -> ScampInstrument:
"""
Add a soundfont playback implementation for this instrument.
:param preset: either a preset number, a tuple of (bank, preset), a string giving a name to search for in the
soundfont, or the string "auto", in which case the name of this instrument is used to search for a preset.
:param soundfont: which soundfont to use. This can be either a path to a soundfont or the name of one of the
soundfonts specified in playback_settings.named_soundfonts. If this instrument belongs to an Ensemble,
"default" means use the Ensemble default; if not, we will fall back to the default provided in
playback_settings.
:param num_channels: how many channels to allocate for managing pitch bends, etc.
:param audio_driver: which driver to use
:param max_pitch_bend: max pitch bend to allow
:param note_on_and_off_only: Deprecated; set the instrument's `start_note_fixed` instead.
:param volume_cc_num: The cc value used for volume changes. Defaults to 11 (expression).
:return: self, for chaining purposes
"""
self.start_note_fixed = _resolve_deprecated_note_on_and_off_only(self.start_note_fixed, note_on_and_off_only)
soundfont = self.ensemble.default_soundfont \
if self.ensemble is not None and soundfont == "default" else soundfont
if isinstance(preset, str):
preset = Ensemble._resolve_preset_from_name(self.name if preset == "auto" else preset, soundfont)
elif isinstance(preset, int):
preset = (0, preset)
self.playback_implementations.append(
SoundfontPlaybackImplementation(bank_and_preset=preset, soundfont=soundfont, num_channels=num_channels,
audio_driver=audio_driver, max_pitch_bend=max_pitch_bend,
volume_cc_num=volume_cc_num)
)
return self
[docs]
def remove_soundfont_playback(self) -> ScampInstrument:
"""
Remove the most recent SoundfontPlaybackImplementation from this instrument.
:return: self, for chaining purposes
"""
for index in reversed(range(len(self.playback_implementations))):
if isinstance(self.playback_implementations[index], SoundfontPlaybackImplementation):
self.playback_implementations.pop(index)
break
return self
[docs]
def add_streaming_midi_playback(self, midi_output_device: int | str = "default", num_channels: int = 8,
midi_output_name: str = None, max_pitch_bend: int = "default",
note_on_and_off_only: bool = False, start_channel: int = 0,
volume_cc_num: int = 11) -> ScampInstrument:
"""
Add a streaming MIDI playback implementation for this instrument.
:param midi_output_device: name or number of the device to use
:param num_channels: how many channels to allocate for managing pitch bends, etc.
:param midi_output_name: name given to the output stream
:param max_pitch_bend: max pitch bend to allow
:param note_on_and_off_only: Deprecated; set the instrument's `start_note_fixed` instead.
:param start_channel: the first channel to use. For instance, if start_channel is 4, and num_channels is 5,
we will use channels (4, 5, 6, 7, 8). NOTE: channel counting in SCAMP starts from 0, so this may show
up as channels 5-9 in your MIDI software.
:param volume_cc_num: The cc value used for volume changes. Defaults to 11 (expression).
:return: self, for chaining purposes
"""
self.start_note_fixed = _resolve_deprecated_note_on_and_off_only(self.start_note_fixed, note_on_and_off_only)
self.playback_implementations.append(
MIDIStreamPlaybackImplementation(midi_output_device=midi_output_device, num_channels=num_channels,
midi_output_name=midi_output_name, max_pitch_bend=max_pitch_bend,
start_channel=start_channel, volume_cc_num=volume_cc_num)
)
return self
[docs]
def remove_streaming_midi_playback(self) -> ScampInstrument:
"""
Remove the most recent MIDIStreamPlaybackImplementation from this instrument.
:return: self, for chaining purposes
"""
for index in reversed(range(len(self.playback_implementations))):
if isinstance(self.playback_implementations[index], MIDIStreamPlaybackImplementation):
self.playback_implementations.pop(index)
break
return self
[docs]
def add_osc_playback(self, port: int, ip_address: str = "127.0.0.1", message_prefix: str = None,
osc_message_addresses: dict = "default"):
"""
Add an OSCPlaybackImplementation for this instrument.
:param port: port to use
:param ip_address: ip address to use
:param message_prefix: the prefix to give to all outgoing osc messages; defaults to the instrument name
with all spaces removed.
:param osc_message_addresses: the specifix message addresses to be used for each type of message. Defaults are
defined in playback_settings
:return: self, for chaining purposes
"""
message_prefix = self.name.replace(" ", "") if message_prefix is None else message_prefix
self.playback_implementations.append(
OSCPlaybackImplementation(port=port, ip_address=ip_address, message_prefix=message_prefix,
osc_message_addresses=osc_message_addresses)
)
return self
[docs]
def remove_osc_playback(self) -> ScampInstrument:
"""
Remove the most recent OSCPlaybackImplementation from this instrument.
:return: self, for chaining purposes
"""
for index in reversed(range(len(self.playback_implementations))):
if isinstance(self.playback_implementations[index], OSCPlaybackImplementation):
self.playback_implementations.pop(index)
break
return self
"""
------------------------------------------------- Other -----------------------------------------------------
"""
[docs]
def set_max_pitch_bend(self, semitones: int) -> None:
"""
Set the max pitch bend for all midi playback implementations on this instrument
"""
for playback_implementation in self.playback_implementations:
playback_implementation.set_max_pitch_bend(semitones)
[docs]
def send_midi_cc(self, cc_number: int, value_from_0_to_1: float) -> None:
"""
Sends a midi cc message to all midi-based playback implementations, affecting all channels this instrument uses.
This is useful for stuff like pedal messages, that we don't really want to bundle with note playback, and that
we want to apply to all channels.
:param cc_number: the cc number from 0 to 127
:param value_from_0_to_1: the value to send, normalized from 0 to 1
"""
for playback_implementation in self.playback_implementations:
if hasattr(playback_implementation, "cc"):
for chan in range(playback_implementation.num_channels):
playback_implementation.cc(chan, cc_number, value_from_0_to_1)
[docs]
def pedal_down(self, press_amount: float = 1.0) -> None:
"""
Presses the sustain pedal, by sending a CC 64 message to all midi-based playback implementations.
:param press_amount: how far down to press the pedal, from 0 to 1 (values in between allow half-pedaling,
if supported by the receiving device)
"""
self._last_pedal_press_amount = press_amount
self.send_midi_cc(64, press_amount)
[docs]
def pedal_up(self) -> None:
"""
Releases the sustain pedal, by sending a CC 64 message of 0 to all midi-based playback implementations.
"""
self.send_midi_cc(64, 0)
[docs]
def pedal_change(self, duration: float = 0.2, press_amount: float = None) -> None:
"""
Quickly lifts and re-presses the sustain pedal, clearing resonating notes. Returns immediately, with the
re-press happening in a forked process after the given duration.
:param duration: how long (in seconds, regardless of tempo) to keep the pedal up before re-pressing it
:param press_amount: how far down to re-press the pedal, from 0 to 1; defaults to the press amount of the
last call to :func:`pedal_down` or :func:`pedal_change`.
"""
press_amount = press_amount if press_amount is not None else self._last_pedal_press_amount
self._last_pedal_press_amount = press_amount
def _pedal_changer():
self.send_midi_cc(64, 0)
wait(duration, units="time")
self.send_midi_cc(64, press_amount)
clock, _ = self._resolve_clock(None, False)
# fork on the master clock so that the re-press isn't killed if the
# current (child) clock ends, and so that "time" is true seconds
pedal_clock = clock.master.fork(_pedal_changer, name="PEDAL_CHANGE")
pedal_clock.description = f"a pedal change on {self.name!r}"
@property
def clef_preference(self):
"""
The clef preference for this instrument. Can be any of:
- "from_name", which picks clef based on the instrument name
- "default", which uses the default clef preferences for an unknown instrument
- the name of a clef
- the name of an instrument whose clef defaults to use
- a list of possible clefs. Each of these choices should be either a valid clef name string or a tuple of (valid clef name string, center pitch).
"""
return self._clef_preference
@clef_preference.setter
def clef_preference(self, value):
old_value = self._clef_preference
self._clef_preference = value
try:
self.resolve_clef_preference()
except RuntimeError as e:
self._clef_preference = old_value
raise e
[docs]
def resolve_clef_preference(self) -> Sequence[str | tuple[str, Real]]:
"""
Resolves the clef preference to a sequence of possible clef choices.
"""
if isinstance(self.clef_preference, str):
if self.clef_preference == "from_name":
# base clef preference on instrument name
if self.name.lower().strip() in engraving_settings.clefs_by_instrument:
return engraving_settings.clefs_by_instrument[self.name.lower().strip()]
else:
# instrument name is not found, so revert to default clef preferences
return engraving_settings.clefs_by_instrument["default"]
elif self.clef_preference == "default":
# just use the default clef preferences
return engraving_settings.clefs_by_instrument["default"]
elif self.clef_preference in engraving_settings.clef_pitch_centers:
# clef preference is the name of a clef
return [self.clef_preference]
elif self.clef_preference.lower().strip() in engraving_settings.clefs_by_instrument:
# clef preference is an instrument name
return engraving_settings.clefs_by_instrument[self.clef_preference.lower().strip()]
raise RuntimeError("Clef preference could not be resolved.")
elif isinstance(self.clef_preference, Sequence):
# if not a string, clef preference should be a sequence of possible clef choices
# each of these choices should be either a valid clef name string or a tuple of
# (valid clef name string, center pitch)
if all(
x in engraving_settings.clef_pitch_centers or hasattr(x, "__len__") and
x[0] in engraving_settings.clef_pitch_centers and isinstance(x[1], Real)
for x in self.clef_preference
):
return self.clef_preference
else:
raise RuntimeError("Clef preference not understood.")
else:
raise RuntimeError("Clef preference not understood.")
@property
def default_spelling_policy(self):
"""
The default spelling policy for notes played back by this instrument. (Can be set with a
:class:`~scamp.spelling.SpellingPolicy`, or a string or tuple interpretable as such via
:func:`~scamp.spelling.SpellingPolicy.interpret`)
"""
return self._default_spelling_policy
@default_spelling_policy.setter
def default_spelling_policy(self, value: SpellingPolicy | str | tuple):
self._default_spelling_policy = SpellingPolicy.interpret(value) if value is not None else None
"""
--------------------------------------------- To / from JSON -------------------------------------------------
"""
def _to_dict(self):
return {
"name": self.name,
"playback_implementations": self.playback_implementations,
"default_spelling_policy": self.default_spelling_policy,
"clef_preference": self.clef_preference,
"start_note_fixed": self.start_note_fixed
}
@classmethod
def _from_dict(cls, json_dict):
return cls(**json_dict)
def __str__(self):
return "ScampInstrument('{}')".format(self.name)
def __repr__(self):
return "ScampInstrument({})".format(", ".join("{}={}".format(k, repr(v)) for k, v in self._to_dict().items()))
@property
def note_info_by_id(self):
"""
Dictionary mapping the id of each note currently sounding on this instrument to the bookkeeping
information kept about it (the clock that started it, its start/end time stamps, its current
parameter values, any ongoing parameter change segments, and its :class:`NoteProperties`).
This is exposed mostly for the benefit of custom :class:`~scamp.playback_implementations.PlaybackImplementation`
subclasses, which may need to consult the state of a note in progress.
"""
return self._note_info_by_id
[docs]
class NoteHandle:
"""
This handle, which is returned by instrument.start_note, allows us to manipulate the note that we have started,
(i.e. by changing pitch, volume, or another other parameter, or by ending the note). You would never create
one of these directly.
:param note_id: the reference id of the note
:param instrument: the instrument playing the note
:ivar note_id: the reference id of the note
:ivar instrument: the instrument playing the note
"""
def __init__(self, note_id: int, instrument: ScampInstrument):
self.note_id: int = note_id
self.instrument: ScampInstrument = instrument
[docs]
def change_parameter(self, param_name: str, target_value_or_values: float | Sequence[float],
transition_length_or_lengths: float | Sequence[float] = 0,
transition_curve_shape_or_shapes: float | Sequence[float] = 0, clock: Clock = None) -> None:
"""
Change a custom playback parameter for this note to a given target value or values, over a given duration and
with a given curve shape.
:param param_name: name of the parameter to change
:param target_value_or_values: either a single value or a list of values to which we want to change the
parameter of interest.
:param transition_length_or_lengths: the duration (in beats) that we want it to take to reach the target value.
The default value of 0 represents an instantaneous change. If multiple target values were given, a list of
durations should be given for each segment.
:param transition_curve_shape_or_shapes: the curve shape used in transitioning to the new target value.
The default value of 0 represents an linear change, a value greater than zero represents late change,
and a value less than 0 represents early change. If multiple target values were given, a list of curve
shapes should be given (unless it is left as the default 0, in which case all segments are linear).
:param clock: The clock with which to interpret the transition timings. The default value of "from_note", which
you likely don't want to change, carries out the timings on the clock on which the note was started.
"""
self.instrument.change_note_parameter(self.note_id, param_name, target_value_or_values,
transition_length_or_lengths, transition_curve_shape_or_shapes, clock)
[docs]
def change_pitch(self, target_value_or_values: float | Sequence[float],
transition_length_or_lengths: float | Sequence[float] = 0,
transition_curve_shape_or_shapes: float | Sequence[float] = 0, clock: Clock = None) -> None:
"""
Change the pitch of this note to a given target value or values, over a given duration and with a given
curve shape. Only works if the note was started with `fixed=False`.
:param target_value_or_values: either a single target pitch or a list of target pitches.
:param transition_length_or_lengths: the duration (in beats) that we want it to take to reach the target pitch.
The default value of 0 represents an instantaneous change. If multiple target values were given, a list of
durations should be given for each segment.
:param transition_curve_shape_or_shapes: the curve shape used in transitioning to the new target pitch.
The default value of 0 represents an linear change, a value greater than zero represents late change,
and a value less than 0 represents early change. If multiple target values were given, a list of curve
shapes should be given (unless it is left as the default 0, in which case all segments are linear).
:param clock: The clock with which to interpret the transition timings. The default value of "from_note", which
you likely don't want to change, carries out the timings on the clock on which the note was started.
"""
self.instrument.change_note_pitch(self.note_id, target_value_or_values, transition_length_or_lengths,
transition_curve_shape_or_shapes, clock)
[docs]
def change_volume(self, target_value_or_values: float | Sequence[float],
transition_length_or_lengths: float | Sequence[float] = 0,
transition_curve_shape_or_shapes: float | Sequence[float] = 0, clock: Clock = None) -> None:
"""
Change the volume of this note to a given target value or values, over a given duration and with a given
curve shape. Requires `fixed=False`, and can only raise volume above the start value if an explicit
`velocity` was given when the note was started.
:param target_value_or_values: either a single target volume or a list of target volumes.
:param transition_length_or_lengths: the duration (in beats) that we want it to take to reach the target volume.
The default value of 0 represents an instantaneous change. If multiple target values were given, a list of
durations should be given for each segment.
:param transition_curve_shape_or_shapes: the curve shape used in transitioning to the new target volume.
The default value of 0 represents an linear change, a value greater than zero represents late change,
and a value less than 0 represents early change. If multiple target values were given, a list of curve
shapes should be given (unless it is left as the default 0, in which case all segments are linear).
:param clock: The clock with which to interpret the transition timings. The default value of "from_note", which
you likely don't want to change, carries out the timings on the clock on which the note was started.
"""
self.instrument.change_note_volume(self.note_id, target_value_or_values, transition_length_or_lengths,
transition_curve_shape_or_shapes, clock)
[docs]
def split(self) -> None:
"""
Adds a split point to this note, causing it later to be rendered as tied pieces.
"""
self.instrument.split_note(self.note_id)
[docs]
def end(self) -> None:
"""
Ends this note.
"""
self.instrument.end_note(self.note_id)
def __repr__(self):
return "NoteHandle({}, {})".format(self.note_id, self.instrument)
[docs]
class ChordHandle:
"""
This handle, returned by instrument.start_chord, allows us to manipulate a chord that we have started,
(i.e. by changing pitch, volume, or another other parameter, or by ending the note). You would never create
one of these directly.
:param note_handles: the handles of the notes that make up this chord
:param intervals: the original pitch intervals between the chord tones
:ivar note_handles: the handles of the notes that make up this chord
"""
def __init__(self, note_handles: Sequence[NoteHandle], intervals: Sequence[float]):
self.note_handles = tuple(note_handles) if not isinstance(note_handles, tuple) else note_handles
self._intervals = tuple(intervals) if not isinstance(intervals, tuple) else intervals
[docs]
def change_parameter(self, param_name: str, target_value_or_values: float | Sequence[float],
transition_length_or_lengths: float | Sequence[float] = 0,
transition_curve_shape_or_shapes: float | Sequence[float] = 0, clock: Clock = None) -> None:
"""
Change a custom playback parameter for all notes in this chord to a given target value or values, over a given
duration and with a given curve shape.
:param param_name: name of the parameter to change
:param target_value_or_values: either a single value or a list of values to which we want to change the
parameter of interest.
:param transition_length_or_lengths: the duration (in beats) that we want it to take to reach the target value.
The default value of 0 represents an instantaneous change. If multiple target values were given, a list of
durations should be given for each segment.
:param transition_curve_shape_or_shapes: the curve shape used in transitioning to the new target value.
The default value of 0 represents an linear change, a value greater than zero represents late change,
and a value less than 0 represents early change. If multiple target values were given, a list of curve
shapes should be given (unless it is left as the default 0, in which case all segments are linear).
:param clock: The clock with which to interpret the transition timings. The default value of "from_note", which
you likely don't want to change, carries out the timings on the clock on which the note was started.
"""
for note_handle in self.note_handles:
note_handle.change_parameter(param_name, target_value_or_values, transition_length_or_lengths,
transition_curve_shape_or_shapes, clock)
[docs]
def change_pitch(self, target_value_or_values: float | Sequence[float],
transition_length_or_lengths: float | Sequence[float] = 0,
transition_curve_shape_or_shapes: float | Sequence[float] = 0, clock: Clock = None) -> None:
"""
Change the pitches of this chord such that the first note of the chord goes to the given target value or values,
over a given duration and with a given curve shape. Only works if the chord was started with `fixed=False`.
:param target_value_or_values: either a single target pitch or a list of target pitches. Note that this is the
pitch that the first note of the chord gets changed to; all of the other notes in the chord follow suit,
maintaining the same interval as before with the first note of the chord.
:param transition_length_or_lengths: the duration (in beats) that we want it to take to reach the target pitch.
The default value of 0 represents an instantaneous change. If multiple target values were given, a list of
durations should be given for each segment.
:param transition_curve_shape_or_shapes: the curve shape used in transitioning to the new target pitch.
The default value of 0 represents an linear change, a value greater than zero represents late change,
and a value less than 0 represents early change. If multiple target values were given, a list of curve
shapes should be given (unless it is left as the default 0, in which case all segments are linear).
:param clock: The clock with which to interpret the transition timings. The default value of "from_note", which
you likely don't want to change, carries out the timings on the clock on which the note was started.
"""
for note_handle, interval in zip(self.note_handles, self._intervals):
this_note_pitch_targets = [target_value + interval for target_value in target_value_or_values] \
if hasattr(target_value_or_values, "__len__") else target_value_or_values + interval
note_handle.change_pitch(this_note_pitch_targets, transition_length_or_lengths,
transition_curve_shape_or_shapes, clock)
[docs]
def change_volume(self, target_value_or_values: float | Sequence[float],
transition_length_or_lengths: float | Sequence[float] = 0,
transition_curve_shape_or_shapes: float | Sequence[float] = 0, clock: Clock = None) -> None:
"""
Change the volume for all notes in this chord to a given target value or values, over a given duration and with
a given curve shape. Requires `fixed=False`, and can only raise volume above the start value if an explicit
`velocity` was given when the chord was started.
:param target_value_or_values: either a single target volume or a list of target volumes.
:param transition_length_or_lengths: the duration (in beats) that we want it to take to reach the target volume.
The default value of 0 represents an instantaneous change. If multiple target values were given, a list of
durations should be given for each segment.
:param transition_curve_shape_or_shapes: the curve shape used in transitioning to the new target volume.
The default value of 0 represents an linear change, a value greater than zero represents late change,
and a value less than 0 represents early change. If multiple target values were given, a list of curve
shapes should be given (unless it is left as the default 0, in which case all segments are linear).
:param clock: The clock with which to interpret the transition timings. The default value of "from_note", which
you likely don't want to change, carries out the timings on the clock on which the note was started.
"""
for note_handle in self.note_handles:
note_handle.change_volume(target_value_or_values, transition_length_or_lengths,
transition_curve_shape_or_shapes, clock)
[docs]
def split(self) -> None:
"""
Adds a split point to this chord, causing it later to be rendered as tied pieces.
"""
for note_handle in self.note_handles:
note_handle.split()
[docs]
def end(self) -> None:
"""
Ends all notes in this chord.
"""
for note_handle in self.note_handles:
note_handle.end()
def __repr__(self):
return "ChordHandle({}, {})".format(self.note_handles, self._intervals)
class _ParameterChangeSegment(EnvelopeSegment):
"""
An interruptable transition of one note parameter, holding the info needed to both play it back and
transcribe it. (This is an implementation detail.)
As an :class:`~expenvelope.envelope_segment.EnvelopeSegment` it is the shape to follow; playback is done
by the clock family's :class:`~scamp._animation._AnimationTickDriver`, which samples the segment live on
each tick while it is registered (unless flat or silent). The transcriber later reconstructs the
notated curve from the ``start_time_stamp``/``end_time_stamp`` recorded here, independent of how
coarsely playback sampled it.
:param parameter_change_function: the function called to apply a sampled value (calls the playback
implementations' change_note_pitch/volume/parameter for a given note; a no-op record for silent notes)
:param start_value: start value of the parameter in the transition
:param target_value: target value of the parameter in the transition
:param transition_length: length of the transition in beats on the clock given
:param transition_curve_shape: curve shape of the transition
:param clock: the clock whose beats the transition is measured in
:param driver: the tick driver for this clock's family, which samples the segment during playback
"""
# EnvelopeSegment compares (and so hashes) by value, but each of these is a distinct live animation
# with its own lifecycle -- identity is what we want, and it lets the tick driver hold them in a set.
__eq__ = object.__eq__
__hash__ = object.__hash__
def __init__(self, parameter_change_function, start_value, target_value, transition_length,
transition_curve_shape, clock, driver):
# set this up as an envelope
super().__init__(0, transition_length, start_value, target_value, transition_curve_shape)
# "do_change_parameter" feels more like an action name
self.do_change_parameter = parameter_change_function
self.clock = clock # the clock whose beats this transition is measured in
self.driver = driver # the tick driver that samples this segment during playback
self.running = False # True between begin() and finish/abort; also gates a stale _finalize
# the beat on self.clock where the envelope starts, so the driver can sample at (clock.beat - this)
self._anim_start_beat = None
self._on_finish = None # callback function on natural finish
# used to claim a single finish (either from _finalize or abort_if_running),
# so the two can race harmlessly
self._finish_lock = Lock()
# the time stamps at which this starts and finishes, from which the transcriber rebuilds the curve
self.start_time_stamp = None
self.end_time_stamp = None
def begin(self, silent=False, on_finish=None):
"""
Start the segment: stamp its start, register it for live sampling by the tick driver (unless flat or
silent), and schedule a beat-anchored endpoint that lands the exact final value and
stamps the end time. Because the endpoint is scheduled as a beat on this clock, it tracks the
note's beat-defined lifetime through any tempo changes.
:param silent: skip the playback sampling (the note is being transcribed but not heard, as with a note
that had a playback adjustment applied). The start/end stamps are still recorded for transcription.
:param on_finish: called once if the segment finishes naturally, used to start the next segment of a
sequence. Not called if the segment is aborted, so aborting the running segment stops the sequence.
"""
self._on_finish = on_finish
self.start_time_stamp = TimeStamp.now(self.clock)
# a zero-length segment has nothing to animate: land the final value and finish immediately
if self.duration == 0:
self.end_time_stamp = self.start_time_stamp
self.do_change_parameter(self.end_level)
if on_finish is not None:
on_finish()
return
self.running = True
# technically stores redundant info from start_time_stamp but since we need it for
# every animation frame it's worth caching.
self._anim_start_beat = self.start_time_stamp.beat_in_clock(self.clock)
# only sample live if necessary; a silent or flat segment just needs its
# endpoint (below) to stamp the end and record the final value.
if not silent and self.end_level != self.start_level:
self.driver.register(self)
self.clock.schedule_action(self._finalize, Moment.at_beat(self._anim_start_beat + self.duration))
def _finalize(self):
"""Scheduled wind down for the natural end of this segment: land the exact final value,
stamp the end, and start the next segment if present."""
# if already aborted (note ended early, or a newer change_pitch/volume/parameter was called
# before this one finished), this finalization is stale; return as a no-op
if not self._claim_finish():
return
# otherwise, wind down: deregister from tick driver, stamp, set final value
self.driver.deregister(self)
self.end_time_stamp = TimeStamp.now(self.clock)
self.do_change_parameter(self.end_level)
# for the natural end path, we run _on_finish (starting the next segment)
if self._on_finish is not None:
self._on_finish()
def _claim_finish(self):
"""
Atomically claim the one-time finish of this segment, either from the natural end (_finalize)
or from an early abort_if_running(). This keeps them from racing.
Returns True to the winner, False otherwise.
"""
with self._finish_lock:
if not self.running:
return False
self.running = False
return True
def abort_if_running(self):
if self._claim_finish():
self.driver.deregister(self)
# save the time stamp at which we aborted as the end time stamp
self.end_time_stamp = TimeStamp.now(self.clock)
# since the units of this envelope are beats in self.clock, see how far we got in the envelope by
# subtracting converting the start and end time stamps to those beats and subtracting
how_far_we_got = self.end_time_stamp.beat_in_clock(self.clock) - \
self.start_time_stamp.beat_in_clock(self.clock)
# Decide, tolerantly, where in the envelope we ended up. how_far_we_got is reconstructed from
# scheduler-time TimeStamps and can end up infinitesimally short due to floating point math.
# A bare "<" comparison would lead to unwanted truncation in that case. This is the kind of
# thing that meaningfully_less/greater_than are designed to guard against.
if meaningfully_greater_than(how_far_we_got, self.start_time) and \
meaningfully_less_than(how_far_we_got, self.end_time):
# genuinely aborted partway through: split there, discarding the rest of the envelope.
# This makes self.end_level the value we ended up at.
self.split_at(how_far_we_got)
elif not meaningfully_greater_than(how_far_we_got, self.start_time):
# this was aborted before it even got going. Later, the transcriber will ignore this nothing segment
self.end_time = self.start_time
self.end_level = self.start_level
return
# otherwise we reached the end (within noise), so leave the segment and its exact end_level untouched
self.do_change_parameter(self.end_level) # set it to where we should be at this point
# the segment's still-queued endpoint will find running False and no-op (see _finalize)
def __repr__(self):
return "_ParameterChangeSegment[{}, {}, {}, {}, {}]".format(
self.start_time_stamp, self.end_time_stamp, self.start_level, self.end_level, self.curve_shape
)