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 .venv/lib/python3.13/site-packages/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.owner._state_machine_state = state_machine.initial_state
        # self.state_machine._prepare(owner)

    def get_state(self) -> typing.Union[str, StrEnum, None]:
        """
        return the current state. one can also access the property `current state`.

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

    def set_state(self, 
                value : typing.Union[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. One can also set state using '=' operator of `current_state` property in which case 
        callbacks will be called. If originally an enumeration for the list of allowed states was supplied, 
        then an enumeration member must be used to set 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:
            previous_state = self.current_state
            next_state = self.descriptor._get_machine_compliant_state(value)
            self.owner._state_machine_state = 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: typing.Union[Property, typing.Callable]) -> bool:
        """
        returns True 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: typing.Union[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: Union[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. One can also set state using '=' operator of current_state property in which case callbacks will be called. If originally an enumeration for the list of allowed states was supplied, then an enumeration member must be used to set 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 .venv/lib/python3.13/site-packages/hololinked/core/state_machine.py

def set_state(self, 
            value : typing.Union[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. One can also set state using '=' operator of `current_state` property in which case 
    callbacks will be called. If originally an enumeration for the list of allowed states was supplied, 
    then an enumeration member must be used to set 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:
        previous_state = self.current_state
        next_state = self.descriptor._get_machine_compliant_state(value)
        self.owner._state_machine_state = 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() -> typing.Union[str, StrEnum, None]

return the current state. one can also access the property current state.

Returns:

Type	Description
current state: str

Source code in .venv/lib/python3.13/site-packages/hololinked/core/state_machine.py

def get_state(self) -> typing.Union[str, StrEnum, None]:
    """
    return the current state. one can also access the property `current state`.

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

contains_object

contains_object(object: Union[Property, Callable]) -> bool

returns True 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 .venv/lib/python3.13/site-packages/hololinked/core/state_machine.py

def contains_object(self, object: typing.Union[Property, typing.Callable]) -> bool:
    """
    returns True 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