src.dackar.RCA.signal_evidence.historian_adapter

Classes

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

Functions

_to_float(value)

Module Contents

class src.dackar.RCA.signal_evidence.historian_adapter.HistorianAdapter

Bases: Protocol

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

get_anomalies(sensor_ids, window_start, window_end)

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.historian_adapter.NullHistorianAdapter

Graceful-degradation adapter used when historian is unavailable.

get_anomalies(sensor_ids, window_start, window_end)

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.historian_adapter.InfileHistorianAdapter(source_path)

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)

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()

Return type:

List[dict]

class src.dackar.RCA.signal_evidence.historian_adapter.OSIsoftPIHistorianAdapter

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)

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]]

src.dackar.RCA.signal_evidence.historian_adapter._to_float(value)

Parameters:

value (object)

Return type:

Optional[float]