src.dackar.knowledge_graph.kg_ingest_fmea_workflow¶

kg_ingest_fmea_workflow.py ───────────────────────────────────────────────────────────────────────────── Ingest parsed FMEA records (output of fmeaParser.parse_fmea_file) into Neo4j.

Graph objects created per FMEA record ────────────────────────────────────── Nodes:

fmea_case one per unique (fmea_source_ref, sheet) combination failure_mode one per record (keyed by failure_mode_id; merged if duplicate) risk_assessment one per record when at least one of severity/occurrence/detection

is present

effect one per record when local_effect text is present

Edges:: fmea_case -[:IDENTIFIES_FAILURE_MODE]-> failure_mode failure_mode -[:HAS_RISK_ASSESSMENT]-> risk_assessment failure_mode -[:LEADS_TO_EFFECT]-> effect failure_mode -[:APPLIES_TO]-> element_usage (see below)

Component-type resolution (APPLIES_TO edges) ───────────────────────────────────────────── FMEA data is class-level: a row for “centrifugal_pump / seal degradation” applies to every centrifugal pump in the plant. During ingestion the component_type value is resolved to individual element_usage node IDs by querying the live KG (mbseSchema v3.1):

MATCH (c:element_usage)-[:instance_of]->(def:element_definition) WHERE toLower(def.domain_category) = toLower($component_type) RETURN c.id AS component_id

An APPLIES_TO edge is created for each matched element_usage. If no usages are found for a type the failure_mode node is still written (with the component_type property set) and a warning is logged so the gap can be addressed when MBSE entities are loaded.

CLI usage ─────────

python -m dackar.knowledge_graph.kg_ingest_fmea_workflow
–schema src/dackar/knowledge_graph/schemas/fmeaSchema.toml –schema src/dackar/knowledge_graph/schemas/mbseSchema.toml –neo4j-uri bolt://localhost:7687 –neo4j-user neo4j –neo4j-pass secret fmea_pump.xlsx fmea_valve.xlsx

Attributes¶

`LOGGER`
`_ch`
`JsonLike`

Functions¶

`_resolve_component_type`(client, component_type, ...)	Return all element_usage IDs whose linked element_definition matches component_type.
`_extract_fmea_ingestion_quality`(fmea_records)	Extract the shared normalization quality report attached by fmeaParser.
`build_fmea_graph`(schema_paths, fmea_records[, client, ...])	Build a Neo4j graph batch from parsed FMEA records.
`ingest_fmea_to_neo4j`(client, schema_paths, fmea_records, *)	Build and ingest a FMEA graph into Neo4j.
`_parse_args`()
`main`()	CLI entry point.

Module Contents¶

src.dackar.knowledge_graph.kg_ingest_fmea_workflow.LOGGER[source]¶

src.dackar.knowledge_graph.kg_ingest_fmea_workflow._ch[source]¶

src.dackar.knowledge_graph.kg_ingest_fmea_workflow.JsonLike[source]¶

src.dackar.knowledge_graph.kg_ingest_fmea_workflow._resolve_component_type(client, component_type, database, cache)[source]¶

Return all element_usage IDs whose linked element_definition matches component_type.

Matches against element_definition.domain_category (mbseSchema v3.1). Results are cached per component_type string to avoid redundant queries.

Parameters:

client (dackar.knowledge_graph.py2neo.Py2Neo) – Active Neo4j connection.
component_type (str) – Equipment class string (e.g. "centrifugal_pump").
database (Optional[str]) – Neo4j target database; None uses the driver default.
cache (Dict[str, List[str]]) – Mutable dict used as an in-process cache across calls.

Returns:

List of element_usage id values (may be empty).

Return type:

List[str]

src.dackar.knowledge_graph.kg_ingest_fmea_workflow._extract_fmea_ingestion_quality(fmea_records)[source]¶

Extract the shared normalization quality report attached by fmeaParser.

Parameters:: fmea_records (Sequence[Dict[str, Any]])
Return type:: Dict[str, Any]

src.dackar.knowledge_graph.kg_ingest_fmea_workflow.build_fmea_graph(schema_paths, fmea_records, client=None, database=None)[source]¶

Build a Neo4j graph batch from parsed FMEA records.

Creates fmea_case, failure_mode, risk_assessment, and effect nodes with their connecting edges. When client is supplied, APPLIES_TO edges to element_usage nodes are also created after resolving component_type against the live KG (mbseSchema v3.1).

Parameters:

schema_paths (Union[str, pathlib.Path, Iterable[Union[str, pathlib.Path]]]) – One or more paths to TOML schema files.
fmea_records (Sequence[Dict[str, Any]]) – Output of fmeaParser.parse_fmea_file().
client (Optional[dackar.knowledge_graph.py2neo.Py2Neo]) – Optional live Py2Neo connection used to resolve component types to element_usage IDs. When None, APPLIES_TO edges are omitted.
database (Optional[str]) – Neo4j target database; None uses the driver default.

Returns:

A two-tuple (nodes, edges) suitable for kg_schema_builder_workflow.ingest_graph_toml().

Return type:

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

src.dackar.knowledge_graph.kg_ingest_fmea_workflow.ingest_fmea_to_neo4j(client, schema_paths, fmea_records, *, database=None, create_constraints=True)[source]¶

Build and ingest a FMEA graph into Neo4j.

Parameters:

client (dackar.knowledge_graph.py2neo.Py2Neo) – Active Py2Neo connection.
schema_paths (Union[str, pathlib.Path, Iterable[Union[str, pathlib.Path]]]) – One or more TOML schema file paths.
fmea_records (Sequence[Dict[str, Any]]) – Parsed FMEA records from fmeaParser.parse_fmea_file().
database (Optional[str]) – Target Neo4j database; None uses the driver default.
create_constraints (bool) – Apply DDL constraints/indexes before ingestion.

Returns:

(node_count, edge_count) written to the database.

Return type:

Tuple[int, int]

src.dackar.knowledge_graph.kg_ingest_fmea_workflow._parse_args()[source]¶

Return type:: argparse.Namespace

src.dackar.knowledge_graph.kg_ingest_fmea_workflow.main()[source]¶

CLI entry point.

Return type:: None