strepsiades
89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Mapping, Optional
|
|
|
|
from .evaluator import guard_matches
|
|
from .invariants import InvariantViolation
|
|
from .model import StateMachineConfig
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Decision:
|
|
event_id: str
|
|
from_state: str
|
|
outcome_index: int
|
|
to_state: Optional[str]
|
|
emit: Optional[Mapping[str, Any]]
|
|
|
|
|
|
class StateMachineRuntime:
|
|
def __init__(self, config: StateMachineConfig):
|
|
self._config = config
|
|
|
|
def dispatch(
|
|
self,
|
|
*,
|
|
current_state: str,
|
|
event_id: str,
|
|
context: Mapping[str, Any],
|
|
) -> Optional[Decision]:
|
|
transition = self._config.transitions.get(event_id)
|
|
if transition is None:
|
|
raise KeyError(f"Unknown event: {event_id}")
|
|
|
|
if transition.from_states and current_state not in transition.from_states and "*" not in transition.from_states:
|
|
return None
|
|
|
|
for index, outcome in enumerate(transition.outcomes):
|
|
if guard_matches(outcome.when, context):
|
|
return Decision(
|
|
event_id=event_id,
|
|
from_state=current_state,
|
|
outcome_index=index,
|
|
to_state=outcome.to,
|
|
emit=outcome.emit,
|
|
)
|
|
|
|
return None
|
|
|
|
def check_invariants(self, snapshot: Mapping[str, Any]) -> list[InvariantViolation]:
|
|
def _match_all(cond: Mapping[str, Any]) -> bool:
|
|
for key, value in cond.items():
|
|
if snapshot.get(key) != value:
|
|
return False
|
|
return True
|
|
|
|
violations: list[InvariantViolation] = []
|
|
for invariant_id, idef in self._config.invariants.items():
|
|
if not isinstance(idef, dict):
|
|
continue
|
|
|
|
should_check = False
|
|
when = idef.get("when")
|
|
when_any = idef.get("when_any")
|
|
if isinstance(when, dict):
|
|
should_check = _match_all(when)
|
|
elif isinstance(when_any, list):
|
|
for cond in when_any:
|
|
if isinstance(cond, dict) and _match_all(cond):
|
|
should_check = True
|
|
break
|
|
|
|
if not should_check:
|
|
continue
|
|
|
|
require = idef.get("require")
|
|
if not isinstance(require, dict):
|
|
continue
|
|
|
|
for rk, rv in require.items():
|
|
if snapshot.get(rk) != rv:
|
|
violations.append(
|
|
InvariantViolation(
|
|
invariant_id=str(invariant_id),
|
|
message=f"require {rk}={rv}, got {snapshot.get(rk)}",
|
|
)
|
|
)
|
|
return violations
|