gig/events.py

# events.py

import soundfile
import sounddevice

import timings
import display

import numpy as np

class Event:
    def __init__(self, at='start'):
        self._at = timings.Timing(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, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._media = media


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 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._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)
        chunksize = min(len(self._data) - self._current_frame, frames)
        if self._fadecurve is not None:
            chunksize = min(chunksize, len(self._fadecurve) - self._fadeframe)
        outdata[:chunksize, self._mapping] = self._data[self._current_frame:self._current_frame + chunksize] * self._gain
        if self._fadecurve is not None:
            outdata[:chunksize, self._mapping] *= self._fadecurve[self._fadeframe:self._fadeframe + chunksize, np.newaxis]
            self._fadeframe += chunksize
        if chunksize < frames:
            outdata[chunksize:, self._mapping] = 0
            raise sounddevice.CallbackStop()
        self._current_frame += chunksize

    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 notify(self, message, fadetime=10, db=30, *args, **kwargs):
        if message == 'fadeout':
            self._fadecurve = np.logspace(0, -db/10, fadetime * self._fs)
            self._fadeframe = 0

class FadeOut(Event):
    def __init__(self, duration=10, db=30, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._duration = duration
        self._db = db

    def start(self):
        print("Fading out")
        self._eventrunner.notify('fadeout', fadetime=self._duration, db=self._db)
        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 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 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, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._name = name

    def prepare(self):
        self._at._timings.register_mark(self._name, None)

    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 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)