unaiverse.hsm.state

What this module does 🟡

Defines the State class, a fundamental HSM building block that pairs an optional Action with a name, waiting time, blocking behaviour, and a human-readable message.

state ¶

█████ █████ ██████ █████ █████ █████ █████ ██████████ ███████████ █████████ ██████████ ░░███ ░░███ ░░██████ ░░███ ░░███ ░░███ ░░███ ░░███░░░░░█░░███░░░░░███ ███░░░░░███░░███░░░░░█ ░███ ░███ ░███░███ ░███ ██████ ░███ ░███ ░███ ░███ █ ░ ░███ ░███ ░███ ░░░ ░███ █ ░ ░███ ░███ ░███░░███░███ ░░░░░███ ░███ ░███ ░███ ░██████ ░██████████ ░░█████████ ░██████ ░███ ░███ ░███ ░░██████ ███████ ░███ ░░███ ███ ░███░░█ ░███░░░░░███ ░░░░░░░░███ ░███░░█ ░███ ░███ ░███ ░░█████ ███░░███ ░███ ░░░█████░ ░███ ░ █ ░███ ░███ ███ ░███ ░███ ░ █ ░░████████ █████ ░░█████░░████████ █████ ░░███ ██████████ █████ █████░░█████████ ██████████ ░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░ ░░░░░ ░░░ ░░░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░░ ░░░░░░░░░░ 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

State ¶

State(name: str, idx: int = -1, action: Action | None = None, waiting_time: float = 0.0, blocking: bool = True, msg: str | None = None)

A single node in a Hybrid State Machine (HSM).

A State encapsulates all the information that the HSM needs in order to reside in or pass through a particular node of the state graph. Each state has a unique name (used as the lookup key inside the machine), an optional Action that is executed every time the state callable is invoked, a configurable waiting_time that delays execution, and a blocking flag that indicates whether the machine should wait for the current action to complete before processing further transitions.

An optional human-readable msg is printed once to the log the first time the state is entered. Any <wildcard> placeholders embedded in msg are resolved at runtime via set_wildcards and apply_wildcards.

A State is callable: invoking it runs the attached Action (if any) and returns that action's result. The owning HSM registers itself with the state via set_state_machine, which also propagates wildcard bindings down to the action.

Attributes:

Name	Type	Description
`name`		Unique name of the state within the HSM.
`id`		Integer index assigned by the HSM. `-1` when not yet assigned.
`action`		The `Action` executed each time this state's callable is invoked, or `None` if the state is purely passive.
`waiting_time`		Minimum number of seconds to remain in this state before the machine considers acting (evaluated by `must_wait`).
`blocking`		Whether the state prevents the machine from transitioning until the action completes.
`msg`		Human-readable description logged once on state entry, or `None`.
`msg_printed`		True after the entry message has been emitted at least once.
`state_machine`		Reference to the owning HSM, set by `set_state_machine`.
`wildcards`		Mapping from placeholder strings (e.g. `"<playlist>"`) to their concrete replacement values.
`msg_with_wildcards`		A snapshot of `msg` before wildcard substitution, used to restore the template for subsequent `apply_wildcards` calls.

Examples:

Creating a state with an action and a log message:

>>> from unaiverse.hsm.state import State
>>> from unaiverse.hsm.action import Action
>>>
>>> action = Action(name="greet", ...)
>>> state = State(name="greeting", idx=0, action=action,
...               waiting_time=2.0, blocking=True,
...               msg="Entering greeting state")
>>> print(state.name)
greeting

A passive state (no action) that simply waits:

>>> idle = State(name="idle", waiting_time=5.0, blocking=False)
>>> idle.has_action()
False

Initialize a State with its identity, optional action, and display properties.

HTML entities inside msg are automatically unescaped so that state descriptions loaded from JSON behaviour files render correctly. A critical log message is emitted if name is one of the reserved names listed in Custom.NOT_ALLOWED_STATE_NAMES.

Parameters:

Name	Type	Description	Default
`name`	`str`	Unique name of the state within the HSM. Must not be one of the reserved names in `Custom.NOT_ALLOWED_STATE_NAMES`.	required
`idx`	`int`	Integer index assigned to the state by the HSM. Defaults to `-1` when no explicit assignment is needed.	`-1`
`action`	`Action \| None`	An optional `Action` instance executed each time the state's callable is invoked. Defaults to `None`.	`None`
`waiting_time`	`float`	Minimum number of seconds the state must remain active before the machine proceeds. Defaults to `0.`, meaning no delay.	`0.0`
`blocking`	`bool`	If `True`, the HSM waits for the action to complete before evaluating outgoing transitions. Defaults to `True`.	`True`
`msg`	`str \| None`	Human-readable description logged once when the state is first entered. HTML entities are unescaped automatically. Defaults to `None`.	`None`

Examples:

>>> state = State(name="fetch_data", idx=1, waiting_time=1.5,
...               blocking=False, msg="Fetching data...")
>>> state.name
'fetch_data'
>>> state.blocking
False

Source code in unaiverse/hsm/state.py

def __init__(self, name: str, idx: int = -1, action: Action | None = None, waiting_time: float = 0.,
             blocking: bool = True, msg: str | None = None):
    """Initialize a ``State`` with its identity, optional action, and display properties.

    HTML entities inside ``msg`` are automatically unescaped so that state
    descriptions loaded from JSON behaviour files render correctly. A critical
    log message is emitted if ``name`` is one of the reserved names listed in
    ``Custom.NOT_ALLOWED_STATE_NAMES``.

    Args:
        name: Unique name of the state within the HSM. Must not be one of the
            reserved names in ``Custom.NOT_ALLOWED_STATE_NAMES``.
        idx: Integer index assigned to the state by the HSM. Defaults to ``-1``
            when no explicit assignment is needed.
        action: An optional ``Action`` instance executed each time the state's
            callable is invoked. Defaults to ``None``.
        waiting_time: Minimum number of seconds the state must remain active
            before the machine proceeds. Defaults to ``0.``, meaning no delay.
        blocking: If ``True``, the HSM waits for the action to complete before
            evaluating outgoing transitions. Defaults to ``True``.
        msg: Human-readable description logged once when the state is first
            entered. HTML entities are unescaped automatically. Defaults to
            ``None``.

    Examples:
        >>> state = State(name="fetch_data", idx=1, waiting_time=1.5,
        ...               blocking=False, msg="Fetching data...")
        >>> state.name
        'fetch_data'
        >>> state.blocking
        False
    """
    self.name = name  # Name of the state (must be unique)
    self.action = action  # Inner state action (it can be None)
    self.id = idx  # Unique ID of the state (-1 if not needed)
    self.waiting_time = waiting_time  # Number of seconds to wait in the current state before acting
    self.starting_time = 0.
    self.blocking = blocking
    self.msg = msg  # Human-readable message associated to this instance of state
    self.msg_printed = False
    self.state_machine = None

    # Fix UNICODE chars
    if self.msg is not None:
        self.msg = html.unescape(self.msg)

    # Message parts replaced by wildcards (commonly assumed to be in the format <value>)
    self.wildcards = {}  # Value-to-value (es: <playlist> to this:and:this)
    self.msg_with_wildcards = self.msg

    if self.name is Custom.NOT_ALLOWED_STATE_NAMES:
        log.critical(f"Invalid state name (not allowed): {self.name}")

name `instance-attribute` ¶

name = name

action `instance-attribute` ¶

action = action

id `instance-attribute` ¶

id = idx

waiting_time `instance-attribute` ¶

waiting_time = waiting_time

starting_time `instance-attribute` ¶

starting_time = 0.0

blocking `instance-attribute` ¶

blocking = blocking

msg `instance-attribute` ¶

msg = msg

msg_printed `instance-attribute` ¶

msg_printed = False

state_machine `instance-attribute` ¶

state_machine = None

wildcards `instance-attribute` ¶

wildcards = {}

msg_with_wildcards `instance-attribute` ¶

msg_with_wildcards = msg

set_state_machine ¶

set_state_machine(hsm: object) -> None

Register the parent state machine that owns this state.

Stores a reference to the HSM, propagates the machine's current wildcard bindings into this state via set_wildcards, and, if the state has an associated Action, registers the same machine with that action so the action can access machine-level context.

This method is called automatically by the HSM when the state is added to it. Direct calls are only necessary in advanced scenarios such as testing or dynamic state reconfiguration.

Parameters:

Name	Type	Description	Default
`hsm`	`object`	The `HybridStateMachine` (or compatible object) that owns this state. The object is expected to expose `get_wildcards()` and `show_blocking_states`.	required

Source code in unaiverse/hsm/state.py

def set_state_machine(self, hsm: object) -> None:
    """Register the parent state machine that owns this state.

    Stores a reference to the HSM, propagates the machine's current wildcard
    bindings into this state via ``set_wildcards``, and, if the state has an
    associated ``Action``, registers the same machine with that action so the
    action can access machine-level context.

    This method is called automatically by the HSM when the state is added to
    it. Direct calls are only necessary in advanced scenarios such as testing or
    dynamic state reconfiguration.

    Args:
        hsm: The ``HybridStateMachine`` (or compatible object) that owns this
            state. The object is expected to expose ``get_wildcards()`` and
            ``show_blocking_states``.
    """
    self.state_machine = hsm
    self.set_wildcards(hsm.get_wildcards())
    if self.action is not None:
        self.action.set_state_machine(hsm)

set_msg ¶

set_msg(msg: str | None) -> None

Set the human-readable message displayed when this state is entered.

If msg is not None, HTML entities are unescaped and both msg and msg_with_wildcards are updated to the new value. When None is passed, both attributes are cleared so no message is printed on the next state entry.

Parameters:

Name	Type	Description	Default
`msg`	`str \| None`	The new display message, or `None` to clear it. HTML entities (e.g. `&`, `<`) are unescaped automatically before storing.	required

Source code in unaiverse/hsm/state.py

def set_msg(self, msg: str | None) -> None:
    """Set the human-readable message displayed when this state is entered.

    If ``msg`` is not ``None``, HTML entities are unescaped and both ``msg``
    and ``msg_with_wildcards`` are updated to the new value. When ``None`` is
    passed, both attributes are cleared so no message is printed on the next
    state entry.

    Args:
        msg: The new display message, or ``None`` to clear it. HTML entities
            (e.g. ``&amp;``, ``&lt;``) are unescaped automatically before
            storing.
    """

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

must_wait ¶

must_wait() -> bool

Return whether the state is still within its mandatory waiting period.

Compares the elapsed time since starting_time was recorded against waiting_time. If no waiting time is configured (waiting_time <= 0), the method always returns False immediately. While the state is still waiting, the elapsed time is emitted at DEBUG log level.

Returns:

Type	Description
`bool`	`True` if the elapsed time since state entry is less than
`bool`	`waiting_time`, meaning the state must not yet transition.
`bool`	`False` otherwise, including when `waiting_time` is zero or
`bool`	negative.

Source code in unaiverse/hsm/state.py

def must_wait(self) -> bool:
    """Return whether the state is still within its mandatory waiting period.

    Compares the elapsed time since ``starting_time`` was recorded against
    ``waiting_time``. If no waiting time is configured (``waiting_time <= 0``),
    the method always returns ``False`` immediately. While the state is still
    waiting, the elapsed time is emitted at ``DEBUG`` log level.

    Returns:
        ``True`` if the elapsed time since state entry is less than
        ``waiting_time``, meaning the state must not yet transition.
        ``False`` otherwise, including when ``waiting_time`` is zero or
        negative.
    """
    if self.waiting_time > 0.:
        if (time.perf_counter() - self.starting_time) >= self.waiting_time:
            return False
        else:
            log.debug(f"Time passing: {(time.perf_counter() - self.starting_time)} seconds")
            return True
    else:
        return False

to_dict ¶

to_dict() -> dict

Serialize the state's configuration to a plain dictionary.

The resulting dictionary uses the same keys expected by the HSM's JSON behaviour format, making it suitable for persistence, transmission, or reconstruction of the state. The action name and its keyword arguments are included when an action is present. Non-ASCII characters in msg are replaced with XML character references so the output is safe for ASCII serialization.

Returns:

Type	Description
`dict`	A dictionary with the following keys:
`dict`	`"action"`: The action's name string, or `None` if no action.
`dict`	`"action_kwargs"`: The action's argument dictionary, or `{}` if no action.
`dict`	`"msg"`: The state message with non-ASCII characters XML-escaped, or `None` if no message is set.
`dict`	`"blocking"`: The blocking flag as a boolean.
`dict`	The time-to-wait key (taken from `Custom.TIME_TO_WAIT_BEFORE_ACTING_ARG_NAMES[0]`): the configured `waiting_time` as a float.

Source code in unaiverse/hsm/state.py

def to_dict(self) -> dict:
    """Serialize the state's configuration to a plain dictionary.

    The resulting dictionary uses the same keys expected by the HSM's JSON
    behaviour format, making it suitable for persistence, transmission, or
    reconstruction of the state. The action name and its keyword arguments are
    included when an action is present. Non-ASCII characters in ``msg`` are
    replaced with XML character references so the output is safe for ASCII
    serialization.

    Returns:
        A dictionary with the following keys:

        - ``"action"``: The action's name string, or ``None`` if no action.
        - ``"action_kwargs"``: The action's argument dictionary, or ``{}`` if no
          action.
        - ``"msg"``: The state message with non-ASCII characters XML-escaped,
          or ``None`` if no message is set.
        - ``"blocking"``: The blocking flag as a boolean.
        - The time-to-wait key (taken from
          ``Custom.TIME_TO_WAIT_BEFORE_ACTING_ARG_NAMES[0]``): the configured
          ``waiting_time`` as a float.
    """
    return {
        "action": self.action.name if self.action is not None else None,
        "action_kwargs": self.action.args if self.action is not None else {},
        "msg": self.msg.encode("ascii",
                               "xmlcharrefreplace").decode("ascii") if self.msg is not None else None,
        "blocking": self.blocking,
        Custom.TIME_TO_WAIT_BEFORE_ACTING_ARG_NAMES[0]: self.waiting_time
    }

has_action ¶

has_action() -> bool

Return whether this state has an associated action.

Returns:

Type	Description
`bool`	`True` if an `Action` instance is attached to this state,
`bool`	`False` if `action` is `None`.

Source code in unaiverse/hsm/state.py

def has_action(self) -> bool:
    """Return whether this state has an associated action.

    Returns:
        ``True`` if an ``Action`` instance is attached to this state,
        ``False`` if ``action`` is ``None``.
    """
    return self.action is not None

get_starting_time ¶

get_starting_time() -> float

Return the timestamp recorded when the state's execution began.

The starting time is set the first time the state callable is invoked (i.e. when the HSM enters the state). It is reset to 0. by reset. The value is used internally by must_wait to calculate elapsed time and is exposed here for external monitoring or testing.

Returns:

Type	Description
`float`	A `float` representing the `time.perf_counter()` value captured at
`float`	state entry, or `0.` if the state has not yet been entered (or has
`float`	been reset).

Source code in unaiverse/hsm/state.py

def get_starting_time(self) -> float:
    """Return the timestamp recorded when the state's execution began.

    The starting time is set the first time the state callable is invoked (i.e.
    when the HSM enters the state). It is reset to ``0.`` by ``reset``. The
    value is used internally by ``must_wait`` to calculate elapsed time and is
    exposed here for external monitoring or testing.

    Returns:
        A ``float`` representing the ``time.perf_counter()`` value captured at
        state entry, or ``0.`` if the state has not yet been entered (or has
        been reset).
    """
    return self.starting_time

reset ¶

reset(reset_message_printing: bool = True) -> None

Reset the state's internal counters so it can be re-entered cleanly.

Sets starting_time back to 0. so that must_wait starts measuring from the next entry, and optionally clears msg_printed so the entry message is displayed again. If an action is attached, its step counter is also reset via action.system_interaction.reset_state().

This method is called by the HSM whenever it transitions back into this state after having left it.

Parameters:

Name	Type	Description	Default
`reset_message_printing`	`bool`	If `True`, `msg_printed` is set to `False` so the entry message will be shown again on the next entry. If `False`, the entry message suppression is preserved. Defaults to `True`.	`True`

Source code in unaiverse/hsm/state.py

def reset(self, reset_message_printing: bool = True) -> None:
    """Reset the state's internal counters so it can be re-entered cleanly.

    Sets ``starting_time`` back to ``0.`` so that ``must_wait`` starts measuring
    from the next entry, and optionally clears ``msg_printed`` so the entry
    message is displayed again. If an action is attached, its step counter is
    also reset via ``action.system_interaction.reset_state()``.

    This method is called by the HSM whenever it transitions back into this
    state after having left it.

    Args:
        reset_message_printing: If ``True``, ``msg_printed`` is set to
            ``False`` so the entry message will be shown again on the next
            entry. If ``False``, the entry message suppression is preserved.
            Defaults to ``True``.
    """
    self.starting_time = 0.
    if reset_message_printing:
        self.msg_printed = False
    if self.action is not None:
        self.action.system_interaction.reset_state()

set_blocking ¶

set_blocking(blocking: bool) -> None

Set the blocking status of this state.

A blocking state prevents the HSM from evaluating outgoing transitions until the state's action has fully completed. A non-blocking state allows the machine to proceed to the next state regardless of the action's outcome. The visual indicator shown in the log message (red or green dot) is also controlled by this flag when state_machine.show_blocking_states is enabled.

Parameters:

Name	Type	Description	Default
`blocking`	`bool`	`True` to make this state block the machine until its action completes; `False` to allow the machine to transition freely.	required

Source code in unaiverse/hsm/state.py

def set_blocking(self, blocking: bool) -> None:
    """Set the blocking status of this state.

    A blocking state prevents the HSM from evaluating outgoing transitions
    until the state's action has fully completed. A non-blocking state allows
    the machine to proceed to the next state regardless of the action's outcome.
    The visual indicator shown in the log message (red or green dot) is also
    controlled by this flag when ``state_machine.show_blocking_states`` is
    enabled.

    Args:
        blocking: ``True`` to make this state block the machine until its
            action completes; ``False`` to allow the machine to transition
            freely.
    """
    self.blocking = blocking

set_wildcards ¶

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

Replace the wildcard substitution table for this state and its action.

Wildcards are placeholder strings embedded in msg (typically formatted as <placeholder>) that are resolved to concrete values at runtime by apply_wildcards. Providing None clears all existing wildcards. The new table is propagated to the attached action (if any) so that action arguments containing wildcards are also substituted correctly.

See apply_wildcards to trigger the actual substitution.

Parameters:

Name	Type	Description	Default
`wildcards`	`dict[str, str \| float \| int] \| None`	A mapping from placeholder strings to their replacement values, or `None` to clear all wildcards. Defaults to an empty dict when `None` is passed.	required

Source code in unaiverse/hsm/state.py

def set_wildcards(self, wildcards: dict[str, str | float | int] | None) -> None:
    """Replace the wildcard substitution table for this state and its action.

    Wildcards are placeholder strings embedded in ``msg`` (typically formatted
    as ``<placeholder>``) that are resolved to concrete values at runtime by
    ``apply_wildcards``. Providing ``None`` clears all existing wildcards. The
    new table is propagated to the attached action (if any) so that action
    arguments containing wildcards are also substituted correctly.

    See ``apply_wildcards`` to trigger the actual substitution.

    Args:
        wildcards: A mapping from placeholder strings to their replacement
            values, or ``None`` to clear all wildcards. Defaults to an empty
            dict when ``None`` is passed.
    """
    self.wildcards = wildcards if wildcards is not None else {}
    if self.action is not None:
        self.action.set_wildcards(self.wildcards)

apply_wildcards ¶

apply_wildcards() -> None

Apply the current wildcard table to the state message and its action.

Restores msg to the pre-substitution template stored in msg_with_wildcards and then performs a str.replace pass for every entry in wildcards. This guarantees that changing the wildcard values and calling apply_wildcards again always produces a fresh result based on the original template rather than the previously substituted string.

If msg_with_wildcards is None, it is first initialised from the current msg. The same substitution is then delegated to the attached action (if any) by calling action.apply_wildcards().

Note

This method mutates msg in place. Call set_wildcards first whenever the substitution table needs to change.

Source code in unaiverse/hsm/state.py

def apply_wildcards(self) -> None:
    """Apply the current wildcard table to the state message and its action.

    Restores ``msg`` to the pre-substitution template stored in
    ``msg_with_wildcards`` and then performs a ``str.replace`` pass for every
    entry in ``wildcards``. This guarantees that changing the wildcard values
    and calling ``apply_wildcards`` again always produces a fresh result based
    on the original template rather than the previously substituted string.

    If ``msg_with_wildcards`` is ``None``, it is first initialised from the
    current ``msg``. The same substitution is then delegated to the attached
    action (if any) by calling ``action.apply_wildcards()``.

    Note:
        This method mutates ``msg`` in place. Call ``set_wildcards`` first
        whenever the substitution table needs to change.
    """
    if self.msg_with_wildcards is None:
        self.msg_with_wildcards = self.msg
    else:
        self.msg = self.msg_with_wildcards

    if self.msg is not None:
        for wildcard_from, wildcard_to in self.wildcards.items():
            self.msg = self.msg.replace(wildcard_from, str(wildcard_to))

    if self.action is not None:
        self.action.apply_wildcards()

get_time_passed ¶

get_time_passed()

Return the elapsed time since the state was entered.

Computes the number of seconds that have passed since starting_time was recorded. If the state has not yet been entered (or has been reset), returns -1. as a sentinel value.

Returns:

Type	Description
	A `float` representing the elapsed seconds since entry, or `-1.`
	if `starting_time` is `0.` (state not yet entered or reset).

Source code in unaiverse/hsm/state.py

def get_time_passed(self):
    """Return the elapsed time since the state was entered.

    Computes the number of seconds that have passed since ``starting_time``
    was recorded. If the state has not yet been entered (or has been reset),
    returns ``-1.`` as a sentinel value.

    Returns:
        A ``float`` representing the elapsed seconds since entry, or ``-1.``
        if ``starting_time`` is ``0.`` (state not yet entered or reset).
    """
    return (time.perf_counter() - self.starting_time) if self.starting_time > 0 else -1.

unaiverse.hsm.state

state ¶

State ¶

name instance-attribute ¶

action instance-attribute ¶

id instance-attribute ¶

waiting_time instance-attribute ¶

starting_time instance-attribute ¶

blocking instance-attribute ¶

msg instance-attribute ¶

msg_printed instance-attribute ¶

state_machine instance-attribute ¶

wildcards instance-attribute ¶

msg_with_wildcards instance-attribute ¶

set_state_machine ¶

set_msg ¶

must_wait ¶

to_dict ¶

has_action ¶

get_starting_time ¶

reset ¶

set_blocking ¶

set_wildcards ¶

apply_wildcards ¶

get_time_passed ¶

name `instance-attribute` ¶

action `instance-attribute` ¶

id `instance-attribute` ¶

waiting_time `instance-attribute` ¶

starting_time `instance-attribute` ¶

blocking `instance-attribute` ¶

msg `instance-attribute` ¶

msg_printed `instance-attribute` ¶

state_machine `instance-attribute` ¶

wildcards `instance-attribute` ¶

msg_with_wildcards `instance-attribute` ¶