src.dackar.RCA.pm_compliance.aggregator

PMComplianceAggregator — assemble the pm_compliance artifact (JSON-serializable dict).

Attributes

_OUTPUT_SCHEMA_PATH

Functions

_output_validator()	Return a cached jsonschema validator for the pm_compliance schema.
_validate_output(artifact)	Validate artifact against schemas/pm_compliance.json (fail-closed).
_window_for_event(event_ts, lookback_days)
_summary_metrics(checks[, reference_dt])
_rollup_risk(primary_fm_id, all_gap, has_overdue, has_fail)	Architecture §3.6 — maintenance_induced_risk and overall_compliance roll-ups.
_compliance_status_md(overdue_days, st, missed_cycles)	Narrative labels for pm_tasks[].compliance_status (architecture §5).
_build_pm_tasks_per_component(raw_rows, checks[, ...])	Architecture §5 — components[].pm_tasks (narrative) alongside pipeline checks.
_apply_frequency_flags(comp_views, raw_rows, ...)	Set pm_frequency_concern on each component (architecture §3.5).
build_pm_compliance(event[, kg_context, export_rows, ...])	Assemble a pm_compliance object validated by schemas/pm_compliance.json.

Module Contents

src.dackar.RCA.pm_compliance.aggregator._OUTPUT_SCHEMA_PATH

src.dackar.RCA.pm_compliance.aggregator._output_validator()

Return a cached jsonschema validator for the pm_compliance schema.

src.dackar.RCA.pm_compliance.aggregator._validate_output(artifact)

Validate artifact against schemas/pm_compliance.json (fail-closed).

Raises ValueError naming the first offending JSON path when the built artifact — including the now-formalized components[] extension — does not conform (MR#56 review A2).

Parameters:

artifact (src.dackar.RCA.pm_compliance.types.JsonDict)

Return type:

None

src.dackar.RCA.pm_compliance.aggregator._window_for_event(event_ts, lookback_days)

Parameters:

• event_ts (str)

• lookback_days (int)

Return type:

tuple[str, str]

src.dackar.RCA.pm_compliance.aggregator._summary_metrics(checks, reference_dt=None)

Parameters:

• checks (List[src.dackar.RCA.pm_compliance.types.JsonDict])

• reference_dt (Optional[datetime.datetime])

Return type:

Dict[str, Any]

src.dackar.RCA.pm_compliance.aggregator._rollup_risk(primary_fm_id, all_gap, has_overdue, has_fail)

Architecture §3.6 — maintenance_induced_risk and overall_compliance roll-ups.

When primary_fm_id is supplied (e.g. after synthesis), risk matches the spec: high = primary in scope gap and PM was overdue; medium = primary gap or overdue; else low. Without primary_fm_id, risk falls back to gap + overdue heuristics.

Parameters:

• primary_fm_id (Optional[str])

• all_gap (Set[str])

• has_overdue (bool)

• has_fail (bool)

Return type:

Tuple[str, str, bool]

src.dackar.RCA.pm_compliance.aggregator._compliance_status_md(overdue_days, st, missed_cycles, raw_compliance_status=None)

Narrative labels for pm_tasks[].compliance_status (architecture §5).

"unknown" check status (no schedule dates in the export) returns "undetermined" so analysts can distinguish it from a genuine pass. The governance-scoring path (checks[].status) is unaffected — it still carries "unknown"; only the component narrative label changes.

Parameters:

• overdue_days (float)

• st (str)

• missed_cycles (int)

• raw_compliance_status (Optional[str])

Return type:

str

src.dackar.RCA.pm_compliance.aggregator._build_pm_tasks_per_component(raw_rows, checks, check_to_coverage_type=None)

Architecture §5 — components[].pm_tasks (narrative) alongside pipeline checks.

check_to_coverage_type (from analyze_scope) maps check_id → "preventive" | "detective". When present, it overrides the export row’s coverage_type field, which defaults to "none".

Parameters:

• raw_rows (List[src.dackar.RCA.pm_compliance.types.JsonDict])

• checks (List[src.dackar.RCA.pm_compliance.types.JsonDict])

• check_to_coverage_type (Optional[Dict[str, str]])

Return type:

Dict[str, List[Dict[str, Any]]]

src.dackar.RCA.pm_compliance.aggregator._apply_frequency_flags(comp_views, raw_rows, kg_context, ratio)

Set pm_frequency_concern on each component (architecture §3.5).

Parameters:

• comp_views (List[Dict[str, Any]])

• raw_rows (List[src.dackar.RCA.pm_compliance.types.JsonDict])

• kg_context (Optional[src.dackar.RCA.pm_compliance.types.JsonDict])

• ratio (float)

Return type:

None

src.dackar.RCA.pm_compliance.aggregator.build_pm_compliance(event, kg_context=None, export_rows=None, config=None, primary_fm_id=None)

Assemble a pm_compliance object validated by schemas/pm_compliance.json.

Parameters:

• event (src.dackar.RCA.pm_compliance.types.JsonDict) – Must include asset_id and timestamp_start (ISO). event_id is optional.

• kg_context (Optional[src.dackar.RCA.pm_compliance.types.JsonDict]) – Optional; supplies failure modes for scope analysis.

• export_rows (Optional[Sequence[src.dackar.RCA.pm_compliance.types.JsonDict]]) – Pre-parsed PM tasks from a CMMS export or adapter (see PMExportTaskRow).

• config (Optional[src.dackar.RCA.pm_compliance.config.PMComplianceConfig]) – Lookback and thresholds.

• primary_fm_id (Optional[str]) – Optional failure mode id (e.g. from the eventual primary hypothesis) to evaluate has_scope_gaps_for_primary_fm and maintenance_induced_risk per PM_Compliance_Module_Architecture.md §3.5–3.6.

Returns:

A pm_compliance artifact conforming to schemas/pm_compliance.json: asset_id, window (start/end), checks[] (per-PM pass/fail/unknown rows for governance scoring), and summary (aggregate counts, compliance_rate over evaluable checks, overall_compliance, maintenance_induced_risk, data_quality_confidence). Optional keys — event_id, assessment_date, look_back_window_days, fmea_pm_linkage_available, data_quality_notes, components[] (per-component detail) and overdue_items[] — appear when applicable. The artifact is schema-validated before return, raising ValueError on a non-conforming build.

Return type:

JsonDict