Gavan Fantom
14a05e8694
PlaybackNext replaces the current playback object with new media. Optionally, it waits until the end of the media. In this case, it only waits until the end of the current loop. EndLoop stops the playback at the end of the current loop. Note that PlaybackNext and EndLoop have no effect if there is no media currently playing.
321 lines
10 KiB
Python
321 lines
10 KiB
Python
# events.py
|
|
|
|
import soundfile
|
|
import sounddevice
|
|
|
|
import timings
|
|
import display
|
|
|
|
import numpy as np
|
|
|
|
import pyglet.image
|
|
|
|
class Event:
|
|
def __init__(self, at='start'):
|
|
self._at = timings.Timing(at)
|
|
|
|
def __repr__(self):
|
|
return '{}(at={})'.format(type(self).__name__, self._at)
|
|
|
|
def set_timings(self, timings):
|
|
self._at.set_timings(timings)
|
|
|
|
def set_audiencecontent(self, audiencecontent):
|
|
self._audiencecontent = audiencecontent
|
|
|
|
def set_eventrunner(self, eventrunner):
|
|
self._eventrunner = eventrunner
|
|
|
|
@property
|
|
def scheduled(self):
|
|
return self._at.scheduled
|
|
|
|
@scheduled.setter
|
|
def scheduled(self, value):
|
|
self._at.scheduled = value
|
|
|
|
@property
|
|
def at(self):
|
|
return self._at.at
|
|
|
|
def prepare(self):
|
|
pass
|
|
|
|
def start(self):
|
|
pass
|
|
|
|
def stop(self):
|
|
pass
|
|
|
|
def notify(self, *args, **kwargs):
|
|
pass
|
|
|
|
class Display(Event):
|
|
def __init__(self, media, background=None, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._media = media
|
|
self._background = background
|
|
self._image = pyglet.image.load(self._media)
|
|
self._image.anchor_x = self._image.width // 2
|
|
self._image.anchor_y = self._image.height // 2
|
|
|
|
def __repr__(self):
|
|
return '{}(at={}, media={}, background={})'.format(type(self).__name__, self._at, self._media, self._background)
|
|
|
|
def start(self):
|
|
self._audiencecontent.new_content(display.DisplayImage(self._image, self._background))
|
|
print("Started displaying image")
|
|
|
|
def stop(self):
|
|
self._audiencecontent.new_content(display.DisplayContent())
|
|
print("Stopped displaying image")
|
|
|
|
def notify(self, message, *args, **kwargs):
|
|
if message == 'newdisplay':
|
|
self._eventrunner.finished(self)
|
|
|
|
class Playback(Event):
|
|
def __init__(self, media, device=None, channels=[1,2], tempo=None, beats=None, leadin=0, loop=False, gain=0, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._media = media
|
|
self._device = device
|
|
self._mapping = np.atleast_1d(np.array(channels, copy=True))
|
|
self._channels = self._mapping.max()
|
|
self._mapping -= 1
|
|
self._tempo = tempo
|
|
self._beats = beats
|
|
self._leadin = leadin
|
|
self._loop = loop
|
|
self._gain = 10**(gain/10) # Convert from dB to linear gain
|
|
self._fadecurve = None
|
|
self._fadeframe = None
|
|
|
|
def __repr__(self):
|
|
return '{}(at={}, media={}, device={}, tempo={}, beats={}, leadin={}, loop={}, gain={})'.format(type(self).__name__, self._at, self._media, self._device, self._tempo, self._beats, self._leadin, self._loop, self._gain)
|
|
|
|
def prepare(self):
|
|
self._at._timings.register_tempo(self._tempo, self._beats, self._at)
|
|
|
|
# def start(self):
|
|
# print("Start playback")
|
|
# data, fs = soundfile.read(self._media)
|
|
# sounddevice.play(data, fs, device=self._device, mapping=self._mapping)
|
|
|
|
def start(self):
|
|
print("Start playback")
|
|
self._data, self._fs = soundfile.read(self._media)
|
|
self._current_frame = 0
|
|
self._fadecurve = None
|
|
self._fadeframe = None
|
|
self._nextdata = None
|
|
self._nextfs = None
|
|
self._delay = False
|
|
|
|
self._stream = sounddevice.OutputStream(
|
|
samplerate=self._fs, device=self._device, channels=self._channels,
|
|
callback=self.callback, finished_callback=self.finished_callback)
|
|
self._stream.start()
|
|
|
|
def callback(self, outdata, frames, time, status):
|
|
if status:
|
|
print(status)
|
|
if self._nextdata is not None and not self._delay:
|
|
self.callback_startnew()
|
|
offset = 0
|
|
while frames > 0:
|
|
chunksize = self.callback_fillchunk(outdata, offset, frames)
|
|
offset += chunksize
|
|
frames -= chunksize
|
|
if frames > 0:
|
|
if self._nextdata is not None:
|
|
self.callback_startnew()
|
|
elif self._loop:
|
|
self._current_frame = 0
|
|
else:
|
|
outdata[offset:, self._mapping] = 0
|
|
raise sounddevice.CallbackStop()
|
|
self._current_frame += frames
|
|
frames = 0
|
|
|
|
def callback_fillchunk(self, outdata, offset, frames):
|
|
chunksize = min(len(self._data) - self._current_frame, frames)
|
|
if self._fadecurve is not None:
|
|
chunksize = min(chunksize, len(self._fadecurve) - self._fadeframe)
|
|
outdata[offset:offset + chunksize, self._mapping] = self._data[self._current_frame:self._current_frame + chunksize] * self._gain
|
|
if self._fadecurve is not None:
|
|
outdata[offset:offset + chunksize, self._mapping] *= self._fadecurve[self._fadeframe:self._fadeframe + chunksize, np.newaxis]
|
|
self._fadeframe += chunksize
|
|
self._current_frame += chunksize
|
|
return chunksize
|
|
|
|
def callback_startnew(self):
|
|
self._data = self._nextdata
|
|
self._fs = self._nextfs
|
|
self._loop = self._nextloop
|
|
self._current_frame = 0
|
|
self._nextdata = None
|
|
self._nextfs = None
|
|
self._fadecurve = None
|
|
self._fadeframe = None
|
|
|
|
def finished_callback(self):
|
|
self._data = None
|
|
self._fadecurve = None
|
|
self._fadeframe = None
|
|
print("Playback finished")
|
|
self._eventrunner.finished(self)
|
|
|
|
def stop(self):
|
|
print("Stop playback")
|
|
self._stream.stop(True)
|
|
self._stream.close(True)
|
|
self._stream = None
|
|
|
|
def replace(self, media, delay=False, loop=False):
|
|
self._nextdata, self._nextfs = soundfile.read(media)
|
|
self._delay = delay
|
|
self._nextloop = loop
|
|
# We really should check that the media is the sampe shape as the
|
|
# existing playback. But for now, let's trust the user. *cough*
|
|
|
|
def notify(self, message, fadetime=10, db=30, media=None, delay=False, *args, **kwargs):
|
|
if message == 'fadeout':
|
|
self._fadecurve = np.logspace(0, -db/10, fadetime * self._fs)
|
|
self._fadeframe = 0
|
|
elif message == 'replaceplayback':
|
|
self.replace(media, delay)
|
|
elif message == 'endloop':
|
|
self._loop = False
|
|
|
|
class FadeOut(Event):
|
|
def __init__(self, duration=10, db=30, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._duration = duration
|
|
self._db = db
|
|
|
|
def __repr__(self):
|
|
return '{}(at={}, duration={}, db={})'.format(type(self).__name__, self._at, self._duration, self._db)
|
|
|
|
def start(self):
|
|
print("Fading out")
|
|
self._eventrunner.notify('fadeout', fadetime=self._duration, db=self._db)
|
|
self._eventrunner.finished(self)
|
|
|
|
class EndLoop(Event):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def __repr__(self):
|
|
return '{}(at={})'.format(type(self).__name__, self._at)
|
|
|
|
def start(self):
|
|
print("Ending loop")
|
|
self._eventrunner.notify('endloop')
|
|
self._eventrunner.finished(self)
|
|
|
|
|
|
class PlaybackNext(Event):
|
|
def __init__(self, media, delay=False, loop=False, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._media = media
|
|
self._delay = delay
|
|
self._loop = loop
|
|
|
|
def __repr__(self):
|
|
return '{}(at={}, media={}, delay={}, loop={})'.format(type(self).__name__, self._at, self._media, self._delay, self._loop)
|
|
|
|
def start(self):
|
|
print("Replacing playback")
|
|
self._eventrunner.notify('replaceplayback', media=self._media, delay=self._delay, loop=self._loop)
|
|
self._eventrunner.finished(self)
|
|
|
|
class SetTempo(Event):
|
|
def __init__(self, tempo=None, beats=None, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._tempo = tempo
|
|
self._beats = beats
|
|
|
|
def __repr__(self):
|
|
return '{}(at={}, tempo={}, beats={})'.format(type(self).__name__, self._at, self._tempo, self._beats)
|
|
|
|
def prepare(self):
|
|
self._at._timings.register_tempo(self._tempo, self._beats, self._at)
|
|
|
|
def start(self):
|
|
self._eventrunner.finished(self)
|
|
|
|
def stop(self):
|
|
pass
|
|
|
|
class SetMark(Event):
|
|
def __init__(self, name, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._name = name
|
|
|
|
def __repr__(self):
|
|
return '{}(at={}, name={})'.format(type(self).__name__, self._at, self._name)
|
|
|
|
def prepare(self):
|
|
self._at._timings.register_mark(self._name, self._at)
|
|
|
|
def start(self):
|
|
self._eventrunner.finished(self)
|
|
|
|
def stop(self):
|
|
pass
|
|
|
|
class ButtonMark(Event):
|
|
def __init__(self, name, prestart=False, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._name = name
|
|
self._prestart = prestart
|
|
|
|
def __repr__(self):
|
|
return '{}(at={}, name={})'.format(type(self).__name__, self._at, self._name)
|
|
|
|
def prepare(self):
|
|
self._at._timings.register_mark(self._name, None, self._prestart)
|
|
|
|
def start(self):
|
|
self._eventrunner.finished(self)
|
|
|
|
def stop(self):
|
|
pass
|
|
|
|
class Lyrics(Event):
|
|
def __init__(self, text, duration=None, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._text = text
|
|
self._duration = duration
|
|
|
|
def __repr__(self):
|
|
return '{}(at={}, text={}, duration={})'.format(type(self).__name__, self._at, self._text, self._duration)
|
|
|
|
def start(self):
|
|
self._audiencecontent.new_content(display.DisplayLyrics(self._text))
|
|
print("Started displaying lyrics")
|
|
print(self._text)
|
|
|
|
def stop(self):
|
|
self._audiencecontent.new_content(display.DisplayContent())
|
|
print("Stopped displaying lyrics")
|
|
|
|
def notify(self, message, *args, **kwargs):
|
|
if message == 'newdisplay':
|
|
self._eventrunner.finished(self)
|
|
|
|
class LightScene(Event):
|
|
def __init__(self, scene, fade=0, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self._scene = scene
|
|
self._fade = fade
|
|
|
|
def __repr__(self):
|
|
return '{}(at={}, scene={}, fade={})'.format(type(self).__name__, self._at, self._scene, self._fade)
|
|
|
|
def start(self):
|
|
scenes = self._eventrunner.get_config('scenes')
|
|
scene = scenes[self._scene]
|
|
scene.activate(self._fade)
|
|
|