'failed',
{
'name': 'loaded',
- 'children': ['running']
- }
+ 'children': ['stopped', 'running']
+ },
+ 'destroyed'
]
TRANSITIONS = [
},
{
'trigger': 'fail',
- 'source': 'loading',
+ 'source': ['loading', 'loaded'],
'dest': 'failed',
- 'after': 'poll_loaded'
},
{
'trigger': 'success',
'source': 'loading',
- 'dest': 'loaded',
- 'after': 'poll_loaded'
+ 'dest': 'loaded_stopped',
},
{
- 'trigger': 'run',
+ 'trigger': 'reload',
'source': 'loaded',
+ 'dest': 'loading',
+ },
+ {
+ 'trigger': 'run',
+ 'source': 'loaded_stopped',
'dest': 'loaded_running',
'after': 'finish_action',
- # if a child has no transitions, then it is bubbled to the parent,
- # and we don't want that. Not useful in that machine precisely.
- 'conditions': ['is_loaded']
},
{
'trigger': 'finish_action',
'source': 'loaded_running',
- 'dest': 'loaded'
+ 'dest': 'loaded_stopped'
+ },
+ {
+ 'trigger': 'destroy',
+ 'source': '*',
+ 'dest': 'destroyed'
}
]
def __init__(self, action, key, **kwargs):
Machine(model=self, states=self.STATES,
transitions=self.TRANSITIONS, initial='initial',
- ignore_invalid_triggers=True, queued=True)
+ ignore_invalid_triggers=True, queued=True,
+ after_state_change=self.notify_state_change)
self.action = action
self.key = key
def is_loaded_or_failed(self):
return self.is_loaded(allow_substates=True) or self.is_failed()
- def callback_music_loaded(self, success):
- if success:
- self.success()
- else:
+ def callback_music_state(self, new_state):
+ # If a music gets unloaded while the action is loaded_running and
+ # depending on the music, it won't be able to do the finish_action.
+ # Can that happen?
+ # a: play 'mp3';
+ # z: wait 'mp3';
+ # e: pause 'mp3';
+ # r: stop 'mp3'; unload_music 'mp3'
+ if new_state == 'failed':
self.fail()
+ elif self.is_loaded(allow_substates=True) and\
+ new_state in ['initial', 'loading']:
+ self.reload(reloading=True)
+ elif self.is_loading() and new_state.startswith('loaded_'):
+ self.success()
# Machine states / events
- def on_enter_loading(self):
+ def on_enter_loading(self, reloading=False):
+ if reloading:
+ return
if hasattr(actions, self.action):
- if 'music' in self.arguments:
- self.arguments['music'].subscribe_loaded(
- self.callback_music_loaded)
+ if 'music' in self.arguments and\
+ self.action not in ['unload_music', 'load_music']:
+ self.arguments['music'].subscribe_state_change(
+ self.callback_music_state)
else:
self.success()
else:
getattr(actions, self.action).run(self,
key_start_time=key_start_time, **self.arguments)
- def poll_loaded(self):
- self.key.callback_action_ready(self,
- self.is_loaded(allow_substates=True))
+ def on_enter_destroyed(self):
+ if 'music' in self.arguments:
+ self.arguments['music'].unsubscribe_state_change(
+ self.callback_music_state)
+
+ def notify_state_change(self, *args, **kwargs):
+ self.key.callback_action_state_changed()
# This one cannot be in the Machine state since it would be queued to run
# *after* the wait is ended...
# https://github.com/kivy/kivy/wiki/Working-with-Python-threads-inside-a-Kivy-application
from kivy.clock import mainthread
-class KeyMachine(Widget):
+class KeyMachine():
STATES = [
'initial',
'configuring',
{
'trigger': 'repeat_protection_finished',
'source': 'loaded_protecting_repeat',
- 'dest': 'loaded'
+ 'dest': 'loaded',
+ 'after': 'callback_action_state_changed'
},
]
- state = StringProperty("")
+ def __setattr__(self, name, value):
+ if hasattr(self, 'initialized') and name == 'state':
+ self.key.update_state(value)
+ super().__setattr__(name, value)
def __init__(self, key, **kwargs):
self.key = key
Machine(model=self, states=self.STATES,
transitions=self.TRANSITIONS, initial='initial',
ignore_invalid_triggers=True, queued=True)
- super(KeyMachine, self).__init__(**kwargs)
+
+ self.initialized = True
# Machine states / events
def is_loaded_or_failed(self):
def key_loaded_callback(self):
self.key.parent.key_loaded_callback()
+ def callback_action_state_changed(self):
+ if self.state not in ['failed', 'loading', 'loaded']:
+ return
+
+ if any(action.is_failed() for action in self.key.actions):
+ self.to_failed()
+ elif any(action.is_loading() for action in self.key.actions):
+ self.to_loading()
+ else:
+ self.to_loaded()
+ self.key_loaded_callback()
class Key(ButtonBehavior, Widget):
else:
raise AttributeError
- def machine_state_changed(self, instance, machine_state):
- self.machine_state = self.machine.state
-
def __init__(self, **kwargs):
self.actions = []
self.current_action = None
self.machine = KeyMachine(self)
- self.machine.bind(state=self.machine_state_changed)
super(Key, self).__init__(**kwargs)
def interrupt(self):
self.current_action.interrupt()
- # Callbacks
- def callback_action_ready(self, action, success):
- if not success:
- self.fail()
- elif all(action.is_loaded_or_failed() for action in self.actions):
- self.success()
-
# Setters
def set_description(self, description):
if description[0] is not None:
'configuring',
'configured',
'loading',
- 'loaded',
- 'failed'
+ 'loaded'
]
TRANSITIONS = [
'source': 'initial',
'dest': 'configuring'
},
- {
- 'trigger': 'fail',
- 'source': 'configuring',
- 'dest': 'failed'
- },
{
'trigger': 'success',
'source': 'configuring',
'source': 'configured',
'dest': 'loading'
},
- {
- 'trigger': 'fail',
- 'source': 'loading',
- 'dest': 'failed'
- },
{
'trigger': 'success',
'source': 'loading',
self.running = []
self.wait_ids = {}
self.open_files = {}
+ self.is_leaving_application = False
Machine(model=self, states=self.STATES,
transitions=self.TRANSITIONS, initial='initial',
- ignore_invalid_triggers=True, queued=True)
+ auto_transitions=False, queued=True)
super(Mapping, self).__init__(**kwargs)
self.keyboard = Window.request_keyboard(self.on_keyboard_closed, self)
self.keyboard.bind(on_key_down=self.on_keyboard_down)
elif 'ctrl' in modifiers and (keycode[0] == 113 or keycode[0] == '99'):
self.leave_application()
sys.exit()
- elif 'ctrl' in modifiers and keycode[0] == 114:
- threading.Thread(name="MSReload", target=self.reload).start()
+ elif 'ctrl' in modifiers and keycode[0] == 114 and self.is_loaded():
+ self.reload()
return True
def leave_application(self):
self.keyboard.unbind(on_key_down=self.on_keyboard_down)
self.stop_all_running()
+ self.is_leaving_application = True
for music in self.open_files.values():
music.stop()
for thread in threading.enumerate():
# Callbacks
def key_loaded_callback(self):
+ if hasattr(self, 'finished_loading'):
+ return
+
+ opacity = int(Config.load_all_musics)
+
result = self.all_keys_ready()
if result == "success":
- self.ready_color = [0, 1, 0, 1]
+ self.ready_color = [0, 1, 0, opacity]
+ self.finished_loading = True
elif result == "partial":
- self.ready_color = [1, 0, 0, 1]
+ self.ready_color = [1, 0, 0, opacity]
+ self.finished_loading = True
else:
- self.ready_color = [1, 165/255, 0, 1]
+ self.ready_color = [1, 165/255, 0, opacity]
## Some global actions
def stop_all_running(self, except_key=None, key_start_time=0):
{
'name': 'loaded',
'children': [
+ 'stopped',
'playing',
'paused',
'stopping'
TRANSITIONS = [
{
'trigger': 'load',
- 'source': 'initial',
- 'dest': 'loading',
- 'after': 'poll_loaded'
+ 'source': ['initial', 'failed'],
+ 'dest': 'loading'
},
{
'trigger': 'fail',
'source': 'loading',
'dest': 'failed'
},
+ {
+ 'trigger': 'unload',
+ 'source': ['failed', 'loaded_stopped'],
+ 'dest': 'initial',
+ },
{
'trigger': 'success',
'source': 'loading',
- 'dest': 'loaded'
+ 'dest': 'loaded_stopped'
},
{
'trigger': 'start_playing',
- 'source': 'loaded',
- 'dest': 'loaded_playing',
- # if a child has no transitions, then it is bubbled to the parent,
- # and we don't want that. Not useful in that machine precisely.
- 'conditions': ['is_loaded']
+ 'source': 'loaded_stopped',
+ 'dest': 'loaded_playing'
},
{
'trigger': 'pause',
},
{
'trigger': 'stopped',
- 'source': '*',
- 'dest': 'loaded',
+ 'source': 'loaded',
+ 'dest': 'loaded_stopped',
'before': 'trigger_stopped_events',
- 'conditions': ['is_in_use']
+ 'unless': 'is_loaded_stopped',
}
]
def __init__(self, filename, mapping, name=None, gain=1):
- Machine(model=self, states=self.STATES,
+ machine = Machine(model=self, states=self.STATES,
transitions=self.TRANSITIONS, initial='initial',
- ignore_invalid_triggers=True)
+ auto_transitions=False,
+ after_state_change=self.notify_state_change)
- self.loaded_callbacks = []
+ self.state_change_callbacks = []
self.mapping = mapping
self.filename = filename
self.name = name or filename
self.initial_volume_factor = gain
self.music_lock = Lock("music__" + filename)
- threading.Thread(name="MSMusicLoad", target=self.load).start()
+ if Config.load_all_musics:
+ threading.Thread(name="MSMusicLoad", target=self.load).start()
def reload_properties(self, name=None, gain=1):
self.name = name or self.filename
if gain != self.initial_volume_factor:
self.initial_volume_factor = gain
- self.reload_music_file()
+ self.stopped()
+ self.unload()
+ self.load(reloading=True)
- def reload_music_file(self):
- with file_lock:
- try:
- if self.filename.startswith("/"):
- filename = self.filename
- else:
- filename = Config.music_path + self.filename
+ # Machine related events
+ def on_enter_initial(self):
+ self.audio_segment = None
- debug_print("Reloading « {} »".format(self.name))
- initial_db_gain = gain(self.initial_volume_factor * 100)
- self.audio_segment = pydub.AudioSegment \
- .from_file(filename) \
- .set_frame_rate(Config.frame_rate) \
- .set_channels(Config.channels) \
- .set_sample_width(Config.sample_width) \
- .apply_gain(initial_db_gain)
- except Exception as e:
- error_print("failed to reload « {} »: {}"\
- .format(self.name, e))
- self.loading_error = e
- self.to_failed()
- else:
- debug_print("Reloaded « {} »".format(self.name))
+ def on_enter_loading(self, reloading=False):
+ if reloading:
+ prefix = 'Rel'
+ prefix_s = 'rel'
+ else:
+ prefix = 'L'
+ prefix_s = 'l'
- # Machine related events
- def on_enter_loading(self):
with file_lock:
+ if self.mapping.is_leaving_application:
+ self.fail()
+ return
+
try:
if self.filename.startswith("/"):
filename = self.filename
else:
filename = Config.music_path + self.filename
- debug_print("Loading « {} »".format(self.name))
+ debug_print("{}oading « {} »".format(prefix, self.name))
self.mixer = self.mapping.mixer or Mixer()
initial_db_gain = gain(self.initial_volume_factor * 100)
self.audio_segment = pydub.AudioSegment \
.apply_gain(initial_db_gain)
self.sound_duration = self.audio_segment.duration_seconds
except Exception as e:
- error_print("failed to load « {} »: {}".format(self.name, e))
+ error_print("failed to {}oad « {} »: {}".format(
+ prefix_s, self.name, e))
self.loading_error = e
self.fail()
else:
self.success()
- debug_print("Loaded « {} »".format(self.name))
+ debug_print("{}oaded « {} »".format(prefix, self.name))
def on_enter_loaded(self):
self.cleanup()
# Machine related states
def is_in_use(self):
- return self.is_loaded(allow_substates=True) and not self.is_loaded()
+ return self.is_loaded(allow_substates=True) and\
+ not self.is_loaded_stopped()
def is_in_use_not_stopping(self):
return self.is_loaded_playing() or self.is_loaded_paused()
+ def is_unloadable(self):
+ return self.is_loaded_stopped() or self.is_failed()
+
# Machine related triggers
def trigger_stopped_events(self):
self.mixer.remove_file(self)
if wait:
self.mapping.add_wait(self.wait_event, wait_id=set_wait_id)
self.wait_end()
- else:
+ elif self.is_loaded(allow_substates=True):
self.stopped()
def abandon_all_effects(self):
self.wait_event.clear()
self.wait_event.wait()
- # Let other subscribe for an event when they are ready
- def subscribe_loaded(self, callback):
- # FIXME: should lock to be sure we have no race, but it makes the
- # initialization screen not showing until everything is loaded
- if self.is_loaded(allow_substates=True):
- callback(True)
- elif self.is_failed():
- callback(False)
- else:
- self.loaded_callbacks.append(callback)
+ # Let other subscribe for state change
+ def notify_state_change(self, **kwargs):
+ for callback in self.state_change_callbacks:
+ callback(self.state)
+
+ def subscribe_state_change(self, callback):
+ if callback not in self.state_change_callbacks:
+ self.state_change_callbacks.append(callback)
+ callback(self.state)
- def poll_loaded(self):
- for callback in self.loaded_callbacks:
- callback(self.is_loaded())
- self.loaded_callbacks = []
+ def unsubscribe_state_change(self, callback):
+ if callback in self.state_change_callbacks:
+ self.state_change_callbacks.remove(callback)
# Callbacks
def finished_callback(self):
projects / perso / Immae / Projets / Python / MusicSampler.git / commitdiff
