src.dackar.RCA.signal_evidence¶

Stage B.5 signal evidence package.

Submodules¶

Classes¶

`SignalEvidenceBuilderContract`	Base class for protocol classes.
`StageCScorerContract`	Base class for protocol classes.
`StageFRefineContract`	Base class for protocol classes.
`HistorianAdapter`	Fetches pre-flagged anomaly records for a set of sensors in a window.
`InfileHistorianAdapter`	Reads pre-flagged anomalies from a JSON/CSV export.
`NullHistorianAdapter`	Graceful-degradation adapter used when historian is unavailable.
`OSIsoftPIHistorianAdapter`	NOT IMPLEMENTED — placeholder shim for a future OSIsoft PI integration.

Functions¶

build_signal_evidence(*, run_id, event, ...[, ...])

Build the Stage B.5 signal-evidence bundle for one RCA run.

Package Contents¶

src.dackar.RCA.signal_evidence.build_signal_evidence(*, run_id, event, telemetry_summary, kg_context, neo4j_client=None, neo4j_database=None, historian_adapter=None, fetch_lookback_hours=72.0, fetch_lookahead_hours=4.0, dedup_tolerance_min=5.0, max_paths=20, max_chains=10)[source]¶

Build the Stage B.5 signal-evidence bundle for one RCA run.

Merges baseline telemetry anomalies with historian-fetched anomalies, builds a component-level propagation DAG from Allen temporal relations and KG reachability, classifies node topology, enumerates and scores propagation chains, and derives per-failure-mode chain-position scores.

Parameters:

run_id (str) – Identifier for this analysis run; echoed into the bundle.
event (JsonDict) – Triggering event; timestamp_start / timestamp and timestamp_end seed the analysis window.
telemetry_summary (JsonDict) – Stage-B summary whose signals[].anomalies supply the baseline anomaly set.
kg_context (JsonDict) – KG context providing components[].monitored_variable_ids (the sensor↔component map) and failure_modes.
neo4j_client (Optional[Any]) – Optional live graph client. When None (the default), is_upstream degrades to False and resolve_edge_type to "mixed", so no propagation edges are built and the DAG, all propagation chains, and per-candidate scores come back empty; a {"type": "topology_unavailable"} entry is added to chain_warnings so consumers can tell this apart from “analyzed, no propagation found”.
neo4j_database (Optional[str]) – Target Neo4j database; None uses the driver default.
historian_adapter (Optional[src.dackar.RCA.signal_evidence.historian_adapter.HistorianAdapter]) – Anomaly source; defaults to NullHistorianAdapter (records a gap per sensor and returns no anomalies).
fetch_lookback_hours (float) – Hours before the event to widen the window (raised to the largest failure-mode expected_latency_max_hours when that is greater).
fetch_lookahead_hours (float) – Hours after the event to widen the window.
dedup_tolerance_min (float) – Minutes within which a historian anomaly is treated as a duplicate of a same-sensor baseline anomaly.
max_paths (int) – Cap on enumerated propagation paths (DFS guard).
max_chains (int) – Cap on scored chains retained in the bundle.

Returns:

run_id, generated_at, augmented_anomaly_set, propagation_chains (scored, ranked), per_candidate_chain_score, dag_topology_summary, chain_coverage, augmented_anomaly_count, historian_anomaly_count, fetch_gaps and chain_warnings.

Return type:

A JSON-serializable dict with keys

class src.dackar.RCA.signal_evidence.SignalEvidenceBuilderContract[source]¶

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

build(*, run_id, event, telemetry_summary, kg_context)[source]¶

Parameters:

run_id (str)
event (JsonDict)
telemetry_summary (JsonDict)
kg_context (JsonDict)

Return type:

JsonDict

class src.dackar.RCA.signal_evidence.StageCScorerContract[source]¶

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

score(event, telemetry_summary, kg_context, operational_context, run_context, signal_evidence=None)[source]¶

Parameters:

event (JsonDict)
telemetry_summary (JsonDict)
kg_context (JsonDict)
operational_context (Optional[JsonDict])
run_context (JsonDict)
signal_evidence (Optional[JsonDict])

Return type:

JsonDict

class src.dackar.RCA.signal_evidence.StageFRefineContract[source]¶

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...

refine_with_evidence(causality_candidates, evidence_bundle, kg_context=None, signal_evidence=None, entity_normalizer_cfg=None)[source]¶

Parameters:

causality_candidates (JsonDict)
evidence_bundle (JsonDict)
kg_context (Optional[JsonDict])
signal_evidence (Optional[JsonDict])
entity_normalizer_cfg (Optional[JsonDict])

Return type:

JsonDict

class src.dackar.RCA.signal_evidence.HistorianAdapter[source]¶

Bases: Protocol

Fetches pre-flagged anomaly records for a set of sensors in a window.

get_anomalies(sensor_ids, window_start, window_end)[source]¶

Return (anomaly_records, gaps) for sensor_ids within the window.

Parameters:

sensor_ids (list[str]) – Sensor/tag identifiers to fetch anomalies for.
window_start (datetime.datetime) – Inclusive UTC-aware start of the query window.
window_end (datetime.datetime) – Inclusive UTC-aware end of the query window.

Returns:

A tuple of the matched AnomalyRecord list and a list of gap dicts (one per sensor with no data / a fetch failure).

Return type:

tuple[list[src.dackar.RCA.signal_evidence.models.AnomalyRecord], list[dict]]

class src.dackar.RCA.signal_evidence.InfileHistorianAdapter(source_path)[source]¶

Reads pre-flagged anomalies from a JSON/CSV export.

Parameters:: source_path (str | pathlib.Path)

source_path¶

get_anomalies(sensor_ids, window_start, window_end)[source]¶

Parameters:

sensor_ids (list[str])
window_start (datetime.datetime)
window_end (datetime.datetime)

Return type:

tuple[list[src.dackar.RCA.signal_evidence.models.AnomalyRecord], list[dict]]

_load_rows()[source]¶

Return type:: List[dict]

class src.dackar.RCA.signal_evidence.NullHistorianAdapter[source]¶

Graceful-degradation adapter used when historian is unavailable.

get_anomalies(sensor_ids, window_start, window_end)[source]¶

Parameters:

sensor_ids (list[str])
window_start (datetime.datetime)
window_end (datetime.datetime)

Return type:

tuple[list[src.dackar.RCA.signal_evidence.models.AnomalyRecord], list[dict]]

class src.dackar.RCA.signal_evidence.OSIsoftPIHistorianAdapter[source]¶

NOT IMPLEMENTED — placeholder shim for a future OSIsoft PI integration.

This adapter is contract-compatible but not wired to the PI Web API. Every call reports each sensor as historian_unavailable and returns no anomalies, so it must not be mistaken for a functional PI integration. Use InfileHistorianAdapter for real data until PI is implemented.

get_anomalies(sensor_ids, window_start, window_end)[source]¶

Parameters:

sensor_ids (list[str])
window_start (datetime.datetime)
window_end (datetime.datetime)

Return type:

tuple[list[src.dackar.RCA.signal_evidence.models.AnomalyRecord], list[dict]]