1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
from transitions.extensions import HierarchicalMachine as Machine
from .helpers import debug_print, error_print
from . import actions
class Action:
STATES = [
'initial',
'loading',
'failed',
{
'name': 'loaded',
'children': ['running']
}
]
TRANSITIONS = [
{
'trigger': 'load',
'source': 'initial',
'dest': 'loading'
},
{
'trigger': 'fail',
'source': 'loading',
'dest': 'failed',
'after': 'poll_loaded'
},
{
'trigger': 'success',
'source': 'loading',
'dest': 'loaded',
'after': 'poll_loaded'
},
{
'trigger': 'run',
'source': 'loaded',
'dest': 'loaded_running',
'after': 'finish_action',
# if a child has no transitions, then it is bubbled to the parent,
# and we don't want that. Not useful in that machine precisely.
'conditions': ['is_loaded']
},
{
'trigger': 'finish_action',
'source': 'loaded_running',
'dest': 'loaded'
}
]
def __init__(self, action, key, **kwargs):
Machine(model=self, states=self.STATES,
transitions=self.TRANSITIONS, initial='initial',
ignore_invalid_triggers=True, queued=True)
self.action = action
self.key = key
self.mapping = key.parent
self.arguments = kwargs
self.sleep_event = None
self.waiting_music = None
def is_loaded_or_failed(self):
return self.is_loaded(allow_substates=True) or self.is_failed()
def callback_music_loaded(self, success):
if success:
self.success()
else:
self.fail()
# Machine states / events
def on_enter_loading(self):
if hasattr(actions, self.action):
if 'music' in self.arguments:
self.arguments['music'].subscribe_loaded(
self.callback_music_loaded)
else:
self.success()
else:
error_print("Unknown action {}".format(self.action))
self.fail()
def on_enter_loaded_running(self, key_start_time):
debug_print(self.description())
if hasattr(actions, self.action):
getattr(actions, self.action).run(self,
key_start_time=key_start_time, **self.arguments)
def poll_loaded(self):
self.key.callback_action_ready(self,
self.is_loaded(allow_substates=True))
# This one cannot be in the Machine state since it would be queued to run
# *after* the wait is ended...
def interrupt(self):
if getattr(actions, self.action, None) and\
hasattr(getattr(actions, self.action), 'interrupt'):
return getattr(getattr(actions, self.action), 'interrupt')(
self, **self.arguments)
def pause(self):
if getattr(actions, self.action, None) and\
hasattr(getattr(actions, self.action), 'pause'):
return getattr(getattr(actions, self.action), 'pause')(
self, **self.arguments)
def unpause(self):
if getattr(actions, self.action, None) and\
hasattr(getattr(actions, self.action), 'unpause'):
return getattr(getattr(actions, self.action), 'unpause')(
self, **self.arguments)
def reset(self):
if getattr(actions, self.action, None) and\
hasattr(getattr(actions, self.action), 'reset'):
return getattr(getattr(actions, self.action), 'reset')(
self, **self.arguments)
# Helpers
def music_list(self, music):
if music is not None:
return [music]
else:
return self.mapping.open_files.values()
def description(self):
if hasattr(actions, self.action):
return getattr(actions, self.action)\
.description(self, **self.arguments)
else:
return _("unknown action {}").format(self.action)
|
