music_sampler/key.py

   1 from kivy.uix.widget import Widget
   2 from kivy.properties import AliasProperty, BooleanProperty, \
   3                             ListProperty, StringProperty
   4 from kivy.uix.behaviors import ButtonBehavior
   5
   6 from .action import Action
   7 from .helpers import debug_print
   8 import time
   9 import threading
  10 from transitions.extensions import HierarchicalMachine as Machine
  11
  12 # All drawing operations should happen in the main thread
  13 # https://github.com/kivy/kivy/wiki/Working-with-Python-threads-inside-a-Kivy-application
  14 from kivy.clock import mainthread
  15
  16 class KeyMachine():
  17     STATES = [
  18         'initial',
  19         'configuring',
  20         'configured',
  21         'loading',
  22         'failed',
  23         {
  24             'name': 'loaded',
  25             'children': [
  26                 'no_config',
  27                 'no_actions',
  28                 'running',
  29                 'protecting_repeat'
  30             ]
  31         }
  32     ]
  33
  34     TRANSITIONS = [
  35         {
  36             'trigger': 'configure',
  37             'source': 'initial',
  38             'dest': 'configuring'
  39         },
  40         {
  41             'trigger': 'fail',
  42             'source': 'configuring',
  43             'dest': 'failed',
  44             'after': 'key_loaded_callback'
  45         },
  46         {
  47             'trigger': 'success',
  48             'source': 'configuring',
  49             'dest': 'configured',
  50             'after': 'load'
  51         },
  52         {
  53             'trigger': 'no_config',
  54             'source': 'configuring',
  55             'dest': 'loaded_no_config',
  56             'after': 'key_loaded_callback'
  57         },
  58         {
  59             'trigger': 'load',
  60             'source': 'configured',
  61             'dest': 'loading'
  62         },
  63         {
  64             'trigger': 'fail',
  65             'source': 'loading',
  66             'dest': 'failed',
  67             'after': 'key_loaded_callback'
  68         },
  69         {
  70             'trigger': 'success',
  71             'source': 'loading',
  72             'dest': 'loaded',
  73             'after': 'key_loaded_callback'
  74         },
  75         {
  76             'trigger': 'no_actions',
  77             'source': 'loading',
  78             'dest': 'loaded_no_actions',
  79             'after': 'key_loaded_callback'
  80         },
  81         {
  82             'trigger': 'reload',
  83             'source': ['loaded','failed'],
  84             'dest': 'configuring',
  85             'after': 'key_loaded_callback'
  86         },
  87         {
  88             'trigger': 'run',
  89             'source': 'loaded',
  90             'dest': 'loaded_running',
  91             'after': ['run_actions', 'finish'],
  92             # if a child, like loaded_no_actions, has no transitions, then it
  93             # is bubbled to the parent, and we don't want that.
  94             'conditions': ['is_loaded']
  95         },
  96         {
  97             'trigger': 'finish',
  98             'source': 'loaded_running',
  99             'dest': 'loaded_protecting_repeat'
 100         },
 101         {
 102             'trigger': 'repeat_protection_finished',
 103             'source': 'loaded_protecting_repeat',
 104             'dest': 'loaded',
 105             'after': 'callback_action_state_changed'
 106         },
 107     ]
 108
 109     def __setattr__(self, name, value):
 110         if hasattr(self, 'initialized') and name == 'state':
 111             self.key.update_state(value)
 112         super().__setattr__(name, value)
 113
 114     def __init__(self, key, **kwargs):
 115         self.key = key
 116
 117         Machine(model=self, states=self.STATES,
 118                 transitions=self.TRANSITIONS, initial='initial',
 119                 ignore_invalid_triggers=True, queued=True)
 120
 121         self.initialized = True
 122
 123     # Machine states / events
 124     def is_loaded_or_failed(self):
 125         return self.is_loaded(allow_substates=True) or self.is_failed()
 126
 127     def is_loaded_inactive(self):
 128         return self.is_loaded_no_config() or self.is_loaded_no_actions()
 129
 130     @mainthread
 131     def on_enter_configuring(self):
 132         self.destroy_actions()
 133         self.key.unset_description()
 134         self.key.unset_color()
 135
 136         if self.key.key_sym in self.key.parent.key_config:
 137             self.key.config = self.key.parent.key_config[self.key.key_sym]
 138
 139             for key_action in self.key.config['actions']:
 140                 self.key.add_action(key_action[0], **key_action[1])
 141
 142             if 'description' in self.key.config['properties']:
 143                 self.key.set_description(self.key.config['properties']['description'])
 144             if 'color' in self.key.config['properties']:
 145                 self.key.set_color(self.key.config['properties']['color'])
 146             self.success()
 147         else:
 148             self.no_config()
 149
 150     def on_enter_loading(self):
 151         if len(self.key.actions) > 0:
 152             for action in self.key.actions:
 153                 action.load()
 154         else:
 155             self.no_actions()
 156
 157     def destroy_actions(self):
 158         for action in self.key.actions:
 159             action.destroy()
 160         self.key.actions = []
 161
 162     def run_actions(self, modifiers):
 163         self.key.parent.parent.ids['KeyList'].append(self.key.key_sym)
 164         debug_print("running actions for {}".format(self.key.key_sym))
 165         start_time = time.time()
 166         self.key.parent.start_running(self.key, start_time)
 167         for self.key.current_action in self.key.actions:
 168             if self.key.parent.keep_running(self.key, start_time):
 169                 self.key.list_actions()
 170                 self.key.current_action.run(start_time)
 171         self.key.list_actions(last_action_finished=True)
 172
 173         self.key.parent.finished_running(self.key, start_time)
 174
 175     def on_enter_loaded_protecting_repeat(self, modifiers):
 176         if self.key.repeat_delay > 0:
 177             self.key.protecting_repeat_timer = threading.Timer(
 178                     self.key.repeat_delay,
 179                     self.key.repeat_protection_finished)
 180             self.key.protecting_repeat_timer.start()
 181         else:
 182             self.key.repeat_protection_finished()
 183
 184     # Callbacks
 185     @mainthread
 186     def key_loaded_callback(self):
 187         self.key.parent.key_loaded_callback()
 188
 189     def callback_action_state_changed(self):
 190         if self.state not in ['failed', 'loading', 'loaded']:
 191             return
 192
 193         if any(action.is_failed() for action in self.key.actions):
 194             self.to_failed()
 195         elif any(action.is_loading() for action in self.key.actions):
 196             self.to_loading()
 197         else:
 198             self.to_loaded()
 199         self.key_loaded_callback()
 200
 201 class Key(ButtonBehavior, Widget):
 202
 203     key_sym = StringProperty(None)
 204     custom_color = ListProperty([0, 1, 0])
 205     description_title = StringProperty("")
 206     description = ListProperty([])
 207     machine_state = StringProperty("")
 208
 209     def get_alias_line_cross_color(self):
 210         if not self.is_failed() and (
 211                 not self.is_loaded(allow_substates=True)\
 212                 or self.is_loaded_running()\
 213                 or self.is_loaded_protecting_repeat()):
 214             return [120/255, 120/255, 120/255, 1]
 215         else:
 216             return [0, 0, 0, 0]
 217
 218     def set_alias_line_cross_color(self):
 219         pass
 220
 221     line_cross_color = AliasProperty(
 222             get_alias_line_cross_color,
 223             set_alias_line_cross_color,
 224             bind=['machine_state'])
 225
 226     def get_alias_line_color(self):
 227         if self.is_loaded_running():
 228             return [0, 0, 0, 1]
 229         else:
 230             return [120/255, 120/255, 120/255, 1]
 231
 232     def set_alias_line_color(self):
 233         pass
 234
 235     line_color = AliasProperty(get_alias_line_color, set_alias_line_color,
 236             bind=['machine_state'])
 237
 238     def get_alias_color(self):
 239         if self.is_loaded_inactive():
 240             return [1, 1, 1, 1]
 241         elif self.is_loaded_protecting_repeat():
 242             return [*self.custom_color, 100/255]
 243         elif self.is_loaded_running():
 244             return [*self.custom_color, 100/255]
 245         elif self.is_loaded(allow_substates=True):
 246             return [*self.custom_color, 1]
 247         elif self.is_failed():
 248             return [0, 0, 0, 1]
 249         else:
 250             return [*self.custom_color, 100/255]
 251     def set_alias_color(self):
 252         pass
 253
 254     color = AliasProperty(get_alias_color, set_alias_color,
 255             bind=['machine_state', 'custom_color'])
 256
 257     def __getattr__(self, name):
 258         if hasattr(self.machine, name):
 259             return getattr(self.machine, name)
 260         else:
 261             raise AttributeError
 262
 263     def __init__(self, **kwargs):
 264         self.actions = []
 265         self.current_action = None
 266         self.machine = KeyMachine(self)
 267
 268         super(Key, self).__init__(**kwargs)
 269
 270     # Kivy events
 271     @mainthread
 272     def update_state(self, value):
 273         self.machine_state = value
 274
 275     def on_key_sym(self, key, key_sym):
 276         if key_sym != "":
 277             self.configure()
 278
 279     def on_press(self):
 280         self.list_actions()
 281
 282     # This one cannot be in the Machine state since it would be queued to run
 283     # *after* the loop is ended...
 284     def interrupt(self):
 285         self.current_action.interrupt()
 286
 287     # Setters
 288     def set_description(self, description):
 289         if description[0] is not None:
 290             self.description_title = str(description[0])
 291         self.description = []
 292         for desc in description[1 :]:
 293             if desc is None:
 294                 self.description.append("")
 295             else:
 296                 self.description.append(str(desc).replace(" ", " "))
 297
 298     def unset_description(self):
 299         self.description_title = ""
 300         self.description = []
 301
 302     def set_color(self, color):
 303         color = [x / 255 for x in color]
 304         self.custom_color = color
 305
 306     def unset_color(self):
 307         self.custom_color = [0, 1, 0]
 308
 309     # Helpers
 310     @property
 311     def repeat_delay(self):
 312          if hasattr(self, 'config') and\
 313                  'repeat_delay' in self.config['properties']:
 314              return self.config['properties']['repeat_delay']
 315          else:
 316              return 0
 317
 318     # Actions handling
 319     def add_action(self, action_name, **arguments):
 320         self.actions.append(Action(action_name, self, **arguments))
 321
 322     def list_actions(self, last_action_finished=False):
 323         not_running = (not self.is_loaded_running())
 324         current_action_seen = False
 325         action_descriptions = []
 326         for action in self.actions:
 327             if not_running:
 328                 state = "inactive"
 329             elif last_action_finished:
 330                 state = "done"
 331             elif current_action_seen:
 332                 state = "pending"
 333             elif action == self.current_action:
 334                 current_action_seen = True
 335                 state = "current"
 336             else:
 337                 state = "done"
 338             action_descriptions.append([action.description(), state])
 339         self.parent.parent.ids['ActionList'].update_list(
 340                 self,
 341                 action_descriptions)