Classification: State Machines from Data

PyStator can classify raw data records into states using declarative rules, then feed those states into the Discovery subsystem to infer state machine transitions automatically. This lets you build an FSM directly from an incoming data stream rather than defining every transition by hand.

Use classification when you have:

• Continuous data feeds (sensor readings, order updates, system metrics) where field values implicitly define the system's current state.
• State rules you can express declaratively — e.g. "overheating when temperature > 100 AND pressure < 50".
• Unknown transitions — you know the states, but want the system to discover which transitions actually occur.

How it works

Raw data record (dict of field values)
        │
        ▼
 StateClassifier  ── evaluates per-state `when` rules ──▶ state name
        │
        ▼
 DataIngester     ── wraps into ObservedStateEvent
        │
        ▼
 InferenceEngine.record_observation()   (existing discovery)
        │
        ▼
 InferenceEngine.build_candidate()      (inferred FSM config)

1. Each State in your YAML config can have an optional when clause — a list of field checks.
2. The StateClassifier evaluates these rules against each incoming data record. States are checked in definition order; the first state whose when checks all pass wins.
3. The DataIngester wraps the classified state into an ObservedStateEvent and feeds it to the existing InferenceEngine.
4. After enough observations, call build_candidate() to get an inferred FSM with transitions, confidence scores, and edge counts.

---

YAML configuration

Classification rules live per-state via the when key. States without when are not classifiable from data — they are only reachable via explicit transitions or as a default.

Classification settings (entity ID field, timestamp field, source label, default state) go in meta.classification.

meta:
  machine_name: sensor_fsm
  version: "1.0.0"
  classification:
    entity_id_field: sensor_id
    timestamp_field: recorded_at
    source: sensor_feed
    default_state: unknown

states:
  - name: overheating
    type: stable
    when:
      - field: temperature
        op: gt
        value: 100
      - field: pressure
        op: lt
        value: 50

  - name: normal
    type: stable
    when:
      - field: temperature
        op: gte
        value: 20
      - field: temperature
        op: lte
        value: 100

  - name: shutdown
    type: terminal
    when:
      - field: status
        op: in
        values: ["off", "shutdown"]

  - name: unknown
    type: stable
    # no `when` — only reachable as default_state

transitions:
  - trigger: cool_down
    source: [overheating]
    dest: normal

meta.classification settings

Field	Default	Description
entity_id_field	entity_id	Record field to extract the entity identifier from.
timestamp_field	(none)	Record field to extract the observation timestamp. If absent, datetime.now(UTC) is used.
source	classifier	Source label attached to each ObservedStateEvent.
default_state	(none)	State name assigned when no when clause matches. None means unclassified records are skipped.

---

Operators

Each check in a when clause is a {field, op, value} dict. The same operators are used in declarative guards.

Operator	Description	Example
eq	Equal	{field: status, op: eq, value: "active"}
neq	Not equal	{field: status, op: neq, value: "disabled"}
gt	Greater than	{field: temperature, op: gt, value: 100}
gte	Greater than or equal	{field: temperature, op: gte, value: 20}
lt	Less than	{field: pressure, op: lt, value: 50}
lte	Less than or equal	{field: pressure, op: lte, value: 100}
in	Value in list	{field: status, op: in, values: ["off", "shutdown"]}
not_in	Value not in list	{field: region, op: not_in, values: ["US", "EU"]}
is_set	Field is not None	{field: error_code, op: is_set}
is_null	Field is None	{field: error_code, op: is_null}
contains	String contains substring	{field: msg, op: contains, value: "error"}
starts_with	String starts with prefix	{field: code, op: starts_with, value: "ERR"}
ends_with	String ends with suffix	{field: file, op: ends_with, value: ".csv"}
matches	Regex search	{field: msg, op: matches, value: "^FATAL"}

---

Composite logic (OR, AND, NOT)

By default, all checks in a when clause are AND'd — every check must pass. For OR or NOT logic, use composites:

states:
  - name: alert
    type: stable
    when:
      # OR: at least one must pass
      - any_of:
          - field: temperature
            op: gt
            value: 100
          - field: pressure
            op: gt
            value: 200

  - name: healthy
    type: stable
    when:
      # NOT: status must NOT be "degraded"
      - negate:
          field: status
          op: eq
          value: degraded
      # AND: temperature in range (explicit all_of, same as top-level AND)
      - all_of:
          - field: temperature
            op: gte
            value: 20
          - field: temperature
            op: lte
            value: 80

Composite	Syntax	Description
any_of	any_of: [check, ...]	OR — at least one child must pass.
all_of	all_of: [check, ...]	AND — all children must pass (nestable).
negate	negate: check	NOT — inverts the child result.

Composites can be nested: an any_of can contain all_of groups, which can contain negate items.

---

Usage

Direct (embedded)

The simplest path — no server, no queue. Classify and ingest in-process:

from pystator import StateMachine
from pystator.discovery import (
    InferenceEngine,
    InferenceThresholds,
    InMemoryDiscoveryStore,
    StateClassifier,
    DataIngester,
)

# 1. Load machine (states with `when` clauses)
machine = StateMachine.from_yaml("sensor_fsm.yaml")

# 2. Create classifier from the machine config
classifier = StateClassifier.from_machine(machine)

# 3. Set up discovery engine
store = InMemoryDiscoveryStore()
engine = InferenceEngine(
    store, store, store, store,
    thresholds=InferenceThresholds(min_edge_count=2, min_confidence=0.5),
)

# 4. Create ingester
ingester = DataIngester(classifier, engine, machine_name="sensor_fsm")

# 5a. Stream records one at a time
result = ingester.ingest_record({
    "sensor_id": "s1",
    "temperature": 105,
    "pressure": 40,
    "recorded_at": "2026-03-25T10:00:00",
})
print(result.state)    # "overheating"
print(result.matched)  # True

# 5b. Or ingest a batch
batch_result = ingester.ingest_batch([
    {"sensor_id": "s1", "temperature": 50, "pressure": 60,
     "recorded_at": "2026-03-25T10:01:00"},
    {"sensor_id": "s1", "temperature": 110, "pressure": 30,
     "recorded_at": "2026-03-25T10:02:00"},
    {"sensor_id": "s2", "temperature": 15, "pressure": 10,
     "status": "shutdown",
     "recorded_at": "2026-03-25T10:00:00"},
])
print(batch_result)
# IngestionResult(total=3, accepted=3, classified=3, skipped=0, errors=0)

# 6. Build the inferred state machine
candidate = engine.build_candidate("sensor_fsm")
for t in candidate.config["transitions"]:
    print(f"  {t['source']} -> {t['dest']}  (count={t.get('meta', {}).get('count')})")

With the Orchestrator (stateful transitions)

Combine classification with the Orchestrator to both discover transitions and drive actual state changes on entities:

from pystator import StateMachine, Orchestrator, InMemoryStateStore
from pystator.discovery import (
    InferenceEngine, InMemoryDiscoveryStore,
    StateClassifier, DataIngester,
)

# Load and wire up
machine = StateMachine.from_yaml("sensor_fsm.yaml")

@machine.guard("is_critical")
def is_critical(ctx):
    return ctx.get("temperature", 0) > 150

@machine.action("send_alert")
def send_alert(ctx):
    print(f"ALERT: {ctx.get('_entity_id')} overheating!")

# Orchestrator for driving real transitions
state_store = InMemoryStateStore()
orchestrator = Orchestrator(
    machine=machine,
    state_store=state_store,
    guards=machine.guard_registry,
    actions=machine.action_registry,
)

# Classifier for detecting state from raw data
classifier = StateClassifier.from_machine(machine)

# Discovery for learning transitions
disc_store = InMemoryDiscoveryStore()
engine = InferenceEngine(disc_store, disc_store, disc_store, disc_store)
ingester = DataIngester(classifier, engine, "sensor_fsm")

def process_reading(reading: dict) -> None:
    """Handle an incoming sensor reading."""
    # Classify raw data into a state
    result = classifier.classify(reading)
    if result.state is None:
        return

    # Feed into discovery (learns transition patterns)
    ingester.ingest_record(reading)

    # Drive real FSM transition via orchestrator
    orchestrator.process_event(
        entity_id=reading["sensor_id"],
        event=f"to_{result.state}",
        context=reading,
    )

# Process incoming data
process_reading({"sensor_id": "s1", "temperature": 50, "pressure": 60})
process_reading({"sensor_id": "s1", "temperature": 110, "pressure": 30})

With the concurrent Worker

For production workloads, use the Worker to process classified events through a durable queue with concurrency, retries, and dead-letter handling:

import asyncio
from pystator import StateMachine
from pystator.discovery import (
    InferenceEngine, InMemoryDiscoveryStore,
    StateClassifier, DataIngester,
)
from pystator.worker import submit_event

machine = StateMachine.from_yaml("sensor_fsm.yaml")
classifier = StateClassifier.from_machine(machine)

# Discovery (optional — for learning transitions)
disc_store = InMemoryDiscoveryStore()
engine = InferenceEngine(disc_store, disc_store, disc_store, disc_store)
ingester = DataIngester(classifier, engine, "sensor_fsm")


async def process_reading(reading: dict) -> None:
    """Classify and enqueue for the Worker."""
    result = classifier.classify(reading)
    if result.state is None:
        return

    # Feed into discovery
    ingester.ingest_record(reading)

    # Submit to the Worker queue (durable, with retries)
    await submit_event(
        machine_name="sensor_fsm",
        entity_id=reading["sensor_id"],
        trigger=f"to_{result.state}",
        context=reading,
    )


async def main():
    readings = [
        {"sensor_id": "s1", "temperature": 50, "pressure": 60},
        {"sensor_id": "s1", "temperature": 110, "pressure": 30},
        {"sensor_id": "s1", "temperature": 70, "pressure": 55},
    ]
    for r in readings:
        await process_reading(r)

asyncio.run(main())

Then run the Worker separately to process queued events with concurrency:

pystator worker --concurrency 5 --poll-interval 500

The Worker atomically claims events, runs them through the Orchestrator, persists state, and handles retries. See Worker guide for full configuration.

Via REST API

Send classified data as events through the API:

# Start the API server
pystator api --host 0.0.0.0 --port 8004

# From your data pipeline (curl, Python requests, etc.)
curl -X POST http://localhost:8004/api/v1/entities/sensor-42/events \
  -H "Content-Type: application/json" \
  -d '{
    "machine_name": "sensor_fsm",
    "trigger": "to_overheating",
    "context": {"temperature": 120, "pressure": 40}
  }'

Or combine classification with the API by running the classifier in your producer and submitting the trigger:

import httpx
from pystator import StateMachine
from pystator.discovery import StateClassifier

machine = StateMachine.from_yaml("sensor_fsm.yaml")
classifier = StateClassifier.from_machine(machine)

def send_reading(reading: dict) -> None:
    result = classifier.classify(reading)
    if result.state is None:
        return

    httpx.post(
        f"http://localhost:8004/api/v1/entities/{reading['sensor_id']}/events",
        json={
            "machine_name": "sensor_fsm",
            "trigger": f"to_{result.state}",
            "context": reading,
        },
    )

---

Programmatic rule sets

You can build a ClassificationRuleSet without a YAML file:

from pystator._types import CheckSpec, State, StateType, WhenCheck
from pystator.discovery.classification import (
    ClassificationRuleSet,
    StateClassifier,
)

hot = State(
    name="hot",
    type=StateType.STABLE,
    when=(WhenCheck(check=CheckSpec(field="temp", op="gt", value=100)),),
)
cold = State(
    name="cold",
    type=StateType.STABLE,
    when=(WhenCheck(check=CheckSpec(field="temp", op="lt", value=20)),),
)

rule_set = ClassificationRuleSet(
    states=(hot, cold),
    default_state="normal",
    entity_id_field="device_id",
    source="device_feed",
)
classifier = StateClassifier(rule_set)

result = classifier.classify({"device_id": "d1", "temp": 150})
print(result.state)  # "hot"

---

ClassificationResult and IngestionResult

ClassificationResult

Returned by StateClassifier.classify():

Field	Type	Description
state	str \| None	Classified state name, or None if no match and no default.
matched	bool	True if a state's when clause matched. False means the default was used.
record	dict	The original data record.

IngestionResult

Returned by DataIngester.ingest_batch():

Field	Type	Description
total	int	Total records processed.
accepted	int	Records stored in the observation store.
classified	int	Records where a state was determined and event was built.
skipped	int	Records with no state (unclassified, no default).
errors	int	Records that caused exceptions during ingestion.

---

See also

• Discovery (inferred FSM) — How the inference engine builds candidate machines from observations.
• Worker — Running the concurrent event-processing service.
• Declarative features — Guards and effects using the same CheckSpec operators.
• Configuration — Full YAML config reference.