Discovery (Python API)

Discovery covers inferred FSM candidates, observation ingestion, shadow diffs, and pluggable stores. Conceptual guide: Discovery (inferred FSM) and Classification.

Below are the main service and store entrypoints; protocols and data models are available in the pystator.discovery package.

Engine and thresholds

InferenceEngine

InferenceEngine(observation_store: ObservationStore, inference_store: InferenceStore, built_candidate_store: BuiltDiscoveryCandidateStore, shadow_store: ShadowResultStore, thresholds: InferenceThresholds | None = None, *, draft_store: DraftStore | None = None)

Primary discovery API: ingest, infer, and shadow-compare.

Parameters:

Name	Type	Description	Default
observation_store	ObservationStore	Store for observed state events.	required
inference_store	InferenceStore	Store for inferred edges.	required
built_candidate_store	BuiltDiscoveryCandidateStore	Store for built (persisted) candidate FSM snapshots.	required
shadow_store	ShadowResultStore	Store for shadow comparison results.	required
thresholds	InferenceThresholds | None	Optional edge-count and confidence thresholds.	None
draft_store	DraftStore | None	Draft scopes before build; defaults to observation_store if it implements draft persistence (same combined store).	None

Source code in src/pystator/discovery/service.py

def __init__(
    self,
    observation_store: ObservationStore,
    inference_store: InferenceStore,
    built_candidate_store: BuiltDiscoveryCandidateStore,
    shadow_store: ShadowResultStore,
    thresholds: InferenceThresholds | None = None,
    *,
    draft_store: DraftStore | None = None,
) -> None:
    """Initialize the inference engine.

    Args:
        observation_store: Store for observed state events.
        inference_store: Store for inferred edges.
        built_candidate_store: Store for built (persisted) candidate FSM snapshots.
        shadow_store: Store for shadow comparison results.
        thresholds: Optional edge-count and confidence thresholds.
        draft_store: Draft scopes before build; defaults to *observation_store* if it
            implements draft persistence (same combined store).
    """
    self._observations = observation_store
    self._inference = inference_store
    self._built_candidates = built_candidate_store
    self._shadow = shadow_store
    self._thresholds = thresholds or InferenceThresholds()
    self._drafts: DraftStore = draft_store or observation_store  # type: ignore[assignment]

record_observation

record_observation(machine_name: str, event: ObservedStateEvent) -> bool

Record an observed state event for a machine.

Parameters:

Name	Type	Description	Default
machine_name	str	Expected machine name (must match event).	required
event	ObservedStateEvent	The observed state event to record.	required

Returns:

Type	Description
bool	True if the observation was successfully stored.

Raises:

Type	Description
ValueError	If the event's machine_name does not match.

Source code in src/pystator/discovery/service.py

def record_observation(self, machine_name: str, event: ObservedStateEvent) -> bool:
    """Record an observed state event for a machine.

    Args:
        machine_name: Expected machine name (must match event).
        event: The observed state event to record.

    Returns:
        True if the observation was successfully stored.

    Raises:
        ValueError: If the event's machine_name does not match.
    """
    if event.machine_name != machine_name:
        raise ValueError(
            f"Event machine_name '{event.machine_name}' does not match "
            f"expected '{machine_name}'"
        )
    return self._observations.add_observation(event)

preview_inference

preview_inference(machine_name: str, *, mode: Literal['inference', 'manual'] = 'inference', entity_ids: list[str] | None = None, min_edge_count: int | None = None, min_confidence: float | None = None) -> dict[str, Any]

Preview inferred edges without creating a candidate.

Parameters:

Name	Type	Description	Default
machine_name	str	Machine to infer edges for.	required
mode	Literal['inference', 'manual']	Inference mode ("inference" or "manual").	'inference'
entity_ids	list[str] | None	Optional filter to specific entities.	None
min_edge_count	int | None	Override minimum edge count threshold.	None
min_confidence	float | None	Override minimum confidence threshold.	None

Returns:

Type	Description
dict[str, Any]	Dict with inferred edges, selected edges, and metadata.

Source code in src/pystator/discovery/service.py

def preview_inference(
    self,
    machine_name: str,
    *,
    mode: Literal["inference", "manual"] = "inference",
    entity_ids: list[str] | None = None,
    min_edge_count: int | None = None,
    min_confidence: float | None = None,
) -> dict[str, Any]:
    """Preview inferred edges without creating a candidate.

    Args:
        machine_name: Machine to infer edges for.
        mode: Inference mode (``"inference"`` or ``"manual"``).
        entity_ids: Optional filter to specific entities.
        min_edge_count: Override minimum edge count threshold.
        min_confidence: Override minimum confidence threshold.

    Returns:
        Dict with inferred edges, selected edges, and metadata.
    """
    normalized_ids = sorted(
        {x.strip() for x in (entity_ids or []) if x and x.strip()}
    )
    if normalized_ids:
        observations: list[ObservedStateEvent] = []
        for entity_id in normalized_ids:
            observations.extend(
                self._observations.list_observations(
                    machine_name, entity_id=entity_id
                )
            )
    else:
        observations = self._observations.list_observations(machine_name)
    edges = self._infer_edges(observations)
    thresholds = self._resolve_thresholds(min_edge_count, min_confidence)
    selected = [
        e
        for e in edges
        if e.count >= thresholds.min_edge_count
        and e.confidence >= thresholds.min_confidence
    ]
    entity_ids_used = sorted(
        {obs.entity_id for obs in observations if obs.entity_id}
    )
    return {
        "machine_name": machine_name,
        "mode": mode,
        "entity_ids_used": entity_ids_used,
        "observation_count": len(observations),
        "edges": edges,
        "selected_edges": selected,
        "thresholds": {
            "min_edge_count": thresholds.min_edge_count,
            "min_confidence": thresholds.min_confidence,
        },
    }

build_candidate

build_candidate(machine_name: str, version: str | None = None, *, mode: Literal['inference', 'manual'] = 'inference', entity_ids: list[str] | None = None, edge_selection: list[tuple[str, str]] | None = None, min_edge_count: int | None = None, min_confidence: float | None = None, draft_id: str | None = None) -> BuiltDiscoveryCandidate

Build and store a candidate machine from observations.

Parameters:

Name	Type	Description	Default
machine_name	str	Machine to build a candidate for.	required
version	str | None	Explicit version string (auto-incremented if omitted).	None
mode	Literal['inference', 'manual']	"inference" uses threshold filtering; "manual" requires explicit edge_selection.	'inference'
entity_ids	list[str] | None	Optional filter to specific entities.	None
edge_selection	list[tuple[str, str]] | None	Required for manual mode; list of (source, dest) pairs to include.	None
min_edge_count	int | None	Override minimum edge count threshold.	None
min_confidence	float | None	Override minimum confidence threshold.	None
draft_id	str | None	Optional originating draft id (recorded in metadata).	None

Returns:

Type	Description
BuiltDiscoveryCandidate	The persisted built discovery candidate.

Raises:

Type	Description
ValueError	If manual mode is used without edge_selection or selection contains unknown edge pairs.

Source code in src/pystator/discovery/service.py

def build_candidate(
    self,
    machine_name: str,
    version: str | None = None,
    *,
    mode: Literal["inference", "manual"] = "inference",
    entity_ids: list[str] | None = None,
    edge_selection: list[tuple[str, str]] | None = None,
    min_edge_count: int | None = None,
    min_confidence: float | None = None,
    draft_id: str | None = None,
) -> BuiltDiscoveryCandidate:
    """Build and store a candidate machine from observations.

    Args:
        machine_name: Machine to build a candidate for.
        version: Explicit version string (auto-incremented if omitted).
        mode: ``"inference"`` uses threshold filtering; ``"manual"``
            requires explicit *edge_selection*.
        entity_ids: Optional filter to specific entities.
        edge_selection: Required for manual mode; list of
            ``(source, dest)`` pairs to include.
        min_edge_count: Override minimum edge count threshold.
        min_confidence: Override minimum confidence threshold.
        draft_id: Optional originating draft id (recorded in metadata).

    Returns:
        The persisted built discovery candidate.

    Raises:
        ValueError: If manual mode is used without *edge_selection*
            or selection contains unknown edge pairs.
    """
    preview = self.preview_inference(
        machine_name,
        mode=mode,
        entity_ids=entity_ids,
        min_edge_count=min_edge_count,
        min_confidence=min_confidence,
    )
    edges: list[InferredEdge] = preview["edges"]
    thresholds = self._resolve_thresholds(min_edge_count, min_confidence)
    selected: list[InferredEdge]
    normalized_selection = sorted(set(edge_selection or []))
    if mode == "manual":
        if not normalized_selection:
            raise ValueError("manual mode requires non-empty edge_selection")
        allowed = set(normalized_selection)
        selected = [
            e for e in edges if (e.source_state, e.destination_state) in allowed
        ]
        if len(selected) != len(allowed):
            existing = {(e.source_state, e.destination_state) for e in edges}
            missing = sorted(allowed - existing)
            raise ValueError(
                "edge_selection contains pairs not present in inferred edges: "
                f"{missing}"
            )
    else:
        selected = [
            e
            for e in edges
            if e.count >= thresholds.min_edge_count
            and e.confidence >= thresholds.min_confidence
        ]
    config = self._build_machine_config(machine_name, selected)
    meta: dict[str, Any] = {
        "edge_count": len(edges),
        "selected_edge_count": len(selected),
        "build_mode": mode,
        "build_scope": {
            "entity_ids": preview["entity_ids_used"],
            "edge_selection": [
                {"source_state": src, "destination_state": dst}
                for (src, dst) in normalized_selection
            ],
        },
        "observation_count": preview["observation_count"],
        "thresholds": {
            "min_edge_count": thresholds.min_edge_count,
            "min_confidence": thresholds.min_confidence,
        },
        "inferred_edges_snapshot": self._edges_to_dict(edges),
        "selected_edges_snapshot": self._edges_to_dict(selected),
    }
    if draft_id:
        meta["draft_id"] = draft_id
    candidate = BuiltDiscoveryCandidate(
        machine_name=machine_name,
        version=version or self._next_version(machine_name),
        status="candidate",
        config=config,
        metadata=meta,
    )
    return self._built_candidates.save_built_candidate(candidate)

list_built_candidates

list_built_candidates(machine_name: str) -> list[BuiltDiscoveryCandidate]

Return all saved built candidates for machine_name, store-defined order.

Source code in src/pystator/discovery/service.py

def list_built_candidates(self, machine_name: str) -> list[BuiltDiscoveryCandidate]:
    """Return all saved built candidates for *machine_name*, store-defined order."""
    return self._built_candidates.list_built_candidates(machine_name)

get_built_candidate

get_built_candidate(machine_name: str, version: str) -> BuiltDiscoveryCandidate | None

Return a specific built candidate version, if present.

Source code in src/pystator/discovery/service.py

def get_built_candidate(
    self, machine_name: str, version: str
) -> BuiltDiscoveryCandidate | None:
    """Return a specific built candidate version, if present."""
    return self._built_candidates.get_built_candidate(machine_name, version)

InferenceThresholds dataclass

InferenceThresholds(min_edge_count: int = 2, min_confidence: float = 0.5)

Store factory

create_discovery_store

create_discovery_store(backend: str, *, connection_string: str | None = None, options: dict[str, Any] | None = None) -> InMemoryDiscoveryStore | SQLiteDiscoveryStore | PostgresDiscoveryStore | MongoDBDiscoveryStore | RedisDiscoveryStore

Source code in src/pystator/discovery/stores/factory.py

def create_discovery_store(
    backend: str,
    *,
    connection_string: str | None = None,
    options: dict[str, Any] | None = None,
) -> (
    InMemoryDiscoveryStore
    | SQLiteDiscoveryStore
    | PostgresDiscoveryStore
    | MongoDBDiscoveryStore
    | RedisDiscoveryStore
):
    opts = options or {}
    normalized = backend.strip().lower()
    if normalized == "memory":
        return InMemoryDiscoveryStore()
    if normalized == "sqlite":
        if not connection_string:
            raise ValueError("connection_string is required for sqlite discovery store")
        return SQLiteDiscoveryStore(connection_string)
    if normalized in {"postgres", "postgresql"}:
        if not connection_string:
            raise ValueError(
                "connection_string is required for postgres discovery store"
            )
        return PostgresDiscoveryStore(connection_string)
    if normalized == "mongodb":
        if not connection_string:
            raise ValueError(
                "connection_string is required for mongodb discovery store"
            )
        return MongoDBDiscoveryStore(
            connection_string,
            database_name=str(opts.get("database_name", "pystator")),
        )
    if normalized == "redis":
        if not connection_string:
            raise ValueError("connection_string is required for redis discovery store")
        return RedisDiscoveryStore(
            connection_string,
            key_prefix=str(opts.get("key_prefix", "pystator:discovery")),
        )
    raise ValueError(f"Unsupported discovery backend: {backend}")

Classification

StateClassifier

StateClassifier(rule_set: ClassificationRuleSet)

Evaluates state when rules against data records to determine states.

States are evaluated in their definition order within the rule set. The first state whose when checks all pass wins.

Parameters:

Name	Type	Description	Default
rule_set	ClassificationRuleSet	The classification rule set containing states and settings.	required

Source code in src/pystator/discovery/classification/_classifier.py

def __init__(self, rule_set: ClassificationRuleSet) -> None:
    self._rule_set = rule_set

rule_set property

rule_set: ClassificationRuleSet

The classification rule set.

classify

classify(record: dict[str, Any]) -> ClassificationResult

Classify a single data record into a state.

Iterates classifiable states in order. For each state, all top-level when items are AND'd together. The first state where all checks pass wins.

Parameters:

Name	Type	Description	Default
record	dict[str, Any]	Dict of field values from the data source.	required

Returns:

Type	Description
ClassificationResult	ClassificationResult with the matched state, or
ClassificationResult	default_state if no rule matched.

Source code in src/pystator/discovery/classification/_classifier.py

def classify(self, record: dict[str, Any]) -> ClassificationResult:
    """Classify a single data record into a state.

    Iterates classifiable states in order. For each state, all
    top-level ``when`` items are AND'd together. The first state
    where all checks pass wins.

    Args:
        record: Dict of field values from the data source.

    Returns:
        ClassificationResult with the matched state, or
        default_state if no rule matched.
    """
    for state in self._rule_set.states:
        if all(evaluate_when_check(wc, record) for wc in state.when):
            logger.debug("Record matched state '%s'", state.name)
            return ClassificationResult(
                state=state.name,
                matched=True,
                record=record,
            )

    logger.debug(
        "No state matched; default_state=%s",
        self._rule_set.default_state,
    )
    return ClassificationResult(
        state=self._rule_set.default_state,
        matched=False,
        record=record,
    )

from_machine classmethod

from_machine(machine: StateMachine, *, entity_id_field: str | None = None, timestamp_field: str | None = None, source: str | None = None, default_state: str | None = None) -> StateClassifier

Create a classifier from a loaded StateMachine.

Extracts states that have when clauses in their definition order. Classification settings are read from machine.meta["classification"] and can be overridden via keyword arguments.

Parameters:

Name	Type	Description	Default
machine	StateMachine	A loaded StateMachine instance.	required
entity_id_field	str | None	Override for entity_id_field.	None
timestamp_field	str | None	Override for timestamp_field.	None
source	str | None	Override for source label.	None
default_state	str | None	Override for default_state.	None

Returns:

Type	Description
StateClassifier	A configured StateClassifier.

Raises:

Type	Description
ValueError	If no states have when clauses.

Source code in src/pystator/discovery/classification/_classifier.py

@classmethod
def from_machine(
    cls,
    machine: StateMachine,
    *,
    entity_id_field: str | None = None,
    timestamp_field: str | None = None,
    source: str | None = None,
    default_state: str | None = None,
) -> StateClassifier:
    """Create a classifier from a loaded StateMachine.

    Extracts states that have ``when`` clauses in their definition
    order. Classification settings are read from
    ``machine.meta["classification"]`` and can be overridden via
    keyword arguments.

    Args:
        machine: A loaded StateMachine instance.
        entity_id_field: Override for entity_id_field.
        timestamp_field: Override for timestamp_field.
        source: Override for source label.
        default_state: Override for default_state.

    Returns:
        A configured StateClassifier.

    Raises:
        ValueError: If no states have ``when`` clauses.
    """
    classifiable = [s for s in machine.states.values() if s.when]
    if not classifiable:
        raise ValueError(
            "No states with 'when' clauses found in machine "
            f"'{machine.meta.get('machine_name', '?')}'"
        )

    meta_cls = machine.meta.get("classification", {})

    rule_set = ClassificationRuleSet(
        states=tuple(classifiable),
        default_state=(
            default_state
            if default_state is not None
            else meta_cls.get("default_state")
        ),
        entity_id_field=(
            entity_id_field or meta_cls.get("entity_id_field", "entity_id")
        ),
        timestamp_field=(
            timestamp_field
            if timestamp_field is not None
            else meta_cls.get("timestamp_field")
        ),
        source=source or meta_cls.get("source", "classifier"),
    )
    return cls(rule_set)

DataIngester

DataIngester(classifier: StateClassifier, engine: InferenceEngine, machine_name: str, *, skip_unclassified: bool = True)

Ingests raw data records through classification into InferenceEngine.

Supports record-by-record streaming and batch (list) ingestion. For each record: classify state -> create ObservedStateEvent -> feed to InferenceEngine.record_observation().

Parameters:

Name	Type	Description	Default
classifier	StateClassifier	StateClassifier with rules for state determination.	required
engine	InferenceEngine	InferenceEngine to feed observations into.	required
machine_name	str	Target machine name for observations.	required
skip_unclassified	bool	If True, silently skip records with no state. If False, raise ValueError for unclassified records.	True

Source code in src/pystator/discovery/classification/_ingester.py

def __init__(
    self,
    classifier: StateClassifier,
    engine: InferenceEngine,
    machine_name: str,
    *,
    skip_unclassified: bool = True,
) -> None:
    self._classifier = classifier
    self._engine = engine
    self._machine_name = machine_name
    self._skip_unclassified = skip_unclassified
    self._ingest_seq: int = 0

classifier property

classifier: StateClassifier

The state classifier.

machine_name property

machine_name: str

Target machine name for observations.

ingest_record

ingest_record(record: dict[str, Any]) -> ClassificationResult

Classify and ingest a single record (streaming).

Parameters:

Name	Type	Description	Default
record	dict[str, Any]	Raw data record with field values.	required

Returns:

Type	Description
ClassificationResult	ClassificationResult for this record.

Raises:

Type	Description
ValueError	If state is None and skip_unclassified is False.
ValueError	If entity_id field is missing from record.

Source code in src/pystator/discovery/classification/_ingester.py

def ingest_record(self, record: dict[str, Any]) -> ClassificationResult:
    """Classify and ingest a single record (streaming).

    Args:
        record: Raw data record with field values.

    Returns:
        ClassificationResult for this record.

    Raises:
        ValueError: If state is None and skip_unclassified is False.
        ValueError: If entity_id field is missing from record.
    """
    result = self._classifier.classify(record)
    if result.state is None:
        if not self._skip_unclassified:
            raise ValueError(
                "No state determined for record and no default_state "
                f"configured: {record}"
            )
        logger.debug("Skipping unclassified record")
        return result

    event = self._build_event(record, result.state)
    self._engine.record_observation(self._machine_name, event)
    return result

ingest_batch

ingest_batch(records: list[dict[str, Any]]) -> IngestionResult

Classify and ingest a batch of records.

Each record is processed independently; one bad record does not abort the batch.

Parameters:

Name	Type	Description	Default
records	list[dict[str, Any]]	List of raw data records.	required

Returns:

Type	Description
IngestionResult	IngestionResult with counts of accepted, skipped, and errors.

Source code in src/pystator/discovery/classification/_ingester.py

def ingest_batch(self, records: list[dict[str, Any]]) -> IngestionResult:
    """Classify and ingest a batch of records.

    Each record is processed independently; one bad record does
    not abort the batch.

    Args:
        records: List of raw data records.

    Returns:
        IngestionResult with counts of accepted, skipped, and errors.
    """
    accepted = 0
    classified = 0
    skipped = 0
    errors = 0

    for record in records:
        try:
            result = self._classifier.classify(record)
            if result.state is None:
                skipped += 1
                continue
            event = self._build_event(record, result.state)
            classified += 1
            if self._engine.record_observation(self._machine_name, event):
                accepted += 1
        except Exception:
            logger.warning(
                "Error ingesting record: %s",
                record,
                exc_info=True,
            )
            errors += 1

    return IngestionResult(
        total=len(records),
        accepted=accepted,
        classified=classified,
        skipped=skipped,
        errors=errors,
    )

Key models (selected)

ObservedStateEvent dataclass

ObservedStateEvent(entity_id: str, machine_name: str, state: str, observed_at: datetime, source: str = 'unknown', source_event_id: str | None = None, metadata: dict[str, Any] = dict(), ingested_at: datetime = _utc_now(), ingest_seq: int = 0)

One observed state for an entity.

Attributes:

Name	Type	Description
entity_id	str	Identifier for the entity being observed.
machine_name	str	FSM machine this observation belongs to.
state	str	The observed state name.
observed_at	datetime	When the state was observed.
source	str	Origin system or label for this observation.
source_event_id	str | None	Optional upstream event ID for traceability.
metadata	dict[str, Any]	Additional user-defined metadata.
ingested_at	datetime	When this event was ingested into the store.
ingest_seq	int	Monotonic sequence number within an ingestion session.

BuiltDiscoveryCandidate dataclass

BuiltDiscoveryCandidate(machine_name: str, version: str, status: str, config: dict[str, Any], created_at: datetime = _utc_now(), metadata: dict[str, Any] = dict(), candidate_sequence: int = 0)

Persisted built FSM snapshot from discovery (a versioned candidate machine).

InferredEdge dataclass

InferredEdge(source_state: str, destination_state: str, count: int, unique_entities: int, unique_sources: int, confidence: float, last_seen_at: datetime | None = None)

Weighted transition inferred from observed sequences.

Imports

from pystator.discovery import (
    InferenceEngine,
    InferenceThresholds,
    create_discovery_store,
    StateClassifier,
    DataIngester,
    ObservedStateEvent,
    BuiltDiscoveryCandidate,
)

Additional exports (DiscoveryDraft, ShadowDiff, PostgresDiscoveryStore, ClassificationResult, …) are listed in pystator.discovery.__all__.