Transitions module¶
¶
DelayedTransition
¶
Abstract DelayedTransition allows configuring skipable pre and post transition execution delays.
delay_after: ApproximateFloat
property
readonly
¶
The amount of time in seconds to delay after execution
delay_before: ApproximateFloat
property
readonly
¶
The amount of time in seconds to delay before execution
name: str
inherited
property
readonly
¶
The name of the transition (including the prefix)
name_only: str
inherited
property
readonly
¶
The name of the transition
name_prefix: Optional[str]
inherited
property
readonly
¶
The name prefix of the transition instance.
target: Optional[str]
inherited
property
readonly
¶
The target state of the transition
transition_function: TransitionFunction
inherited
property
readonly
¶
The transition function
__init__(self, transition_function, name=None, target=None, delay_before=0.0, delay_after=0.0, name_prefix=None)
special
¶
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name |
Optional[str] |
The transition name |
None |
transition_function |
TransitionFunction |
The transition function to call upon execution |
required |
target |
Optional[str] |
The target state |
None |
delay_before |
Union[cr_kyoushi.simulation.model.ApproximateFloat, float] |
The pre execution delay to configure |
0.0 |
delay_after |
Union[cr_kyoushi.simulation.model.ApproximateFloat, float] |
The post execution delay to configure |
0.0 |
name_prefix |
Optional[str] |
A prefix for the transition name |
None |
Source code in simulation/transitions.py
def __init__(
self,
transition_function: TransitionFunction,
name: Optional[str] = None,
target: Optional[str] = None,
delay_before: Union[ApproximateFloat, float] = 0.0,
delay_after: Union[ApproximateFloat, float] = 0.0,
name_prefix: Optional[str] = None,
):
"""
Args:
name: The transition name
transition_function: The transition function to call upon execution
target: The target state
delay_before: The pre execution delay to configure
delay_after: The post execution delay to configure
name_prefix: A prefix for the transition name
"""
super().__init__(transition_function, name, target, name_prefix)
if isinstance(delay_before, float):
delay_before = ApproximateFloat.convert(delay_before)
if isinstance(delay_after, float):
delay_after = ApproximateFloat.convert(delay_after)
self._delay_before = delay_before
self._delay_after = delay_after
execute(self, log, current_state, context)
¶
Delayed transition execution sleeps before and after executing the transition.
Both delays can be configured during initialization.
The delays use a special sleep
function that registers a SIGINT
signal handler to make it possible to interrupt and skip the sleep
phase.
Note
When running in a terminal press Ctrl+C to fast forward.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
log |
BoundLogger |
The bound logger initialized with transition specific information |
required |
current_state |
str |
The calling states name |
required |
context |
Union[pydantic.main.BaseModel, Dict[str, Any]] |
The state machine context |
required |
Returns:
Type | Description |
---|---|
Optional[str] |
The state the machine has moved to |
Exceptions:
Type | Description |
---|---|
TransitionExecutionError |
If a transition error occurs for which we can fallback into a valid state |
Source code in simulation/transitions.py
def execute(
self,
log: BoundLogger,
current_state: str,
context: Context,
) -> Optional[str]:
"""
Delayed transition execution sleeps before and after executing the transition.
Both delays can be configured during initialization.
The delays use a special [`sleep`][cr_kyoushi.simulation.util.sleep] function that registers a `SIGINT`
signal handler to make it possible to interrupt and skip the sleep
phase.
!!! Note
When running in a terminal press ++ctrl+c++ to
*fast forward*.
Args:
log: The bound logger initialized with transition specific information
current_state: The calling states name
context: The state machine context
Returns:
The state the machine has moved to
Raises:
TransitionExecutionError: If a transition error occurs for which we can fallback into a valid state
"""
sleep(self.delay_before)
next_state = super().execute(log, current_state, context)
sleep(self.delay_after)
return next_state
NoopTransition
¶
No noperation transition that only changes the current state.
name: str
inherited
property
readonly
¶
The name of the transition (including the prefix)
name_only: str
inherited
property
readonly
¶
The name of the transition
name_prefix: Optional[str]
inherited
property
readonly
¶
The name prefix of the transition instance.
target: Optional[str]
inherited
property
readonly
¶
The target state of the transition
transition_function: TransitionFunction
inherited
property
readonly
¶
The transition function
__init__(self, name='noop', target=None, name_prefix=None)
special
¶
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name |
str |
The name of the transition |
'noop' |
target |
Optional[str] |
The name of the target state |
None |
name_prefix |
Optional[str] |
A prefix for the transition name |
None |
Source code in simulation/transitions.py
def __init__(
self,
name: str = "noop",
target: Optional[str] = None,
name_prefix: Optional[str] = None,
):
"""
Args:
name: The name of the transition
target: The name of the target state
name_prefix: A prefix for the transition name
"""
super().__init__(noop, name=name, target=target, name_prefix=name_prefix)
execute(self, log, current_state, context)
inherited
¶
Transition execution function called by the state machine.
The default behavior is to directly call the configured
transition_function
Info
This function can be overridden or extended to change how
transitions are executed. Actual transition implementations
should be done via the supplied
transition_function
.
Warning
Don't forget to call super().execute(current_state, context)
or self.transition_function(current_state, context)
.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
log |
BoundLogger |
The bound logger initialized with transition specific information |
required |
current_state |
str |
The calling states name |
required |
context |
Union[pydantic.main.BaseModel, Dict[str, Any]] |
The state machine context |
required |
Returns:
Type | Description |
---|---|
Optional[str] |
The state the machine has moved to |
Exceptions:
Type | Description |
---|---|
TransitionExecutionError |
If a transition error occurs for which we can fallback into a valid state |
Source code in simulation/transitions.py
def execute(
self,
log: BoundLogger,
current_state: str,
context: Context,
) -> Optional[str]:
"""Transition execution function called by the state machine.
The default behavior is to directly call the configured
[`transition_function`][cr_kyoushi.simulation.transitions.Transition.transition_function]
!!! Info
This function can be overridden or extended to change **how**
transitions are executed. Actual transition implementations
should be done via the supplied
[`transition_function`][cr_kyoushi.simulation.transitions.Transition.transition_function].
!!! Warning
Don't forget to call `#!python super().execute(current_state, context)`
or `#!python self.transition_function(current_state, context)`.
Args:
log: The bound logger initialized with transition specific information
current_state: The calling states name
context: The state machine context
Returns:
The state the machine has moved to
Raises:
TransitionExecutionError: If a transition error occurs for which we can fallback into a valid state
"""
self.transition_function(log, current_state, context, self.target)
return self.target
Transition
¶
Abstract base Transition class describes a transition from one state into another
name: str
property
readonly
¶
The name of the transition (including the prefix)
name_only: str
property
readonly
¶
The name of the transition
name_prefix: Optional[str]
property
readonly
¶
The name prefix of the transition instance.
target: Optional[str]
property
readonly
¶
The target state of the transition
transition_function: TransitionFunction
property
readonly
¶
The transition function
__init__(self, transition_function, name=None, target=None, name_prefix=None)
special
¶
Parameters:
Name | Type | Description | Default |
---|---|---|---|
transition_function |
TransitionFunction |
The transition function to call upon execution |
required |
name |
Optional[str] |
The transition name |
None |
target |
Optional[str] |
The target state |
None |
name_prefix |
Optional[str] |
A prefix for the transition name |
None |
Source code in simulation/transitions.py
def __init__(
self,
transition_function: TransitionFunction,
name: Optional[str] = None,
target: Optional[str] = None,
name_prefix: Optional[str] = None,
):
"""
Args:
transition_function: The transition function to call upon execution
name: The transition name
target: The target state
name_prefix: A prefix for the transition name
"""
if name is None:
if isinstance(transition_function, FunctionType):
name = transition_function.__name__.lower()
else:
name = transition_function.__class__.__name__.lower()
self._transition_function: TransitionFunction = transition_function
self._name: str = name
self._name_prefix: Optional[str] = name_prefix
self._target: Optional[str] = target
execute(self, log, current_state, context)
¶
Transition execution function called by the state machine.
The default behavior is to directly call the configured
transition_function
Info
This function can be overridden or extended to change how
transitions are executed. Actual transition implementations
should be done via the supplied
transition_function
.
Warning
Don't forget to call super().execute(current_state, context)
or self.transition_function(current_state, context)
.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
log |
BoundLogger |
The bound logger initialized with transition specific information |
required |
current_state |
str |
The calling states name |
required |
context |
Union[pydantic.main.BaseModel, Dict[str, Any]] |
The state machine context |
required |
Returns:
Type | Description |
---|---|
Optional[str] |
The state the machine has moved to |
Exceptions:
Type | Description |
---|---|
TransitionExecutionError |
If a transition error occurs for which we can fallback into a valid state |
Source code in simulation/transitions.py
def execute(
self,
log: BoundLogger,
current_state: str,
context: Context,
) -> Optional[str]:
"""Transition execution function called by the state machine.
The default behavior is to directly call the configured
[`transition_function`][cr_kyoushi.simulation.transitions.Transition.transition_function]
!!! Info
This function can be overridden or extended to change **how**
transitions are executed. Actual transition implementations
should be done via the supplied
[`transition_function`][cr_kyoushi.simulation.transitions.Transition.transition_function].
!!! Warning
Don't forget to call `#!python super().execute(current_state, context)`
or `#!python self.transition_function(current_state, context)`.
Args:
log: The bound logger initialized with transition specific information
current_state: The calling states name
context: The state machine context
Returns:
The state the machine has moved to
Raises:
TransitionExecutionError: If a transition error occurs for which we can fallback into a valid state
"""
self.transition_function(log, current_state, context, self.target)
return self.target
TransitionFunction
¶
__call__(self, log, current_state, context, target)
special
¶
Parameters:
Name | Type | Description | Default |
---|---|---|---|
log |
BoundLogger |
The bound logger initialized with transition specific information |
required |
current_state |
str |
The calling states name |
required |
context |
Union[pydantic.main.BaseModel, Dict[str, Any]] |
The state machine context |
required |
target |
Optional[str] |
The target state |
required |
Exceptions:
Type | Description |
---|---|
TransitionExecutionError |
If a transition error occurs for which we can fallback into a valid state |
Source code in simulation/transitions.py
def __call__(
self,
log: BoundLogger,
current_state: str,
context: Context,
target: Optional[str],
):
"""
Args:
log: The bound logger initialized with transition specific information
current_state: The calling states name
context: The state machine context
target: The target state
Raises:
TransitionExecutionError: If a transition error occurs for which we can fallback into a valid state
"""
delayed_transition(name=None, target=None, delay_before=0.0, delay_after=0.0, name_prefix=None)
¶
Transition decorator that can be used to turn a
TransitionFunction
into a
DelayedTransition
.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name |
Optional[str] |
The name of the transition |
None |
target |
Optional[str] |
The target states name |
None |
delay_before |
Union[cr_kyoushi.simulation.model.ApproximateFloat, float] |
The pre execution delay to configure |
0.0 |
delay_after |
Union[cr_kyoushi.simulation.model.ApproximateFloat, float] |
The post execution delay to configure |
0.0 |
name_prefix |
Optional[str] |
A prefix for the transition name |
None |
Examples:
@delayed_transition(name="example", target="next", delay_before=1.5)
def example(current_state, context, target):
...
# the above is equivalent to
def example_func(current_state, context, target):
...
example = DelayedTransition(example_func, name="example", target="next", delay_before=1.5)
Returns:
Type | Description |
---|---|
Callable[[cr_kyoushi.simulation.transitions.TransitionFunction], cr_kyoushi.simulation.transitions.DelayedTransition] |
A decorator that turns a |
Source code in simulation/transitions.py
def delayed_transition(
name: Optional[str] = None,
target: Optional[str] = None,
delay_before: Union[ApproximateFloat, float] = 0.0,
delay_after: Union[ApproximateFloat, float] = 0.0,
name_prefix: Optional[str] = None,
) -> Callable[[TransitionFunction], DelayedTransition]:
"""Transition decorator that can be used to turn a
[`TransitionFunction`][cr_kyoushi.simulation.transitions.TransitionFunction] into a
[`DelayedTransition`][cr_kyoushi.simulation.transitions.DelayedTransition].
Args:
name: The name of the transition
target: The target states name
delay_before: The pre execution delay to configure
delay_after: The post execution delay to configure
name_prefix: A prefix for the transition name
Example:
```python
@delayed_transition(name="example", target="next", delay_before=1.5)
def example(current_state, context, target):
...
# the above is equivalent to
def example_func(current_state, context, target):
...
example = DelayedTransition(example_func, name="example", target="next", delay_before=1.5)
```
Returns:
A decorator that turns a [`TransitionFunction`][cr_kyoushi.simulation.transitions.TransitionFunction]
into a DelayedTransition initialized with the given args.
"""
def decorator(func: TransitionFunction) -> DelayedTransition:
return DelayedTransition(
transition_function=func,
target=target,
name=name,
delay_before=delay_before,
delay_after=delay_after,
name_prefix=name_prefix,
)
return decorator
noop(log, current_state, context, target)
¶
No operation transition function
Source code in simulation/transitions.py
def noop(log: BoundLogger, current_state: str, context: Context, target: Optional[str]):
"""No operation transition function"""
transition(name=None, target=None, name_prefix=None)
¶
Transition decorator that can be used to turn a
TransitionFunction
into a
Transition
.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
name |
Optional[str] |
The name of the transition |
None |
target |
Optional[str] |
The target states name |
None |
name_prefix |
Optional[str] |
A prefix for the transition name |
None |
Examples:
@transition(name="example", target="next")
def example(current_state, context, target):
...
# the above is equivalent to
def example_func(current_state, context, target):
...
example = Transition(example_func, name="example", target="next")
Returns:
Type | Description |
---|---|
Callable[[cr_kyoushi.simulation.transitions.TransitionFunction], cr_kyoushi.simulation.transitions.Transition] |
A decorator that turns a |
Source code in simulation/transitions.py
def transition(
name: Optional[str] = None,
target: Optional[str] = None,
name_prefix: Optional[str] = None,
) -> Callable[[TransitionFunction], Transition]:
"""Transition decorator that can be used to turn a
[`TransitionFunction`][cr_kyoushi.simulation.transitions.TransitionFunction] into a
[`Transition`][cr_kyoushi.simulation.transitions.Transition].
Args:
name: The name of the transition
target: The target states name
name_prefix: A prefix for the transition name
Example:
```python
@transition(name="example", target="next")
def example(current_state, context, target):
...
# the above is equivalent to
def example_func(current_state, context, target):
...
example = Transition(example_func, name="example", target="next")
```
Returns:
A decorator that turns a [`TransitionFunction`][cr_kyoushi.simulation.transitions.TransitionFunction]
into a Transition initialized with the given args.
"""
def decorator(func: TransitionFunction) -> Transition:
return Transition(
transition_function=func, target=target, name=name, name_prefix=name_prefix
)
return decorator