path: root/music_sampler/key.py

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 .helpers import debug_print
import time
import threading
from transitions.extensions import HierarchicalMachine as Machine

# All drawing operations should happen in the main thread
# https://github.com/kivy/kivy/wiki/Working-with-Python-threads-inside-a-Kivy-application
from kivy.clock import mainthread

class KeyMachine():
    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',
            'after': 'callback_action_state_changed'
        },
    ]

    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)

        self.initialized = True

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

    @mainthread
    def on_enter_configuring(self):
        self.destroy_actions()
        self.key.unset_description()
        self.key.unset_color()

        if self.key.key_sym in self.key.parent.key_config:
            self.key.config = self.key.parent.key_config[self.key.key_sym]

            for key_action in self.key.config['actions']:
                self.key.add_action(key_action[0], **key_action[1])

            if 'description' in self.key.config['properties']:
                self.key.set_description(self.key.config['properties']['description'])
            if 'color' in self.key.config['properties']:
                self.key.set_color(self.key.config['properties']['color'])
            self.success()
        else:
            self.no_config()

    def on_enter_loading(self):
        if len(self.key.actions) > 0:
            for action in self.key.actions:
                action.load()
        else:
            self.no_actions()

    def destroy_actions(self):
        for action in self.key.actions:
            action.destroy()
        self.key.actions = []

    def run_actions(self, modifiers):
        self.key.parent.parent.ids['KeyList'].append(self.key.key_sym)
        debug_print("running actions for {}".format(self.key.key_sym))
        start_time = time.time()
        self.key.parent.start_running(self.key, start_time)
        for self.key.current_action in self.key.actions:
            if self.key.parent.keep_running(self.key, start_time):
                self.key.list_actions()
                self.key.current_action.run(start_time)
        self.key.list_actions(last_action_finished=True)

        self.key.parent.finished_running(self.key, start_time)

    def on_enter_loaded_protecting_repeat(self, modifiers):
        if self.key.repeat_delay > 0:
            self.key.protecting_repeat_timer = threading.Timer(
                    self.key.repeat_delay,
                    self.key.repeat_protection_finished)
            self.key.protecting_repeat_timer.start()
        else:
            self.key.repeat_protection_finished()

    # Callbacks
    @mainthread
    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):

    key_sym = StringProperty(None)
    custom_color = ListProperty([0, 1, 0])
    description_title = StringProperty("")
    description = ListProperty([])
    machine_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=['machine_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=['machine_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=['machine_state', 'custom_color'])

    def __getattr__(self, name):
        if hasattr(self.machine, name):
            return getattr(self.machine, name)
        else:
            raise AttributeError

    def __init__(self, **kwargs):
        self.actions = []
        self.current_action = None
        self.machine = KeyMachine(self)

        super(Key, self).__init__(**kwargs)

    # Kivy events
    @mainthread
    def update_state(self, value):
        self.machine_state = value

    def on_key_sym(self, key, key_sym):
        if key_sym != "":
            self.configure()

    def on_press(self):
        self.list_actions()

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

    # 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 unset_description(self):
        self.description_title = ""
        self.description = []

    def set_color(self, color):
        color = [x / 255 for x in color]
        self.custom_color = color

    def unset_color(self):
        self.custom_color = [0, 1, 0]

    # Helpers
    @property
    def repeat_delay(self):
         if hasattr(self, 'config') and\
                 'repeat_delay' in self.config['properties']:
             return self.config['properties']['repeat_delay']
         else:
             return 0

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