본문 바로가기

Python

python-statemachine guard # validators # cond # unless

728x90
반응형

StateMachine Guard Call Flow

Context

  • Library: python-statemachine 2.5.0 (see venv/lib/python3.11/site-packages)
  • Goal: control whether a transition executes before on_*/after_* hooks run.

Guard Types

validators
* Purpose: fail-fast validation. Raise on invalid payloads or environment.
* Behavior: executed first; any exception stops the transition immediately.
* Return value: ignored. Success is "no exception". Exceptions propagate.

cond
* Purpose: allow transition only when every guard returns truthy.
* Behavior: evaluated after validators. If any returns falsy, the transition
is skipped silently (state remains unchanged, no further callbacks).
* Return value: truthy keeps evaluating; falsy vetoes.

unless
* Purpose: veto list that should return False. Any truthy return blocks.
* Behavior: runs after cond. Works like not cond. Useful for safety
interlocks or emergency stops.
* Return value: falsy allows proceed; truthy vetoes.

Signature & Arguments

  • Register guards as method names (strings), direct callables, or iterables of either.
  • Each guard receives the event's positional/keyword arguments plus:
    event_data, machine, model, transition, state, source, target.
  • Async callables are supported. If any guard is awaitable, the engine switches
    to async processing automatically.

Call Order For A Successful Transition

  1. Validators
  2. cond
  3. unless
  4. before_transition and before_<event>
  5. source.exit (skipped for internal transitions)
  6. on_transition and on_<event>
  7. Assign target state
  8. target.enter (skipped for internal transitions)
  9. after_<event> and after_transition

If any guard raises, returns falsy (cond), or truthy (unless), steps 4-9 are
skipped and the state remains at the source.

Worked Example

from statemachine import State
from crp_core.comm.czmq.message import Message

class PickingZoneScenario(ScenarioBase):
    picking_vision_capture = State("PickingVisionCapture")
    classify_boxes = State("ClassifyBoxes")

    start_classify_boxes = picking_vision_capture.to(
        classify_boxes,
        validators=["validate_capture_payload"],
        cond=["camera_ready", "ai_idle"],
        unless=["safety_lock_engaged"],
    )

    def validate_capture_payload(self, cmd: Message, **_):
        if "image_path" not in cmd.param:
            raise ValueError("camera response missing image_path")

    def camera_ready(self, **_):
        return self.camera_controller.is_ready()

    def ai_idle(self, **_):
        return self.picking_ai_controller.is_idle()

    def safety_lock_engaged(self, **_):
        return self.config.get("emergency_stop", False)

    def before_start_classify_boxes(self, cmd: Message, **_):
        self.logger.debug("Preparing classify run for %s", cmd.param.get("image_path"))

    def after_start_classify_boxes(self, cmd: Message, **_):
        self.picking_ai_controller.send_picking_ai_req_operation(cmd.param)

python-statemachine Guard Hooks

Python-statemachine (2.5.0) evaluates guard callbacks before executing a transition. Guards decide whether the transition continues to the before_*, on_*, after_*, and state entry/exit hooks.

Call Order

  1. validators
  2. cond
  3. unless
  4. before_transition & before_<event>
  5. source.exit (skipped if internal=True)
  6. on_transition & on_<event>
  7. Assign target state
  8. target.enter (skipped if internal=True)
  9. after_<event> & after_transition

Any guard failure (exception, falsy cond, or truthy unless) stops here; the state stays at the source and later callbacks do not run.

Guard Catalog

Validators

  • Use when you need a hard failure.
  • Execution: first.
  • Success condition: no exception raised.
  • Failure: raise ValueError, RuntimeError, etc. to abort.

Conditions (cond)

  • Use when all prerequisites must be true.
  • Execution: after validators.
  • Success condition: every callable returns truthy.
  • Failure: any falsy value vetoes silently.

Unless

  • Use when you maintain a stop-list (e.g., safety interlocks).
  • Execution: after cond.
  • Success condition: every callable returns falsy.
  • Failure: any truthy value blocks silently.

Arguments & Registration

Each guard receives the event's arguments plus event_data, machine, model, transition, state, source, and target. Register guards by:

  • String name of a method on the state machine ("camera_ready").
  • Passing a callable (instance method, @classmethod, @staticmethod, or free function).
  • Mixing the above in iterables; async callables are supported automatically.

Scenario Example

from statemachine import State
from crp_core.comm.czmq.message import Message

class PickingZoneScenario(ScenarioBase):
    picking_vision_capture = State("PickingVisionCapture")
    classify_boxes = State("ClassifyBoxes")

    start_classify_boxes = picking_vision_capture.to(
        classify_boxes,
        validators=["validate_capture_payload"],
        cond=["camera_ready", "ai_idle"],
        unless=["safety_lock_engaged"],
    )

    def validate_capture_payload(self, cmd: Message, **_):
        if "image_path" not in cmd.param:
            raise ValueError("camera response missing image_path")

    def camera_ready(self, **_):
        return self.camera_controller.is_ready()

    def ai_idle(self, **_):
        return self.picking_ai_controller.is_idle()

    def safety_lock_engaged(self, **_):
        return self.config.get("emergency_stop", False)

Behavior:

  • Missing image_path raises and stops the transition.
  • Any falsy cond keeps the state in PickingVisionCapture without an exception.
  • Truthy safety_lock_engaged vetoes regardless of cond results.

Classmethod Guards

class PickingZoneScenario(ScenarioBase):
    start_classify_boxes = picking_vision_capture.to(
        classify_boxes,
        validators=[cls.validate_payload],
        cond=["camera_ready", cls.ai_idle],
        unless=[cls.safety_lock_engaged],
    )

    @classmethod
    def validate_payload(cls, cmd: Message, **_):
        if "image_path" not in cmd.param:
            raise ValueError("camera response missing image_path")

    def camera_ready(self, **_):
        return self.camera_controller.is_ready()

    @classmethod
    def ai_idle(cls, **_):
        return cls.global_ai_state.is_idle()

    @classmethod
    def safety_lock_engaged(cls, **_):
        return cls.config_store.get("emergency_stop", False)

When python-statemachine resolves the guard name, class methods are bound automatically with cls set to the class.

Cross-Class Guards

class GlobalGuards:
    @classmethod
    def payload_valid(cls, cmd: Message, **_):
        if "image_path" not in cmd.param:
            raise ValueError("Missing image_path")

    @classmethod
    def system_ready(cls, **_):
        return cls.status_store.ready

    @classmethod
    def safety_lock(cls, **_):
        return cls.status_store.emergency_stop

class PickingZoneScenario(ScenarioBase):
    start_classify_boxes = picking_vision_capture.to(
        classify_boxes,
        validators=[GlobalGuards.payload_valid],
        cond=[GlobalGuards.system_ready, "camera_ready"],
        unless=[GlobalGuards.safety_lock],
    )

    def camera_ready(self, **_):
        return self.camera_controller.is_ready()

Passing callables from other classes works because the engine executes the provided callable directly; no listener registration is needed.

Troubleshooting

  • Use guards for silent vetoes and validators for explicit failures.
  • before_* hooks never cancel transitions—raise or guard instead.
  • Inspect registered callbacks via machine._callbacks.str(<key>) for debugging (statemachine/callbacks.py).

Source References

  • Transition definitions: venv/lib/python3.11/site-packages/statemachine/transition.py
  • Callback registry: .../statemachine/callbacks.py
  • Engine call order: .../statemachine/engines/sync.py

Observed Effects

  • Missing image_path raises from validate_capture_payload; transition aborts
    with an exception and no callbacks execute.
  • If either camera_ready or ai_idle returns falsy, the transition is ignored;
    the machine stays in PickingVisionCapture, no exception.
  • If safety_lock_engaged returns truthy, the unless veto fires and the
    transition is skipped regardless of cond results.

Troubleshooting

  • Need a silent veto? Use cond/unless. Need an explicit error? Use a validator
    and raise.
  • Returning False from before_* has no effect on transition flow in
    python-statemachine 2.x. Use guards or raise instead.
  • To inspect guard registration, call print(machine._callbacks.str(<key>)) where
    <key> is the callback group key (see statemachine/callbacks.py).
728x90
반응형