hololinked.core.state_machine.BoundFSM

A FSM bound to a Thing instance, returned when accessed as a instance attribute (self.state_machine). There is no need to instantiate this class directly.

Source code in hololinked/hololinked/core/state_machine.py

class BoundFSM:
    """
    A FSM bound to a `Thing` instance, returned when accessed as a instance attribute (`self.state_machine`).
    There is no need to instantiate this class directly.
    """

    def __init__(self, owner: Thing, state_machine: StateMachine) -> None:
        self.descriptor = state_machine
        self.push_state_change_event = state_machine.push_state_change_event
        self.owner = owner
        self.logger = state_machine.logger

    def get_state(self) -> str | StrEnum | None:
        """
        return the current state, one can also access it using the property `current state`.

        Returns
        -------
        str
            current state of the state machine
        """
        try:
            return self.owner._state_machine_state
        except AttributeError:
            return self.initial_state

    def set_state(self, value: str | StrEnum | Enum, push_event: bool = True, skip_callbacks: bool = False) -> None:
        """
        set state of state machine. Also triggers state change callbacks if `skip_callbacks=False` and pushes a state
        change event when `push_event=True` (when __init__ argument `push_state_change_event=True`).
        One can also set state using the '=' operator of the `current_state` property,
        in which case `skip_callbacks=False` and `push_event=True` will be used.

        If originally an enumeration for the list of allowed states was supplied,
        then an enumeration member must be used to set the state. If a list of strings were supplied,
        then a string is accepted.

        Raises
        ------
        ValueError:
            if the state is not found in the allowed states
        """
        if value in self.states:
            given_state = self.descriptor._get_machine_compliant_state(value)
            if given_state == self.current_state:
                return
            previous_state = self.current_state
            next_state = self.descriptor._get_machine_compliant_state(value)
            self.owner._state_machine_state = next_state
            self.logger.info(f"state changed from {previous_state} to {next_state}")
            if push_event and self.push_state_change_event and hasattr(self.owner, "event_publisher"):
                self.owner.state  # just acces to trigger the observable event
            if skip_callbacks:
                return
            if previous_state in self.on_exit:
                for func in self.on_exit[previous_state]:
                    func(self.owner)
            if next_state in self.on_enter:
                for func in self.on_enter[next_state]:
                    func(self.owner)
        else:
            raise ValueError("given state '{}' not in set of allowed states : {}.".format(value, self.states))

    current_state = property(get_state, set_state, None, doc="""read and write current state of the state machine""")

    def contains_object(self, object: Property | Callable) -> bool:
        """
        Check if specified object is found in any of the state machine states.
        Supply unbound method for checking methods, as state machine is specified at class level
        when the methods are unbound.
        """
        return self.descriptor.contains_object(object)

    def __hash__(self):
        return hash(
            self.owner.id
            + (str(state) for state in self.states)
            + str(self.initial_state)
            + self.owner.__class__.__name__
        )

    def __str__(self):
        return f"StateMachine(owner={self.owner.__class__.__name__} id={self.owner.id} initial_state={self.initial_state}, states={self.states})"

    def __eq__(self, other) -> bool:
        if not isinstance(other, StateMachine):
            return False
        return (
            self.states == other.states
            and self.initial_state == other.initial_state
            and self.owner.__class__ == other.owner.__class__
            and self.owner.id == other.owner.id
        )

    def __contains__(self, state: str | StrEnum) -> bool:
        return state in self.descriptor

    @property
    def initial_state(self):
        """initial state of the machine"""
        return self.descriptor.initial_state

    @property
    def states(self):
        """list of allowed states"""
        return self.descriptor.states

    @property
    def on_enter(self):
        """callbacks to execute when a certain state is entered"""
        return self.descriptor.on_enter

    @property
    def on_exit(self):
        """callbacks to execute when certain state is exited"""
        return self.descriptor.on_exit

    @property
    def machine(self):
        """the machine specification with state as key and objects as list"""
        return self.descriptor.machine

Functions

set_state

set_state(value: str | StrEnum | Enum, push_event: bool = True, skip_callbacks: bool = False) -> None

set state of state machine. Also triggers state change callbacks if skip_callbacks=False and pushes a state change event when push_event=True (when init argument push_state_change_event=True). One can also set state using the '=' operator of the current_state property, in which case skip_callbacks=False and push_event=True will be used.

If originally an enumeration for the list of allowed states was supplied, then an enumeration member must be used to set the state. If a list of strings were supplied, then a string is accepted.

Raises:

Type	Description
ValueError:	if the state is not found in the allowed states

Source code in hololinked/hololinked/core/state_machine.py

def set_state(self, value: str | StrEnum | Enum, push_event: bool = True, skip_callbacks: bool = False) -> None:
    """
    set state of state machine. Also triggers state change callbacks if `skip_callbacks=False` and pushes a state
    change event when `push_event=True` (when __init__ argument `push_state_change_event=True`).
    One can also set state using the '=' operator of the `current_state` property,
    in which case `skip_callbacks=False` and `push_event=True` will be used.

    If originally an enumeration for the list of allowed states was supplied,
    then an enumeration member must be used to set the state. If a list of strings were supplied,
    then a string is accepted.

    Raises
    ------
    ValueError:
        if the state is not found in the allowed states
    """
    if value in self.states:
        given_state = self.descriptor._get_machine_compliant_state(value)
        if given_state == self.current_state:
            return
        previous_state = self.current_state
        next_state = self.descriptor._get_machine_compliant_state(value)
        self.owner._state_machine_state = next_state
        self.logger.info(f"state changed from {previous_state} to {next_state}")
        if push_event and self.push_state_change_event and hasattr(self.owner, "event_publisher"):
            self.owner.state  # just acces to trigger the observable event
        if skip_callbacks:
            return
        if previous_state in self.on_exit:
            for func in self.on_exit[previous_state]:
                func(self.owner)
        if next_state in self.on_enter:
            for func in self.on_enter[next_state]:
                func(self.owner)
    else:
        raise ValueError("given state '{}' not in set of allowed states : {}.".format(value, self.states))

get_state

get_state() -> str | StrEnum | None

return the current state, one can also access it using the property current state.

Returns:

Type	Description
str	current state of the state machine

Source code in hololinked/hololinked/core/state_machine.py

def get_state(self) -> str | StrEnum | None:
    """
    return the current state, one can also access it using the property `current state`.

    Returns
    -------
    str
        current state of the state machine
    """
    try:
        return self.owner._state_machine_state
    except AttributeError:
        return self.initial_state

contains_object

contains_object(object: Property | Callable) -> bool

Check if specified object is found in any of the state machine states. Supply unbound method for checking methods, as state machine is specified at class level when the methods are unbound.

Source code in hololinked/hololinked/core/state_machine.py

def contains_object(self, object: Property | Callable) -> bool:
    """
    Check if specified object is found in any of the state machine states.
    Supply unbound method for checking methods, as state machine is specified at class level
    when the methods are unbound.
    """
    return self.descriptor.contains_object(object)

Attributes

current_state

str
instance-attribute, writable
read and write current state of the state machine

initial_state

str
instance-attribute, read-only
initial state of the state machine

states

List[str]
instance-attribute, read-only
list of allowed states

on_enter

typing.Dict
instance-attribute, read-only
callbacks to execute when a certain state is entered

on_exit

str
instance-attribute, read-only
callbacks to execute when a certain state is exited

machine

typing.Dict[str, List[Callable | Property]]
instance-attribute, read-only
state machine definition, i.e. list of allowed properties and actions for each state