from enum import Enum
from typing import TYPE_CHECKING
from typing import Any
from typing import Dict
from typing import Generator
from typing import List
from typing import cast
from weakref import ref
from .callbacks import CallbackGroup
from .callbacks import CallbackPriority
from .callbacks import CallbackSpecList
from .event import _expand_event_id
from .exceptions import InvalidDefinition
from .exceptions import StateMachineError
from .i18n import _
from .invoke import normalize_invoke_callbacks
from .transition import Transition
from .transition_list import TransitionList
if TYPE_CHECKING:
from .statemachine import StateChart
class _TransitionBuilder:
def __init__(self, state: "State"):
self._state = state
def itself(self, **kwargs):
return self.__call__(self._state, **kwargs)
def __call__(self, *states: "State", **kwargs):
raise NotImplementedError
class _ToState(_TransitionBuilder):
def __call__(self, *states: "State | NestedStateFactory | None", **kwargs):
transitions = TransitionList(
Transition(self._state, cast("State | None", state), **kwargs) for state in states
)
self._state.transitions.add_transitions(transitions)
return transitions
class _FromState(_TransitionBuilder):
def any(self, **kwargs):
"""Create transitions from all non-final states (reversed)."""
return self.__call__(AnyState(), **kwargs)
def __call__(self, *states: "State | NestedStateFactory", **kwargs):
transitions = TransitionList()
for origin in states:
state = cast(State, origin)
transition = Transition(state, self._state, **kwargs)
state.transitions.add_transitions(transition)
transitions.add_transitions(transition)
return transitions
class NestedStateFactory(type):
def __new__( # type: ignore [misc]
cls, classname, bases, attrs, name="", **kwargs
) -> "State":
if not bases:
new_cls = super().__new__(cls, classname, bases, attrs) # type: ignore [return-value]
new_cls._factory_kwargs = kwargs # type: ignore [attr-defined]
return new_cls # type: ignore [return-value]
# Inherit factory kwargs from base classes (e.g., parallel=True from State.Parallel)
inherited_kwargs: dict = {}
for base in bases:
inherited_kwargs.update(getattr(base, "_factory_kwargs", {}))
inherited_kwargs.update(kwargs)
states = []
history = []
callbacks = {}
for key, value in attrs.items():
if isinstance(value, HistoryState):
value._set_id(key)
history.append(value)
elif isinstance(value, State):
value._set_id(key)
states.append(value)
elif isinstance(value, TransitionList):
value.add_event(_expand_event_id(key))
elif callable(value):
callbacks[key] = value
return State(
name=name, states=states, history=history, _callbacks=callbacks, **inherited_kwargs
)
@classmethod
def to(cls, *args: "State | NestedStateFactory", **kwargs) -> "_ToState": # pragma: no cover
"""Create transitions to the given target states.
.. note: This method is only a type hint for mypy.
The actual implementation belongs to the :ref:`State` class.
"""
return _ToState(State())
@classmethod
def from_( # pragma: no cover
cls, *args: "State | NestedStateFactory", **kwargs
) -> "_FromState":
"""Create transitions from the given target states (reversed).
.. note: This method is only a type hint for mypy.
The actual implementation belongs to the :ref:`State` class.
"""
return _FromState(State())
[docs]
class State:
"""
A State in a :ref:`StateMachine` describes a particular behavior of the machine.
When we say that a machine is “in” a state, it means that the machine behaves
in the way that state describes.
Args:
name: A human-readable representation of the state. Default is derived
from the name of the variable assigned to the state machine class.
The name is derived from the id using this logic::
name = id.replace("_", " ").capitalize()
value: A specific value to the storage and retrieval of states.
If specified, you can use It to map a more friendly representation to a low-level
value.
initial: Set ``True`` if the ``State`` is the initial one. There must be one and only
one initial state in a statemachine. Defaults to ``False``.
If not specified, the default initial state is the first child state in document order.
final: Set ``True`` if represents a final state. A machine can have
optionally many final states. Final states have no :ref:`transition` starting from It.
Defaults to ``False``.
enter: One or more callbacks assigned to be executed when the state is entered.
See :ref:`actions`.
exit: One or more callbacks assigned to be executed when the state is exited.
See :ref:`actions`.
State is a core component on how this library implements an expressive API to declare
StateMachines.
>>> from statemachine import State
Given a few states...
>>> draft = State(name="Draft", initial=True)
>>> producing = State("Producing")
>>> closed = State('Closed', final=True)
Transitions are declared using the :func:`State.to` or :func:`State.from_` (reversed) methods.
>>> draft.to(producing)
TransitionList([Transition('Draft', 'Producing', event=[], internal=False, initial=False)])
The result is a :ref:`TransitionList`.
Don't worry about this internal class.
But the good thing is that it implements the ``OR`` operator to combine transitions,
so you can use the ``|`` syntax to compound a list of transitions and assign
to the same event.
You can declare all transitions for a state in one single line ...
>>> transitions = draft.to(draft) | producing.to(closed)
... and you can append additional transitions for a state to previous definitions.
>>> transitions |= closed.to(draft)
>>> [(t.source.name, t.target.name) for t in transitions]
[('Draft', 'Draft'), ('Producing', 'Closed'), ('Closed', 'Draft')]
There are handy shortcuts that you can use to express this same set of transitions.
The first one, ``draft.to(draft)``, is also called a :ref:`self-transition`, and can be
expressed using an alternative syntax:
>>> draft.to.itself()
TransitionList([Transition('Draft', 'Draft', event=[], internal=False, initial=False)])
You can even pass a list of target states to declare at once all transitions starting
from the same state.
>>> transitions = draft.to(draft, producing, closed)
>>> [(t.source.name, t.target.name) for t in transitions]
[('Draft', 'Draft'), ('Draft', 'Producing'), ('Draft', 'Closed')]
Sometimes it's easier to use the :func:`State.from_` method:
>>> transitions = closed.from_(draft, producing, closed)
>>> [(t.source.name, t.target.name) for t in transitions]
[('Draft', 'Closed'), ('Producing', 'Closed'), ('Closed', 'Closed')]
"""
[docs]
class Compound(metaclass=NestedStateFactory):
"Uses the class namespace to build a :ref:`State` instance of a compound state"
[docs]
class Parallel(metaclass=NestedStateFactory, parallel=True):
"Uses the class namespace to build a :ref:`State` instance of a parallel state"
def __init__(
self,
name: str = "",
value: Any = None,
initial: bool = False,
final: bool = False,
parallel: bool = False,
states: "List[State] | None" = None,
history: "List[HistoryState] | None" = None,
enter: Any = None,
exit: Any = None,
invoke: Any = None,
donedata: Any = None,
_callbacks: Any = None,
):
self.name = name
self.value = value
self._parallel = parallel
self.states = states or []
self.history = history or []
self.is_atomic = bool(not self.states)
self._initial = initial
self._final = final
self.is_active = False
self._id: str = ""
self._callbacks = _callbacks
self.parent: "State | None" = None
self.transitions = TransitionList()
self._specs = CallbackSpecList()
self.enter = self._specs.grouper(CallbackGroup.ENTER).add(
enter, priority=CallbackPriority.INLINE
)
self.exit = self._specs.grouper(CallbackGroup.EXIT).add(
exit, priority=CallbackPriority.INLINE
)
self.invoke = self._specs.grouper(CallbackGroup.INVOKE).add(
normalize_invoke_callbacks(invoke), priority=CallbackPriority.INLINE
)
if donedata is not None:
if not final:
raise InvalidDefinition(_("'donedata' can only be specified on final states."))
self.enter.add(donedata, priority=CallbackPriority.INLINE)
self.document_order = 0
self._init_states()
def _init_states(self):
for state in self.states:
state.parent = self
state._initial = state.initial or self.parallel
setattr(self, state.id, state)
for history in self.history:
history.parent = self
setattr(self, history.id, history)
def __eq__(self, other):
return (
isinstance(other, State)
and self.name == other.name
and self.id == other.id
or (self.value == other)
)
def __hash__(self):
return hash(repr(self))
def _setup(self):
self.enter.add("on_enter_state", priority=CallbackPriority.GENERIC, is_convention=True)
self.enter.add(f"on_enter_{self.id}", priority=CallbackPriority.NAMING, is_convention=True)
self.exit.add("on_exit_state", priority=CallbackPriority.GENERIC, is_convention=True)
self.exit.add(f"on_exit_{self.id}", priority=CallbackPriority.NAMING, is_convention=True)
self.invoke.add("on_invoke_state", priority=CallbackPriority.GENERIC, is_convention=True)
self.invoke.add(
f"on_invoke_{self.id}", priority=CallbackPriority.NAMING, is_convention=True
)
def _on_event_defined(self, event: str, transition: Transition, states: List["State"]):
"""Called by statemachine factory when an event is defined having a transition
starting from this state.
"""
pass
def __repr__(self):
return (
f"{type(self).__name__}({self.name!r}, id={self.id!r}, value={self.value!r}, "
f"initial={self.initial!r}, final={self.final!r}, parallel={self.parallel!r})"
)
def __str__(self):
return self.name
def __get__(self, machine, owner):
if machine is None:
return self
return self.for_instance(machine=machine, cache=machine._states_for_instance)
def __set__(self, instance, value):
raise StateMachineError(
_("State overriding is not allowed. Trying to add '{}' to {}").format(value, self.id)
)
def for_instance(self, machine: "StateChart", cache: Dict["State", "State"]) -> "State":
if self not in cache:
cache[self] = InstanceState(self, machine)
return cache[self]
@property
def id(self) -> str:
return self._id
def _set_id(self, id: str) -> "State":
self._id = id
if self.value is None:
self.value = id
if not self.name:
self.name = self._id.replace("_", " ").capitalize()
return self
@property
def to(self) -> _ToState:
"""Create transitions to the given target states."""
return _ToState(self)
@property
def from_(self) -> _FromState:
"""Create transitions from the given target states (reversed)."""
return _FromState(self)
@property
def initial(self):
return self._initial
@property
def final(self):
return self._final
@property
def parallel(self):
return self._parallel
@property
def is_compound(self):
return bool(self.states) and not self.parallel
@property
def is_history(self):
return isinstance(self, HistoryState)
def ancestors(self, parent: "State | None" = None) -> Generator["State", None, None]: # noqa: UP043
selected = self.parent
while selected:
if parent and selected == parent:
break
yield selected
selected = selected.parent
def is_descendant(self, state: "State") -> bool:
return state in self.ancestors()
class InstanceState(State):
""" """
def __init__(
self,
state: State,
machine: "StateChart",
):
self._state = ref(state)
self._machine = ref(machine)
self._init_states()
def _ref(self) -> State:
"""Dereference the weakref, raising if the referent has been collected."""
state = self._state()
assert state is not None
return state
@property
def name(self):
return self._ref().name
@property
def value(self):
return self._ref().value
@property
def transitions(self):
return self._ref().transitions
@property
def enter(self):
return self._ref().enter
@property
def exit(self):
return self._ref().exit
@property
def invoke(self):
return self._ref().invoke
def __eq__(self, other):
return self._ref() == other
def __hash__(self):
return hash(repr(self._ref()))
def __repr__(self):
return repr(self._ref())
@property
def initial(self):
return self._ref()._initial
@property
def final(self):
return self._ref()._final
@property
def id(self) -> str:
return (self._state() or self)._id # type: ignore[union-attr]
@property
def is_active(self):
machine = self._machine()
assert machine is not None
return self.value in machine.configuration_values
@property
def is_atomic(self):
return self._ref().is_atomic
@property
def parent(self):
return self._ref().parent
@property
def states(self):
return self._ref().states
@property
def history(self):
return self._ref().history
@property
def parallel(self):
return self._ref().parallel
@property
def is_compound(self):
return self._ref().is_compound
@property
def document_order(self):
return self._ref().document_order
class AnyState(State):
"""A special state that works as a "ANY" placeholder.
It is used as the "From" state of a transtion,
until the state machine class is evaluated.
"""
def _on_event_defined(self, event: str, transition: Transition, states: List[State]):
for state in states:
if state.final:
continue
new_transition = transition._copy_with_args(source=state, event=event)
state.transitions.add_transitions(new_transition)
class HistoryType(str, Enum):
"""Type of history recorded by a :class:`HistoryState`."""
SHALLOW = "shallow"
"""Remembers only the direct children of the compound state.
If the remembered child is itself a compound, it re-enters from its initial state."""
DEEP = "deep"
"""Remembers the exact leaf (atomic) state across the entire nested hierarchy.
Re-entering restores the full ancestor chain down to that leaf."""
@property
def is_deep(self) -> bool:
return self == HistoryType.DEEP
[docs]
class HistoryState(State):
def __init__(
self, name: str = "", value: Any = None, type: "str | HistoryType" = HistoryType.SHALLOW
):
super().__init__(name=name, value=value)
self.type = HistoryType(type)
self.is_active = False