import threading import time from transitions.extensions import HierarchicalMachine as Machine from . import debug_print, error_print class Action: ACTION_TYPES = [ 'command', 'interrupt_wait', 'pause', 'play', 'seek', 'stop', 'stop_all_actions', 'unpause', 'volume', 'wait', ] STATES = [ 'initial', 'loading', 'failed', { 'name': 'loaded', 'children': ['running'] } ] TRANSITIONS = [ { 'trigger': 'load', 'source': 'initial', 'dest': 'loading' }, { 'trigger': 'fail', 'source': 'loading', 'dest': 'failed' }, { 'trigger': 'success', 'source': 'loading', 'dest': 'loaded' }, { 'trigger': 'run', 'source': 'loaded', 'dest': 'loaded_running', 'after': 'finish_action' }, { 'trigger': 'interrupt', 'source': 'loaded_running', 'dest': 'loaded', 'before': 'trigger_interrupt' }, { 'trigger': 'finish_action', 'source': 'loaded_running', 'dest': 'loaded' } ] def __init__(self, action, key, **kwargs): Machine(model=self, states=self.STATES, transitions=self.TRANSITIONS, initial='initial', ignore_invalid_triggers=True, queued=True) self.action = action self.key = key self.mapping = key.parent self.arguments = kwargs self.sleep_event = None self.waiting_music = None self.load() def ready(self): return self.is_loaded(allow_substates=True) def callback_loaded(self, success): if success: self.success() else: self.fail() # Machine states / events def on_enter_loading(self): if self.action in self.ACTION_TYPES: if 'music' in self.arguments: self.arguments['music'].subscribe_loaded(self.callback_loaded) else: self.success() else: error_print("Unknown action {}".format(self.action)) self.fail() def on_enter_loaded_running(self): debug_print(self.description()) getattr(self, self.action)(**self.arguments) def trigger_interrupt(self): if getattr(self, self.action + "_interrupt", None): return getattr(self, self.action + "_interrupt")(**self.arguments) # Helpers def music_list(self, music): if music is not None: return [music] else: return self.mapping.open_files.values() def description(self): if getattr(self, self.action + "_print", None): return getattr(self, self.action + "_print")(**self.arguments) else: return "unknown action {}".format(self.action) # Actions def command(self, command="", **kwargs): # FIXME: todo pass def pause(self, music=None, **kwargs): for music in self.music_list(music): if music.is_loaded_playing(): music.pause() def unpause(self, music=None, **kwargs): for music in self.music_list(music): if music.is_loaded_paused(): music.unpause() def play(self, music=None, fade_in=0, start_at=0, restart_if_running=False, volume=100, loop=0, **kwargs): for music in self.music_list(music): if restart_if_running: if music.is_in_use(): music.stop() music.play( volume=volume, fade_in=fade_in, start_at=start_at, loop=loop) elif not music.is_in_use(): music.play( volume=volume, fade_in=fade_in, start_at=start_at, loop=loop) def seek(self, music=None, value=0, delta=False, **kwargs): for music in self.music_list(music): music.seek(value=value, delta=delta) def interrupt_wait(self, wait_id=None): self.mapping.interrupt_wait(wait_id) def stop(self, music=None, fade_out=0, wait=False, set_wait_id=None, **kwargs): previous = None for music in self.music_list(music): if music.is_loaded_paused() or music.is_loaded_playing(): if previous is not None: previous.stop(fade_out=fade_out) previous = music else: music.stop(fade_out=fade_out) if previous is not None: self.waiting_music = previous previous.stop( fade_out=fade_out, wait=wait, set_wait_id=set_wait_id) def stop_all_actions(self, **kwargs): self.mapping.stop_all_running() def volume(self, music=None, value=100, fade=0, delta=False, **kwargs): if music is not None: music.set_volume(value, delta=delta, fade=fade) else: self.mapping.set_master_volume(value, delta=delta, fade=fade) def wait(self, duration=0, music=None, set_wait_id=None, **kwargs): if set_wait_id is not None: self.mapping.add_wait_id(set_wait_id, self) self.sleep_event = threading.Event() if music is not None: music.wait_end() threading.Timer(duration, self.sleep_event.set).start() self.sleep_event.wait() # Action messages def command_print(self, command="", **kwargs): return "running command {}".format(command) def interrupt_wait_print(self, wait_id=None, **kwargs): return "interrupt wait with id {}".format(wait_id) def pause_print(self, music=None, **kwargs): if music is not None: return "pausing « {} »".format(music.name) else: return "pausing all musics" def unpause_print(self, music=None, **kwargs): if music is not None: return "unpausing « {} »".format(music.name) else: return "unpausing all musics" def play_print(self, music=None, fade_in=0, start_at=0, restart_if_running=False, volume=100, loop=0, **kwargs): message = "starting " if music is not None: message += "« {} »".format(music.name) else: message += "all musics" if start_at != 0: message += " at {}s".format(start_at) if fade_in != 0: message += " with {}s fade_in".format(fade_in) message += " at volume {}%".format(volume) if loop > 0: message += " {} times".format(loop + 1) elif loop < 0: message += " in loop" if restart_if_running: message += " (restarting if already running)" return message def stop_print(self, music=None, fade_out=0, wait=False, set_wait_id=None, **kwargs): message = "stopping " if music is not None: message += "music « {} »".format(music.name) else: message += "all musics" if fade_out > 0: message += " with {}s fadeout".format(fade_out) if wait: if set_wait_id is not None: message += " (waiting the end of fadeout, with id {})"\ .format(set_wait_id) else: message += " (waiting the end of fadeout)" return message def stop_all_actions_print(self, **kwargs): return "stopping all actions" def seek_print(self, music=None, value=0, delta=False, **kwargs): if delta: if music is not None: return "moving music « {} » by {:+d}s" \ .format(music.name, value) else: return "moving all musics by {:+d}s" \ .format(value) else: if music is not None: return "moving music « {} » to position {}s" \ .format(music.name, value) else: return "moving all musics to position {}s" \ .format(value) def volume_print(self, music=None, value=100, delta=False, fade=0, **kwargs): message = "" if delta: if music is not None: message += "{:+d}% to volume of « {} »" \ .format(value, music.name) else: message += "{:+d}% to volume" \ .format(value) else: if music is not None: message += "setting volume of « {} » to {}%" \ .format(music.name, value) else: message += "setting volume to {}%" \ .format(value) if fade > 0: message += " with {}s fade".format(fade) return message def wait_print(self, duration=0, music=None, set_wait_id=None, **kwargs): message = "" if music is None: message += "waiting {}s" \ .format(duration) elif duration == 0: message += "waiting the end of « {} »" \ .format(music.name) else: message += "waiting the end of « {} » + {}s" \ .format(music.name, duration) if set_wait_id is not None: message += " (setting id = {})".format(set_wait_id) return message # Interruptions (only for non-"atomic" actions) def wait_interrupt(self, duration=0, music=None, **kwargs): if self.sleep_event is not None: self.sleep_event.set() if music is not None: music.wait_event.set() def stop_interrupt(self, music=None, fade_out=0, wait=False, set_wait_id=None, **kwargs): if self.waiting_music is not None: self.waiting_music.wait_event.set()