Declaring states in __init__ in a dict

[icen1]

Currently i am trying to declare states in a map like states_storage['Start'] = State('Start', initial=True) but this doesn't actually make the states, instead in the factory.py file it says there are no states, instead when i do Start = State('Start', initial=True) it works as intended. Is there a way to get working with dicts? The example given is simplified, the actual program is a bit more complex

[fgmacedo]

Can you give more context on what you're trying to solve by using a dict to store states? Is your SM definition dynamic?

A helper that you can use is the States class, which is used to create State instances from an Enum, but can be adapted to create states from a dict: https://python-statemachine.readthedocs.io/en/latest/api.html#states-class

from statemachine import StateMachine, State
from statemachine.states import States

class SM(StateMachine):

    states = States({
        name: State(initial=idx == 0) for idx, name in enumerate(["initial", "final"])
    })

    finish = states.initial.to(states.final)

[icen1]

@fgmacedo Yes that's what i am trying to do, make the definition dynamic on a specific input. So far i managed to do it by setting an environment variable. The page you sent is very helpful, but i can't do it through an intialisation function, which means the states would have to be specified in the environment/variable somehow. Do you have any suggestions? Debugging it i found it was becauses the states are created on importing the class and not its intialisation.

[fgmacedo]

In fact, the syntax is best suited for SM constructed by code at module level and not dynamically. So you can use the expressive API to declare states, transitions and events.

Another possible solution is to have a "constructor" method that returns the state machine class.

If you provide a concrete example, I can try to figure out something.

[icen1]

@fgmacedo so currently what i am trying to do is have the definition dynamic. With every object has its own set of states based on a variable and their opposing traits (think markov chains), but at the moment this is not possible as they are declared in the class level, which means all instances of the class will have those exact states and it won't be based on the newer version of the set of traits. To solve this i need to declare the states using an instance variable, which i beleive is not possible. Currently i declare each object in its own process using multiprocessing

[fgmacedo]

The factory.StateMachineMetaclass plays a crucial role in generating StateMachine classes from a static code definition. Thanks to Python's dynamic capabilities, each metaclass, including the fundamental one, type, has the ability to create classes. Consequently, in Python, a class is essentially an instance of a metaclass.

To illustrate this concept, I have prepared a simple example for you. This example demonstrates how to craft a "dynamic factory" leveraging the metaclass to produce a new StateMachine class, based on a dictionary detailing the states and transitions of the state machine. Although this example does not comprehensively address all parameters for creating States and Transitions, it serves as an excellent starting point.

Should this approach resonate well with the community, we might consider integrating such a factory into the library itself.

Please upvote the answer if you find this to be useful.

Dynamic created machine

A demonstration of how to create a :ref:StateMachine from a dynamic source, like loading
from a file or a database, or even from a remote API, and then using it as a regular
:class:StateMachine. We will only show the intermediate step that is a dictionary that
represents the machine.

from statemachine.factory import StateMachineMetaclass
from statemachine.state import State
from statemachine.statemachine import StateMachine


def create_machine_class_from_definition(name: str, definition: dict):
    """
    Creates a StateMachine class from a dictionary definition, using the StateMachineMetaclass metaclass.
    It maps the definition to the StateMachineMetaclass parameters and then creates the class.

    Example usage to generate a traffic light machine:

    >>> machine = create_machine_class_from_definition(
    ...     "TrafficLightMachine",
    ...     {
    ...         "states": {
    ...             "green": {"initial": True},
    ...             "yellow": {},
    ...             "red": {},
    ...         },
    ...         "events": {
    ...             "change": [
    ...                 {"from": "green", "to": "yellow"},
    ...                 {"from": "yellow", "to": "red"},
    ...                 {"from": "red", "to": "green"},
    ...             ]
    ...         },
    ...     }
    ... )

    """

    states_instances = {
        state_id: State(**state_kwargs)
        for state_id, state_kwargs in definition["states"].items()
    }

    events = {}
    for event_name, transitions in definition["events"].items():
        for transition_data in transitions:
            source = states_instances[transition_data["from"]]
            target = states_instances[transition_data["to"]]

            transition = source.to(target, event=event_name)

            if event_name in events:
                events[event_name] |= transition
            else:
                events[event_name] = transition


    attrs_mapper = {**states_instances, **events}

    return StateMachineMetaclass(name, (StateMachine,), attrs_mapper)

Putting It all together

CampaignMachine = create_machine_class_from_definition(
    "CampaignMachine",
    {
        "states": {
            "draft": {"initial": True, "final": False},
            "producing": {"initial": False, "final": False},
            "closed": {"initial": False, "final": True},
        },
        "events": {
            "add_job": [
                {"from": "draft", "to": "draft"},
                {"from": "producing", "to": "producing"},
            ],
            "produce": [{"from": "draft", "to": "producing"}],
            "deliver": [{"from": "producing", "to": "closed"}],
        },
    }
)

Asserting campaign machine declaration

assert CampaignMachine.draft.initial
assert not CampaignMachine.draft.final

assert not CampaignMachine.producing.initial
assert not CampaignMachine.producing.final

assert not CampaignMachine.closed.initial
assert CampaignMachine.closed.final

Testing our campaign machine

sm = CampaignMachine()
res = sm.send("produce")

assert sm.draft.is_active is False
assert sm.producing.is_active is True
assert sm.closed.is_active is False
assert sm.current_state == sm.producing
assert sm.current_state_value == "producing"

[fgmacedo]

Hi @icen1, how are you? It's possible and can be done in different ways, so I don't have a "right/preferred" method. One possible solution is to use an "extra_kwargs" to include extra properties on the created class. Since StateMachineMetaclass supports "name convention" to bind callbacks to actions/guards, it is enough to match names and you have a glue that binds them to the events and states, like on_enter_<state.id> or on_<event.name>. Then, we use the dict declaration to include "cond", "unless" on the transition params creation.

Let's update the example!

Dynamic created machine

A demonstration of how to create a :ref:StateMachine from a dynamic source, like loading
from a file or a database, or even from a remote API, and then using it as a regular
:class:StateMachine. We will only show the intermediate step that is a dictionary that
represents the machine.

def create_machine_class_from_definition(name: str, definition: dict, **extra_kwargs):
    """
    Creates a StateMachine class from a dictionary definition, using the StateMachineMetaclass metaclass.
    It maps the definition to the StateMachineMetaclass parameters and then creates the class.

    Example usage with a traffic light machine:

    >>> machine = create_machine_class_from_definition(
    ...     "TrafficLightMachine",
    ...     {
    ...         "states": {
    ...             "green": {"initial": True},
    ...             "yellow": {},
    ...             "red": {},
    ...         },
    ...         "events": {
    ...             "change": [
    ...                 {"from": "green", "to": "yellow"},
    ...                 {"from": "yellow", "to": "red"},
    ...                 {"from": "red", "to": "green"},
    ...             ]
    ...         },
    ...     }
    ... )

    """

    states_instances = {
        state_id: State(**state_kwargs)
        for state_id, state_kwargs in definition["states"].items()
    }

    events = {}
    for event_name, transitions in definition["events"].items():
        for transition_data in transitions:
            source = states_instances[transition_data["from"]]
            target = states_instances[transition_data["to"]]

            transition = source.to(
                target, 
                event=event_name, 
                cond=transition_data.get("cond"),
                unless=transition_data.get("unless"),
            )

            if event_name in events:
                events[event_name] |= transition
            else:
                events[event_name] = transition


    attrs_mapper = {**extra_kwargs, **states_instances, **events}

    return StateMachineMetaclass(name, (StateMachine,), attrs_mapper)

Putting It all together

def on_enter_draft(self):
    if not hasattr(self, "jobs"):
        self.jobs = []


def on_add_job(self, job):
    self.jobs.append(job)


def has_any_job(self) -> bool:
    return len(self.jobs) > 0



CampaignMachine = create_machine_class_from_definition(
    "CampaignMachine",
    {
        "states": {
            "draft": {"initial": True, "final": False},
            "producing": {"initial": False, "final": False},
            "closed": {"initial": False, "final": True},
        },
        "events": {
            "add_job": [
                {"from": "draft", "to": "draft"},
                {"from": "producing", "to": "producing", },
            ],
            "produce": [{"from": "draft", "to": "producing", "cond": "has_any_job"}],
            "deliver": [{"from": "producing", "to": "closed"}],
        },
    },
    on_enter_draft=on_enter_draft,
    on_add_job=on_add_job,
    has_any_job=has_any_job,
)

Asserting campaign machine declaration

assert CampaignMachine.draft.initial
assert not CampaignMachine.draft.final

assert not CampaignMachine.producing.initial
assert not CampaignMachine.producing.final

assert not CampaignMachine.closed.initial
assert CampaignMachine.closed.final

Testing our campaign machine

sm = CampaignMachine()
assert sm.draft.is_active
assert sm.jobs == []

try:
    res = sm.send("produce")
except TransitionNotAllowed as e:
    pass

sm.add_job("job1")

assert sm.jobs == ["job1"]

res = sm.send("produce")

assert sm.draft.is_active is False
assert sm.producing.is_active is True
assert sm.closed.is_active is False
assert sm.current_state == sm.producing
assert sm.current_state_value == "producing"

[s-celles]

Good idea, I would do create_machine_class_from_definition a StateMachine class method or build a StateMachineFactory using a factory pattern

Following #421

On the other side I think StateMachine should have a to_dict() method (as Pandas DataFrames have) and build method on top of that a to_json() (for most python-statemachine objects... at least States and Events).

https://stackoverflow.com/questions/3768895/how-to-make-a-class-json-serializable can give some ideas.

With such mechanism a round-trip would be possible ie build a StateMachine dynamically from dict, export this SM to dict...

This will be very useful for supervision / telemetry of a running SM.

[ghost]

This is wonderful, thank you!

[fgmacedo]

Do you think that this should be included in the lib?

[icen1]

@fgmacedo I believe you might have been asking for @scls19fr's opinion, but I am all for it! I had to work around serialization myself in the project I was working on, so I am sure it will be very useful to others. The create_machine_class_from_definition is extremely useful and versatile too. I would love to see it in the library if that's feasible.