src.dackar.RCA.doc_parsers.fmeaParser

fmeaParser.py ───────────────────────────────────────────────────────────────────────────── FMEA spreadsheet (CSV / Excel) → canonical FMEA record list.

Each non-empty input row becomes a dict with normalised canonical field names. Column headings are matched against a configurable regex map so that plant- specific naming variations (e.g. “SEV” vs “Severity”) are handled transparently.

Supported formats ───────────────── • CSV — any delimiter recognised by csv.Sniffer; UTF-8 or latin-1 • .xlsx — via openpyxl (already a project dependency) • .xls — via xlrd (already a project dependency) • Multi-sheet workbooks — all sheets parsed; records carry _sheet

Output schema (canonical keys) ─────────────────────────────── Required (hard-fail if absent):

fmea_source_ref str source filename (set from the input path) component_type str equipment class (e.g. “centrifugal_pump”) failure_mode_id str FM:<slug(component_type)>:<slug(failure_mode_name)> failure_mode_name str human-readable failure mode label

Optional:

failure_mechanism str physical mechanism (fatigue, corrosion, …) local_effect str local / end effect description severity int 1–10 occurrence int 1–10 detection int 1–10 rpn int explicit or derived = S × O × D expected_latency_min_hours float converted from min_days × 24 expected_latency_max_hours float converted from max_days × 24 expected_anomaly_pattern str normalised to enum values expected_symptoms list[str] split from local_effect text corrective_actions list[str] notes str _sheet str Excel sheet name; None for CSV _row_index int 1-based row number after header

Attributes

LOGGER
_ch
DEFAULT_COLUMN_MAP
PROFILE_COLUMN_MAPS
_ANOMALY_PATTERN_ENUM
_EFFECT_SPLIT_RE

Exceptions

_RowValidationSkip

Signal that a single data row should be skipped and reported.

Classes

FmeaColumnResolver

Resolve actual spreadsheet column headers to canonical field names.

Functions

_slug(text)	Return a lowercase, underscore-separated identifier-safe string.
_norm(value)	Strip and lower a cell value.
_to_int(value)
_to_float(value)
_resolve_anomaly_pattern(raw)
_split_effect_to_symptoms(effect_text)
_split_actions(raw)
_split_causes(raw)
_build_column_map(*, profile_name[, column_map_override])
_build_record(cells, row_index, fmea_source_ref, sheet)	Convert a resolved cell dict into a canonical FMEA record.
_rows_from_csv(path)	Read a CSV file and return [(None, rows)] where rows is a list of
_rows_from_xlsx(path)	Read all sheets from an .xlsx workbook using openpyxl.
_rows_from_xls(path)	Read all sheets from an .xls workbook using xlrd.
parse_fmea_file(path, *[, column_map_override, ...])	Parse a FMEA spreadsheet into a list of canonical FMEA record dicts.
parse_fmea_files(paths, **kwargs)	Parse multiple FMEA files and return a combined record list.
_merge_ingestion_reports(current, incoming)

Module Contents

src.dackar.RCA.doc_parsers.fmeaParser.LOGGER

src.dackar.RCA.doc_parsers.fmeaParser._ch

src.dackar.RCA.doc_parsers.fmeaParser.DEFAULT_COLUMN_MAP: Dict[str, List[str]]

src.dackar.RCA.doc_parsers.fmeaParser.PROFILE_COLUMN_MAPS: Dict[str, Dict[str, List[str]]]

src.dackar.RCA.doc_parsers.fmeaParser._ANOMALY_PATTERN_ENUM

src.dackar.RCA.doc_parsers.fmeaParser._EFFECT_SPLIT_RE

src.dackar.RCA.doc_parsers.fmeaParser._slug(text)

Return a lowercase, underscore-separated identifier-safe string.

Parameters:

text (str)

Return type:

str

src.dackar.RCA.doc_parsers.fmeaParser._norm(value)

Strip and lower a cell value.

Parameters:

value (Any)

Return type:

str

src.dackar.RCA.doc_parsers.fmeaParser._to_int(value)

Parameters:

value (Any)

Return type:

Optional[int]

src.dackar.RCA.doc_parsers.fmeaParser._to_float(value)

Parameters:

value (Any)

Return type:

Optional[float]

src.dackar.RCA.doc_parsers.fmeaParser._resolve_anomaly_pattern(raw)

Parameters:

raw (Optional[str])

Return type:

Optional[str]

src.dackar.RCA.doc_parsers.fmeaParser._split_effect_to_symptoms(effect_text)

Parameters:

effect_text (Optional[str])

Return type:

List[str]

src.dackar.RCA.doc_parsers.fmeaParser._split_actions(raw)

Parameters:

raw (Optional[str])

Return type:

List[str]

src.dackar.RCA.doc_parsers.fmeaParser._split_causes(raw)

Parameters:

raw (Optional[str])

Return type:

List[str]

src.dackar.RCA.doc_parsers.fmeaParser._build_column_map(*, profile_name, column_map_override=None)

Parameters:

• profile_name (str)

• column_map_override (Optional[Dict[str, List[str]]])

Return type:

Dict[str, List[str]]

class src.dackar.RCA.doc_parsers.fmeaParser.FmeaColumnResolver(column_map)

Resolve actual spreadsheet column headers to canonical field names.

Resolution is purely regex-based: each header is tested against every pattern list in the column map. For a given header the first canonical field (in the column map’s insertion order) whose pattern matches wins. If two headers resolve to the same canonical field, the later one is ignored and a warning is logged.

Parameters:

column_map (Dict[str, List[str]]) – Merged column map (defaults + overrides).

_map

resolve(headers)

Return a dict mapping canonical field name → 0-based column index.

Unrecognised headers are silently ignored.

Parameters:

headers (Sequence[str]) – Raw header strings from the spreadsheet.

Returns:

{canonical_field: col_index} for every resolved column.

Return type:

Dict[str, int]

validate_required(resolved, source)

Raise ValueError if required fields are missing.

Parameters:

• resolved (Dict[str, int]) – Output of resolve().

• source (str) – Human-readable source description for the error message.

Raises:

ValueError – If any of component_type, failure_mode_name, or failure_mechanism cannot be resolved, with a list of all detected canonical fields included.

Return type:

None

exception src.dackar.RCA.doc_parsers.fmeaParser._RowValidationSkip

Bases: Exception

Signal that a single data row should be skipped and reported.

Raised by _build_record() when a row is structurally present but has a blank required cell (e.g. failure_mechanism). The parse loop catches it, counts the row, logs a warning, and continues — so one bad row no longer aborts the whole file (and, via parse_fmea_files(), the whole batch). A missing required column is a different, file-level error still raised by FmeaColumnResolver.validate_required().

src.dackar.RCA.doc_parsers.fmeaParser._build_record(cells, row_index, fmea_source_ref, sheet)

Convert a resolved cell dict into a canonical FMEA record.

Returns None for rows that are entirely empty or have no component_type / failure_mode_name after stripping.

Parameters:

• cells (Dict[str, Any])

• row_index (int)

• fmea_source_ref (str)

• sheet (Optional[str])

Return type:

Optional[Dict[str, Any]]

src.dackar.RCA.doc_parsers.fmeaParser._rows_from_csv(path)

Read a CSV file and return [(None, rows)] where rows is a list of cell lists (all strings).

Parameters:

path (pathlib.Path)

Return type:

List[Tuple[Optional[str], List[List[str]]]]

src.dackar.RCA.doc_parsers.fmeaParser._rows_from_xlsx(path)

Read all sheets from an .xlsx workbook using openpyxl.

Parameters:

path (pathlib.Path)

Return type:

List[Tuple[Optional[str], List[List[Any]]]]

src.dackar.RCA.doc_parsers.fmeaParser._rows_from_xls(path)

Read all sheets from an .xls workbook using xlrd.

Parameters:

path (pathlib.Path)

Return type:

List[Tuple[Optional[str], List[List[Any]]]]

src.dackar.RCA.doc_parsers.fmeaParser.parse_fmea_file(path, *, column_map_override=None, sheet_filter=None, profile_name='auto', include_normalization_metadata=True)

Parse a FMEA spreadsheet into a list of canonical FMEA record dicts.

Parameters:

• path (str | pathlib.Path) – Path to a .csv, .xlsx, or .xls file.

• column_map_override (Optional[Dict[str, List[str]]]) – Optional dict that is merged into DEFAULT_COLUMN_MAP. Keys must be canonical field names; values are lists of additional regex patterns to try before the defaults. Use this to add plant-specific column naming without replacing the default patterns.

• sheet_filter (Optional[Sequence[str]]) – For multi-sheet workbooks, only parse the sheets whose names are in this list. Pass None (default) to parse all sheets.

• profile_name (str) – FMEA format profile forwarded to normalize_fmea_records() (e.g. "auto", "aiag_4th", "aiag_5th", "mil_std_1629a", "iec_60812", "nuclear_generic"). It also selects profile-specific header patterns via PROFILE_COLUMN_MAPS.

• include_normalization_metadata (bool) – When True (default), attach the per-field _field_quality tags, _normalization_profile, and the shared _fmea_ingestion_quality report to each record; when False these normalization-metadata keys are stripped.

Returns:

fmea_source_ref, component_type, failure_mode_id, failure_mode_name.

Return type:

List of record dicts. Each dict contains at minimum

Raises:

• ValueError – If required columns (component_type, failure_mode_name, failure_mechanism) cannot be resolved in a sheet, or if the file extension is not recognised. A row whose failure_mechanism cell is blank is skipped and counted in the ingestion-quality report’s rows_skipped_missing_mechanism rather than raising.

• FileNotFoundError – If path does not exist.

src.dackar.RCA.doc_parsers.fmeaParser.parse_fmea_files(paths, **kwargs)

Parse multiple FMEA files and return a combined record list.

Parameters:

• paths (Sequence[str | pathlib.Path]) – Iterable of file paths.

• **kwargs (Any) – Forwarded to parse_fmea_file().

Returns:

Combined list of all records from all files.

Return type:

List[Dict[str, Any]]

src.dackar.RCA.doc_parsers.fmeaParser._merge_ingestion_reports(current, incoming)

Parameters:

• current (Optional[Dict[str, Any]])

• incoming (Dict[str, Any])

Return type:

Dict[str, Any]