src.dackar.RCA.log_pattern_recognition.rca_pattern_search.extractor¶

IncidentExtractor — event normalisation and fingerprint derivation.

Converts raw alarm / SOE / anomaly records into a unified event list and derives the three fingerprint representations (event_set, event_seq, freq_vec) used by all downstream similarity metrics.

Attributes¶

`_log`
`_DEFAULT_SEVERITY_THRESHOLD`

Classes¶

IncidentExtractor

Converts raw source records to UnifiedEvents and derives IncidentFingerprints.

Functions¶

`_expand_window`(window_start, window_end, beta)	Applies beta buffer expansion symmetrically to a time window.
`_compute_density`(events, window_start, window_end)	Computes event density over the given (unexpanded) window.
`_dominant_asset`(events)	Returns the most frequently occurring asset_id among events, or '' if empty.
`_parse_ts`(value)	Coerces a value to datetime.
`_derive_soe_end_timestamps`(records)	Derives timestamp_end for each SOE record as the timestamp of the next

Module Contents¶

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.extractor._log[source]¶

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.extractor._DEFAULT_SEVERITY_THRESHOLD = 0.5[source]¶

class src.dackar.RCA.log_pattern_recognition.rca_pattern_search.extractor.IncidentExtractor(config)[source]¶

Converts raw source records to UnifiedEvents and derives IncidentFingerprints.

Two public entry points:: to_unified_events() — normalise all three sources into a flat list extract() — full pipeline: expand window → filter → fingerprint

_derive_fingerprint() is a staticmethod so that indexer.py can call it directly on pre-filtered episode event lists without instantiating an extractor.

Parameters:: config (src.dackar.RCA.log_pattern_recognition.rca_pattern_search.config.SearchConfig)

config[source]¶

to_unified_events(alarm_log, soe_log, telemetry_summaries, *, anomaly_severity_threshold=_DEFAULT_SEVERITY_THRESHOLD)[source]¶

Converts raw schema records to a flat UnifiedEvent list.

Applies filtering defaults:

Alarms with state == “suppressed” are excluded.
Anomalies with promoted_to_kg_event == False are excluded. If the field is absent, severity_score >= anomaly_severity_threshold is used as the inclusion gate.
All SOE records are included.

Parameters:

alarm_log (dict) – Dict with key “alarms”: list of alarm dicts.
soe_log (dict) – Dict with key “records”: list of SOE record dicts.
telemetry_summaries (list[dict]) – List of telemetry summary dicts, each with key “anomalies”: list of anomaly dicts.
anomaly_severity_threshold (float) – Fallback gate when promoted_to_kg_event absent.

Returns:

Flat list of UnifiedEvent, unsorted.

Return type:

list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent]

extract(alarm_log, soe_log, telemetry_summaries, incident_id, window_start, window_end, metadata=None)[source]¶

Full extraction pipeline for a single query incident.

Steps:

Apply beta buffer to compute expanded window.
Call to_unified_events() for all sources.
Filter events to those with timestamp_start in expanded window.
Compute density over the expanded window (consistent with historical episode density).
Derive event_set, event_seq, freq_vec via _derive_fingerprint().

Parameters:

alarm_log (dict) – Raw alarm log dict.
soe_log (dict) – Raw SOE log dict.
telemetry_summaries (list[dict]) – List of telemetry summary dicts.
incident_id (str) – Identifier for this query incident.
window_start (datetime.datetime) – Incident window start (before buffer expansion).
window_end (datetime.datetime) – Incident window end (before buffer expansion).
metadata (Optional[dict]) – Optional dict; “known_rca” and “asset_id” are read if present.

Returns:

IncidentFingerprint with expanded window stored as window_start/end.

Return type:

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.IncidentFingerprint

static _derive_fingerprint(events, freq_threshold)[source]¶

Derives the three similarity representations from a list of events.

High-frequency event types (count > freq_threshold) are excluded from event_set and event_seq but retained in freq_vec.

Parameters:

events (list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent]) – Events belonging to a single episode or incident.
freq_threshold (int) – Count above which a type is considered high-frequency.

Returns:

(event_set, event_seq, freq_vec)

Return type:

tuple[frozenset[str], list[str], dict[str, int]]

_parse_alarm_log(alarm_log)[source]¶

Parameters:: alarm_log (dict)
Return type:: list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent]

_parse_soe_log(soe_log)[source]¶

Parameters:: soe_log (dict)
Return type:: list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent]

_parse_telemetry_summary(summary, severity_threshold)[source]¶

Parameters:

summary (dict)
severity_threshold (float)

Return type:

list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent]

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.extractor._expand_window(window_start, window_end, beta)[source]¶

Applies beta buffer expansion symmetrically to a time window.

E_search_start = window_start - beta * duration E_search_end = window_end + beta * duration

Parameters:

window_start (datetime.datetime)
window_end (datetime.datetime)
beta (float)

Return type:

tuple[datetime.datetime, datetime.datetime]

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.extractor._compute_density(events, window_start, window_end)[source]¶

Computes event density over the given (unexpanded) window.

rho = N_events_in_window / window_duration_seconds

Returns 0.0 if duration is zero or negative.

Parameters:

events (list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent])
window_start (datetime.datetime)
window_end (datetime.datetime)

Return type:

float

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.extractor._dominant_asset(events)[source]¶

Returns the most frequently occurring asset_id among events, or ‘’ if empty.

Parameters:: events (list[src.dackar.RCA.log_pattern_recognition.rca_pattern_search.models.UnifiedEvent])
Return type:: str

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.extractor._parse_ts(value)[source]¶

Coerces a value to datetime.

Accepts: datetime objects, ISO 8601 strings. Returns None for None, pandas NaT, unparseable strings, or unknown types.

Return type:: Optional[datetime.datetime]

src.dackar.RCA.log_pattern_recognition.rca_pattern_search.extractor._derive_soe_end_timestamps(records)[source]¶

Derives timestamp_end for each SOE record as the timestamp of the next record with the same signal_id but a different transition value.

This models the “opposing transition” pairing (e.g. trip → reset) without requiring an explicit transition vocabulary.

Returns:: Mapping from record_id (str) to timestamp_end (datetime | None).
Parameters:: records (list[dict])
Return type:: dict[str, Optional[datetime.datetime]]