Belle II Software development
Machine Class Reference
Inheritance diagram for Machine:
AlgorithmMachine CalibrationMachine

Public Member Functions

 __init__ (self, states=None, initial_state="default_initial")
 
 add_state (self, state, enter=None, exit=None)
 
 initial_state (self)
 
 initial_state (self, state)
 
 state (self)
 
 state (self, state)
 
 add_transition (self, trigger, source, dest, conditions=None, before=None, after=None)
 
 __getattr__ (self, name, **kwargs)
 
 get_transitions (self, source)
 
 get_transition_dict (self, state, transition)
 
 save_graph (self, filename, graphname)
 

Static Public Member Functions

 default_condition (**kwargs)
 

Public Attributes

dict states = {}
 Valid states for this machine.
 
 initial_state = initial_state
 Pointless docstring since it's a property.
 
 transitions = defaultdict(list)
 Allowed transitions between states.
 
 state = dest
 Current State of machine.
 

Protected Member Functions

 _trigger (self, transition_name, transition_dict, **kwargs)
 

Static Protected Member Functions

 _callback (func, **kwargs)
 

Protected Attributes

dict _initial_state = State(initial_state)
 Actual attribute holding initial state for this machine.
 
dict _state = self.initial_state
 Actual attribute holding the Current state.
 

Detailed Description

Parameters:
  states (list[str]): A list of possible states of the machine.
  initial_state (str):

Base class for a final state machine wrapper.
Implements the framework that a more complex machine can inherit from.

The `transitions` attribute is a dictionary of trigger name keys, each value of
which is another dictionary of 'source' states, 'dest' states, and 'conditions'
methods. 'conditions' should be a list of callables or a single one. A transition is
valid if it goes from an allowed state to an allowed state.
Conditions are optional but must be a callable that returns True or False based
on some state of the machine. They cannot have input arguments currently.

Every condition/before/after callback function MUST take ``**kwargs`` as the only
argument (except ``self`` if it's a class method). This is because it's basically
impossible to determine which arguments to pass to which functions for a transition.
Therefore this machine just enforces that every function should simply take ``**kwargs``
and use the dictionary of arguments (even if it doesn't need any arguments).

This also means that if you call a trigger with arguments e.g. ``machine.walk(speed=5)``
you MUST use the keyword arguments rather than positional ones. So ``machine.walk(5)``
will *not* work.

Definition at line 144 of file state_machines.py.

Constructor & Destructor Documentation

◆ __init__()

__init__ ( self,
states = None,
initial_state = "default_initial" )
Basic Setup of states and initial_state.

Definition at line 171 of file state_machines.py.

171 def __init__(self, states=None, initial_state="default_initial"):
172 """
173 Basic Setup of states and initial_state.
174 """
175
176 self.states = {}
177 if states:
178 for state in states:
179 self.add_state(state)
180 if initial_state != "default_initial":
181
182 self.initial_state = initial_state
183 else:
184 self.add_state(initial_state)
185
186 self._initial_state = State(initial_state)
187
188
189 self._state = self.initial_state
190
191 self.transitions = defaultdict(list)
192

Member Function Documentation

◆ __getattr__()

__getattr__ ( self,
name,
** kwargs )
Allows us to create a new method for each trigger on the fly.
If there is no trigger name in the machine to match, then the normal
AttributeError is called.

Definition at line 306 of file state_machines.py.

306 def __getattr__(self, name, **kwargs):
307 """
308 Allows us to create a new method for each trigger on the fly.
309 If there is no trigger name in the machine to match, then the normal
310 AttributeError is called.
311 """
312 possible_transitions = self.get_transitions(self.state)
313 if name not in possible_transitions:
314 raise AttributeError(f"{name} does not exist in transitions for state {self.state}.")
315 transition_dict = self.get_transition_dict(self.state, name)
316 # \cond silence doxygen warning about _trigger
317 return partial(self._trigger, name, transition_dict, **kwargs)
318 # \endcond
319

◆ _callback()

_callback ( func,
** kwargs )
staticprotected
Calls a condition/before/after.. function using arguments passed (or not).

Definition at line 344 of file state_machines.py.

344 def _callback(func, **kwargs):
345 """
346 Calls a condition/before/after.. function using arguments passed (or not).
347 """
348 return func(**kwargs)
349

◆ _trigger()

_trigger ( self,
transition_name,
transition_dict,
** kwargs )
protected
Runs the transition logic. Callbacks are evaluated in the order:
conditions -> before -> <new state set here> -> after.

Definition at line 320 of file state_machines.py.

320 def _trigger(self, transition_name, transition_dict, **kwargs):
321 """
322 Runs the transition logic. Callbacks are evaluated in the order:
323 conditions -> before -> <new state set here> -> after.
324 """
325 dest, conditions, before_callbacks, after_callbacks = (
326 transition_dict["dest"],
327 transition_dict["conditions"],
328 transition_dict["before"],
329 transition_dict["after"]
330 )
331 # Returns True only if every condition returns True when called
332 if all(map(lambda condition: self._callback(condition, **kwargs), conditions)):
333 for before_func in before_callbacks:
334 self._callback(before_func, **kwargs)
335
336 self.state = dest
337 for after_func in after_callbacks:
338 self._callback(after_func, **kwargs)
339 else:
340 raise ConditionError(f"Transition '{transition_name}' called for but one or more conditions "
341 "evaluated False")
342
STL class.

◆ add_state()

add_state ( self,
state,
enter = None,
exit = None )
Adds a single state to the list of possible ones.
Should be a unique string or a State object with a unique name.

Definition at line 193 of file state_machines.py.

193 def add_state(self, state, enter=None, exit=None):
194 """
195 Adds a single state to the list of possible ones.
196 Should be a unique string or a State object with a unique name.
197 """
198 if isinstance(state, str):
199 self.add_state(State(state, enter, exit))
200 elif isinstance(state, State):
201 if state.name not in self.states.keys():
202 self.states[state.name] = state
203 else:
204 B2WARNING(f"You asked to add a state {state} but it was already in the machine states.")
205 else:
206 B2WARNING(f"You asked to add a state {state} but it wasn't a State or str object")
207

◆ add_transition()

add_transition ( self,
trigger,
source,
dest,
conditions = None,
before = None,
after = None )
Adds a single transition to the dictionary of possible ones.
Trigger is the method name that begins the transition between the
source state and the destination state.

The condition is an optional function that returns True or False
depending on the current state/input.

Definition at line 264 of file state_machines.py.

264 def add_transition(self, trigger, source, dest, conditions=None, before=None, after=None):
265 """
266 Adds a single transition to the dictionary of possible ones.
267 Trigger is the method name that begins the transition between the
268 source state and the destination state.
269
270 The condition is an optional function that returns True or False
271 depending on the current state/input.
272 """
273 transition_dict = {}
274 try:
275 source = self.states[source]
276 dest = self.states[dest]
277 transition_dict["source"] = source
278 transition_dict["dest"] = dest
279 except KeyError as err:
280 B2WARNING("Tried to add a transition where the source or dest isn't in the list of states")
281 raise err
282 if conditions:
283 if isinstance(conditions, (list, tuple, set)):
284 transition_dict["conditions"] = list(conditions)
285 else:
286 transition_dict["conditions"] = [conditions]
287 else:
288 transition_dict["conditions"] = [Machine.default_condition]
289
290 if not before:
291 before = []
292 if isinstance(before, (list, tuple, set)):
293 transition_dict["before"] = list(before)
294 else:
295 transition_dict["before"] = [before]
296
297 if not after:
298 after = []
299 if isinstance(after, (list, tuple, set)):
300 transition_dict["after"] = list(after)
301 else:
302 transition_dict["after"] = [after]
303
304 self.transitions[trigger].append(transition_dict)
305

◆ default_condition()

default_condition ( ** kwargs)
static
Method to always return True.

Definition at line 258 of file state_machines.py.

258 def default_condition(**kwargs):
259 """
260 Method to always return True.
261 """
262 return True
263

◆ get_transition_dict()

get_transition_dict ( self,
state,
transition )
Returns the transition dictionary for a state and transition out of it.

Definition at line 361 of file state_machines.py.

361 def get_transition_dict(self, state, transition):
362 """
363 Returns the transition dictionary for a state and transition out of it.
364 """
365 transition_dicts = self.transitions[transition]
366 for transition_dict in transition_dicts:
367 if transition_dict["source"] == state:
368 return transition_dict
369 else:
370 raise KeyError(f"No transition from state {state} with the name {transition}.")
371

◆ get_transitions()

get_transitions ( self,
source )
Returns allowed transitions from a given state.

Definition at line 350 of file state_machines.py.

350 def get_transitions(self, source):
351 """
352 Returns allowed transitions from a given state.
353 """
354 possible_transitions = []
355 for transition, transition_dicts in self.transitions.items():
356 for transition_dict in transition_dicts:
357 if transition_dict["source"] == source:
358 possible_transitions.append(transition)
359 return possible_transitions
360

◆ initial_state() [1/2]

initial_state ( self)
The initial state of the machine. Needs a special property to prevent trying to run on_enter callbacks when set.

Definition at line 209 of file state_machines.py.

209 def initial_state(self):
210 """
211 The initial state of the machine. Needs a special property to prevent trying to run on_enter callbacks when set.
212 """
213 return self._initial_state
214

◆ initial_state() [2/2]

initial_state ( self,
state )
 

Definition at line 216 of file state_machines.py.

216 def initial_state(self, state):
217 """
218 """
219 if state in self.states.keys():
220 self._initial_state = self.states[state]
221
222 self._state = self.states[state]
223 else:
224 raise KeyError(f"Attempted to set state to '{state}' which is not in the 'states' attribute!")
225

◆ save_graph()

save_graph ( self,
filename,
graphname )
Does a simple dot file creation to visualise states and transiitons.

Definition at line 372 of file state_machines.py.

372 def save_graph(self, filename, graphname):
373 """
374 Does a simple dot file creation to visualise states and transiitons.
375 """
376 with open(filename, "w") as dotfile:
377 dotfile.write("digraph " + graphname + " {\n")
378 for state in self.states.keys():
379 dotfile.write('"' + state + '" [shape=ellipse, color=black]\n')
380 for trigger, transition_dicts in self.transitions.items():
381 for transition in transition_dicts:
382 dotfile.write('"' + transition["source"].name + '" -> "' +
383 transition["dest"].name + '" [label="' + trigger + '"]\n')
384 dotfile.write("}\n")
385
386

◆ state() [1/2]

state ( self)
        The current state of the machine. Actually a `property` decorator. It will call the exit method of the
        current state and enter method of the new one. To get around the behaviour e.g. for setting initial states,
        either use the `initial_state` property or directly set the _state attribute itself (at your own risk!).

Definition at line 227 of file state_machines.py.

227 def state(self):
228 """
229 The current state of the machine. Actually a `property` decorator. It will call the exit method of the
230 current state and enter method of the new one. To get around the behaviour e.g. for setting initial states,
231 either use the `initial_state` property or directly set the _state attribute itself (at your own risk!).
232 """
233 return self._state
234

◆ state() [2/2]

state ( self,
state )
 

Definition at line 236 of file state_machines.py.

236 def state(self, state):
237 """
238 """
239 if isinstance(state, str):
240 state_name = state
241 else:
242 state_name = state.name
243
244 try:
245 state = self.states[state_name]
246 # Run exit callbacks of current state
247 for callback in self.state.on_exit:
248 callback(prior_state=self.state, new_state=state)
249 # Run enter callbacks of new state
250 for callback in state.on_enter:
251 callback(prior_state=self.state, new_state=state)
252 # Set the state
253 self._state = state
254 except KeyError:
255 raise MachineError(f"Attempted to set state to '{state}' which not in the 'states' attribute!")
256

Member Data Documentation

◆ _initial_state

dict _initial_state = State(initial_state)
protected

Actual attribute holding initial state for this machine.

Definition at line 186 of file state_machines.py.

◆ _state

dict _state = self.initial_state
protected

Actual attribute holding the Current state.

Definition at line 189 of file state_machines.py.

◆ initial_state

initial_state = initial_state

Pointless docstring since it's a property.

Definition at line 182 of file state_machines.py.

◆ state

state = dest

Current State of machine.

Definition at line 336 of file state_machines.py.

◆ states

dict states = {}

Valid states for this machine.

Definition at line 176 of file state_machines.py.

◆ transitions

transitions = defaultdict(list)

Allowed transitions between states.

Definition at line 191 of file state_machines.py.


The documentation for this class was generated from the following file: