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 140 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 167 of file state_machines.py.

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

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 302 of file state_machines.py.

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

◆ _callback()

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

Definition at line 340 of file state_machines.py.

340 def _callback(func, **kwargs):
341 """
342 Calls a condition/before/after.. function using arguments passed (or not).
343 """
344 return func(**kwargs)
345

◆ _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 316 of file state_machines.py.

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

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

◆ 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 260 of file state_machines.py.

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

◆ default_condition()

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

Definition at line 254 of file state_machines.py.

254 def default_condition(**kwargs):
255 """
256 Method to always return True.
257 """
258 return True
259

◆ get_transition_dict()

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

Definition at line 357 of file state_machines.py.

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

◆ get_transitions()

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

Definition at line 346 of file state_machines.py.

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

◆ 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 205 of file state_machines.py.

205 def initial_state(self):
206 """
207 The initial state of the machine. Needs a special property to prevent trying to run on_enter callbacks when set.
208 """
209 return self._initial_state
210

◆ initial_state() [2/2]

initial_state ( self,
state )
 

Definition at line 212 of file state_machines.py.

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

◆ save_graph()

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

Definition at line 368 of file state_machines.py.

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

◆ 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 223 of file state_machines.py.

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

◆ state() [2/2]

state ( self,
state )
 

Definition at line 232 of file state_machines.py.

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

Member Data Documentation

◆ _initial_state

dict _initial_state = State(initial_state)
protected

Actual attribute holding initial state for this machine.

Definition at line 182 of file state_machines.py.

◆ _state

dict _state = self.initial_state
protected

Actual attribute holding the Current state.

Definition at line 185 of file state_machines.py.

◆ initial_state

initial_state = initial_state

Pointless docstring since it's a property.

Definition at line 178 of file state_machines.py.

◆ state

state = dest

Current State of machine.

Definition at line 332 of file state_machines.py.

◆ states

dict states = {}

Valid states for this machine.

Definition at line 172 of file state_machines.py.

◆ transitions

transitions = defaultdict(list)

Allowed transitions between states.

Definition at line 187 of file state_machines.py.


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