src.dackar.RCA.doc_parsers.fmea_normalizer¶

fmea_normalizer.py ───────────────────────────────────────────────────────────────────────────── Normalization helpers for parsed FMEA rows.

This module is intentionally format-aware but lightweight: it takes the already-parsed row dictionaries produced by fmeaParser.parse_fmea_file and applies profile detection, derivation rules, and per-field quality tagging.

Attributes¶

`JsonDict`
`CANONICAL_REQUIRED_FIELDS`
`CANONICAL_STANDARD_OPTIONAL_FIELDS`
`CANONICAL_ENRICHMENT_FIELDS`
`FIELD_STATUS_PRESENT`
`FIELD_STATUS_DERIVED`
`FIELD_STATUS_NLP`
`FIELD_STATUS_MISSING_CRITICAL`
`FIELD_STATUS_MISSING_OPTIONAL`
`FIELD_STATUS_MISSING_ENRICHMENT`
`_ANOMALY_PATTERN_KEYWORDS`
`FMEA_FORMAT_PROFILES`

Classes¶

`FmeaFormatProfile`
`AiagFmeaProfile`
`Aiag5thFmeaProfile`
`MilStd1629aProfile`
`Iec60812Profile`
`NuclearGenericProfile`
`AutoDetectProfile`

Functions¶

`classify_anomaly_pattern`(text)	Map free text describing an anomaly to a canonical enum value.
`_split_listish`(value)
`_to_float`(value)
`_to_int`(value)
`_derive_rpn`(row)
`_derive_occurrence_from_lambda`(row)
`_derive_severity_from_criticality`(row)
`_derive_end_effect`(row)
`_split_cause_effect_text`(text)	Heuristic split for mixed cause/effect prose.
`_mark_nlp_field`(row, field_name)
`_derive_potential_causes`(row)
`_derive_local_effect_from_mechanism`(row)
`_infer_pattern_from_local_effect`(row)
`_profile_by_name`(name)
`_autodetect_profile`(records)
`normalize_fmea_records`(records, *[, profile_name])	Normalize parsed records and attach field-level quality status.

Module Contents¶

src.dackar.RCA.doc_parsers.fmea_normalizer.JsonDict[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.CANONICAL_REQUIRED_FIELDS = ('component_type', 'failure_mode_name', 'failure_mechanism')[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.CANONICAL_STANDARD_OPTIONAL_FIELDS = ('local_effect', 'system_effect', 'end_effect', 'potential_causes', 'detection_method',...[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.CANONICAL_ENRICHMENT_FIELDS = ('expected_latency_min_hours', 'expected_latency_max_hours', 'expected_anomaly_pattern',...[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.FIELD_STATUS_PRESENT = 'present_native'[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.FIELD_STATUS_DERIVED = 'derived'[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.FIELD_STATUS_NLP = 'nlp_inferred'[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.FIELD_STATUS_MISSING_CRITICAL = 'missing_critical'[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.FIELD_STATUS_MISSING_OPTIONAL = 'missing_optional'[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.FIELD_STATUS_MISSING_ENRICHMENT = 'missing_enrichment'[source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer._ANOMALY_PATTERN_KEYWORDS: Dict[str, str][source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.classify_anomaly_pattern(text)[source]¶

Map free text describing an anomaly to a canonical enum value.

Returns the matched enum string, or None when no keyword matches so the caller decides the fallback: the parser records "unknown" for a value that was explicitly provided but is unrecognised, whereas the normalizer leaves the field missing so it is flagged in the ingestion-quality report.

Parameters:: text (Any)
Return type:: Optional[str]

src.dackar.RCA.doc_parsers.fmea_normalizer._split_listish(value)[source]¶

Parameters:: value (Any)
Return type:: List[str]

src.dackar.RCA.doc_parsers.fmea_normalizer._to_float(value)[source]¶

Parameters:: value (Any)
Return type:: Optional[float]

src.dackar.RCA.doc_parsers.fmea_normalizer._to_int(value)[source]¶

Parameters:: value (Any)
Return type:: Optional[int]

src.dackar.RCA.doc_parsers.fmea_normalizer._derive_rpn(row)[source]¶

Parameters:: row (JsonDict)
Return type:: Optional[int]

src.dackar.RCA.doc_parsers.fmea_normalizer._derive_occurrence_from_lambda(row)[source]¶

Parameters:: row (JsonDict)
Return type:: Optional[int]

src.dackar.RCA.doc_parsers.fmea_normalizer._derive_severity_from_criticality(row)[source]¶

Parameters:: row (JsonDict)
Return type:: Optional[int]

src.dackar.RCA.doc_parsers.fmea_normalizer._derive_end_effect(row)[source]¶

Parameters:: row (JsonDict)
Return type:: Optional[str]

src.dackar.RCA.doc_parsers.fmea_normalizer._split_cause_effect_text(text)[source]¶

Heuristic split for mixed cause/effect prose. Returns (causes, effect_text, used_nlp_heuristic).

Parameters:: text (str)
Return type:: Tuple[List[str], Optional[str], bool]

src.dackar.RCA.doc_parsers.fmea_normalizer._mark_nlp_field(row, field_name)[source]¶

Parameters:

row (JsonDict)
field_name (str)

Return type:

None

src.dackar.RCA.doc_parsers.fmea_normalizer._derive_potential_causes(row)[source]¶

Parameters:: row (JsonDict)
Return type:: Optional[List[str]]

src.dackar.RCA.doc_parsers.fmea_normalizer._derive_local_effect_from_mechanism(row)[source]¶

Parameters:: row (JsonDict)
Return type:: Optional[str]

src.dackar.RCA.doc_parsers.fmea_normalizer._infer_pattern_from_local_effect(row)[source]¶

Parameters:: row (JsonDict)
Return type:: Optional[str]

class src.dackar.RCA.doc_parsers.fmea_normalizer.FmeaFormatProfile[source]¶

name: str[source]¶

required_fields: Sequence[str] = ('component_type', 'failure_mode_name', 'failure_mechanism')[source]¶

optional_fields: Sequence[str] = ('local_effect', 'system_effect', 'end_effect', 'potential_causes', 'detection_method',...[source]¶

enrichment_fields: Sequence[str] = ('expected_latency_min_hours', 'expected_latency_max_hours', 'expected_anomaly_pattern',...[source]¶

derived_fields: Dict[str, Callable[[JsonDict], Any]][source]¶

class src.dackar.RCA.doc_parsers.fmea_normalizer.AiagFmeaProfile[source]¶

Bases: FmeaFormatProfile

name: str = 'aiag_4th'[source]¶

derived_fields: Dict[str, Callable[[JsonDict], Any]][source]¶

class src.dackar.RCA.doc_parsers.fmea_normalizer.Aiag5thFmeaProfile[source]¶

Bases: AiagFmeaProfile

name: str = 'aiag_5th'[source]¶

class src.dackar.RCA.doc_parsers.fmea_normalizer.MilStd1629aProfile[source]¶

Bases: FmeaFormatProfile

name: str = 'mil_std_1629a'[source]¶

derived_fields: Dict[str, Callable[[JsonDict], Any]][source]¶

class src.dackar.RCA.doc_parsers.fmea_normalizer.Iec60812Profile[source]¶

Bases: FmeaFormatProfile

name: str = 'iec_60812'[source]¶

derived_fields: Dict[str, Callable[[JsonDict], Any]][source]¶

class src.dackar.RCA.doc_parsers.fmea_normalizer.NuclearGenericProfile[source]¶

Bases: FmeaFormatProfile

name: str = 'nuclear_generic'[source]¶

derived_fields: Dict[str, Callable[[JsonDict], Any]][source]¶

class src.dackar.RCA.doc_parsers.fmea_normalizer.AutoDetectProfile[source]¶

Bases: FmeaFormatProfile

name: str = 'auto'[source]¶

derived_fields: Dict[str, Callable[[JsonDict], Any]][source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer.FMEA_FORMAT_PROFILES: Dict[str, Callable[[], FmeaFormatProfile]][source]¶

src.dackar.RCA.doc_parsers.fmea_normalizer._profile_by_name(name)[source]¶

Parameters:: name (str)
Return type:: FmeaFormatProfile

src.dackar.RCA.doc_parsers.fmea_normalizer._autodetect_profile(records)[source]¶

Parameters:: records (Sequence[JsonDict])
Return type:: Tuple[str, float]

src.dackar.RCA.doc_parsers.fmea_normalizer.normalize_fmea_records(records, *, profile_name='auto')[source]¶

Normalize parsed records and attach field-level quality status.

Parameters:

records (Sequence[JsonDict])
profile_name (str)

Return type:

Tuple[List[JsonDict], JsonDict]