+++ /dev/null
-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)