src.dackar.RCA.orchestrators.kg_context_builder

kg_context_builder — KGContextBuilderConfig and Neo4jKGContextBuilder.

Extracted from rca_reasoning_orchestrator.py. The parent module re-exports both names for backward-compatible imports.

Attributes

JsonDict
LOGGER

Classes

KGContextBuilderConfig
Neo4jKGContextBuilder

Functions

utcnow_iso()

Module Contents

src.dackar.RCA.orchestrators.kg_context_builder.JsonDict

src.dackar.RCA.orchestrators.kg_context_builder.LOGGER

src.dackar.RCA.orchestrators.kg_context_builder.utcnow_iso()

Return type:

str

class src.dackar.RCA.orchestrators.kg_context_builder.KGContextBuilderConfig

max_hops: int = 2

max_past_events: int = 10

max_documents: int = 20

include_documents: bool = True

include_past_events: bool = True

include_safety_functions: bool = True

include_oe_documents: bool = True

max_oe_documents: int = 10

doc_window_days_before: int = 90

doc_window_days_after: int = 7

past_event_window_days: int = 3650

class src.dackar.RCA.orchestrators.kg_context_builder.Neo4jKGContextBuilder(client, database=None, config=None)

Parameters:

• client (dackar.knowledge_graph.py2neo.Py2Neo)

• database (Optional[str])

• config (Optional[KGContextBuilderConfig])

client

database = None

config

build(event, telemetry_summary, operational_context, pm_compliance, run_context, focus_component_ids=None)

Parameters:

• event (JsonDict)

• telemetry_summary (JsonDict)

• operational_context (Optional[JsonDict])

• pm_compliance (Optional[JsonDict])

• run_context (JsonDict)

• focus_component_ids (Optional[List[str]])

Return type:

JsonDict

_record_truncation(*, family, total_matched, cap, dropped_ids, retained=None)

Record per-family truncation stats into self._truncation_stats (P-1).

total_matched is None when the pre-truncation total is unknown (e.g. a DB-side LIMIT); in that case truncation is inferred from an at-cap retained count. Otherwise a family is truncated when more items matched than the cap allows.

Parameters:

• family (str)

• total_matched (Optional[int])

• cap (int)

• dropped_ids (List[str])

• retained (Optional[int])

Return type:

None

_basic_input_checks(event, telemetry_summary)

Parameters:

• event (JsonDict)

• telemetry_summary (JsonDict)

Return type:

None

_resolve_seed_nodes(event, telemetry_summary)

Parameters:

• event (JsonDict)

• telemetry_summary (JsonDict)

Return type:

JsonDict

_resolve_monitored_variable(sensor_id, monitored_variable_id)

Parameters:

• sensor_id (Optional[str])

• monitored_variable_id (Optional[str])

Return type:

List[JsonDict]

_expand_neighborhood(seed_component_ids)

Parameters:

seed_component_ids (Set[str])

Return type:

JsonDict

_estimate_path_strength(hop_count, rel_types)

Parameters:

• hop_count (int)

• rel_types (List[str])

Return type:

float

_fetch_failure_modes(component_ids)

Parameters:

component_ids (List[str])

Return type:

List[JsonDict]

_fetch_safety_functions(component_ids)

Fetch safety function nodes in the KG that are linked to any component in component_ids via standard nuclear-plant relation types.

The query is intentionally permissive on relation direction: different KG schemas place the directed arrow in either direction between a component and its associated safety function. Both directions are matched.

Returns a list of dicts conforming to kg_context.json#/safety_functions.

Parameters:

component_ids (List[str])

Return type:

List[JsonDict]

_fetch_kg_snapshot_version()

Return a stable version string that identifies the current state of the Neo4j KG.

Strategy (in priority order): 1. CALL dbms.components() — returns the Neo4j server version. Combined

with the highest last_modified timestamp across all nodes this gives a reproducible snapshot key tied to both the software and data state.

2. Highest node last_modified timestamp alone (if dbms.components fails).

3. Synthetic ISO timestamp fallback — used when the graph has no last_modified properties and the procedure call is unavailable.

The returned string is recorded in kg_context.kg_snapshot_version so that every RCA run can be replayed against the exact KG state that existed at query time (§6 Model Governance).

Return type:

str

_fetch_oe_documents(failure_mode_ids, component_types)

Return OE report nodes from the KG relevant to this event’s failure modes and component types.

OE documents are fleet-wide and timeless — no date window is applied. Retrieval is via two soft-match paths (ordered by specificity):

1. Direct FM linkage: (oe_document)-[:APPLICABLE_TO]->(failure_mode)

2. Component-type soft match: oe_document.applicable_component_types overlaps with the subgraph component types.

plant_scope (pwr_only / bwr_only) is surfaced as metadata but is NOT used as a hard filter — many components are shared across reactor types.

Parameters:

• failure_mode_ids (List[str])

• component_types (List[str])

Return type:

List[JsonDict]

_fetch_documents(asset_ids, component_ids, event, operational_context)

Parameters:

• asset_ids (List[str])

• component_ids (List[str])

• event (JsonDict)

• operational_context (Optional[JsonDict])

Return type:

List[JsonDict]

_fetch_past_events(target_event_id, asset_ids, component_ids, failure_mode_ids, event)

Parameters:

• target_event_id (str)

• asset_ids (List[str])

• component_ids (List[str])

• failure_mode_ids (List[str])

• event (JsonDict)

Return type:

List[JsonDict]

_parse_event_time(event)

Parameters:

event (JsonDict)

Return type:

Optional[datetime.datetime]

_compute_time_distance_days(created_at, event_time)

Parameters:

• created_at (Optional[str])

• event_time (Optional[datetime.datetime])

Return type:

Optional[int]