unaiverse.hsm.hsm

What this module does 🔴

Implements the HybridStateMachine, the core behaviour engine that manages states, transitions, teleports, policies, wildcards, and serialization for agent behaviour.

hsm ¶

█████ █████ ██████ █████ █████ █████ █████ ██████████ ███████████ █████████ ██████████ ░░███ ░░███ ░░██████ ░░███ ░░███ ░░███ ░░███ ░░███░░░░░█░░███░░░░░███ ███░░░░░███░░███░░░░░█ ░███ ░███ ░███░███ ░███ ██████ ░███ ░███ ░███ ░███ █ ░ ░███ ░███ ░███ ░░░ ░███ █ ░ ░███ ░███ ░███░░███░███ ░░░░░███ ░███ ░███ ░███ ░██████ ░██████████ ░░█████████ ░██████
░███ ░███ ░███ ░░██████ ███████ ░███ ░░███ ███ ░███░░█ ░███░░░░░███ ░░░░░░░░███ ░███░░█
░███ ░███ ░███ ░░█████ ███░░███ ░███ ░░░█████░ ░███ ░ █ ░███ ░███ ███ ░███ ░███ ░ █ ░░████████ █████ ░░█████░░████████ █████ ░░███ ██████████ █████ █████░░█████████ ██████████ ░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░ ░░░░░ ░░░ ░░░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░░ ░░░░░░░░░░ A Collectionless AI Project (https://collectionless.ai) Registration/Login: https://unaiverse.io Code Repositories: https://github.com/collectionlessai/ Main Developers: Stefano Melacci (Project Leader), Christian Di Maio, Tommaso Guidi

HybridStateMachine ¶

HybridStateMachine(actionable: object, wildcards: dict[str, str | float | int] | None = None, policy: Callable[[list[Action]], tuple[int, Interaction | None]] | None = None)

A hybrid state machine that orchestrates agent behaviour through states and transitions.

HybridStateMachine (HSM) combines classical finite-state-machine semantics with support for multi-step, async actions and external interaction requests. Each state can carry an inner action (a method on the actionable object that runs while the machine is in that state) and is connected to other states via transitions, each guarded by an Action. At every execution tick, the machine selects a feasible transition action according to a configurable policy, executes it, and advances to the next state when the action succeeds.

State machines can be defined programmatically (using add_state and add_transit) or loaded from a JSON file (using load). They can also be saved (save), composed (include), and visualised (to_graphviz, save_pdf).

Key concepts:

Wildcards: placeholder strings (e.g. "<role>") embedded in action arguments and state messages that are replaced at runtime with concrete values (see set_wildcards, update_wildcard, apply_wildcards).
Policy: a callable that, given the list of currently feasible actions, returns the index of the action to execute and an optional Interaction (see set_policy). The default policy is first-requested or first-ready.
Policy filter: an optional second callable that can veto or override the policy's choice (see set_policy_filter).
Teleports: transitions that are hidden in the visual graph but still run normally (added with add_teleport or add_global_teleport).

Attributes:

Name	Type	Description
`initial_state`	`str \| None`	Name of the first state the machine enters.
`prev_state`	`str \| None`	Name of the state the machine was in before the most recent transition.
`limbo_state`	`str \| None`	Name of the state the machine was in when a multi-step action started. While a multi-step action is running, `state` is `None` and `limbo_state` holds the originating state.
`state`	`str \| None`	Name of the current state, or `None` while a transition action is in progress.
`role`	`str \| None`	Role string assigned to the agent running this machine (e.g. `"teacher"`).
`enabled`	`bool`	Whether the machine will execute actions when `act` / `act_transitions` / `act_states` are called.
`states`	`dict[str, State]`	Mapping from state name to `State` object.
`transitions`	`dict[str, dict[str, list[Action]]]`	Nested mapping `{from_state: {to_state: [Action, ...]}}` describing all transitions.
`actionable`	`object`	The object whose methods serve as the actions of this machine.
`wildcards`	`dict[str, str \| float \| int] \| None`	Mapping from wildcard placeholder to its current replacement value.
`policy`		Callable that selects which action to execute.
`policy_filter`		Optional callable that overrides the policy's decision.
`welcome_msg`		Message printed once when the machine reaches its initial state, with wildcards already substituted. `None` after it has been printed.
`show_blocking_states`		Whether colored markers are appended to blocking-state log messages (see `show_marks_in_blocking_state_messages`).
`show_action_completion`		Whether tick symbols are appended to action log messages (see `show_ticks_in_action_messages`).
`show_action_request_info`		Whether requester/UUID info is appended to action log messages (see `show_request_info_in_action_messages`).

Examples:

Build a minimal two-state machine and wire it to an actionable object:

>>> class MyAgent:
...     async def do_work(self): ...
...     async def finish(self): ...
>>>
>>> agent = MyAgent()
>>> hsm = HybridStateMachine(actionable=agent)
>>> hsm.add_transit("idle", "working", action="do_work")
>>> hsm.add_transit("working", "done", action="finish")
>>> print(hsm.get_state_name())
idle

Load a machine from a JSON file and attach it to an actionable:

>>> hsm = HybridStateMachine(actionable=agent).load("my_behaviour.json")

Initialize a HybridStateMachine with an actionable object, optional wildcards, and an optional policy.

All internal state is reset: no states, no transitions, no current action. The Custom.DEFAULT_WILDCARDS are merged in via add_wildcards immediately after construction so that built-in placeholders (such as the role wildcard) are always available. Wildcards are then applied to any existing messages and the welcome message. set_actionable is called last, propagating actionable to all action and state objects (currently none, since the machine is empty at this point).

Parameters:

Name	Type	Description	Default
`actionable`	`object`	The object whose methods serve as actions. Every `Action` in the machine holds a reference to this object and calls the method whose name matches the action name. May be `None` when constructing a temporary or placeholder machine (e.g. in `load`).	required
`wildcards`	`dict[str, str \| float \| int] \| None`	Initial wildcard dictionary. Merged with `Custom.DEFAULT_WILDCARDS`; keys in `wildcards` take precedence only when they do not collide with defaults already set by `add_wildcards`. Defaults to `None` (equivalent to an empty dict).	`None`
`policy`	`Callable[[list[Action]], tuple[int, Interaction \| None]] \| None`	Callable with signature `(actions_list: list[Action]) -> (int, Interaction \| None)` that selects which action to execute. Must return the index of the chosen action and an optional `Interaction`, or `-1` and `None` to select nothing. Defaults to `None`, which installs the built-in first-requested-or-first-ready policy.	`None`

Examples:

>>> hsm = HybridStateMachine(actionable=my_agent)

Provide a custom policy that always picks the last action in the list:

>>> def last_action_policy(actions):
...     return len(actions) - 1, None
>>> hsm = HybridStateMachine(actionable=my_agent, policy=last_action_policy)

Source code in unaiverse/hsm/hsm.py

def __init__(self, actionable: object, wildcards: dict[str, str | float | int] | None = None,
             policy: Callable[[list[Action]], tuple[int, Interaction | None]] | None = None):
    """Initialize a ``HybridStateMachine`` with an actionable object, optional wildcards, and an optional policy.

    All internal state is reset: no states, no transitions, no current action. The
    ``Custom.DEFAULT_WILDCARDS`` are merged in via ``add_wildcards`` immediately after
    construction so that built-in placeholders (such as the role wildcard) are always
    available. Wildcards are then applied to any existing messages and the welcome
    message. ``set_actionable`` is called last, propagating ``actionable`` to all
    action and state objects (currently none, since the machine is empty at this point).

    Args:
        actionable: The object whose methods serve as actions. Every ``Action`` in the
            machine holds a reference to this object and calls the method whose name
            matches the action name. May be ``None`` when constructing a temporary or
            placeholder machine (e.g. in ``load``).
        wildcards: Initial wildcard dictionary. Merged with ``Custom.DEFAULT_WILDCARDS``;
            keys in ``wildcards`` take precedence only when they do not collide with
            defaults already set by ``add_wildcards``. Defaults to ``None`` (equivalent
            to an empty dict).
        policy: Callable with signature
            ``(actions_list: list[Action]) -> (int, Interaction | None)`` that selects
            which action to execute. Must return the index of the chosen action and an
            optional ``Interaction``, or ``-1`` and ``None`` to select nothing. Defaults
            to ``None``, which installs the built-in first-requested-or-first-ready
            policy.

    Examples:
        >>> hsm = HybridStateMachine(actionable=my_agent)

        Provide a custom policy that always picks the last action in the list:

        >>> def last_action_policy(actions):
        ...     return len(actions) - 1, None
        >>> hsm = HybridStateMachine(actionable=my_agent, policy=last_action_policy)
    """

    # States are identified by strings, and then handled as State object with possibly and integer ID and action
    self.initial_state: str | None = None  # Initial state of the machine
    self.prev_state: str | None = None  # Previous state
    self.limbo_state: str | None = None  # When an action takes more than a step to complete, we are in "limbo"
    self.state: str | None = None  # Current state
    self.role: str | None = None  # Role of the agent in the state machine (e.g., teacher, student, etc.)
    self.enabled: bool = True
    self.states: dict[str, State] = {}  # State name to State object

    # Actions (transitions) are handled as Action objects in-between state strings
    self.transitions: dict[str, dict[str, list[Action]]] = {}  # Pair-of-states to the actions between them
    self.actionable: object = None  # The object on whose methods are actions that the machine calls
    self.wildcards: dict[str, str | float | int] | None = wildcards \
        if wildcards is not None else {}  # From a wildcards string to a specific value (used in action arguments)
    self.policy = policy if policy is not None else self.__policy_first_requested_or_first_ready
    self.policy_filter = None
    self.policy_filter_opts = {}
    self.welcome_msg = None
    self.welcome_msg_with_wildcards = None
    self.show_blocking_states = False
    self.show_action_completion = False
    self.show_action_request_info = False

    # Running data
    self.__action: Action | None = None  # Action that is being executed (could take more than a step to complete)
    self.__last_completed_action: Action | None = None
    self.__cur_feasible_actions_status: dict | None = None  # Store info of the executed action (for multi-steps)
    self.__id_to_state: list[State] = []  # Map from state ID to State object
    self.__id_to_action: list[Action] = []  # Map from action ID to Action object
    self.__state_changed = False  # Internal flag
    self.__id_to_original_state_msg: list[tuple[str | None, str | None]] = []
    self.__id_to_original_action_msg: list[str | None] = []

    # Forcing default wildcards
    self.add_wildcards(Custom.DEFAULT_WILDCARDS)

    # Applying to the welcome message, state, actions, if needed
    self.apply_wildcards()

    # Forcing output function
    self.__debug_messages_active = False

    self.set_actionable(actionable)

initial_state `instance-attribute` ¶

initial_state: str | None = None

prev_state `instance-attribute` ¶

prev_state: str | None = None

limbo_state `instance-attribute` ¶

limbo_state: str | None = None

state `instance-attribute` ¶

state: str | None = None

role `instance-attribute` ¶

role: str | None = None

enabled `instance-attribute` ¶

enabled: bool = True

states `instance-attribute` ¶

states: dict[str, State] = {}

transitions `instance-attribute` ¶

transitions: dict[str, dict[str, list[Action]]] = {}

actionable `instance-attribute` ¶

actionable: object = None

wildcards `instance-attribute` ¶

wildcards: dict[str, str | float | int] | None = wildcards if wildcards is not None else {}

policy `instance-attribute` ¶

policy = policy if policy is not None else __policy_first_requested_or_first_ready

policy_filter `instance-attribute` ¶

policy_filter = None

policy_filter_opts `instance-attribute` ¶

policy_filter_opts = {}

welcome_msg `instance-attribute` ¶

welcome_msg = None

welcome_msg_with_wildcards `instance-attribute` ¶

welcome_msg_with_wildcards = None

show_blocking_states `instance-attribute` ¶

show_blocking_states = False

show_action_completion `instance-attribute` ¶

show_action_completion = False

show_action_request_info `instance-attribute` ¶

show_action_request_info = False

show_ticks_in_action_messages ¶

show_ticks_in_action_messages(do_it: bool = True) -> None

Enable or disable tick symbols appended to action log messages upon completion.

When enabled, a status character from Custom.ACTION_TICKS_PER_STATUS is printed to the user-facing log after each action completes, giving a quick visual indicator of success or failure. This setting corresponds to the show_action_ticks_after_messages option stored in the serialised JSON.

Parameters:

Name	Type	Description	Default
`do_it`	`bool`	`True` to enable tick display, `False` to suppress it. Defaults to `True`.	`True`

Source code in unaiverse/hsm/hsm.py

def show_ticks_in_action_messages(self, do_it: bool = True) -> None:
    """Enable or disable tick symbols appended to action log messages upon completion.

    When enabled, a status character from ``Custom.ACTION_TICKS_PER_STATUS`` is
    printed to the user-facing log after each action completes, giving a quick
    visual indicator of success or failure. This setting corresponds to the
    ``show_action_ticks_after_messages`` option stored in the serialised JSON.

    Args:
        do_it: ``True`` to enable tick display, ``False`` to suppress it.
            Defaults to ``True``.
    """
    self.show_action_completion = do_it

show_marks_in_blocking_state_messages ¶

show_marks_in_blocking_state_messages(do_it: bool = True) -> None

Enable or disable colored markers appended to blocking-state log messages.

When enabled, a colored marker is added to the log message of every blocking state to make it visually distinct from non-blocking state messages. This corresponds to the highlight_blocking_states_in_messages option in the serialised JSON.

Parameters:

Name	Type	Description	Default
`do_it`	`bool`	`True` to enable marker display, `False` to suppress it. Defaults to `True`.	`True`

Source code in unaiverse/hsm/hsm.py

def show_marks_in_blocking_state_messages(self, do_it: bool = True) -> None:
    """Enable or disable colored markers appended to blocking-state log messages.

    When enabled, a colored marker is added to the log message of every blocking
    state to make it visually distinct from non-blocking state messages. This
    corresponds to the ``highlight_blocking_states_in_messages`` option in the
    serialised JSON.

    Args:
        do_it: ``True`` to enable marker display, ``False`` to suppress it.
            Defaults to ``True``.
    """
    self.show_blocking_states = do_it

show_request_info_in_action_messages ¶

show_request_info_in_action_messages(do_it: bool = True) -> None

Enable or disable requester and UUID info appended to action log messages.

When enabled, the identity of the requester and the UUID of the interaction are included in action-related log output, which is useful for tracing which external entity triggered a given action. This corresponds to the show_action_request_after_messages option in the serialised JSON.

Parameters:

Name	Type	Description	Default
`do_it`	`bool`	`True` to enable request info display, `False` to suppress it. Defaults to `True`.	`True`

Source code in unaiverse/hsm/hsm.py

def show_request_info_in_action_messages(self, do_it: bool = True) -> None:
    """Enable or disable requester and UUID info appended to action log messages.

    When enabled, the identity of the requester and the UUID of the interaction
    are included in action-related log output, which is useful for tracing which
    external entity triggered a given action. This corresponds to the
    ``show_action_request_after_messages`` option in the serialised JSON.

    Args:
        do_it: ``True`` to enable request info display, ``False`` to suppress it.
            Defaults to ``True``.
    """
    self.show_action_request_info = do_it

set_welcome_message ¶

set_welcome_message(msg: str | None) -> None

Set a message that is printed exactly once when the machine first reaches its initial state.

The raw message is stored in welcome_msg after HTML-unescaping (via html.unescape), and a copy is kept in welcome_msg_with_wildcards as the source for future wildcard substitution. When act executes and detects that the current state equals initial_state, it logs welcome_msg and then sets it to None so it is never shown again. Calling this method with None clears any previously set welcome message.

Parameters:

Name	Type	Description	Default
`msg`	`str \| None`	The welcome message string, which may contain HTML entities and wildcard placeholders. Pass `None` to remove the welcome message.	required

Source code in unaiverse/hsm/hsm.py

def set_welcome_message(self, msg: str | None) -> None:
    """Set a message that is printed exactly once when the machine first reaches its initial state.

    The raw message is stored in ``welcome_msg`` after HTML-unescaping (via
    ``html.unescape``), and a copy is kept in ``welcome_msg_with_wildcards`` as the
    source for future wildcard substitution. When ``act`` executes and detects that the
    current state equals ``initial_state``, it logs ``welcome_msg`` and then sets it to
    ``None`` so it is never shown again. Calling this method with ``None`` clears any
    previously set welcome message.

    Args:
        msg: The welcome message string, which may contain HTML entities and wildcard
            placeholders. Pass ``None`` to remove the welcome message.
    """

    if msg is not None:
        self.welcome_msg = html.unescape(msg)
        self.welcome_msg_with_wildcards = self.welcome_msg
    else:
        self.welcome_msg = None
        self.welcome_msg_with_wildcards = None

to_dict ¶

to_dict() -> dict

Serialize the state machine's current configuration into a nested dictionary.

The resulting dictionary mirrors the JSON schema used by save and load. It contains four top-level keys:

"machine": role, initial state, welcome message, and display option flags.
"states": one entry per state, produced by State.to_dict().
"transitions": ordinary (non-teleport) transitions, grouped by source state.
"teleports": teleport transitions, grouped by source state. If a teleport action originates from every state in the machine (except the destination), it is collapsed under the special "all" key so that the JSON stays compact and round-trips cleanly through load.

The welcome message, if present, is ASCII-safe-encoded with XML character references so that non-ASCII characters survive the JSON round-trip.

Returns:

Type	Description
`dict`	A dictionary representation of the machine, suitable for serialisation with
`dict`	`json.dumps`. See `__str__` for the JSON string form.

Examples:

>>> d = hsm.to_dict()
>>> print(d["machine"]["initial_state"])
idle

Source code in unaiverse/hsm/hsm.py

def to_dict(self) -> dict:
    """Serialize the state machine's current configuration into a nested dictionary.

    The resulting dictionary mirrors the JSON schema used by ``save`` and ``load``.
    It contains four top-level keys:

    - ``"machine"``: role, initial state, welcome message, and display option flags.
    - ``"states"``: one entry per state, produced by ``State.to_dict()``.
    - ``"transitions"``: ordinary (non-teleport) transitions, grouped by source state.
    - ``"teleports"``: teleport transitions, grouped by source state. If a teleport
      action originates from every state in the machine (except the destination), it is
      collapsed under the special ``"all"`` key so that the JSON stays compact and
      round-trips cleanly through ``load``.

    The welcome message, if present, is ASCII-safe-encoded with XML character
    references so that non-ASCII characters survive the JSON round-trip.

    Returns:
        A dictionary representation of the machine, suitable for serialisation with
        ``json.dumps``. See ``__str__`` for the JSON string form.

    Examples:
        >>> d = hsm.to_dict()
        >>> print(d["machine"]["initial_state"])
        idle
    """

    # Inverting the organization of the transition matrix, and considering teleport actions only
    teleport_inv_transitions = {}
    action_that_are_teleports = []
    for src, dests in self.transitions.items():
        for dest, actions in dests.items():
            if dest == src:
                continue  # Skipping self loops for teleports (they should not be there at all, if so, we filter)
            for act in actions:
                if act.is_teleport():
                    action_that_are_teleports.append(act)
                    if dest not in teleport_inv_transitions:
                        teleport_inv_transitions[dest] = {}
                    if src not in teleport_inv_transitions[dest]:
                        teleport_inv_transitions[dest][src] = []
                    teleport_inv_transitions[dest][src].append(act)

    # Parsing destination states: if the same action connects this to all the possible states, then it is an
    # "all"-like teleport
    teleport_transitions = {Custom.ALL_STATES_NAME: {}}  # Keep "all" on top (it is handled like a new state)
    for dest, srcs in teleport_inv_transitions.items():

        # For each source state reaching the current destination...
        for src, teleports in srcs.items():

            # For each teleport linking source to the current destination...
            for tel in teleports:
                if tel.get_mark() == "considered":  # Skipping already considered ones
                    continue

                tel.set_mark("considered")  # Marking
                tel_to_list = tel.to_list()
                _found_teleports = [(src, tel)]

                # Let's see if a teleport equivalent to the considered one connects all sources
                for _src, _teleports in srcs.items():
                    if src == _src:
                        continue  # Already considered case

                    _found = False
                    for _tel in _teleports:
                        if _tel.get_mark() == "considered":  # Skipping already considered ones
                            continue

                        if tel_to_list == _tel.to_list():
                            _tel.set_mark("considered")  # Marking
                            _found_teleports.append((_src, _tel))
                            _found = True
                            break
                    if not _found:
                        break

                # Distinguishing
                if len(_found_teleports) == (len(self.states) - 1):
                    if Custom.ALL_STATES_NAME not in teleport_transitions:
                        teleport_transitions[Custom.ALL_STATES_NAME] = {}
                    if dest not in teleport_transitions[Custom.ALL_STATES_NAME]:
                        teleport_transitions[Custom.ALL_STATES_NAME][dest] = []
                    teleport_transitions[Custom.ALL_STATES_NAME][dest].append(tel)  # Append a single one
                else:
                    for _src, _tel in _found_teleports:
                        if _src not in teleport_transitions:
                            teleport_transitions[_src] = {}
                        if dest not in teleport_transitions[_src]:
                            teleport_transitions[_src][dest] = []
                        teleport_transitions[_src][dest].append(_tel)

    # Clearing mark
    for tel in action_that_are_teleports:
        tel.clear_mark()

    return {
        'machine': {
            'role': self.role,
            'initial_state': self.initial_state,
            'msg': self.welcome_msg_with_wildcards.encode("ascii", "xmlcharrefreplace").decode(
                    "ascii") if self.welcome_msg_with_wildcards is not None else None,
            'options': {
                'highlight_blocking_states_in_messages': self.show_blocking_states,
                'show_action_ticks_after_messages': self.show_action_completion,
                'show_action_request_after_messages': self.show_action_request_info
            }
        },
        'states': {
            state.name: state.to_dict() for state in self.__id_to_state
        },
        'transitions': {
            from_state: [{
                "on": act.to_dict(),
                "goto": to_state
            } for to_state, action_list in to_states.items()
                for act in action_list if not act.is_teleport()
            ] for from_state, to_states in self.transitions.items() if len(to_states) > 0
        },
        'teleports': {
            from_state: [{
                "on": act.to_dict(),
                "goto": to_state
            } for to_state, action_list in to_states.items()
                for act in action_list
            ] for from_state, to_states in teleport_transitions.items() if len(to_states) > 0
        }
    }

set_actionable ¶

set_actionable(obj: object) -> None

Set the object on which the state machine's actions are performed.

Updates self.actionable and propagates the new reference to every Action already registered in the machine, including inner state actions. This allows the same machine topology to be reattached to a different object without rebuilding the whole machine from scratch. If obj is None the call is a no-op.

Parameters:

Name	Type	Description	Default
`obj`	`object`	The object instance to set as the new `actionable`. Each action in the machine will call methods on this object by name.	required

Source code in unaiverse/hsm/hsm.py

def set_actionable(self, obj: object) -> None:
    """Set the object on which the state machine's actions are performed.

    Updates ``self.actionable`` and propagates the new reference to every ``Action``
    already registered in the machine, including inner state actions. This allows the
    same machine topology to be reattached to a different object without rebuilding
    the whole machine from scratch. If ``obj`` is ``None`` the call is a no-op.

    Args:
        obj: The object instance to set as the new ``actionable``. Each action in the
            machine will call methods on this object by name.
    """
    if obj is None:
        return

    self.actionable = obj

    for state_obj in self.states.values():
        if state_obj.action is not None:
            state_obj.action.actionable = obj

set_policy ¶

set_policy(policy_fcn: Callable[[int, Interaction | None, list[Action], dict], tuple[int, Interaction | None]] | None) -> None

Set the policy callable used to select which action to execute in the current state.

The policy is invoked by act_transitions with the list of feasible Action objects available from the current state. It must return the index of the chosen action (into that list) and an Interaction that drives the execution, or -1 and None to indicate that no action should be executed this tick. If policy_fcn is None, the machine falls back to the built-in first-requested-or-first-ready policy on the next call to act_transitions.

The policy can be overridden at runtime by a policy filter; see set_policy_filter.

Parameters:

Name	Type	Description	Default
`policy_fcn`	`Callable[[int, Interaction \| None, list[Action], dict], tuple[int, Interaction \| None]] \| None`	Callable with signature `(actions_list: list[Action]) -> (int, Interaction \| None)`. Returns `(-1, None)` to select no action.	required

Source code in unaiverse/hsm/hsm.py

def set_policy(self, policy_fcn: Callable[
    [int, Interaction | None, list[Action], dict], tuple[int, Interaction | None]] | None) -> None:
    """Set the policy callable used to select which action to execute in the current state.

    The policy is invoked by ``act_transitions`` with the list of feasible ``Action``
    objects available from the current state. It must return the index of the chosen
    action (into that list) and an ``Interaction`` that drives the execution, or
    ``-1`` and ``None`` to indicate that no action should be executed this tick. If
    ``policy_fcn`` is ``None``, the machine falls back to the built-in
    first-requested-or-first-ready policy on the next call to ``act_transitions``.

    The policy can be overridden at runtime by a policy filter; see
    ``set_policy_filter``.

    Args:
        policy_fcn: Callable with signature
            ``(actions_list: list[Action]) -> (int, Interaction | None)``. Returns
            ``(-1, None)`` to select no action.
    """
    self.policy = policy_fcn

set_policy_filter ¶

set_policy_filter(filter_fcn: Callable[[int, Interaction | None, list[Action], dict], tuple[int, Interaction | None]] | None, filter_fcn_opts: dict) -> None

Set a filter callable that can veto or redirect the primary policy's decision.

After the main policy selects an action index and an Interaction, the filter (if set) is called with those results, the full list of feasible actions, and a mutable options dict. It may return a different index and interaction, or (-1, None) to suppress execution for this tick. The filter is called inside a try/except in act_transitions; any exception is logged and the original policy decision is kept. Passing None for filter_fcn disables filtering.

The provided filter_fcn_opts dict is cleared immediately on assignment, then stored as self.policy_filter_opts so that the caller shares the same object and can update options in place between ticks.

Parameters:

Name	Type	Description	Default
`filter_fcn`	`Callable[[int, Interaction \| None, list[Action], dict], tuple[int, Interaction \| None]] \| None`	Callable with signature `(idx: int, interaction: Interaction \| None, actions_list: list[Action], opts: dict) -> (int, Interaction \| None)`. Returns `(-1, None)` to suppress the selected action. Pass `None` to remove the current filter.	required
`filter_fcn_opts`	`dict`	A mutable dictionary of caller-defined options forwarded to the filter on every invocation. It is cleared upon registration.	required

Source code in unaiverse/hsm/hsm.py

def set_policy_filter(self, filter_fcn: Callable[
    [int, Interaction | None, list[Action], dict], tuple[int, Interaction | None]] | None,
                      filter_fcn_opts: dict) -> None:
    """Set a filter callable that can veto or redirect the primary policy's decision.

    After the main policy selects an action index and an ``Interaction``, the filter
    (if set) is called with those results, the full list of feasible actions, and a
    mutable options dict. It may return a different index and interaction, or
    ``(-1, None)`` to suppress execution for this tick. The filter is called inside a
    try/except in ``act_transitions``; any exception is logged and the original policy
    decision is kept. Passing ``None`` for ``filter_fcn`` disables filtering.

    The provided ``filter_fcn_opts`` dict is cleared immediately on assignment, then
    stored as ``self.policy_filter_opts`` so that the caller shares the same object
    and can update options in place between ticks.

    Args:
        filter_fcn: Callable with signature
            ``(idx: int, interaction: Interaction | None, actions_list: list[Action],
            opts: dict) -> (int, Interaction | None)``. Returns ``(-1, None)`` to
            suppress the selected action. Pass ``None`` to remove the current filter.
        filter_fcn_opts: A mutable dictionary of caller-defined options forwarded to
            the filter on every invocation. It is cleared upon registration.
    """
    self.policy_filter = filter_fcn
    self.policy_filter_opts = filter_fcn_opts
    self.policy_filter_opts.clear()

set_wildcards ¶

set_wildcards(wildcards: dict[str, str | float | int] | None, apply: bool = False) -> None

Replace the entire wildcard dictionary and propagate it to all actions and states.

The new dictionary completely replaces self.wildcards; existing entries not present in wildcards are lost. The wildcard reference is then propagated to every Action and State object already registered in the machine via their respective set_wildcards methods. If apply is True, apply_wildcards is called immediately afterwards to substitute the new values into all messages and action arguments.

Use add_wildcards to merge new entries without discarding existing ones, and update_wildcard to change a single entry by key.

Parameters:

Name	Type	Description	Default
`wildcards`	`dict[str, str \| float \| int] \| None`	The new wildcard mapping. `None` is treated as an empty dict.	required
`apply`	`bool`	If `True`, immediately apply the wildcards to actions, states, and the welcome message after updating. Defaults to `False`.	`False`

Source code in unaiverse/hsm/hsm.py

def set_wildcards(self, wildcards: dict[str, str | float | int] | None, apply: bool = False) -> None:
    """Replace the entire wildcard dictionary and propagate it to all actions and states.

    The new dictionary completely replaces ``self.wildcards``; existing entries not
    present in ``wildcards`` are lost. The wildcard reference is then propagated to
    every ``Action`` and ``State`` object already registered in the machine via their
    respective ``set_wildcards`` methods. If ``apply`` is ``True``, ``apply_wildcards``
    is called immediately afterwards to substitute the new values into all messages
    and action arguments.

    Use ``add_wildcards`` to merge new entries without discarding existing ones, and
    ``update_wildcard`` to change a single entry by key.

    Args:
        wildcards: The new wildcard mapping. ``None`` is treated as an empty dict.
        apply: If ``True``, immediately apply the wildcards to actions, states, and the
            welcome message after updating. Defaults to ``False``.
    """
    wildcards = wildcards if wildcards is not None else {}

    # Saving wildcard dictionary, completely replacing the previous one
    self.wildcards = wildcards

    # Propagating the reference to the new dictionary to actions and states.
    for action in self.__id_to_action:
        action.set_wildcards(self.wildcards)
    for state in self.__id_to_state:
        state.set_wildcards(self.wildcards)

    if apply:
        self.apply_wildcards()

apply_wildcards ¶

apply_wildcards() -> None

Apply the current wildcard substitutions to all actions, states, and the welcome message.

Iterates over every registered Action and State and calls their apply_wildcards methods so that any placeholder found in their messages or arguments is replaced with the corresponding value from self.wildcards. If a welcome message is set, welcome_msg_with_wildcards (the original template) is first restored, then each wildcard replacement is applied in turn, and the result is stored in welcome_msg.

This method does not alter self.wildcards itself. It is called automatically by set_wildcards and add_wildcards when their apply parameter is True, and by set_role after updating the role wildcard.

Source code in unaiverse/hsm/hsm.py

def apply_wildcards(self) -> None:
    """Apply the current wildcard substitutions to all actions, states, and the welcome message.

    Iterates over every registered ``Action`` and ``State`` and calls their
    ``apply_wildcards`` methods so that any placeholder found in their messages or
    arguments is replaced with the corresponding value from ``self.wildcards``. If a
    welcome message is set, ``welcome_msg_with_wildcards`` (the original template) is
    first restored, then each wildcard replacement is applied in turn, and the result
    is stored in ``welcome_msg``.

    This method does not alter ``self.wildcards`` itself. It is called automatically
    by ``set_wildcards`` and ``add_wildcards`` when their ``apply`` parameter is
    ``True``, and by ``set_role`` after updating the role wildcard.
    """

    # Propagating the reference to the new dictionary to actions and states.
    for action in self.__id_to_action:
        action.apply_wildcards()
    for state in self.__id_to_state:
        state.apply_wildcards()

    # If there is a welcome message that includes wildcards...
    if self.welcome_msg is not None:
        self.welcome_msg = self.welcome_msg_with_wildcards  # Restore before updating
        for wildcard_from, wildcard_to in self.wildcards.items():
            self.welcome_msg = self.welcome_msg.replace(wildcard_from, str(wildcard_to))  # Update

set_role ¶

set_role(role: str) -> None

Set the role string for this machine and propagate it through all wildcards and messages.

Stores role in self.role, then updates the Custom.ROLE_WILDCARD entry in the wildcard dictionary with the new value, and finally calls apply_wildcards so that every action argument and state message that references the role placeholder is refreshed. This is the canonical way to assign or change the agent role at runtime; direct assignment to self.role bypasses wildcard propagation.

Parameters:

Name	Type	Description	Default
`role`	`str`	The role string to assign (e.g. `"teacher"`, `"student"`).	required

Examples:

>>> hsm.set_role("teacher")
>>> print(hsm.role)
teacher

Source code in unaiverse/hsm/hsm.py

def set_role(self, role: str) -> None:
    """Set the role string for this machine and propagate it through all wildcards and messages.

    Stores ``role`` in ``self.role``, then updates the ``Custom.ROLE_WILDCARD`` entry
    in the wildcard dictionary with the new value, and finally calls ``apply_wildcards``
    so that every action argument and state message that references the role placeholder
    is refreshed. This is the canonical way to assign or change the agent role at
    runtime; direct assignment to ``self.role`` bypasses wildcard propagation.

    Args:
        role: The role string to assign (e.g. ``"teacher"``, ``"student"``).

    Examples:
        >>> hsm.set_role("teacher")
        >>> print(hsm.role)
        teacher
    """
    self.role = role
    self.update_wildcard(Custom.ROLE_WILDCARD, self.role)
    self.apply_wildcards()

get_wildcards ¶

get_wildcards() -> dict

Return the wildcard dictionary currently used by the state machine.

The returned object is the live internal dictionary shared with all registered actions and states. Callers should not mutate it directly; use update_wildcard, add_wildcards, or set_wildcards instead.

Returns:

Type	Description
`dict`	The `dict` mapping wildcard placeholder strings to their current replacement
`dict`	values (strings, floats, or ints).

Source code in unaiverse/hsm/hsm.py

def get_wildcards(self) -> dict:
    """Return the wildcard dictionary currently used by the state machine.

    The returned object is the live internal dictionary shared with all registered
    actions and states. Callers should not mutate it directly; use ``update_wildcard``,
    ``add_wildcards``, or ``set_wildcards`` instead.

    Returns:
        The ``dict`` mapping wildcard placeholder strings to their current replacement
        values (strings, floats, or ints).
    """
    return self.wildcards

add_wildcards ¶

add_wildcards(wildcards: dict[str, str | float | int | list[str]], apply: bool = False) -> None

Merge new wildcard entries into the existing wildcard dictionary.

Uses dict.update on self.wildcards, so keys present in both are overwritten by the new values. Existing keys not present in wildcards are preserved. If apply is True, apply_wildcards is called after the merge, substituting updated values into all action arguments, state messages, and the welcome message. Unlike set_wildcards, this method never discards existing entries.

Parameters:

Name	Type	Description	Default
`wildcards`	`dict[str, str \| float \| int \| list[str]]`	Dictionary of wildcard entries to add or update. Values may be strings, floats, ints, or lists of strings.	required
`apply`	`bool`	If `True`, immediately apply the updated wildcards after merging. Defaults to `False`.	`False`

Source code in unaiverse/hsm/hsm.py

def add_wildcards(self, wildcards: dict[str, str | float | int | list[str]], apply: bool = False) -> None:
    """Merge new wildcard entries into the existing wildcard dictionary.

    Uses ``dict.update`` on ``self.wildcards``, so keys present in both are
    overwritten by the new values. Existing keys not present in ``wildcards`` are
    preserved. If ``apply`` is ``True``, ``apply_wildcards`` is called after the
    merge, substituting updated values into all action arguments, state messages, and
    the welcome message. Unlike ``set_wildcards``, this method never discards existing
    entries.

    Args:
        wildcards: Dictionary of wildcard entries to add or update. Values may be
            strings, floats, ints, or lists of strings.
        apply: If ``True``, immediately apply the updated wildcards after merging.
            Defaults to ``False``.
    """

    # Update dictionary
    self.wildcards.update(wildcards)
    if apply:
        self.apply_wildcards()

update_wildcard ¶

update_wildcard(wildcard_key: str, wildcard_value: str | float | int, apply: bool = False) -> None

Update the value of a single existing wildcard entry.

If wildcard_key is not already present in self.wildcards, the error is logged but no exception is raised (the call is silently ignored). When the key exists, its value is updated in place; because all actions and states share the same dictionary reference, they see the change immediately even without calling apply_wildcards. If apply is True, apply_wildcards is called afterwards to refresh all messages and action arguments.

Parameters:

Name	Type	Description	Default
`wildcard_key`	`str`	The placeholder string that must already exist in `self.wildcards` (e.g. `"<role>"`).	required
`wildcard_value`	`str \| float \| int`	The new replacement value for the wildcard.	required
`apply`	`bool`	If `True`, immediately apply wildcards after updating the value. Defaults to `False`.	`False`

Note

No exception is raised for an unknown wildcard_key; the condition is logged as an error and the method returns without modifying state.

Source code in unaiverse/hsm/hsm.py

def update_wildcard(self, wildcard_key: str, wildcard_value: str | float | int, apply: bool = False) -> None:
    """Update the value of a single existing wildcard entry.

    If ``wildcard_key`` is not already present in ``self.wildcards``, the error is
    logged but no exception is raised (the call is silently ignored). When the key
    exists, its value is updated in place; because all actions and states share the
    same dictionary reference, they see the change immediately even without calling
    ``apply_wildcards``. If ``apply`` is ``True``, ``apply_wildcards`` is called
    afterwards to refresh all messages and action arguments.

    Args:
        wildcard_key: The placeholder string that must already exist in
            ``self.wildcards`` (e.g. ``"<role>"``).
        wildcard_value: The new replacement value for the wildcard.
        apply: If ``True``, immediately apply wildcards after updating the value.
            Defaults to ``False``.

    Note:
        No exception is raised for an unknown ``wildcard_key``; the condition is
        logged as an error and the method returns without modifying state.
    """
    if wildcard_key not in self.wildcards:
        log.error(f"{wildcard_key} is not a valid wildcard")
    else:
        self.wildcards[wildcard_key] = wildcard_value
        if apply:
            self.apply_wildcards()

get_action_step_idx ¶

get_action_step_idx() -> int

Return the current step index of the action being executed.

Reads the step index from the Interaction object stored in the internal feasible-actions-status dict. For single-step actions the index is 0 while the action is running and -1 when idle. For multi-step actions it advances with each call to act_transitions.

Returns:

Type	Description
`int`	An integer representing the current step index (>= 0), or `-1` if no
`int`	action is currently being executed.

Source code in unaiverse/hsm/hsm.py

def get_action_step_idx(self) -> int:
    """Return the current step index of the action being executed.

    Reads the step index from the ``Interaction`` object stored in the internal
    feasible-actions-status dict. For single-step actions the index is ``0`` while
    the action is running and ``-1`` when idle. For multi-step actions it advances
    with each call to ``act_transitions``.

    Returns:
        An integer representing the current step index (>= 0), or ``-1`` if no
        action is currently being executed.
    """
    return self.__cur_feasible_actions_status['selected_interaction'].get_step_idx() \
        if self.__action is not None else -1

is_busy_acting ¶

is_busy_acting() -> bool

Return whether the state machine is currently mid-action.

Delegates to get_action_step_idx and returns True when the step index is

= 0, meaning an action has been started but has not yet completed. During this time self.state is None and self.limbo_state holds the originating state.

Returns:

Type	Description
`bool`	`True` if an action is in progress, `False` otherwise.

Source code in unaiverse/hsm/hsm.py

def is_busy_acting(self) -> bool:
    """Return whether the state machine is currently mid-action.

    Delegates to ``get_action_step_idx`` and returns ``True`` when the step index is
    >= 0, meaning an action has been started but has not yet completed. During this
    time ``self.state`` is ``None`` and ``self.limbo_state`` holds the originating
    state.

    Returns:
        ``True`` if an action is in progress, ``False`` otherwise.
    """
    return self.get_action_step_idx() >= 0

add_state ¶

add_state(state: str, action: str = None, args: dict | None = None, state_id: int | None = None, waiting_time: float | None = None, blocking: bool | None = None, msg: str | None = None, msg_action: str | None = None) -> None

Add a new state to the machine, or update an existing state's configuration.

If state is not yet registered, a new State object is created and appended to the internal ordered list, and a new Action is created for the inner state action (if action is given). If state already exists, the existing entry is updated in place: parameters passed as None fall back to the current values stored on the existing State object.

When the first state is added and no current state is set yet, set_state is called automatically to make this the initial and current state.

The wildcard dictionary shared by the machine is propagated to both the new State and its inner Action.

Parameters:

Name	Type	Description	Default
`state`	`str`	Name of the state to add or update.	required
`action`	`str`	Name of the method on `actionable` to call as the inner state action. Pass `None` to create a state with no inner action.	`None`
`args`	`dict \| None`	Keyword arguments forwarded to `action` when it is called. Defaults to an empty dict.	`None`
`state_id`	`int \| None`	Explicit integer ID for the state. If `None`, the next sequential ID is assigned automatically. Defaults to `None`.	`None`
`waiting_time`	`float \| None`	Minimum time (in seconds) the machine must remain in this state before transitions are considered. `None` inherits the existing value or defaults to `0.0`. Defaults to `None`.	`None`
`blocking`	`bool \| None`	If `True`, the `act` loop stops after entering this state. `None` inherits the existing value or defaults to `True`. Defaults to `None`.	`None`
`msg`	`str \| None`	Human-readable message associated with the state, shown in log output. `None` inherits the existing value. Defaults to `None`.	`None`
`msg_action`	`str \| None`	Human-readable message for the inner state action. Only used when a new `Action` is created (i.e. when `action` is not `None`). Defaults to `None`.	`None`

Examples:

>>> hsm.add_state("processing", action="run_model",
...               args={"timeout": 30}, blocking=False,
...               msg="Processing user input...")

Source code in unaiverse/hsm/hsm.py

def add_state(self, state: str, action: str = None, args: dict | None = None, state_id: int | None = None,
              waiting_time: float | None = None, blocking: bool | None = None,
              msg: str | None = None, msg_action: str | None = None) -> None:
    """Add a new state to the machine, or update an existing state's configuration.

    If ``state`` is not yet registered, a new ``State`` object is created and appended
    to the internal ordered list, and a new ``Action`` is created for the inner state
    action (if ``action`` is given). If ``state`` already exists, the existing entry is
    updated in place: parameters passed as ``None`` fall back to the current values
    stored on the existing ``State`` object.

    When the first state is added and no current state is set yet, ``set_state`` is
    called automatically to make this the initial and current state.

    The wildcard dictionary shared by the machine is propagated to both the new
    ``State`` and its inner ``Action``.

    Args:
        state: Name of the state to add or update.
        action: Name of the method on ``actionable`` to call as the inner state action.
            Pass ``None`` to create a state with no inner action.
        args: Keyword arguments forwarded to ``action`` when it is called. Defaults to
            an empty dict.
        state_id: Explicit integer ID for the state. If ``None``, the next sequential
            ID is assigned automatically. Defaults to ``None``.
        waiting_time: Minimum time (in seconds) the machine must remain in this state
            before transitions are considered. ``None`` inherits the existing value or
            defaults to ``0.0``. Defaults to ``None``.
        blocking: If ``True``, the ``act`` loop stops after entering this state.
            ``None`` inherits the existing value or defaults to ``True``. Defaults to
            ``None``.
        msg: Human-readable message associated with the state, shown in log output.
            ``None`` inherits the existing value. Defaults to ``None``.
        msg_action: Human-readable message for the inner state action. Only used when
            a new ``Action`` is created (i.e. when ``action`` is not ``None``).
            Defaults to ``None``.

    Examples:
        >>> hsm.add_state("processing", action="run_model",
        ...               args={"timeout": 30}, blocking=False,
        ...               msg="Processing user input...")
    """
    if args is None:
        args = {}
    sta_obj = None
    if state_id is None:
        if state not in self.states:
            state_id = len(self.__id_to_state)
        else:
            sta_obj = self.states[state]
            state_id = sta_obj.id
    if action is None:
        act = sta_obj.action if sta_obj is not None else None
    else:
        act = Action(name=action, args=args, idx=len(self.__id_to_action),
                     actionable=self.actionable, avoid_changing_ready=True,
                     msg=msg_action)
        act.set_wildcards(self.wildcards)
        self.__id_to_action.append(act)
    if waiting_time is None:
        waiting_time = sta_obj.waiting_time if sta_obj is not None else 0.  # Default waiting time
    if blocking is None:
        blocking = sta_obj.blocking if sta_obj is not None else True  # Default blocking
    if msg is None:
        msg = sta_obj.msg_with_wildcards if sta_obj is not None else None

    sta = State(name=state, idx=state_id, action=act, waiting_time=waiting_time, blocking=blocking, msg=msg)
    sta.set_wildcards(self.wildcards)
    if state not in self.states:
        self.__id_to_state.append(sta)
    else:
        self.__id_to_state[state_id] = sta
    self.states[state] = sta

    if len(self.__id_to_state) == 1 and self.state is None:
        self.set_state(sta.name)

    sta.set_state_machine(self)

get_state_name ¶

get_state_name(consider_limbo: bool = False) -> str | None

Return the name of the current state of the machine.

During a multi-step action, self.state is None and the originating state is stored in self.limbo_state. When consider_limbo is True and self.state is None, this method returns limbo_state instead, giving callers a non-None state name even while an action is in progress.

Parameters:

Name	Type	Description	Default
`consider_limbo`	`bool`	If `True`, return `limbo_state` when the machine is mid-transition and `state` is `None`. Defaults to `False`.	`False`

Returns:

Type	Description
`str \| None`	The name of the current (or limbo) state as a string, or `None` if no
`str \| None`	state has been set and `consider_limbo` is `False`.

Source code in unaiverse/hsm/hsm.py

def get_state_name(self, consider_limbo: bool = False) -> str | None:
    """Return the name of the current state of the machine.

    During a multi-step action, ``self.state`` is ``None`` and the originating state
    is stored in ``self.limbo_state``. When ``consider_limbo`` is ``True`` and
    ``self.state`` is ``None``, this method returns ``limbo_state`` instead, giving
    callers a non-``None`` state name even while an action is in progress.

    Args:
        consider_limbo: If ``True``, return ``limbo_state`` when the machine is
            mid-transition and ``state`` is ``None``. Defaults to ``False``.

    Returns:
        The name of the current (or limbo) state as a string, or ``None`` if no
        state has been set and ``consider_limbo`` is ``False``.
    """

    if not consider_limbo:
        return self.state
    else:
        if self.state is None and self.limbo_state is not None:
            return self.limbo_state
        else:
            return self.state

get_state ¶

get_state() -> State | None

Return the State object corresponding to the current state.

When the machine is mid-transition (a multi-step action is running), self.state is None and this method therefore returns None as well. Use get_state_name(consider_limbo=True) to obtain the originating state name in that situation.

Returns:

Type	Description
`State \| None`	The `State` object for the current state, or `None` if no state is active
`State \| None`	(e.g., while a multi-step action is in progress or before any state has been set).

Source code in unaiverse/hsm/hsm.py

def get_state(self) -> 'State | None':
    """Return the ``State`` object corresponding to the current state.

    When the machine is mid-transition (a multi-step action is running), ``self.state``
    is ``None`` and this method therefore returns ``None`` as well. Use
    ``get_state_name(consider_limbo=True)`` to obtain the originating state name in that
    situation.

    Returns:
        The ``State`` object for the current state, or ``None`` if no state is active
        (e.g., while a multi-step action is in progress or before any state has been set).
    """
    return self.states[self.state] if self.state is not None else None

get_all_states ¶

get_all_states() -> list[State]

Return the ordered list of all State objects registered in the machine.

The list is ordered by the integer ID assigned to each state at creation time (either explicitly via the state_id parameter of add_state, or sequentially). This is the same order in which states appear in the serialised JSON. See also get_states for a plain list of state name strings.

Returns:

Type	Description
`list[State]`	A list of `State` objects in creation order. The returned object is the
`list[State]`	machine's internal list; do not modify it directly.

Source code in unaiverse/hsm/hsm.py

def get_all_states(self) -> list['State']:
    """Return the ordered list of all ``State`` objects registered in the machine.

    The list is ordered by the integer ID assigned to each state at creation time
    (either explicitly via the ``state_id`` parameter of ``add_state``, or
    sequentially). This is the same order in which states appear in the serialised
    JSON. See also ``get_states`` for a plain list of state name strings.

    Returns:
        A list of ``State`` objects in creation order. The returned object is the
        machine's internal list; do not modify it directly.
    """
    return self.__id_to_state

get_all_actions ¶

get_all_actions() -> list[Action]

Return the ordered list of every Action object registered in the machine.

This list includes both transition actions (those that guard edges between states) and inner state actions (those attached to a state via the action parameter of add_state). Actions are ordered by their integer ID, which is assigned at creation time. See get_action to retrieve the action currently being executed.

Returns:

Type	Description
`list[Action]`	A list of `Action` objects in creation order. The returned object is the
`list[Action]`	machine's internal list; do not modify it directly.

Source code in unaiverse/hsm/hsm.py

def get_all_actions(self) -> list['Action']:
    """Return the ordered list of every ``Action`` object registered in the machine.

    This list includes both *transition actions* (those that guard edges between states)
    and *inner state actions* (those attached to a state via the ``action`` parameter of
    ``add_state``). Actions are ordered by their integer ID, which is assigned at
    creation time. See ``get_action`` to retrieve the action currently being executed.

    Returns:
        A list of ``Action`` objects in creation order. The returned object is the
        machine's internal list; do not modify it directly.
    """
    return self.__id_to_action

get_action ¶

get_action() -> Action | None

Return the Action object that is currently being executed.

An action is considered "current" from the moment it is selected by the policy until it completes (status 0). For single-step actions this window is a single call to act_transitions; for multi-step actions it spans multiple calls. When no action is running, this method returns None. See is_busy_acting for a boolean convenience check.

Returns:

Type	Description
`Action \| None`	The active `Action` object, or `None` if no action is currently in
`Action \| None`	progress.

Source code in unaiverse/hsm/hsm.py

def get_action(self) -> 'Action | None':
    """Return the ``Action`` object that is currently being executed.

    An action is considered "current" from the moment it is selected by the policy
    until it completes (status ``0``). For single-step actions this window is a single
    call to ``act_transitions``; for multi-step actions it spans multiple calls. When
    no action is running, this method returns ``None``. See ``is_busy_acting`` for a
    boolean convenience check.

    Returns:
        The active ``Action`` object, or ``None`` if no action is currently in
        progress.
    """
    return self.__action

get_action_name ¶

get_action_name() -> str | None

Return the name of the action currently being executed.

This is a convenience wrapper around get_action that returns only the name string instead of the full Action object. The name corresponds to the method name on actionable that is being called.

Returns:

Type	Description
`str \| None`	The name string of the active action, or `None` if no action is currently
`str \| None`	in progress.

Source code in unaiverse/hsm/hsm.py

def get_action_name(self) -> str | None:
    """Return the name of the action currently being executed.

    This is a convenience wrapper around ``get_action`` that returns only the name
    string instead of the full ``Action`` object. The name corresponds to the method
    name on ``actionable`` that is being called.

    Returns:
        The name string of the active action, or ``None`` if no action is currently
        in progress.
    """
    return self.__action.name if self.__action is not None else None

get_last_completed_action_name ¶

get_last_completed_action_name() -> str | None

Return the name of the most recently completed transition action.

The internal reference is updated every time act_transitions receives status 0 (action fully done) from the action callable. It persists across state changes and is not reset by reset_state. It is None only before any action has ever completed.

Returns:

Type	Description
`str \| None`	The name string of the last successfully completed action, or `None` if no
`str \| None`	action has completed yet in the lifetime of this machine.

Source code in unaiverse/hsm/hsm.py

def get_last_completed_action_name(self) -> str | None:
    """Return the name of the most recently completed transition action.

    The internal reference is updated every time ``act_transitions`` receives status
    ``0`` (action fully done) from the action callable. It persists across state
    changes and is not reset by ``reset_state``. It is ``None`` only before any
    action has ever completed.

    Returns:
        The name string of the last successfully completed action, or ``None`` if no
        action has completed yet in the lifetime of this machine.
    """
    return self.__last_completed_action.name if self.__last_completed_action is not None else None

reset_state ¶

reset_state() -> None

Reset the state machine to its initial state, clearing all transient execution state.

The following fields are cleared or reset:

state is set back to initial_state.
limbo_state and prev_state are set to None.
The currently executing action (if any) is abandoned without completing it.
All step counters and interaction queues on every registered transition action and inner state action are cleared via clear_interactions and system_interaction.reset_state().

Note

__last_completed_action is not cleared, so get_last_completed_action_name still returns the name of the last action that completed before the reset.

Source code in unaiverse/hsm/hsm.py

def reset_state(self) -> None:
    """Reset the state machine to its initial state, clearing all transient execution state.

    The following fields are cleared or reset:

    - ``state`` is set back to ``initial_state``.
    - ``limbo_state`` and ``prev_state`` are set to ``None``.
    - The currently executing action (if any) is abandoned without completing it.
    - All step counters and interaction queues on every registered transition action
      and inner state action are cleared via ``clear_interactions`` and
      ``system_interaction.reset_state()``.

    Note:
        ``__last_completed_action`` is not cleared, so ``get_last_completed_action_name``
        still returns the name of the last action that completed before the reset.
    """
    self.state = self.initial_state
    self.limbo_state = None
    self.prev_state = None
    self.__action = None
    for act in self.__id_to_action:
        act.clear_interactions()
        act.system_interaction.reset_state()
    for s in self.__id_to_state:
        if s.action is not None:
            s.action.clear_interactions()
            s.action.system_interaction.reset_state()

get_states ¶

get_states() -> list

Return a list of all state names defined in the machine, in creation order.

The returned list contains plain strings (state names), not State objects. Use get_all_states to obtain the full State objects, or states to access the underlying name-to-State mapping directly.

Returns:

Type	Description
`list`	A list of state name strings ordered by state creation sequence.

Source code in unaiverse/hsm/hsm.py

def get_states(self) -> list:
    """Return a list of all state names defined in the machine, in creation order.

    The returned list contains plain strings (state names), not ``State`` objects.
    Use ``get_all_states`` to obtain the full ``State`` objects, or ``states`` to
    access the underlying name-to-``State`` mapping directly.

    Returns:
        A list of state name strings ordered by state creation sequence.
    """
    return [s.name for s in self.__id_to_state]

set_state ¶

set_state(state: str) -> None

Forcibly set the current state of the machine, bypassing normal transition logic.

limbo_state is cleared, prev_state is updated to the outgoing state, and state is assigned the new value. If an action was in progress, its associated Interaction is reset and the action reference is discarded. If no initial state has been set yet, initial_state is also assigned to state.

This method is intended for programmatic overrides (e.g., during machine setup or after a reset_state call). Normal runtime transitions are handled by act_transitions, which calls this method internally after a successful action.

Parameters:

Name	Type	Description	Default
`state`	`str`	The name of the state to activate. Must already be registered in the machine (either in `transitions` or in `states`).	required

Raises:

Type	Description
`ValueError`	If `state` is not a known state of this machine.

Examples:

>>> hsm.set_state("idle")
>>> print(hsm.state)
idle

Source code in unaiverse/hsm/hsm.py

def set_state(self, state: str) -> None:
    """Forcibly set the current state of the machine, bypassing normal transition logic.

    ``limbo_state`` is cleared, ``prev_state`` is updated to the outgoing state, and
    ``state`` is assigned the new value. If an action was in progress, its associated
    ``Interaction`` is reset and the action reference is discarded. If no initial state
    has been set yet, ``initial_state`` is also assigned to ``state``.

    This method is intended for programmatic overrides (e.g., during machine setup or
    after a ``reset_state`` call). Normal runtime transitions are handled by
    ``act_transitions``, which calls this method internally after a successful action.

    Args:
        state: The name of the state to activate. Must already be registered in the
            machine (either in ``transitions`` or in ``states``).

    Raises:
        ValueError: If ``state`` is not a known state of this machine.

    Examples:
        >>> hsm.set_state("idle")
        >>> print(hsm.state)
        idle
    """
    if state in self.transitions or state in self.states:
        self.limbo_state = None
        self.prev_state = self.state
        self.state = state
        if self.__action is not None:
            self.__cur_feasible_actions_status['selected_interaction'].reset_status()
            self.__action = None
        if self.initial_state is None:
            self.initial_state = state
    else:
        raise ValueError("Unknown state: " + str(state))

are_debug_messages_active ¶

are_debug_messages_active() -> bool

Return whether debug messages are currently active for this machine.

When debug mode is active, the machine appends tick symbols to action messages, colored markers to blocking-state messages, requester info to action log lines, and auto-generated labels to states and actions that have none. See set_debug_messages_active for details.

Returns:

Type	Description
`bool`	`True` if debug mode is enabled, `False` otherwise.

Source code in unaiverse/hsm/hsm.py

def are_debug_messages_active(self) -> bool:
    """Return whether debug messages are currently active for this machine.

    When debug mode is active, the machine appends tick symbols to action messages,
    colored markers to blocking-state messages, requester info to action log lines, and
    auto-generated labels to states and actions that have none. See
    ``set_debug_messages_active`` for details.

    Returns:
        ``True`` if debug mode is enabled, ``False`` otherwise.
    """
    return self.__debug_messages_active

set_debug_messages_active ¶

set_debug_messages_active(yes: bool) -> None

Enable or disable debug mode for this machine.

When yes is True, the following are enabled in one call:

Tick symbols appended to action log messages (show_ticks_in_action_messages).
Colored markers appended to blocking-state log messages (show_marks_in_blocking_state_messages).
Requester and UUID info appended to action log messages (show_request_info_in_action_messages).
Auto-generated human-readable labels for states and actions that have none (generate_auto_messages), with force=True so that existing messages are prefixed rather than discarded.

When yes is False, all four features are disabled and the original messages stored at enable-time are restored. Wildcards are then re-applied to the restored messages. This method is idempotent: calling it twice in the same direction is safe.

Parameters:

Name	Type	Description	Default
`yes`	`bool`	`True` to activate debug mode, `False` to deactivate it and restore original messages.	required

Source code in unaiverse/hsm/hsm.py

def set_debug_messages_active(self, yes: bool) -> None:
    """Enable or disable debug mode for this machine.

    When ``yes`` is ``True``, the following are enabled in one call:

    - Tick symbols appended to action log messages (``show_ticks_in_action_messages``).
    - Colored markers appended to blocking-state log messages
      (``show_marks_in_blocking_state_messages``).
    - Requester and UUID info appended to action log messages
      (``show_request_info_in_action_messages``).
    - Auto-generated human-readable labels for states and actions that have none
      (``generate_auto_messages``), with ``force=True`` so that existing messages are
      prefixed rather than discarded.

    When ``yes`` is ``False``, all four features are disabled and the original messages
    stored at enable-time are restored. Wildcards are then re-applied to the restored
    messages. This method is idempotent: calling it twice in the same direction is safe.

    Args:
        yes: ``True`` to activate debug mode, ``False`` to deactivate it and restore
            original messages.
    """
    self.__debug_messages_active = yes

    if yes:
        self.show_ticks_in_action_messages(True)
        self.show_marks_in_blocking_state_messages(True)
        self.show_request_info_in_action_messages(True)

        # Replace original messages
        self.generate_auto_messages(force=True)
    else:
        self.show_ticks_in_action_messages(False)
        self.show_marks_in_blocking_state_messages(False)
        self.show_request_info_in_action_messages(False)

        # Restore original messages
        if len(self.__id_to_original_state_msg) > 0:
            for i, state in enumerate(self.__id_to_state):
                state.set_msg(self.__id_to_original_state_msg[i][0])
                if state.action is not None:
                    state.action.set_msg(self.__id_to_original_state_msg[i][1])
        if len(self.__id_to_original_action_msg) > 0:
            for i, action in enumerate(self.__id_to_action):
                action.set_msg(self.__id_to_original_action_msg[i])
        self.__id_to_original_state_msg.clear()
        self.__id_to_original_action_msg.clear()

        # Apply wildcards to the restored messages
        self.apply_wildcards()

generate_auto_messages ¶

generate_auto_messages(states: bool = True, actions: bool = True, force: bool = False) -> None

Auto-generate human-readable messages for states and actions that currently have none.

For each state (and its inner action) or transition action that has no message set, a label is derived from the state/action name by capitalising it and replacing underscores with spaces, prefixed with a pin or rocket emoji. When force is True, existing messages are not skipped but are instead prepended with the auto-generated label and kept in square brackets (e.g. "-> my_state [Original text]").

The original messages are saved internally before any modification so that set_debug_messages_active(False) can restore them exactly. Calling this method twice for states (or for actions) has no effect on the second call because the saved originals list is populated only once.

Parameters:

Name	Type	Description	Default
`states`	`bool`	If `True`, generate messages for states and their inner state actions. Defaults to `True`.	`True`
`actions`	`bool`	If `True`, generate messages for transition actions. Defaults to `True`.	`True`
`force`	`bool`	If `True`, overwrite existing messages, appending the original text in square brackets. Defaults to `False`.	`False`

Note

This method is called automatically by set_debug_messages_active(True) with force=True. Calling it manually before enabling debug mode is supported but may interact with the saved-original-message restoration performed when debug mode is later disabled.

Source code in unaiverse/hsm/hsm.py

def generate_auto_messages(self, states: bool = True, actions: bool = True, force: bool = False) -> None:
    """Auto-generate human-readable messages for states and actions that currently have none.

    For each state (and its inner action) or transition action that has no message set,
    a label is derived from the state/action name by capitalising it and replacing
    underscores with spaces, prefixed with a pin or rocket emoji. When ``force`` is
    ``True``, existing messages are not skipped but are instead prepended with the
    auto-generated label and kept in square brackets (e.g. ``"-> my_state [Original text]"``).

    The original messages are saved internally before any modification so that
    ``set_debug_messages_active(False)`` can restore them exactly. Calling this method
    twice for states (or for actions) has no effect on the second call because the
    saved originals list is populated only once.

    Args:
        states: If ``True``, generate messages for states and their inner state
            actions. Defaults to ``True``.
        actions: If ``True``, generate messages for transition actions. Defaults to
            ``True``.
        force: If ``True``, overwrite existing messages, appending the original text
            in square brackets. Defaults to ``False``.

    Note:
        This method is called automatically by ``set_debug_messages_active(True)``
        with ``force=True``. Calling it manually before enabling debug mode is
        supported but may interact with the saved-original-message restoration
        performed when debug mode is later disabled.
    """
    if states is True and len(self.__id_to_original_state_msg) == 0:
        for state in self.__id_to_state:
            original1 = state.msg_with_wildcards
            original2 = state.action.msg_with_wildcards if state.action is not None else None

            if state.msg_with_wildcards is None:
                state.set_msg("📍 " + state.name.replace('_', ' ').capitalize())
            elif force is True:
                state.set_msg("📍 " + state.name.replace('_', ' ').capitalize() +
                              " [" + state.msg_with_wildcards + "]")
            if state.action is not None:
                if state.action.msg_with_wildcards is None:
                    state.action.set_msg("📍 " + state.action.name.replace('_', ' ').capitalize())
                elif force is True:
                    state.action.set_msg("📍 " + state.action.name.replace('_', ' ').capitalize() +
                                         " [" + state.action.msg_with_wildcards + "]")

            self.__id_to_original_state_msg.append((original1, original2))
    if actions is True and len(self.__id_to_original_action_msg) == 0:
        for action in self.__id_to_action:
            original = action.msg_with_wildcards

            if action.msg_with_wildcards is None:
                action.set_msg("🚀 " + action.name.replace('_', ' ').capitalize())
            elif force:
                action.set_msg("🚀 " + action.name.replace('_', ' ').capitalize() +
                               " [" + action.msg_with_wildcards + "]")

            self.__id_to_original_action_msg.append(original)

add_global_teleport ¶

add_global_teleport(to_state: str, action: str, args: dict | None = None, ready: bool = True, msg: str | None = None, high_priority: bool = False, total_time: float | str = 0.0, timeout: float | str = 0.0, delay: float | str = 0.0) -> None

Add a teleport transition from every state in the machine to a single destination state.

This is a shorthand for add_teleport with from_state set to the special Custom.ALL_STATES_NAME sentinel (typically "all"). The underlying add_transit call expands the wildcard source by iterating over every state currently registered and adding an individual teleport transition for each one (except the destination itself, which is skipped automatically).

Teleports are visually hidden in to_graphviz output but execute normally during act_transitions. This makes them suitable for global "escape" or "interrupt" transitions (for example, moving every state back to an error handler on a particular action).

Parameters:

Name	Type	Description	Default
`to_state`	`str`	The name of the destination state. If this state is not yet registered, `add_transit` creates it automatically.	required
`action`	`str`	The name of the method on `actionable` that guards this transition.	required
`args`	`dict \| None`	Keyword arguments forwarded to `action` when it is called. Defaults to an empty dict.	`None`
`ready`	`bool`	Whether the action is immediately selectable by the policy. Defaults to `True`.	`True`
`msg`	`str \| None`	Human-readable label for the action shown in log output. Defaults to `None`.	`None`
`high_priority`	`bool`	If `True`, the policy selects this action before any non-high-priority action. Defaults to `False`.	`False`
`total_time`	`float \| str`	Maximum wall-clock duration (seconds) for this transition from the moment it starts. `0.` means no limit. Defaults to `0.`.	`0.0`
`timeout`	`float \| str`	Maximum cumulative wait time (seconds) before the action is abandoned. `0.` means no timeout. Defaults to `0.`.	`0.0`
`delay`	`float \| str`	Minimum time (seconds) that must have elapsed in the source state before this transition is considered. `0.` means no delay. Defaults to `0.`.	`0.0`

Examples:

>>> # Jump to "error" from any state when "handle_error" fires.
>>> hsm.add_global_teleport(to_state="error", action="handle_error")

Source code in unaiverse/hsm/hsm.py

def add_global_teleport(self, to_state: str,
                        action: str, args: dict | None = None, ready: bool = True,
                        msg: str | None = None,
                        high_priority: bool = False,
                        total_time: float | str = 0., timeout: float | str = 0., delay: float | str = 0., ) -> None:
    """Add a teleport transition from every state in the machine to a single destination state.

    This is a shorthand for ``add_teleport`` with ``from_state`` set to the special
    ``Custom.ALL_STATES_NAME`` sentinel (typically ``"all"``). The underlying
    ``add_transit`` call expands the wildcard source by iterating over every state
    currently registered and adding an individual teleport transition for each one
    (except the destination itself, which is skipped automatically).

    Teleports are visually hidden in ``to_graphviz`` output but execute normally
    during ``act_transitions``. This makes them suitable for global "escape" or
    "interrupt" transitions (for example, moving every state back to an error handler
    on a particular action).

    Args:
        to_state: The name of the destination state. If this state is not yet
            registered, ``add_transit`` creates it automatically.
        action: The name of the method on ``actionable`` that guards this transition.
        args: Keyword arguments forwarded to ``action`` when it is called. Defaults to
            an empty dict.
        ready: Whether the action is immediately selectable by the policy. Defaults to
            ``True``.
        msg: Human-readable label for the action shown in log output. Defaults to
            ``None``.
        high_priority: If ``True``, the policy selects this action before any
            non-high-priority action. Defaults to ``False``.
        total_time: Maximum wall-clock duration (seconds) for this transition from the
            moment it starts. ``0.`` means no limit. Defaults to ``0.``.
        timeout: Maximum cumulative wait time (seconds) before the action is abandoned.
            ``0.`` means no timeout. Defaults to ``0.``.
        delay: Minimum time (seconds) that must have elapsed in the source state before
            this transition is considered. ``0.`` means no delay. Defaults to ``0.``.

    Examples:
        >>> # Jump to "error" from any state when "handle_error" fires.
        >>> hsm.add_global_teleport(to_state="error", action="handle_error")
    """
    self.add_teleport(from_state=Custom.ALL_STATES_NAME, to_state=to_state, action=action, args=args,
                      ready=ready, act_id=None, high_priority=high_priority,
                      msg=msg, total_time=total_time, timeout=timeout, delay=delay)

add_teleport ¶

add_teleport(from_state: str, to_state: str, action: str, args: dict | None = None, ready: bool = True, act_id: int | None = None, msg: str | None = None, avoid_changing_ready: bool = False, high_priority: bool = False, total_time: float | str = 0.0, timeout: float | str = 0.0, delay: float | str = 0.0) -> None

Add a hidden (teleport) transition between two states.

A teleport is functionally identical to a normal transition but is marked with the teleport flag so that it is rendered with a transparent edge (or omitted from the visual graph in to_graphviz). It runs normally during act_transitions and is picked up by the policy like any other feasible action. This is useful for background "escape" edges that would clutter the visual layout.

This is a thin wrapper around add_transit that forces teleport=True. For a shorthand that adds the teleport from every state, use add_global_teleport.

Parameters:

Name	Type	Description	Default
`from_state`	`str`	The name of the source state. Pass `Custom.ALL_STATES_NAME` to add the teleport from every currently registered state.	required
`to_state`	`str`	The name of the destination state. If it does not exist yet, it is created automatically by `add_transit`.	required
`action`	`str`	The name of the method on `actionable` that guards this transition.	required
`args`	`dict \| None`	Keyword arguments forwarded to `action` when it is called. Defaults to an empty dict.	`None`
`ready`	`bool`	Whether the action is immediately selectable. Defaults to `True`.	`True`
`act_id`	`int \| None`	Explicit integer ID for the action. `None` assigns the next sequential ID. Defaults to `None`.	`None`
`msg`	`str \| None`	Human-readable label for the action. Defaults to `None`.	`None`
`avoid_changing_ready`	`bool`	If `True`, internal rules do not alter the `ready` flag after creation. Defaults to `False`.	`False`
`high_priority`	`bool`	If `True`, the policy prefers this action over non-priority actions. Defaults to `False`.	`False`
`total_time`	`float \| str`	Maximum wall-clock duration (seconds) for this transition. `0.` means no limit. Defaults to `0.`.	`0.0`
`timeout`	`float \| str`	Maximum cumulative wait time (seconds) before giving up. `0.` means no timeout. Defaults to `0.`.	`0.0`
`delay`	`float \| str`	Minimum time (seconds) that must elapse in `from_state` before this transition is considered. `0.` means no delay. Defaults to `0.`.	`0.0`

Raises:

Type	Description
`ValueError`	If an identical action (same name and args) already exists on the `from_state` -> `to_state` edge.
`FileNotFoundError`	If `to_state` ends with `".json"` and the referenced file does not exist.

Source code in unaiverse/hsm/hsm.py

def add_teleport(self, from_state: str, to_state: str,
                 action: str, args: dict | None = None, ready: bool = True,
                 act_id: int | None = None, msg: str | None = None, avoid_changing_ready: bool = False,
                 high_priority: bool = False,
                 total_time: float | str = 0., timeout: float | str = 0., delay: float | str = 0., ) -> None:
    """Add a hidden (teleport) transition between two states.

    A teleport is functionally identical to a normal transition but is marked with the
    ``teleport`` flag so that it is rendered with a transparent edge (or omitted from
    the visual graph in ``to_graphviz``). It runs normally during ``act_transitions``
    and is picked up by the policy like any other feasible action. This is useful for
    background "escape" edges that would clutter the visual layout.

    This is a thin wrapper around ``add_transit`` that forces ``teleport=True``. For a
    shorthand that adds the teleport from *every* state, use ``add_global_teleport``.

    Args:
        from_state: The name of the source state. Pass ``Custom.ALL_STATES_NAME`` to
            add the teleport from every currently registered state.
        to_state: The name of the destination state. If it does not exist yet, it is
            created automatically by ``add_transit``.
        action: The name of the method on ``actionable`` that guards this transition.
        args: Keyword arguments forwarded to ``action`` when it is called. Defaults to
            an empty dict.
        ready: Whether the action is immediately selectable. Defaults to ``True``.
        act_id: Explicit integer ID for the action. ``None`` assigns the next
            sequential ID. Defaults to ``None``.
        msg: Human-readable label for the action. Defaults to ``None``.
        avoid_changing_ready: If ``True``, internal rules do not alter the ``ready``
            flag after creation. Defaults to ``False``.
        high_priority: If ``True``, the policy prefers this action over non-priority
            actions. Defaults to ``False``.
        total_time: Maximum wall-clock duration (seconds) for this transition.
            ``0.`` means no limit. Defaults to ``0.``.
        timeout: Maximum cumulative wait time (seconds) before giving up. ``0.`` means
            no timeout. Defaults to ``0.``.
        delay: Minimum time (seconds) that must elapse in ``from_state`` before this
            transition is considered. ``0.`` means no delay. Defaults to ``0.``.

    Raises:
        ValueError: If an identical action (same name and args) already exists on the
            ``from_state`` -> ``to_state`` edge.
        FileNotFoundError: If ``to_state`` ends with ``".json"`` and the referenced
            file does not exist.
    """
    self.add_transit(from_state=from_state, to_state=to_state, action=action, args=args,
                     ready=ready, act_id=act_id, avoid_changing_ready=avoid_changing_ready,
                     msg=msg, teleport=True, high_priority=high_priority,
                     total_time=total_time, timeout=timeout, delay=delay)

add_transit ¶

add_transit(from_state: str, to_state: str, action: str, args: dict | None = None, ready: bool = True, act_id: int | None = None, msg: str | None = None, avoid_changing_ready: bool = False, teleport: bool = False, high_priority: bool = False, total_time: float | str = 0.0, timeout: float | str = 0.0, delay: float | str = 0.0) -> None

Add a transition between two states with an associated action.

This is the primary method for defining the machine's topology. It handles three special cases transparently:

Wildcard source (from_state == Custom.ALL_STATES_NAME): the transition is expanded and added individually from every state currently registered in the machine (except the destination).
Sub-machine file (to_state ends with ".json"): the referenced JSON file is loaded as a separate HybridStateMachine, state-name clashes are resolved by appending ".1", ".2", etc., and the sub-machine is then merged into the current one via include. The edge from from_state is wired to the sub-machine's initial state.
Ordinary transition: the source and destination states are registered (if not already present) and a new Action is appended to the edge.

Parameters:

Name	Type	Description	Default
`from_state`	`str`	Name of the source state. Use `Custom.ALL_STATES_NAME` to add the transition from every registered state.	required
`to_state`	`str`	Name of the destination state. If it ends with `".json"`, the file is loaded as a sub-machine and included in the current machine.	required
`action`	`str`	Name of the method on `actionable` that guards this transition.	required
`args`	`dict \| None`	Keyword arguments forwarded to `action` when it is called. Defaults to an empty dict.	`None`
`ready`	`bool`	Whether the action is immediately selectable by the policy. Defaults to `True`.	`True`
`act_id`	`int \| None`	Explicit integer ID for the new action. `None` assigns the next sequential ID automatically. Defaults to `None`.	`None`
`msg`	`str \| None`	Human-readable label for the action shown in log output. Defaults to `None`.	`None`
`avoid_changing_ready`	`bool`	If `True`, internal rules will not alter the `ready` flag of the new action after creation. Defaults to `False`.	`False`
`teleport`	`bool`	If `True`, the transition is hidden in the visual graph produced by `to_graphviz` but executes normally. Defaults to `False`.	`False`
`high_priority`	`bool`	If `True`, the policy selects this action before any non-high-priority action in the feasible set. Defaults to `False`.	`False`
`total_time`	`float \| str`	Maximum wall-clock duration in seconds for this transition from the moment it starts. `0.` or a wildcard string means no limit. Defaults to `0.`.	`0.0`
`timeout`	`float \| str`	Maximum cumulative time in seconds the policy will keep retrying this transition before giving up. `0.` means no timeout. Defaults to `0.`.	`0.0`
`delay`	`float \| str`	Minimum time in seconds that must elapse in `from_state` before this transition is considered by the policy. `0.` means no delay. Defaults to `0.`.	`0.0`

Raises:

Type	Description
`ValueError`	If an identical action (same name and args) already exists on the `from_state` -> `to_state` edge.
`FileNotFoundError`	If `to_state` ends with `".json"` and the referenced file cannot be found on disk.

Examples:

>>> hsm.add_transit("idle", "working", action="start_task",
...                 args={"retries": 3}, msg="Starting task...")
>>> # Embed a sub-machine from a JSON file:
>>> hsm.add_transit("working", "sub_behaviours.json", action="enter_sub")

Source code in unaiverse/hsm/hsm.py

def add_transit(self, from_state: str, to_state: str,
                action: str, args: dict | None = None, ready: bool = True,
                act_id: int | None = None, msg: str | None = None,
                avoid_changing_ready: bool = False,
                teleport: bool = False,
                high_priority: bool = False,
                total_time: float | str = 0., timeout: float | str = 0., delay: float | str = 0.,) -> None:
    """Add a transition between two states with an associated action.

    This is the primary method for defining the machine's topology. It handles three
    special cases transparently:

    1. **Wildcard source** (``from_state == Custom.ALL_STATES_NAME``): the transition
       is expanded and added individually from every state currently registered in the
       machine (except the destination).
    2. **Sub-machine file** (``to_state`` ends with ``".json"``): the referenced JSON
       file is loaded as a separate ``HybridStateMachine``, state-name clashes are
       resolved by appending ``".1"``, ``".2"``, etc., and the sub-machine is then
       merged into the current one via ``include``. The edge from ``from_state``
       is wired to the sub-machine's initial state.
    3. **Ordinary transition**: the source and destination states are registered (if
       not already present) and a new ``Action`` is appended to the edge.

    Args:
        from_state: Name of the source state. Use ``Custom.ALL_STATES_NAME`` to add
            the transition from every registered state.
        to_state: Name of the destination state. If it ends with ``".json"``, the file
            is loaded as a sub-machine and included in the current machine.
        action: Name of the method on ``actionable`` that guards this transition.
        args: Keyword arguments forwarded to ``action`` when it is called. Defaults to
            an empty dict.
        ready: Whether the action is immediately selectable by the policy. Defaults to
            ``True``.
        act_id: Explicit integer ID for the new action. ``None`` assigns the next
            sequential ID automatically. Defaults to ``None``.
        msg: Human-readable label for the action shown in log output. Defaults to
            ``None``.
        avoid_changing_ready: If ``True``, internal rules will not alter the ``ready``
            flag of the new action after creation. Defaults to ``False``.
        teleport: If ``True``, the transition is hidden in the visual graph produced by
            ``to_graphviz`` but executes normally. Defaults to ``False``.
        high_priority: If ``True``, the policy selects this action before any
            non-high-priority action in the feasible set. Defaults to ``False``.
        total_time: Maximum wall-clock duration in seconds for this transition from the
            moment it starts. ``0.`` or a wildcard string means no limit. Defaults to
            ``0.``.
        timeout: Maximum cumulative time in seconds the policy will keep retrying this
            transition before giving up. ``0.`` means no timeout. Defaults to ``0.``.
        delay: Minimum time in seconds that must elapse in ``from_state`` before this
            transition is considered by the policy. ``0.`` means no delay. Defaults to
            ``0.``.

    Raises:
        ValueError: If an identical action (same name and args) already exists on the
            ``from_state`` -> ``to_state`` edge.
        FileNotFoundError: If ``to_state`` ends with ``".json"`` and the referenced
            file cannot be found on disk.

    Examples:
        >>> hsm.add_transit("idle", "working", action="start_task",
        ...                 args={"retries": 3}, msg="Starting task...")
        >>> # Embed a sub-machine from a JSON file:
        >>> hsm.add_transit("working", "sub_behaviours.json", action="enter_sub")
    """

    # Handling special state names
    if from_state == Custom.ALL_STATES_NAME:
        for from_state_obj in self.__id_to_state:
            if from_state_obj.name == from_state:
                continue

            # Adding a transition to the dest state
            self.add_transit(from_state=from_state_obj.name, to_state=to_state, action=action, args=args,
                             ready=ready, act_id=None, avoid_changing_ready=avoid_changing_ready,
                             msg=msg, teleport=teleport, high_priority=high_priority,
                             total_time=total_time, timeout=timeout, delay=delay)
        return

    # Plugging a previously loaded HSM
    if to_state.lower().endswith(".json"):
        if not os.path.exists(to_state):
            raise FileNotFoundError(f"Cannot find {to_state}")

        file_name = to_state
        hsm = HybridStateMachine(self.actionable).load(file_name)

        # First, we avoid name clashes, renaming already-used-state-names in original_name~1 (or ~2, or ~3, ...)
        hsm_states = list(hsm.states.keys())  # Keep the list(...) thing, since we need a copy here (it will change)
        for state in hsm_states:
            renamed_state = state
            i = 1
            while renamed_state in self.states or (i > 1 and renamed_state in hsm.states):
                renamed_state = state + "." + str(i)
                i += 1

            if hsm.initial_state == state:
                hsm.initial_state = renamed_state
            if hsm.prev_state == state:
                hsm.prev_state = renamed_state
            if hsm.state == state:
                hsm.state = renamed_state
            if hsm.limbo_state == state:
                hsm.limbo_state = renamed_state

            hsm.states[renamed_state] = hsm.states[state]
            if renamed_state != state:
                del hsm.states[state]
            hsm.transitions[renamed_state] = hsm.transitions[state]
            if renamed_state != state:
                del hsm.transitions[state]

            for to_states in hsm.transitions.values():
                if state in to_states:
                    to_states[renamed_state] = to_states[state]
                    if renamed_state != state:
                        del to_states[state]

        # Saving
        initial_state_was_set = self.initial_state is not None
        state_was_set = self.state is not None

        # Include actions/states from another HSM
        self.include(hsm)

        # Adding a transition to the initial state of the given HSM
        self.add_transit(from_state=from_state, to_state=hsm.initial_state, action=action, args=args,
                         ready=ready, act_id=None, msg=msg, teleport=teleport, high_priority=high_priority,
                         total_time=total_time, timeout=timeout, delay=delay)

        # Restoring
        self.initial_state = from_state if not initial_state_was_set else self.initial_state
        self.state = from_state if not state_was_set else self.state
        return

    # Adding a new transition
    if from_state not in self.transitions:
        if from_state not in self.states:
            self.add_state(from_state, action=None)
        self.transitions[from_state] = {}
    if to_state not in self.transitions:
        if to_state not in self.states:
            self.add_state(to_state, action=None)
        self.transitions[to_state] = {}
    if args is None:
        args = {}
    if act_id is None:
        act_id = len(self.__id_to_action)

    # Clearing
    if to_state not in self.transitions[from_state]:
        self.transitions[from_state][to_state] = []

    # Checking
    existing_action_list = self.transitions[from_state][to_state]
    for existing_action in existing_action_list:
        if existing_action.same_as(name=action, args=args):
            raise ValueError(f"Repeated transition from {from_state} to {to_state}: "
                             f"{existing_action.to_list()}")

    # Adding the new action
    new_action = Action(name=action, args=args, idx=act_id, actionable=self.actionable, ready=ready, msg=msg,
                        avoid_changing_ready=avoid_changing_ready,
                        teleport=teleport, high_priority=high_priority,
                        max_duration=total_time, retry_timeout=timeout, delay=delay)
    self.transitions[from_state][to_state].append(new_action)
    self.__id_to_action.append(new_action)

    new_action.set_state_machine(self)  # This will also share the same wildcards dictionary

include ¶

include(hsm: HybridStateMachine, make_a_copy: bool = False) -> None

Merge all states and transitions from another machine into this one.

This is the primary composition primitive for building complex machines from smaller, reusable components. The merge proceeds in three steps:

Wildcards from hsm are merged into this machine via add_wildcards.
Every state in hsm (including its inner state action and configuration) is added to this machine via add_state, using a fresh sequential ID.
Every transition in hsm is added via add_transit, with avoid_changing_ready=True so that the existing ready flags are preserved verbatim.

When make_a_copy is True, the runtime execution state of hsm (state, prev_state, initial_state, limbo_state, the welcome message, and the display flags) is also copied onto this machine, effectively turning this machine into a live duplicate of hsm.

This method is called internally by add_transit when to_state ends with ".json" to embed a sub-machine.

Parameters:

Name	Type	Description	Default
`hsm`	`HybridStateMachine`	The `HybridStateMachine` whose states and transitions are merged into this machine. It is not modified.	required
`make_a_copy`	`bool`	If `True`, also copy the runtime execution state of `hsm` (current state, initial state, etc.) onto this machine. Defaults to `False`.	`False`

Examples:

>>> base = HybridStateMachine(actionable=agent)
>>> base.add_transit("a", "b", action="do_ab")
>>> ext = HybridStateMachine(actionable=agent)
>>> ext.add_transit("b", "c", action="do_bc")
>>> base.include(ext)
>>> print(base.get_states())
['a', 'b', 'c']

Source code in unaiverse/hsm/hsm.py

def include(self, hsm: 'HybridStateMachine', make_a_copy: bool = False) -> None:
    """Merge all states and transitions from another machine into this one.

    This is the primary composition primitive for building complex machines from
    smaller, reusable components. The merge proceeds in three steps:

    1. Wildcards from ``hsm`` are merged into this machine via ``add_wildcards``.
    2. Every state in ``hsm`` (including its inner state action and configuration) is
       added to this machine via ``add_state``, using a fresh sequential ID.
    3. Every transition in ``hsm`` is added via ``add_transit``, with
       ``avoid_changing_ready=True`` so that the existing ``ready`` flags are
       preserved verbatim.

    When ``make_a_copy`` is ``True``, the runtime execution state of ``hsm``
    (``state``, ``prev_state``, ``initial_state``, ``limbo_state``, the welcome
    message, and the display flags) is also copied onto this machine, effectively
    turning this machine into a live duplicate of ``hsm``.

    This method is called internally by ``add_transit`` when ``to_state`` ends with
    ``".json"`` to embed a sub-machine.

    Args:
        hsm: The ``HybridStateMachine`` whose states and transitions are merged into
            this machine. It is not modified.
        make_a_copy: If ``True``, also copy the runtime execution state of ``hsm``
            (current state, initial state, etc.) onto this machine. Defaults to
            ``False``.

    Examples:
        >>> base = HybridStateMachine(actionable=agent)
        >>> base.add_transit("a", "b", action="do_ab")
        >>> ext = HybridStateMachine(actionable=agent)
        >>> ext.add_transit("b", "c", action="do_bc")
        >>> base.include(ext)
        >>> print(base.get_states())
        ['a', 'b', 'c']
    """

    # Copying wildcards
    self.add_wildcards(hsm.get_wildcards())

    # Adding states before adding transitions, so that we also add inner state actions, if any
    for _state in hsm.states.values():
        self.add_state(state=_state.name,
                       action=_state.action.name if _state.action is not None else None,
                       waiting_time=_state.waiting_time,
                       args=_state.action.args if _state.action is not None else None,
                       state_id=None,
                       blocking=_state.blocking,
                       msg=_state.msg_with_wildcards)

    # Copy all the transitions of the HSM
    for _from_state, _to_states in hsm.transitions.items():
        for _to_state, _action_list in _to_states.items():
            for _action in _action_list:
                total_time = 0.
                timeout = 0.
                delay = 0.
                if isinstance(_action.get_total_time(), str) or _action.get_total_time() > 0:
                    total_time = _action.get_total_time()
                if isinstance(_action.get_timeout(), str) or _action.get_timeout() > 0.:
                    timeout = _action.get_timeout()
                if isinstance(_action.get_delay(), str) or _action.get_delay() > 0.:
                    delay = _action.get_delay()
                self.add_transit(from_state=_from_state, to_state=_to_state, action=_action.name,
                                 args=_action.args, ready=_action.ready, teleport=_action.is_teleport(),
                                 high_priority=_action.is_high_priority,
                                 act_id=None, msg=_action.msg_with_wildcards, avoid_changing_ready=True,
                                 total_time=total_time, timeout=timeout, delay=delay)

    if make_a_copy:
        self.state = hsm.state
        self.prev_state = hsm.state
        self.initial_state = hsm.initial_state
        self.limbo_state = hsm.limbo_state
        self.set_welcome_message(hsm.welcome_msg_with_wildcards)
        self.show_blocking_states = hsm.show_blocking_states
        self.show_action_completion = hsm.show_action_completion
        self.show_action_request_info = hsm.show_action_request_info

must_wait ¶

must_wait() -> bool

Checks if the current state is in a waiting period before any transitions can occur.

Returns:

Type	Description
`bool`	A boolean indicating if the state machine must wait.

Source code in unaiverse/hsm/hsm.py

def must_wait(self) -> bool:
    """Checks if the current state is in a waiting period before any transitions can occur.

    Returns:
        A boolean indicating if the state machine must wait.
    """
    if self.state is not None:
        return self.states[self.state].must_wait()
    else:
        return False

is_enabled ¶

is_enabled() -> bool

A simple getter to check if the state machine is currently enabled to run.

Returns:

Type	Description
`bool`	True if the state machine is enabled, False otherwise.

Source code in unaiverse/hsm/hsm.py

def is_enabled(self) -> bool:
    """A simple getter to check if the state machine is currently enabled to run.

    Returns:
        True if the state machine is enabled, False otherwise.
    """
    return self.enabled

enable ¶

enable(yes_or_not: bool) -> None

Enables or disables the state machine. When disabled, the act_states and act_transitions methods will not perform any actions.

Parameters:

Name	Type	Description	Default
`yes_or_not`	`bool`	A boolean to enable (`True`) or disable (`False`) the state machine.	required

Source code in unaiverse/hsm/hsm.py

def enable(self, yes_or_not: bool) -> None:
    """Enables or disables the state machine. When disabled, the `act_states` and `act_transitions` methods will
    not perform any actions.

    Args:
        yes_or_not: A boolean to enable (`True`) or disable (`False`) the state machine.
    """
    self.enabled = yes_or_not

act_states `async` ¶

act_states() -> None

Executes the inner action of the current state, if one exists. This method is for actions that occur upon entering a state but do not cause an immediate transition. It only runs if the state machine is enabled (async).

Source code in unaiverse/hsm/hsm.py

async def act_states(self) -> None:
    """Executes the inner action of the current state, if one exists. This method is for actions that occur upon
    entering a state but do not cause an immediate transition. It only runs if the state machine is enabled (async).
    """
    if not self.enabled:
        return

    if self.state is not None:  # When in the middle of an action, the state is Nones
        await self.states[self.state]()  # Run the action (if any)

act_ghost_transition `async` ¶

act_ghost_transition(to_state: str) -> int

Source code in unaiverse/hsm/hsm.py

async def act_ghost_transition(self, to_state: str) -> int:

    # Forcing to go back to current state if in the middle of a multistep action
    from_state = self.state
    if from_state is None:
        from_state = self.limbo_state
    self.set_state(from_state)  # This will also reset the current interaction (if any)

    # Setting up the ghost action (nop) as only possible action from the current state
    original_transitions = self.transitions[from_state]
    self.transitions[from_state] = \
        {to_state: [Action(name="nop", args={}, actionable=self.actionable, ready=True, teleport=True)]}

    # Acting the ghost action (nop)
    ret = await self.act_transitions()

    # Restoring original transitions
    self.transitions[from_state] = original_transitions
    return ret

get_time_spent_in_current_state ¶

get_time_spent_in_current_state()

Source code in unaiverse/hsm/hsm.py

def get_time_spent_in_current_state(self):
    state = self.state
    if state is None:
        state = self.limbo_state
    if state is None:
        return -1.
    else:
        return self.states[self.state].get_time_passed()

act_transitions `async` ¶

act_transitions(only_the_ones_with_interactions: bool = False) -> int

This is the core execution loop for transitions. It finds all feasible actions from the current state and, using a policy, selects and executes one. It handles single-step and multistep actions, managing state changes, timeouts, and failed executions. It returns an integer status code indicating the outcome (e.g., transition done, try again, move to next action) (async).

Parameters:

Name	Type	Description	Default
`only_the_ones_with_interactions`	`bool`	A boolean to consider only actions that have pending interactions.	`False`

Returns:

Type	Description
`int`	An integer status code: `0` for a successful transition, `1` to retry the same action, `2` to move to the
`int`	next action, or `-1` if no actions were found.

Source code in unaiverse/hsm/hsm.py

async def act_transitions(self, only_the_ones_with_interactions: bool = False) -> int:
    """This is the core execution loop for transitions. It finds all feasible actions from the current state and,
    using a policy, selects and executes one. It handles single-step and multistep actions, managing state changes,
    timeouts, and failed executions. It returns an integer status code indicating the outcome (e.g., transition
    done, try again, move to next action) (async).

    Args:
        only_the_ones_with_interactions: A boolean to consider only actions that have pending interactions.

    Returns:
        An integer status code: `0` for a successful transition, `1` to retry the same action, `2` to move to the
        next action, or `-1` if no actions were found.
    """
    if not self.enabled:
        return -1

    # Collecting list of feasible actions, wait flags, etc. (from the current state)
    if self.__cur_feasible_actions_status is None:
        if self.state is None:
            return -1

        actions_list = []
        to_state_list = []
        attempts_to_serve_an_interaction_list = []

        for to_state, action_list in self.transitions[self.state].items():
            for i, action in enumerate(action_list):

                # Checking is_ready will check if streams are ready and if the interaction was completed meanwhile
                if (action.is_ready(consider_interactions=True,
                                    delay_starting_time=self.states[self.state].starting_time) and
                        (not only_the_ones_with_interactions or
                         len(action.interactions.get_interactions(doable_only=True)) > 0)):
                    actions_list.append(action)
                    to_state_list.append(to_state)
                    attempts_to_serve_an_interaction_list.append(0)

        if len(actions_list) > 0:
            self.__cur_feasible_actions_status = {
                'actions_list': actions_list,
                'to_state_list': to_state_list,
                'selected_idx': 0,
                'selected_interaction': None,
                'attempts_to_serve_an_interaction_list': attempts_to_serve_an_interaction_list
            }
    else:

        # Reloading the already computed set of actions, wait flags, etc. (when in the middle of an action)
        actions_list = self.__cur_feasible_actions_status['actions_list']
        to_state_list = self.__cur_feasible_actions_status['to_state_list']
        attempts_to_serve_an_interaction_list = (
            self.__cur_feasible_actions_status)['attempts_to_serve_an_interaction_list']

    # Using the selected policy to decide what action to apply
    while len(actions_list) > 0:
        skip_action_due_to_filter = False

        # It there was an already selected action (for example a multistep action), then continue with it,
        # otherwise, select a new one following a certain policy (actually, first-come first-served)
        if self.__action is None:
            log.statem(f"List of actions to choose from:\n   " +
                       "\n   ".join([a.to_code_str().replace("\n", "\n   ")
                                     for a in actions_list]), state=self.get_state_name())

            # Naive policy: take the first action that is ready
            _idx, _interaction = self.policy(actions_list)

            if _idx < 0:
                log.statem(f"Selected no actions", state=self.get_state_name())

                # No actions were applied
                self.__cur_feasible_actions_status = None
                self.__state_changed = False
                return -1  # Early stop
            else:
                if _interaction is not None:
                    log.statem(f"Selected {actions_list[_idx].to_code_str()}, "
                               f"{_interaction.to_code_str(True)}",
                               state=self.get_state_name())
                else:
                    log.statem(f"Selected {actions_list[_idx].to_code_str()}, no-interactions",
                               state=self.get_state_name())

            # Revisiting decisions due to the policy filter
            if self.policy_filter is not None:
                try:
                    _idx_f, _interactions_f = self.policy_filter(_idx, _interaction,
                                                                 actions_list, self.policy_filter_opts)

                    # Keep this part of code: needed to stabilize potential object duplicates (JS)
                    if _interactions_f != _interaction and _interactions_f is not None:
                        _interactions_f = self.actionable.im.get_interaction(uuid=_interactions_f.uuid)
                except Exception as e:
                    log.error(f"Skipping policy filter due to exception: {e}",
                              state=self.get_state_name())
                    _idx_f = _idx
                    _interactions_f = _interaction

                if _idx_f != _idx or _interactions_f != _interaction:
                    if _idx_f < 0:
                        log.statem(f"Filter selected no actions among {[a.name for a in actions_list]}")

                        # No actions were applied
                        # self.__cur_feasible_actions_status = None
                        # self.__state_changed = False
                        # return -1  # Early stop
                        skip_action_due_to_filter = True
                    else:
                        _idx = _idx_f
                        _interaction = _interactions_f

                        if _interaction is not None:
                            log.statem(
                                f"Filter selected {actions_list[_idx].to_code_str()}, {_interaction.__str__()}",
                                state=self.get_state_name())
                        else:
                            log.statem(f"Filter selected {actions_list[_idx].to_code_str()}, "
                                       f"no-interactions", state=self.get_state_name())
                else:
                    log.statem(f"Filter confirmed the selection", state=self.get_state_name())

            # Saving current action
            self.limbo_state = self.state
            self.state = None
            self.__action = actions_list[_idx]

            # Forcing system interaction if no external interactions were selected
            if _interaction is None:
                _interaction = self.__action.system_interaction

            _interaction.reset_state()  # Resetting
            self.__cur_feasible_actions_status['selected_idx'] = _idx
            self.__cur_feasible_actions_status['selected_interaction'] = _interaction

        # References
        action = self.__action
        idx = self.__cur_feasible_actions_status['selected_idx']
        interaction = self.__cur_feasible_actions_status['selected_interaction']

        # If this action has an associated Interaction, set it as current on the IM
        # (this configures stdin/stdout for the action)
        # If the Interaction is (1) None, or (2) a system interaction with no streams at all, or
        # (3) a received interaction with no streams, then the stdin/stdout/... will be set back to default streams
        if self.actionable.im.current is None or self.actionable.im.current != interaction:
            self.actionable.im.set_current(interaction)

        log.statem(f">>> ACTION {self.__action.name}...", state=self.get_state_name(True))
        if len(self.__action.get_list_of_interactions()) > 0:
            log.statem(str(self.__action.get_list_of_interactions()))

        # Calling action, getting back status.
        # Status can be one of these:
        # 0: action fully done;
        # 1: try again this action;
        # 2: failed, move to another action.
        status = await action(interaction=interaction) if not skip_action_due_to_filter else 2

        if status == 0:
            log.statem(f"+++ ACTION {self.__action.name} correctly completed", state=self.get_state_name(True))
        elif status == 1:
            log.statem(f"~~~ ACTION {self.__action.name} will be run again", state=self.get_state_name(True))
        else:
            log.statem(f"--- ACTION {self.__action.name} failed", state=self.get_state_name(True))

        if action.msg is not None and self.show_action_completion:
            log.user(Custom.ACTION_TICKS_PER_STATUS[status])

        # Post-call operations
        if status == 0:  # Done

            # Memorizing tags of data used in this action call
            interaction.record_data_tags()

            # State transition
            self.prev_state = self.limbo_state
            self.state = to_state_list[idx]
            self.limbo_state = None

            # Complete the associated Interaction (if any) via Interaction Manager (IM)
            if interaction is not None:
                # The IM on the actionable (agent) will handle the completion, also saving the destination state
                await self.actionable.im.complete_current(self.state, CompletionReason.OK)

            # Update status
            self.__state_changed = self.state != self.prev_state  # Checking if we are on a self-loop or not
            self.__last_completed_action = self.__action  # This will be set also if the state does not change

            # If we moved to another state
            # (this is not true anymore: "clearing all the pending annotations for the next possible actions")
            if self.__state_changed:
                log.statem(f">>> MOVING TO STATE: {self.state}", state=self.get_state_name())
                # for to_state, action_list in self.transitions[self.state].items():
                #    for i, act in enumerate(action_list):
                #        act.clear_requests()

                # Propagating (trying to propagate forward the residual requests)
                list_of_residual_interactions = self.__action.get_list_of_interactions()
                propagated_requests = []
                for interaction in list_of_residual_interactions:
                    interaction.from_state = None
                    interaction.to_state = None
                    if self.request_action(interaction):
                        _interaction = Interaction()
                        propagated_requests.append(_interaction.from_dict(interaction.to_dict()))
                for _interaction in propagated_requests:
                    list_of_residual_interactions.remove(_interaction)  # Clearing propagated requests

                # if len(propagated_requests) > 0:
                #    print(f"Reached state {self.state}, "
                #          f"propagated these requests taken from {self.__action.name}, and
                #          now starting from here {propagated_requests}")

            self.states[self.prev_state].reset(self.__state_changed)  # Reset starting time
            interaction.reset_state()
            self.__action = None  # Clearing
            self.__cur_feasible_actions_status = None

            return 0  # Transition done, no need to check other actions!

        elif status == 1:  # Try again the same action (either a new step or an already done-and-failed one)

            # Memorizing tags of data used in this action call
            interaction.record_data_tags()

            # Update status
            self.__state_changed = False
            # if self.prev_state is not None:
            #    self.states[self.prev_state].reset()  # Reset starting time

            return 1  # Transition not-done: no need to check other actions, the current one will be run again

        elif status == 2:  # Move to the next action (or to the next request of the same action)

            # Clearing request
            if interaction is not None:
                self.__action.interactions.move_interaction_to_back(interaction)  # Rotating to avoid starvation
                attempts_to_serve_an_interaction_list[idx] += 1

            # Back to the original state
            self.state = self.limbo_state
            self.limbo_state = None

            # Purging action from the current list
            if interaction is None or attempts_to_serve_an_interaction_list[idx] >= len(self.__action.interactions):
                del actions_list[idx]
                del to_state_list[idx]
                del attempts_to_serve_an_interaction_list[idx]

            # Update status
            self.__state_changed = False
            interaction.reset_state()
            self.actionable.im.set_current_as_paused()
            self.__action = None  # Clearing

            continue  # Move to the next action
        else:
            raise ValueError("Unexpected status: " + str(status))

    # No actions were applied
    self.__cur_feasible_actions_status = None
    self.__state_changed = False
    return -1

act `async` ¶

act() -> bool

A high-level method that combines act_states and act_transitions to run the state machine. It repeatedly processes states and transitions until a blocking state is reached or all feasible actions have been tried, thus ensuring a complete processing cycle in one call (async).

Returns:

Type	Description
`bool`	True if, during this whole 'act', the state changed at least once.

Source code in unaiverse/hsm/hsm.py

async def act(self) -> bool:
    """A high-level method that combines `act_states` and `act_transitions` to run the state machine. It repeatedly
    processes states and transitions until a blocking state is reached or all feasible actions have been tried,
    thus ensuring a complete processing cycle in one call (async).

    Returns:
        True if, during this whole 'act', the state changed at least once.
    """

    # It keeps processing states and actions, until all the current feasible actions fail
    # (also when a step of a multistep action is executed) or a blocking state is reached
    changed_state = False
    starting_state = self.state
    while True:
        if self.welcome_msg is not None and self.state is not None and self.state == self.initial_state:
            log.user(self.welcome_msg)
            self.set_welcome_message(None)

        await self.act_states()
        ret = await self.act_transitions(self.must_wait())
        if self.state != starting_state:
            changed_state = True
        if ret != 0 or (self.state is not None and self.states[self.state].blocking):
            break
    return changed_state

get_state_changed ¶

get_state_changed() -> bool

Returns an internal flag that indicates if a state transition has occurred in the last execution cycle. This can be used by an external loop to know when to re-evaluate the state machine's context.

Returns:

Type	Description
`bool`	True if the state has changed, False otherwise.

Source code in unaiverse/hsm/hsm.py

def get_state_changed(self) -> bool:
    """Returns an internal flag that indicates if a state transition has occurred in the last execution cycle.
    This can be used by an external loop to know when to re-evaluate the state machine's context.

    Returns:
        True if the state has changed, False otherwise.
    """
    return self.__state_changed

transition_exists ¶

transition_exists(name: str, only_ready: bool = True) -> bool

Source code in unaiverse/hsm/hsm.py

def transition_exists(self, name: str, only_ready: bool = True) -> bool:
    state = self.state if self.state is not None else self.limbo_state
    trans = self.transitions[state]
    for _, action_list in trans.items():
        for action in action_list:
            if (not only_ready or action.is_ready()) and action.name == name:
                return True
    return False

request_action ¶

request_action(interaction: Interaction | None = None, **kwargs) -> bool

Allows an external entity to request a specific action. The request is validated by a signature checker (if one exists) and then queued on the corresponding action. This method enables dynamic, external triggers for state machine transitions.

Parameters:

Name	Type	Description	Default
`interaction`	`Interaction \| None`	The interaction object.	`None`

Returns:

Type	Description
`bool`	True if the request was accepted and queued, False otherwise.

Source code in unaiverse/hsm/hsm.py

def request_action(self, interaction: Interaction | None = None, **kwargs) -> bool:
    """Allows an external entity to request a specific action. The request is validated by a signature checker
    (if one exists) and then queued on the corresponding action. This method enables dynamic, external triggers for
    state machine transitions.

    Args:
        interaction: The interaction object.

    Returns:
        True if the request was accepted and queued, False otherwise.
    """
    log.misc(f"Received an action request with this interaction: {interaction}", state=self.get_state_name())

    # Getting data
    action_name = interaction.action_name
    args = interaction.action_kwargs

    # If state is not provided, the current state is assumed
    from_state = interaction.from_state
    if from_state is None:
        # If the request arrives in the middle of a multistep action, we need to check limbo state
        from_state = self.state if self.state is not None else self.limbo_state
    if from_state not in self.transitions:
        log.error(f"Request not accepted: not valid source state ({from_state})",
                  state=self.get_state_name())
        return False

    # If the destination state is not provided, all the possible destination from the current state are considered
    to_state = interaction.to_state
    if to_state is not None and to_state not in self.transitions[from_state]:
        log.error(f"Request not accepted: not valid destination state ({to_state})",
                  state=self.get_state_name())
        return False
    to_states = self.transitions[from_state].keys() if to_state is None else [to_state]

    for to_state in to_states:
        action_list = self.transitions[from_state][to_state]
        for i, action in enumerate(action_list):
            if action.same_as(name=action_name, args=args):
                log.misc(f"Requested action ({action_name}) found in state {from_state}, "
                         f"adding interaction to the queue",
                         state=self.get_state_name())

                # Action found, let's save the suggestion
                if action.add_interaction(interaction):
                    return True
                else:
                    log.error("Requested action does not allow interactions")
                    return False  # If the action does not support interactions

    # If the action was not found
    log.error(f"Requested action ({action_name}) not found "
              f"(from_state={from_state}, cur_state={self.get_state_name()})",
              state=self.get_state_name())
    return False

wait_for_all_actions_that_start_with ¶

wait_for_all_actions_that_start_with(prefix: str) -> None

Sets the ready flag to False for all actions whose name begins with a given prefix. This method is used to programmatically disable a group of actions, effectively pausing them.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	The string prefix to match against action names.	required

Source code in unaiverse/hsm/hsm.py

def wait_for_all_actions_that_start_with(self, prefix: str) -> None:
    """Sets the `ready` flag to `False` for all actions whose name begins with a given prefix. This method is used
    to programmatically disable a group of actions, effectively pausing them.

    Args:
        prefix: The string prefix to match against action names.
    """
    for state, to_states in self.transitions.items():
        for to_state, action_list in to_states.items():
            for i, action in enumerate(action_list):
                if action.name.startswith(prefix):
                    action.set_as_not_ready()

wait_for_all_actions_that_include_an_arg ¶

wait_for_all_actions_that_include_an_arg(arg_name: str) -> None

Sets the ready flag to False for all actions that include a specific argument name in their signature. This provides another way to programmatically disable actions.

Parameters:

Name	Type	Description	Default
`arg_name`	`str`	The name of the argument to look for.	required

Source code in unaiverse/hsm/hsm.py

def wait_for_all_actions_that_include_an_arg(self, arg_name: str) -> None:
    """Sets the `ready` flag to `False` for all actions that include a specific argument name in their signature.
    This provides another way to programmatically disable actions.

    Args:
        arg_name: The name of the argument to look for.
    """
    for state, to_states in self.transitions.items():
        for to_state, action_list in to_states.items():
            for i, action in enumerate(action_list):
                if arg_name in action.args:
                    action.set_as_not_ready()

save ¶

save(filename: str, only_if_changed: object | None = None) -> bool

Saves the state machine's current configuration to a JSON file. It can optionally check if the configuration has changed before saving to avoid redundant file writes.

Parameters:

Name	Type	Description	Default
`filename`	`str`	The path to the file to save to.	required
`only_if_changed`	`object \| None`	An optional object to compare against for changes. If a change is not detected, the file is not written.	`None`

Returns:

Type	Description
`bool`	True if the file was written, False otherwise.

Source code in unaiverse/hsm/hsm.py

def save(self, filename: str, only_if_changed: object | None = None) -> bool:
    """Saves the state machine's current configuration to a JSON file. It can optionally check if the configuration
    has changed before saving to avoid redundant file writes.

    Args:
        filename: The path to the file to save to.
        only_if_changed: An optional object to compare against for changes. If a change is not detected, the file
            is not written.

    Returns:
        True if the file was written, False otherwise.
    """
    if only_if_changed is not None and os.path.exists(filename):
        try:
            existing = HybridStateMachine(actionable=only_if_changed).load(filename)
            if str(existing) == str(self):
                return False
        except Exception as e:  # If load fails, we assume it changed
            log.error(f"Error while reloading the exising machine from {filename}, "
                      f"assuming it changed: {e}")

    with open(filename, 'w', encoding='utf-8') as file:
        file.write(str(self))
    return True

load ¶

load(filename_or_hsm_as_string: str | TextIOWrapper | IO[str]) -> HybridStateMachine

Loads a state machine's configuration from a JSON file or a JSON string. It reconstructs the states, actions, and transitions from the serialized data. This method is critical for persistence and for loading pre-defined state machine models.

Parameters:

Name	Type	Description	Default
`filename_or_hsm_as_string`	`str \| TextIOWrapper \| IO[str]`	The path to the JSON file or a JSON string representation of the state machine.	required

Returns:

Type	Description
`HybridStateMachine`	The loaded `HybridStateMachine` object (self).

Source code in unaiverse/hsm/hsm.py

def load(self, filename_or_hsm_as_string: str | io.TextIOWrapper | IO[str]) -> 'HybridStateMachine':
    """Loads a state machine's configuration from a JSON file or a JSON string. It reconstructs the states,
    actions, and transitions from the serialized data. This method is critical for persistence and for loading
    pre-defined state machine models.

    Args:
        filename_or_hsm_as_string: The path to the JSON file or a JSON string representation of the state machine.

    Returns:
        The loaded `HybridStateMachine` object (self).
    """

    # Loading the whole file
    if (isinstance(filename_or_hsm_as_string, importlib.resources.abc.Traversable) or
            isinstance(filename_or_hsm_as_string, io.TextIOWrapper)):

        # Safe way to load when this file is packed in a pip package
        hsm_data = json.load(filename_or_hsm_as_string)
    else:

        # Ordinary case
        if os.path.exists(filename_or_hsm_as_string) and os.path.isfile(filename_or_hsm_as_string):
            with open(filename_or_hsm_as_string, 'r', encoding="utf-8") as file:
                hsm_data = json.load(file)
        else:

            # Assuming it is a string
            hsm_data = json.loads(filename_or_hsm_as_string)

    # Backward compatibility
    if "machine" not in hsm_data:
        return self.load_backward_compat(hsm_data)

    # Getting state info
    try:
        self.initial_state = hsm_data['machine']['initial_state']
        self.state = self.initial_state
        self.prev_state = None
        self.limbo_state = None
        self.set_role(hsm_data['machine']['role'])
        self.set_welcome_message(hsm_data['machine'].get('msg', None))
        self.show_blocking_states = (
            hsm_data['machine']['options'].get('highlight_blocking_states_in_messages', False))
        self.show_action_completion = (
            hsm_data['machine']['options'].get('show_action_ticks_after_messages', False))
        self.show_action_request_info = (
            hsm_data['machine']['options'].get('show_action_request_after_messages', False))
    except Exception as e:
        log.critical(f"Invalid JSON data format when loading the state machine [{e}]")

    self.states = {}
    self.transitions = {}

    # Getting states
    if 'states' not in hsm_data:
        log.critical(f"Invalid JSON data format when loading the state machine (missing state list)")

    for state, state_action in hsm_data['states'].items():
        act_name = state_action.get("action", None)
        act_args = state_action.get("action_kwargs", {})
        blocking = state_action.get("blocking", True)
        msg = state_action.get("msg", None)
        waiting_time = 0.
        for k in Custom.TIME_TO_WAIT_BEFORE_ACTING_ARG_NAMES:
            if k in state_action:
                waiting_time = state_action[k]
                break

        self.add_state(state, action=act_name, args=act_args,
                       waiting_time=waiting_time, blocking=blocking, msg=msg)

    # Getting ordinary transitions (before teleports, to ensure higher priority)
    for from_state, list_of_to_state_dicts in hsm_data['transitions'].items():
        for to_state_dict in list_of_to_state_dicts:
            to_state = to_state_dict["goto"]
            action_dict = to_state_dict["on"]

            act_name = action_dict["action"]
            act_args = action_dict.get("action_kwargs", {})
            msg = action_dict.get("msg", None)
            act_ready = action_dict.get("ready", True)
            high_priority = action_dict.get("high_priority", False)
            total_time = 0.
            for k in Custom.SECONDS_ARG_NAMES:
                if k in action_dict:
                    total_time = action_dict[k]
                    break
            timeout = 0.
            for k in Custom.TIMEOUT_ARG_NAMES:
                if k in action_dict:
                    timeout = action_dict[k]
                    break
            delay = 0.
            for k in Custom.DELAY_ARG_NAMES:
                if k in action_dict:
                    delay = action_dict[k]
                    break

            self.add_transit(from_state, to_state,
                             action=act_name, args=act_args, ready=act_ready, msg=msg,
                             avoid_changing_ready="ready" in action_dict, total_time=total_time,
                             high_priority=high_priority,
                             timeout=timeout, delay=delay)

    # Getting teleports (do it after getting ordinary transitions, so teleports will have lower priority)
    if 'teleports' in hsm_data:

        # Replicating the "all" teleport over all source states
        if Custom.ALL_STATES_NAME in hsm_data['teleports']:
            list_of_to_state_dicts = hsm_data['teleports'][Custom.ALL_STATES_NAME]

            for to_state_dict in list_of_to_state_dicts:
                to_state = to_state_dict["goto"]
                action_dict = to_state_dict["on"]

                act_name = action_dict["action"]
                act_args = action_dict.get("action_kwargs", {})
                msg = action_dict.get("msg", None)
                act_ready = action_dict.get("ready", True)
                high_priority = action_dict.get("high_priority", False)
                total_time = 0.
                for k in Custom.SECONDS_ARG_NAMES:
                    if k in action_dict:
                        total_time = action_dict[k]
                        break
                timeout = 0.
                for k in Custom.TIMEOUT_ARG_NAMES:
                    if k in action_dict:
                        timeout = action_dict[k]
                        break
                delay = 0.
                for k in Custom.DELAY_ARG_NAMES:
                    if k in action_dict:
                        delay = action_dict[k]
                        break

                # Looping over all states
                for from_state_obj in self.__id_to_state:
                    from_state = from_state_obj.name

                    # Skipping the destination state
                    if from_state == to_state:
                        continue

                    self.add_transit(from_state, to_state,
                                     action=act_name, args=act_args, ready=act_ready, msg=msg,
                                     avoid_changing_ready="ready" in action_dict, teleport=True,
                                     high_priority=high_priority,
                                     total_time=total_time, timeout=timeout, delay=delay)

        # Handling all the ordinary teleports
        for from_state, list_of_to_state_dicts in hsm_data['teleports'].items():

            # Skipping "all"-like teleport (already handled)
            if from_state == Custom.ALL_STATES_NAME:
                continue

            for to_state_dict in list_of_to_state_dicts:
                to_state = to_state_dict["goto"]
                action_dict = to_state_dict["on"]

                act_name = action_dict["action"]
                act_args = action_dict.get("action_kwargs", {})
                msg = action_dict.get("msg", None)
                act_ready = action_dict.get("ready", True)
                high_priority = action_dict.get("high_priority", False)
                total_time = 0.
                for k in Custom.SECONDS_ARG_NAMES:
                    if k in action_dict:
                        total_time = action_dict[k]
                timeout = 0.
                for k in Custom.TIMEOUT_ARG_NAMES:
                    if k in action_dict:
                        timeout = action_dict[k]
                delay = 0.
                for k in Custom.DELAY_ARG_NAMES:
                    if k in action_dict:
                        delay = action_dict[k]

                self.add_transit(from_state, to_state,
                                 action=act_name, args=act_args, ready=act_ready, msg=msg,
                                 avoid_changing_ready="ready" in action_dict, teleport=True,
                                 high_priority=high_priority,
                                 total_time=total_time, timeout=timeout, delay=delay)

    return self

load_backward_compat ¶

load_backward_compat(hsm_data: dict) -> HybridStateMachine

Loads a state machine's configuration from a JSON file or a JSON string. It reconstructs the states, actions, and transitions from the serialized data. This method is critical for persistence and for loading pre-defined state machine models.

Parameters:

Name	Type	Description	Default
`hsm_data`	`dict`	The dict with an HSM loaded from a JSON file.	required

Returns:

Type	Description
`HybridStateMachine`	The loaded `HybridStateMachine` object (self).

Source code in unaiverse/hsm/hsm.py

def load_backward_compat(self, hsm_data: dict) -> 'HybridStateMachine':
    """Loads a state machine's configuration from a JSON file or a JSON string. It reconstructs the states,
    actions, and transitions from the serialized data. This method is critical for persistence and for loading
    pre-defined state machine models.

    Args:
        hsm_data: The dict with an HSM loaded from a JSON file.

    Returns:
        The loaded `HybridStateMachine` object (self).
    """

    # Getting state info
    self.initial_state = hsm_data['initial_state']
    self.state = hsm_data['state']
    self.prev_state = hsm_data['prev_state']
    self.limbo_state = hsm_data['limbo_state']
    self.set_role(hsm_data.get('role', 'unknown'))
    self.set_welcome_message(hsm_data.get('welcome_msg', None))
    self.show_blocking_states = hsm_data.get('highlight_blocking_states_in_messages', False)
    self.show_action_completion = hsm_data.get('show_action_ticks_after_messages', False)
    self.show_action_request_info = hsm_data.get('show_action_request_after_messages', False)

    # Getting states
    self.states = {}
    if 'state_actions' in hsm_data:
        for state, state_action_list in hsm_data['state_actions'].items():
            if len(state_action_list) == 3:  # Backward compatibility
                act_name, act_args, state_id = state_action_list
                waiting_time = 0.
                blocking = True
                msg = None
            elif len(state_action_list) == 4:  # Backward compatibility
                act_name, act_args, state_id, waiting_time_or_blocking = state_action_list
                if isinstance(waiting_time_or_blocking, bool):
                    waiting_time = 0.
                    blocking = waiting_time_or_blocking
                else:
                    waiting_time = waiting_time_or_blocking
                    blocking = True
                msg = None
            elif len(state_action_list) == 5:  # Backward compatibility
                act_name, act_args, state_id, blocking, waiting_time = state_action_list
                msg = None
            else:
                act_name, act_args, state_id, blocking, waiting_time, msg = state_action_list

            # Recall that state_id can be set to -1 in the original file, meaning "automatically set the state_id"
            self.add_state(state, action=act_name, args=act_args,
                           state_id=state_id if state_id >= 0 else None,
                           waiting_time=waiting_time, blocking=blocking, msg=msg)

    # Getting transitions
    self.transitions = {}
    for from_state, to_states in hsm_data['transitions'].items():
        for to_state, action_list in to_states.items():
            for action_list_tuple in action_list:
                if len(action_list_tuple) == 4:
                    act_name, act_args, act_ready, act_id = action_list_tuple
                    msg = None
                else:
                    act_name, act_args, act_ready, act_id, msg = action_list_tuple

                # Recall that act_id can be set to -1 in the original file, meaning "automatically set the act_id"
                self.add_transit(from_state, to_state,
                                 action=act_name, args=act_args, ready=act_ready,
                                 act_id=act_id if act_id >= 0 else None, msg=msg,
                                 avoid_changing_ready=True)
    return self

to_graphviz ¶

to_graphviz() -> Digraph

Generates a Graphviz Digraph object representing the state machine's structure. This method visualizes states as nodes and transitions as edges. It includes details such as node shapes (diamond for initial state, oval for others), styles (filled for blocking states), and labels for both states and transitions. The labels for actions include their names and arguments, formatted to wrap lines for readability.

Returns:

Type	Description
`Digraph`	A `graphviz.Digraph` object ready for rendering.

Source code in unaiverse/hsm/hsm.py

def to_graphviz(self) -> graphviz.Digraph:
    """Generates a Graphviz `Digraph` object representing the state machine's structure. This method visualizes
    states as nodes and transitions as edges. It includes details such as node shapes (diamond for initial state,
    oval for others), styles (filled for blocking states), and labels for both states and transitions. The labels
    for actions include their names and arguments, formatted to wrap lines for readability.

    Returns:
        A `graphviz.Digraph` object ready for rendering.
    """
    graph = graphviz.Digraph()
    graph.attr('node', fontsize='8')
    for state, state_obj in self.states.items():
        action = state_obj.action
        if action is not None:
            s = "("
            for i, (k, v) in enumerate(action.args.items()):
                s += str(k) + "=" + (str(v) if not isinstance(v, str) else ("'" + v + "'"))
                if i < len(action.args) - 1:
                    s += ", "
            s += ")"
            label = action.name + s
            if len(label) > 40:
                tokens = label.split(" ")
                z = ""
                i = 0
                done = False
                while i < len(tokens):
                    z += (" " if i > 0 else "") + tokens[i]
                    if not done and i < (len(tokens) - 1) and len(z + tokens[i + 1]) > 40:
                        z += "\n    "
                        done = True
                    i += 1
                label = z
            suffix = "\n" + label
        else:
            suffix = ""
        if state == self.initial_state:
            graph.attr('node', shape='diamond')
        else:
            graph.attr('node', shape='oval')
        if self.states[state].blocking:
            graph.attr('node', style='filled')
        else:
            graph.attr('node', style='solid')
        graph.node(state, state + suffix, _attributes={'id': "node" + str(state_obj.id)})

    for from_state, to_states in self.transitions.items():
        j = 1
        for to_state, action_list in to_states.items():
            for action in action_list:
                args = action.args
                s = "("
                for i, (k, v) in enumerate(args.items()):
                    s += str(k) + "=" + (str(v) if not isinstance(v, str) else ("'" + str(v) + "'"))
                    if i < len(args) - 1:
                        s += ", "
                s += ")"

                special_args = {}
                if isinstance(action.get_total_time(), str) or action.get_total_time() > 0:
                    total_time = action.get_total_time()
                    special_args[Custom.SECONDS_ARG_NAMES[0]] = total_time
                if isinstance(action.get_timeout(), str) or action.get_timeout() > 0.:
                    timeout = action.get_timeout()
                    special_args[Custom.TIMEOUT_ARG_NAMES[0]] = timeout
                if isinstance(action.get_delay(), str) or action.get_delay() > 0.:
                    delay = action.get_delay()
                    special_args[Custom.DELAY_ARG_NAMES[0]] = delay

                if len(special_args) > 0:
                    s += "\n["
                    for i, (k, v) in enumerate(special_args.items()):
                        s += str(k) + "=" + (str(v) if not isinstance(v, str) else ("'" + str(v) + "'"))
                        if i < len(special_args) - 1:
                            s += ", "
                    s += "]"

                label = str(j) + "." + action.name + s
                j += 1
                if len(label) > 40:
                    tokens = label.split(" ")
                    z = ""
                    i = 0
                    done = False
                    while i < len(tokens):
                        z += (" " if i > 0 else "") + tokens[i]
                        if not done and i < (len(tokens) - 1) and len(z + tokens[i + 1]) > 40:
                            z += "\n"
                            done = True
                        i += 1
                    label = z
                edge_color = "#00000050" if action.is_teleport() else "#000000"
                graph.edge(from_state, to_state, label=" " + label + " ", fontsize='8',
                           style='dashed' if not action.is_ready() else 'solid',
                           color=edge_color,
                           fontcolor=edge_color,
                           _attributes={'id': "edge" + str(action.id)})
    return graph

save_pdf ¶

save_pdf(filename: str) -> bool

Saves the state machine's Graphviz representation as a PDF file. It calls to_graphviz() to create the graph and then uses the Graphviz library's render method to generate the PDF.

Parameters:

Name	Type	Description	Default
`filename`	`str`	The path and name of the PDF file to save.	required

Returns:

Type	Description
`bool`	True if the file was successfully saved, False otherwise.

Source code in unaiverse/hsm/hsm.py

def save_pdf(self, filename: str) -> bool:
    """Saves the state machine's Graphviz representation as a PDF file. It calls `to_graphviz()` to create the
    graph and then uses the Graphviz library's `render` method to generate the PDF.

    Args:
        filename: The path and name of the PDF file to save.

    Returns:
        True if the file was successfully saved, False otherwise.
    """
    if filename.lower().endswith(".pdf"):
        filename = filename[0:-4]

    try:
        self.to_graphviz().render(filename, format='pdf', cleanup=True)
        return True
    except Exception as e:
        log.error(f"Error while saving to PDF {e}")
        return False

print_actions ¶

print_actions(state: str | None = None) -> None

Prints a list of all transitions and their associated actions from a given state. If no state is provided, it defaults to the current state. This method is useful for quickly inspecting the available transitions from a specific point in the state machine's flow.

Parameters:

Name	Type	Description	Default
`state`	`str \| None`	The name of the state from which to print actions. Defaults to the current state.	`None`

Source code in unaiverse/hsm/hsm.py

def print_actions(self, state: str | None = None) -> None:
    """Prints a list of all transitions and their associated actions from a given state. If no state is provided,
    it defaults to the current state. This method is useful for quickly inspecting the available transitions from
    a specific point in the state machine's flow.

    Args:
        state: The name of the state from which to print actions. Defaults to the current state.
    """
    state = (self.state if self.state is not None else self.limbo_state) if state is None else state
    for to_state, action_list in self.transitions[state].items():
        if action_list is None or len(action_list) == 0:
            log.user(f"{state}, no actions")
        for action in action_list:
            log.user(f"{state} --> {to_state} {action}")

unaiverse.hsm.hsm

hsm ¶

HybridStateMachine ¶

initial_state instance-attribute ¶

prev_state instance-attribute ¶

limbo_state instance-attribute ¶

state instance-attribute ¶

role instance-attribute ¶

enabled instance-attribute ¶

states instance-attribute ¶

transitions instance-attribute ¶

actionable instance-attribute ¶

wildcards instance-attribute ¶

policy instance-attribute ¶

policy_filter instance-attribute ¶

policy_filter_opts instance-attribute ¶

welcome_msg instance-attribute ¶

welcome_msg_with_wildcards instance-attribute ¶

show_blocking_states instance-attribute ¶

show_action_completion instance-attribute ¶

show_action_request_info instance-attribute ¶

show_ticks_in_action_messages ¶

show_marks_in_blocking_state_messages ¶

show_request_info_in_action_messages ¶

set_welcome_message ¶

to_dict ¶

set_actionable ¶

set_policy ¶

set_policy_filter ¶

set_wildcards ¶

apply_wildcards ¶

set_role ¶

get_wildcards ¶

add_wildcards ¶

update_wildcard ¶

get_action_step_idx ¶

is_busy_acting ¶

add_state ¶

get_state_name ¶

get_state ¶

get_all_states ¶

get_all_actions ¶

get_action ¶

get_action_name ¶

get_last_completed_action_name ¶

reset_state ¶

get_states ¶

set_state ¶

are_debug_messages_active ¶

set_debug_messages_active ¶

generate_auto_messages ¶

add_global_teleport ¶

add_teleport ¶

add_transit ¶

include ¶

must_wait ¶

is_enabled ¶

enable ¶

act_states async ¶

act_ghost_transition async ¶

get_time_spent_in_current_state ¶

act_transitions async ¶

act async ¶

get_state_changed ¶

transition_exists ¶

request_action ¶

wait_for_all_actions_that_start_with ¶

wait_for_all_actions_that_include_an_arg ¶

save ¶

load ¶

load_backward_compat ¶

to_graphviz ¶

save_pdf ¶

print_actions ¶

initial_state `instance-attribute` ¶

prev_state `instance-attribute` ¶

limbo_state `instance-attribute` ¶

state `instance-attribute` ¶

role `instance-attribute` ¶

enabled `instance-attribute` ¶

states `instance-attribute` ¶

transitions `instance-attribute` ¶

actionable `instance-attribute` ¶

wildcards `instance-attribute` ¶

policy `instance-attribute` ¶

policy_filter `instance-attribute` ¶

policy_filter_opts `instance-attribute` ¶

welcome_msg `instance-attribute` ¶

welcome_msg_with_wildcards `instance-attribute` ¶

show_blocking_states `instance-attribute` ¶

show_action_completion `instance-attribute` ¶

show_action_request_info `instance-attribute` ¶

act_states `async` ¶

act_ghost_transition `async` ¶

act_transitions `async` ¶

act `async` ¶