Kiwi Cogs

A simple and easy to use state machine library.

• Github repository: https://github.com/mopeyjellyfish/KiwiCogs/
• Documentation https://mopeyjellyfish.github.io/KiwiCogs/

Installation

Pip

pip install -U kiwi-cogs

Poetry

poetry add kiwi-cogs

Quick start

Events

Example configuration:

light_config = {
    "name": "lights",
    "initial": "green",
    "states": {
        "green": {
            "events": {"NEXT": {"target": "yellow"}},
        },
        "yellow": {"events": {"NEXT": {"target": "red"}}},
        "red": {"events": {"NEXT": {"target": "green"}}},
    },
}

Usage:

light_machine await Machine.create(light_config)
assert traffic_light.initial_state.value == "green"
yellow_state = await traffic_light.event("NEXT")
assert yellow_state.value == "yellow"
red_state = await traffic_light.event("NEXT")
assert red_state.value == "red"
green_state = await traffic_light.event("NEXT")
assert green_state.value == "green"

Transitions

Example configuration:

async def entered(_):
    print("entered state!")


async def log(_):
    print("LOG!")


def exited(_):
    print("exited!")


def is_adult(context, _):
    age = context.get("age")
    return age is not None and age >= 18


def is_child(context, _):
    age = context.get("age")
    return age is not None and age < 18


def log_age(context):
    age = context.get("age")
    print(f"User is {age} old!")


def age_determined(context):
    age = context.get("age")
    print(f"Users age has been determined as: {age}")


age_config = {
        "name": "age",
        "context": {"age": None},  # age unknown
        "initial": "unknown",
        "states": {
            "unknown": {
                "transitions": [
                    {"target": "adult", "cond": is_adult},
                    {"target": "child", "cond": is_child},
                ],
                "entry": [log, entered],
                "exit": age_determined,
            },
            "adult": {"type": "final", "entry": log_age},
            "child": {"type": "final", "entry": log_age},
        },
    }

Usage:

age_machine await Machine.create(age_config)
assert age_machine.state.value == "unknown"
context = {"age": 18}
await age_machine.with_context(context=context)
assert age_machine.state.value == "adult"

Hierarchical machine

Example configuration:

def is_walking(context, _):
    return context["speed"] <= 11


def is_running(context, _):
    return context["speed"] > 11


walk_states = {
        "initial": "start",
        "states": {
            "start": {
                "transitions": [ # resolved in order
                    {"target": "walking", "cond": is_walking},
                    {"target": "running", "cond": is_running},
                ],
            },
            "walking": {"events": {"CROSSED": {"target": "crossed"}}},
            "running": {"events": {"CROSSED": {"target": "crossed"}}},
            "crossed": {},
        },
    }


pedestrian_states = {
        "initial": "walk",
        "states": {
            "walk": {"events": {"PED_COUNTDOWN": {"target": "wait"}}, **walk_states},
            "wait": {"events": {"PED_COUNTDOWN": {"target": "stop"}}},
            "stop": {},
            "blinking": {},
        },
    }


crossing_config = {
        "name": "light",
        "initial": "green",
        "context": {"speed": 10},
        "states": {
            "green": {"events": {"TIMER": {"target": "yellow"}}},
            "yellow": {"events": {"TIMER": {"target": "red"}}},
            "red": {"events": {"TIMER": {"target": "green"}}, **pedestrian_states},
        },
        "events": {
            "POWER_OUTAGE": {"target": ".red.blinking"},
            "POWER_RESTORED": {"target": ".red"},
        },
    }

Example usage:

crossing = await Machine.create(crossing_config)

assert crossing.initial_state.value == "green"
assert crossing.state.type == "atomic"
await crossing.event("TIMER")
assert crossing.state.value == "yellow"
assert crossing.state.type == "atomic"
await crossing.event("TIMER")
assert crossing.state.value == {"red": {"walk": "walking"}}
await crossing.event("CROSSED")
assert crossing.state.value == {"red": {"walk": "crossed"}}
assert crossing.state.type == "compound"
await crossing.event("PED_COUNTDOWN")
assert crossing.state.value == {"red": "wait"}
await crossing.event("PED_COUNTDOWN")
assert crossing.state.value == {"red": "stop"}
await crossing.event("TIMER")
assert crossing.initial_state.value == "green"
assert crossing.state.type == "atomic"