from .action import Action
from . import debug_print
import time
+import threading
from transitions.extensions import HierarchicalMachine as Machine
class Key(ButtonBehavior, Widget):
'failed',
{
'name': 'loaded',
- 'children': ['no_config', 'no_actions', 'running']
+ 'children': [
+ 'no_config',
+ 'no_actions',
+ 'running',
+ 'protecting_repeat'
+ ]
}
]
},
{
'trigger': 'reload',
- 'source': 'loaded',
+ 'source': ['loaded','failed'],
'dest': 'configuring',
'after': 'key_loaded_callback'
},
'trigger': 'run',
'source': 'loaded',
'dest': 'loaded_running',
- 'after': 'finish',
+ '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)
description = ListProperty([])
state = StringProperty("")
+ 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(allow_substates=True):
return [*self.custom_color, 1]
elif self.is_failed():
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)
else:
self.no_actions()
- def on_enter_loaded_running(self, modifiers):
+ 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)
- action_number = 0
for self.current_action in self.actions:
if self.parent.keep_running(self, start_time):
- self.list_actions(action_number=action_number + 0.5)
+ self.list_actions()
self.current_action.run()
- action_number += 1
- self.list_actions(action_number=action_number)
+ 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):
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("")
def add_action(self, action_name, **arguments):
self.actions.append(Action(action_name, self, **arguments))
- def list_actions(self, action_number=0):
- self.parent.parent.ids['ActionList'].update_list(self, action_number)
-
+ 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)