X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=music_sampler%2Fkey.py;fp=music_sampler%2Fkey.py;h=66e792d514ed2ac31a10a273dc95605ff2d5b8ac;hb=63ba5a8dc2aa4ec3e6f203b0ba4db249ecf0b00e;hp=0000000000000000000000000000000000000000;hpb=35bde798b6cda13579337b0ec5a803fdd5eab19a;p=perso%2FImmae%2FProjets%2FPython%2FMusicSampler.git diff --git a/music_sampler/key.py b/music_sampler/key.py new file mode 100644 index 0000000..66e792d --- /dev/null +++ b/music_sampler/key.py @@ -0,0 +1,280 @@ +from kivy.uix.widget import Widget +from kivy.properties import AliasProperty, BooleanProperty, \ + ListProperty, StringProperty +from kivy.uix.behaviors import ButtonBehavior + +from .action import Action +from . import debug_print +import time +import threading +from transitions.extensions import HierarchicalMachine as Machine + +class Key(ButtonBehavior, Widget): + STATES = [ + 'initial', + 'configuring', + 'configured', + 'loading', + 'failed', + { + 'name': 'loaded', + 'children': [ + 'no_config', + 'no_actions', + 'running', + 'protecting_repeat' + ] + } + ] + + TRANSITIONS = [ + { + 'trigger': 'configure', + 'source': 'initial', + 'dest': 'configuring' + }, + { + 'trigger': 'fail', + 'source': 'configuring', + 'dest': 'failed', + 'after': 'key_loaded_callback' + }, + { + 'trigger': 'success', + 'source': 'configuring', + 'dest': 'configured', + 'after': 'load' + }, + { + 'trigger': 'no_config', + 'source': 'configuring', + 'dest': 'loaded_no_config', + 'after': 'key_loaded_callback' + }, + { + 'trigger': 'load', + 'source': 'configured', + 'dest': 'loading' + }, + { + 'trigger': 'fail', + 'source': 'loading', + 'dest': 'failed', + 'after': 'key_loaded_callback' + }, + { + 'trigger': 'success', + 'source': 'loading', + 'dest': 'loaded', + 'after': 'key_loaded_callback' + }, + { + 'trigger': 'no_actions', + 'source': 'loading', + 'dest': 'loaded_no_actions', + 'after': 'key_loaded_callback' + }, + { + 'trigger': 'reload', + 'source': ['loaded','failed'], + 'dest': 'configuring', + 'after': 'key_loaded_callback' + }, + { + 'trigger': 'run', + 'source': 'loaded', + 'dest': 'loaded_running', + 'after': ['run_actions', 'finish'], + # if a child, like loaded_no_actions, has no transitions, then it + # is bubbled to the parent, and we don't want that. + 'conditions': ['is_loaded'] + }, + { + 'trigger': 'finish', + 'source': 'loaded_running', + 'dest': 'loaded_protecting_repeat' + }, + { + 'trigger': 'repeat_protection_finished', + 'source': 'loaded_protecting_repeat', + 'dest': 'loaded' + }, + ] + + key_sym = StringProperty(None) + custom_color = ListProperty([0, 1, 0]) + description_title = StringProperty("") + description = ListProperty([]) + state = StringProperty("") + + def get_alias_line_cross_color(self): + if not self.is_failed() and ( + not self.is_loaded(allow_substates=True)\ + or self.is_loaded_running()\ + or self.is_loaded_protecting_repeat()): + return [120/255, 120/255, 120/255, 1] + else: + return [0, 0, 0, 0] + + def set_alias_line_cross_color(self): + pass + + line_cross_color = AliasProperty( + get_alias_line_cross_color, + set_alias_line_cross_color, + bind=['state']) + + def get_alias_line_color(self): + if self.is_loaded_running(): + return [0, 0, 0, 1] + else: + return [120/255, 120/255, 120/255, 1] + + def set_alias_line_color(self): + pass + + line_color = AliasProperty(get_alias_line_color, set_alias_line_color, + bind=['state']) + + def get_alias_color(self): + if self.is_loaded_inactive(): + return [1, 1, 1, 1] + elif self.is_loaded_protecting_repeat(): + return [*self.custom_color, 100/255] + elif self.is_loaded_running(): + return [*self.custom_color, 100/255] + elif self.is_loaded(allow_substates=True): + return [*self.custom_color, 1] + elif self.is_failed(): + return [0, 0, 0, 1] + else: + return [*self.custom_color, 100/255] + def set_alias_color(self): + pass + + color = AliasProperty(get_alias_color, set_alias_color, + bind=['state', 'custom_color']) + + def __init__(self, **kwargs): + self.actions = [] + self.current_action = None + + Machine(model=self, states=self.STATES, + transitions=self.TRANSITIONS, initial='initial', + ignore_invalid_triggers=True, queued=True) + super(Key, self).__init__(**kwargs) + + # Kivy events + def on_key_sym(self, key, key_sym): + if key_sym != "": + self.configure() + + def on_press(self): + self.list_actions() + + # Machine states / events + def is_loaded_or_failed(self): + return self.is_loaded(allow_substates=True) or self.is_failed() + + def is_loaded_inactive(self): + return self.is_loaded_no_config() or self.is_loaded_no_actions() + + def on_enter_configuring(self): + if self.key_sym in self.parent.key_config: + self.config = self.parent.key_config[self.key_sym] + + self.actions = [] + for key_action in self.config['actions']: + self.add_action(key_action[0], **key_action[1]) + + if 'description' in self.config['properties']: + self.set_description(self.config['properties']['description']) + if 'color' in self.config['properties']: + self.set_color(self.config['properties']['color']) + self.success() + else: + self.no_config() + + def on_enter_loading(self): + if len(self.actions) > 0: + for action in self.actions: + action.load() + else: + self.no_actions() + + def run_actions(self, modifiers): + self.parent.parent.ids['KeyList'].append(self.key_sym) + debug_print("running actions for {}".format(self.key_sym)) + start_time = time.time() + self.parent.start_running(self, start_time) + for self.current_action in self.actions: + if self.parent.keep_running(self, start_time): + self.list_actions() + self.current_action.run(start_time) + self.list_actions(last_action_finished=True) + + self.parent.finished_running(self, start_time) + + def on_enter_loaded_protecting_repeat(self, modifiers): + if 'repeat_delay' in self.config['properties']: + self.protecting_repeat_timer = threading.Timer( + self.config['properties']['repeat_delay'], + self.repeat_protection_finished) + self.protecting_repeat_timer.start() + else: + self.repeat_protection_finished() + + # This one cannot be in the Machine state since it would be queued to run + # *after* the loop is ended... + def interrupt(self): + self.current_action.interrupt() + + # Callbacks + def key_loaded_callback(self): + self.parent.key_loaded_callback() + + 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: + self.description_title = str(description[0]) + self.description = [] + for desc in description[1 :]: + if desc is None: + self.description.append("") + else: + self.description.append(str(desc).replace(" ", " ")) + + def set_color(self, color): + color = [x / 255 for x in color] + self.custom_color = color + + # Actions handling + def add_action(self, action_name, **arguments): + self.actions.append(Action(action_name, self, **arguments)) + + def list_actions(self, last_action_finished=False): + not_running = (not self.is_loaded_running()) + current_action_seen = False + action_descriptions = [] + for action in self.actions: + if not_running: + state = "inactive" + elif last_action_finished: + state = "done" + elif current_action_seen: + state = "pending" + elif action == self.current_action: + current_action_seen = True + state = "current" + else: + state = "done" + action_descriptions.append([action.description(), state]) + self.parent.parent.ids['ActionList'].update_list( + self, + action_descriptions)