src.dackar.knowledge_graph.kg_schema_builder_workflow

Attributes

LOGGER
_BASE_SCHEMA_PATH

Classes

GraphBatch

In-memory accumulator for nodes and edges before bulk Neo4j ingestion.

Functions

load_toml_schema(path)	Load a TOML schema file and return its contents as a dict.
_load_base_schema()	Load and return the knowledge-graph meta-schema (baseSchema.json).
validate_schema_document(schema[, source])	Validate a single parsed TOML schema the way KG.importGraphSchema does.
load_and_merge_schemas(schema_paths)	Load one or more TOML schema files and merge them into a single schema dict.
_schema_props(spec)	Extract the property list from a node or relation schema spec.
_node_primary_key(spec)	Determine the primary key property name for a node spec.
_label_candidates(name)	Generate a list of label name variants to try when resolving a schema label.
resolve_node_label(schema, *candidates)	Resolve the first candidate name that exists as a node label in the schema.
resolve_node_label_strict(schema, *candidates)	Resolve the first candidate name that exists as a node label in the schema.
relation_endpoint_map(schema)	Build a lookup from relation type name to its declared endpoint pairs.
_is_primitive(value)	Return True if value is a Neo4j-native scalar type (str, int, float, or bool).
sanitize_value(value)	Convert an arbitrary Python value to a Neo4j-compatible property value.
sanitize_props(props)	Sanitize all values in a property dict, dropping None entries.
generate_ddl_from_schema(schema)	Generate Neo4j DDL statements (constraints and indexes) from a merged schema.
apply_schema_constraints(client, schema_paths[, database])	Load schemas and apply all DDL constraints and indexes to the database.
_prefix(value, prefix)	Build a namespaced node id by prepending prefix to value.
_truthy(value)	Return True if value is non-empty and not a sentinel null-like string.
_norm_text_key(value)	Normalise a text string into a stable, lowercase key.
_safe_rel(g, src, dst, preferred, fallback[, attrs])	Add an edge using preferred relation type, falling back to fallback if not in schema.
_safe_ptr_entity_rel(g, src, dst[, attrs])	ProcessedTextRecord -> entity relation chooser.
_safe_ptr_failure_mode_rel(g, src, dst[, attrs])	Add a ProcessedTextRecord → failure_mode edge using the best available relation type.
build_graph_from_workflow_artifacts([schema_paths, ...])	Translate all RCA workflow artifacts into a graph of nodes and edges.
build_primary_key_map(schema)	Build a {node_label: primary_key} map from a (merged) TOML schema.
ingest_graph_toml(client, nodes, edges[, database, ...])	Bulk-upsert a node list and edge list into Neo4j.

Module Contents

src.dackar.knowledge_graph.kg_schema_builder_workflow.LOGGER

src.dackar.knowledge_graph.kg_schema_builder_workflow._BASE_SCHEMA_PATH

src.dackar.knowledge_graph.kg_schema_builder_workflow.load_toml_schema(path)

Load a TOML schema file and return its contents as a dict.

Parameters:

path (Union[str, pathlib.Path]) – Filesystem path to the .toml schema file.

Returns:

Parsed TOML document as a nested dictionary.

Return type:

Dict[str, Any]

src.dackar.knowledge_graph.kg_schema_builder_workflow._load_base_schema()

Load and return the knowledge-graph meta-schema (baseSchema.json).

Return type:

Dict[str, Any]

src.dackar.knowledge_graph.kg_schema_builder_workflow.validate_schema_document(schema, source='<schema>')

Validate a single parsed TOML schema the way KG.importGraphSchema does.

Runs the per-file checks that the interactive KG loader applies on import, so malformed production schemas fail at load time instead of much later during DDL generation, graph building, or the batch MERGE:

1. Structural validation against the meta-schema baseSchema.json.

2. Property data types restricted to ALLOWED_SCHEMA_TYPES.

3. Each node’s effective primary key (an explicit primary_key or the "id" default) must name a property that is declared on that node and is non-optional — otherwise the batch MERGE would later key on a null value.

Cross-schema integrity (endpoint resolution, duplicate node labels / relation triples) is intentionally not checked here: it is a whole-set property, and partial sets (e.g. the FMEA CLI loading only fmeaSchema + mbseSchema) legitimately reference node labels defined in sibling schemas. load_and_merge_schemas() enforces the whole-set duplicate checks after merging.

Parameters:

• schema (Dict[str, Any]) – A single parsed TOML schema document.

• source (str) – Human-readable origin (path) used in error messages.

Raises:

• jsonschema.ValidationError – If the document violates baseSchema.json.

• ValueError – If a property uses a disallowed type, or a node’s effective primary key is missing or optional.

Return type:

None

src.dackar.knowledge_graph.kg_schema_builder_workflow.load_and_merge_schemas(schema_paths)

Load one or more TOML schema files and merge them into a single schema dict.

The merged dict has top-level keys "node" and "relation". A node label may be defined only once across the merged files, so a duplicate node key raises. Relation identity, however, is the (name, from_entity, to_entity) triple (matching KG._crossSchemasCheck()): a generic verb such as caused_by or recommends_action may be reused across schemas as long as it connects a different pair of node labels. Each relation name therefore maps to a list of its spec dicts, preserving every distinct endpoint pair; only an exact-duplicate triple raises.

Each file is validated by validate_schema_document() as it is loaded (structure against baseSchema.json, allowed property types, and the primary-key semantic check), so a malformed production schema fails here rather than silently later during DDL/build/write.

Parameters:

schema_paths (Union[str, pathlib.Path, Iterable[Union[str, pathlib.Path]]]) – A single path or an iterable of paths to .toml files.

Returns:

A merged schema dict {"node": {label: spec}, "relation": {name: [spec, ...]}}.

Raises:

• jsonschema.ValidationError – If a file violates baseSchema.json.

• ValueError – If a file uses a disallowed property type or an invalid primary key (see validate_schema_document()), or if the same node label, or the same relation (name, from_entity, to_entity) triple, appears more than once across the merged files.

Return type:

Dict[str, Any]

src.dackar.knowledge_graph.kg_schema_builder_workflow._schema_props(spec)

Extract the property list from a node or relation schema spec.

Checks both "node_properties" and "properties" keys for compatibility with different TOML schema conventions.

Parameters:

spec (Dict[str, Any]) – A single node or relation entry from the merged schema.

Returns:

List of property descriptor dicts, or an empty list if none are defined.

Return type:

List[Dict[str, Any]]

src.dackar.knowledge_graph.kg_schema_builder_workflow._node_primary_key(spec)

Determine the primary key property name for a node spec.

Honours an explicit primary_key declaration on the node spec first (the meta-schema’s MERGE-key contract). Otherwise selects the first non-optional property; falls back to "id" if present in the property list, then to the first listed property name, and finally to the hard-coded default "id".

Parameters:

spec (Dict[str, Any]) – A single node entry from the merged schema.

Returns:

The property name to use as the primary key.

Return type:

str

src.dackar.knowledge_graph.kg_schema_builder_workflow._label_candidates(name)

Generate a list of label name variants to try when resolving a schema label.

Produces the original name, its lowercase form, a PascalCase conversion, and a snake_case conversion (from PascalCase input) to account for naming conventions used across different TOML schemas.

Parameters:

name (str) – Base label name.

Returns:

List of candidate strings (original, lowercase, PascalCase, snake_case), deduplicated while preserving order.

Return type:

List[str]

src.dackar.knowledge_graph.kg_schema_builder_workflow.resolve_node_label(schema, *candidates)

Resolve the first candidate name that exists as a node label in the schema.

For each candidate, tries the original name and common variants (lowercase, PascalCase, snake_case). If none match, returns the first candidate unchanged as a safe fallback.

Parameters:

• schema (Dict[str, Any]) – Merged schema dict (must contain a "node" key).

• *candidates (str) – One or more preferred label names, in priority order.

Returns:

The matched label string from the schema, or candidates[0] if no match is found.

Return type:

str

src.dackar.knowledge_graph.kg_schema_builder_workflow.resolve_node_label_strict(schema, *candidates)

Resolve the first candidate name that exists as a node label in the schema.

Identical to resolve_node_label() but raises KeyError instead of silently falling back when no candidate matches. Use this wherever a miss should be a hard error (e.g. building the labels registry at graph construction time).

Parameters:

• schema (Dict[str, Any]) – Merged schema dict (must contain a "node" key).

• *candidates (str) – One or more preferred label names, in priority order.

Returns:

The matched label string from the schema.

Raises:

KeyError – If no variant of any candidate is found in the schema’s node registry. The error message lists the available node labels so the caller can diagnose schema/TOML mismatches immediately.

Return type:

str

src.dackar.knowledge_graph.kg_schema_builder_workflow.relation_endpoint_map(schema)

Build a lookup from relation type name to its declared endpoint pairs.

A relation name may be reused across schemas for different endpoint pairs (relation identity is the (name, from_entity, to_entity) triple), so each name maps to a list of (from_entity, to_entity) tuples, one per declared definition. Only definitions that declare both from_entity and to_entity are included; exact-duplicate pairs are collapsed.

Accepts both the list-valued "relation" section produced by load_and_merge_schemas() and a single-schema dict whose relation values are plain spec dicts.

Parameters:

schema (Dict[str, Any]) – Merged schema dict (must contain a "relation" key).

Returns:

Dict mapping each relation name to a list of (source_entity, target_entity) tuples of entity type strings.

Return type:

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

src.dackar.knowledge_graph.kg_schema_builder_workflow._is_primitive(value)

Return True if value is a Neo4j-native scalar type (str, int, float, or bool).

Parameters:

value (Any) – Any Python object.

Returns:

True if value can be stored directly as a Neo4j property scalar.

Return type:

bool

src.dackar.knowledge_graph.kg_schema_builder_workflow.sanitize_value(value)

Convert an arbitrary Python value to a Neo4j-compatible property value.

Conversion rules: - datetime / date → ISO-8601 string. - dict → JSON string (sorted keys, UTF-8). - list of primitives → kept as-is; mixed/complex lists → JSON string. - None and primitive scalars → returned unchanged. - Anything else → str(value).

Parameters:

value (Any) – Python value to sanitize.

Returns:

A value safe for storage as a Neo4j node or relationship property.

Return type:

Any

src.dackar.knowledge_graph.kg_schema_builder_workflow.sanitize_props(props)

Sanitize all values in a property dict, dropping None entries.

Parameters:

props (Dict[str, Any]) – Raw property dict potentially containing non-Neo4j-native types.

Returns:

New dict with all values converted by sanitize_value() and keys whose value was None removed.

Return type:

Dict[str, Any]

src.dackar.knowledge_graph.kg_schema_builder_workflow.generate_ddl_from_schema(schema)

Generate Neo4j DDL statements (constraints and indexes) from a merged schema.

For every node label a UNIQUE constraint is created on that label’s primary-key property (as resolved by _node_primary_key()), so an explicit primary_key such as document_id is enforced rather than a hard-coded id. Additionally, a CREATE INDEX statement is emitted for each property that carries "indexed": true in its spec.

Parameters:

schema (Dict[str, Any]) – Merged schema dict as returned by load_and_merge_schemas().

Returns:

List of Cypher DDL strings ready to be executed against Neo4j.

Raises:

ValueError – If a node label or indexed-property name is not a safe Neo4j identifier (see dackar.knowledge_graph.py2neo._safe_token()).

Return type:

List[str]

src.dackar.knowledge_graph.kg_schema_builder_workflow.apply_schema_constraints(client, schema_paths, database=None)

Load schemas and apply all DDL constraints and indexes to the database.

Combines load_and_merge_schemas(), generate_ddl_from_schema(), and Py2Neo.query() into a single convenience call.

Parameters:

• client (dackar.knowledge_graph.py2neo.Py2Neo) – Active Py2Neo connection.

• schema_paths (Union[str, pathlib.Path, Iterable[Union[str, pathlib.Path]]]) – One or more paths to TOML schema files.

• database (Optional[str]) – Target database name; uses the driver default when None.

Return type:

None

class src.dackar.knowledge_graph.kg_schema_builder_workflow.GraphBatch(schema=None)

In-memory accumulator for nodes and edges before bulk Neo4j ingestion.

Nodes are keyed by their id string; duplicate additions are merged. Edges are keyed by (source_id, target_id, rel_type) and support optional schema-based endpoint validation.

Parameters:

schema (Optional[Dict[str, Any]])

schema

nodes: Dict[str, Dict[str, Any]]

edges: Dict[Tuple[str, str, str], Dict[str, Any]]

relation_map

primary_key_map

add_node(node_id, label, attrs=None)

Add or merge a node into the batch.

If a node with node_id already exists its attributes are updated with the new values (shallow merge after sanitization).

Parameters:

• node_id (str) – Unique identifier for the node (used as the id property).

• label (str) – Neo4j label for the node.

• attrs (Optional[Dict[str, Any]]) – Optional property dict; id is always set from node_id.

Returns:

The node_id string, for convenience when chaining calls.

Return type:

str

Notes

node_id is the one canonical identity for the node: it is both the id property and the value the edge builder stores as the edge endpoint. When the label’s schema declares a natural primary key (e.g. document_id), that property is populated with the same node_id so the MERGE key the ingestion batch uses lines up with the edge-endpoint match and Neo4j never merges on a null key. For the default id key this is a no-op.

add_edge(src, dst, rel_type, attrs=None, allow_untyped=True)

Add or update a directed edge in the batch.

Endpoint labels are looked up from already-added nodes. When rel_type is declared in the schema its expected endpoint types are validated.

Parameters:

• src (str) – Node id of the source node (must already be in the batch).

• dst (str) – Node id of the target node (must already be in the batch).

• rel_type (str) – Relationship type name.

• attrs (Optional[Dict[str, Any]]) – Optional property dict for the relationship.

• allow_untyped (bool) – When True, relationship types not declared in the schema are accepted. When False, an undeclared type raises ValueError.

Raises:

• KeyError – If src or dst has not been added to the batch yet.

• ValueError – If the schema declares endpoint types for rel_type and the actual node labels do not match, or if allow_untyped is False and rel_type is not in the schema.

Return type:

None

as_lists()

Return the accumulated nodes and edges as plain lists.

Returns:

A two-tuple (nodes, edges) where each element is a list of dicts suitable for passing to Py2Neo.upsert_nodes_batch() and Py2Neo.upsert_edges_batch() respectively.

Return type:

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

src.dackar.knowledge_graph.kg_schema_builder_workflow._prefix(value, prefix)

Build a namespaced node id by prepending prefix to value.

Strips whitespace, collapses internal spaces to underscores, and skips empty values. Normalising spaces ensures the resulting ID is safe to use unquoted in Cypher — FM:loss_of_lubrication rather than FM:loss of lubrication. Casing is preserved so that structured IDs (e.g. CMP-001) are not altered.

Strings already namespaced with this prefix ("<prefix>:...") are returned unchanged to avoid double-prefixing. A colon appearing elsewhere in the value (e.g. a time "12:30" or an OPCTX/PM context value) no longer suppresses namespacing, so distinct raw values can no longer collide onto the same un-prefixed id.

Parameters:

• value (Optional[str]) – Raw identifier string (e.g. "pump-101" or "loss of lubrication").

• prefix (str) – Namespace prefix (e.g. "ASSET").

Returns:

A string like "ASSET:pump-101", or None if value is falsy or blank after stripping.

Return type:

Optional[str]

src.dackar.knowledge_graph.kg_schema_builder_workflow._truthy(value)

Return True if value is non-empty and not a sentinel null-like string.

Treats "unknown", "none", and "null" as falsy in addition to standard Python falsy values.

Parameters:

value (Any) – Any value to test.

Returns:

True if the value is considered meaningfully present.

Return type:

bool

src.dackar.knowledge_graph.kg_schema_builder_workflow._norm_text_key(value)

Normalise a text string into a stable, lowercase key.

Collapses internal whitespace, strips leading/trailing whitespace, and lowercases the result. Used to derive deterministic node ids from free-text labels.

Parameters:

value (Optional[str]) – Input string to normalise, or None.

Returns:

Normalised string, or None if value is None or blank.

Return type:

Optional[str]

src.dackar.knowledge_graph.kg_schema_builder_workflow._safe_rel(g, src, dst, preferred, fallback, attrs=None)

Add an edge using preferred relation type, falling back to fallback if not in schema.

Parameters:

• g (GraphBatch) – The GraphBatch to add the edge to.

• src (str) – Source node id.

• dst (str) – Target node id.

• preferred (str) – Preferred relationship type name.

• fallback (str) – Fallback relationship type name used when preferred is not declared in the schema’s relation map.

• attrs (Optional[Dict[str, Any]]) – Optional property dict for the relationship.

Return type:

None

src.dackar.knowledge_graph.kg_schema_builder_workflow._safe_ptr_entity_rel(g, src, dst, attrs=None)

ProcessedTextRecord -> entity relation chooser. Avoid using document-scoped relations like ‘mentions’ when the schema types them as condition_report/work_order -> element_usage.

Parameters:

• g (GraphBatch)

• src (str)

• dst (str)

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

Return type:

None

src.dackar.knowledge_graph.kg_schema_builder_workflow._safe_ptr_failure_mode_rel(g, src, dst, attrs=None)

Add a ProcessedTextRecord → failure_mode edge using the best available relation type.

Tries "supports_hypothesis", "references_failure_mode", and "caused_by" in order, picking the first whose endpoint types match the actual node labels. Falls back to an untyped "references_failure_mode" edge if none match.

Parameters:

• g (GraphBatch) – The GraphBatch to add the edge to.

• src (str) – Source node id (a ProcessedTextRecord node).

• dst (str) – Target node id (a failure_mode node).

• attrs (Optional[Dict[str, Any]]) – Optional property dict for the relationship.

Return type:

None

src.dackar.knowledge_graph.kg_schema_builder_workflow.build_graph_from_workflow_artifacts(schema_paths=None, *, event=None, kg_context=None, telemetry_summary=None, evidence_bundle=None, causality_candidates=None, rca_card=None, operational_context=None, pm_compliance=None, documents=None, processed_text_records=None)

Translate all RCA workflow artifacts into a graph of nodes and edges.

Iterates over every supplied artifact, creates typed nodes for each entity (assets, components, failure modes, events, telemetry signals, anomalies, causal candidates, evidence snippets, etc.), and links them with labelled directed edges. All artifacts are optional; only those provided contribute nodes and edges.

Parameters:

• schema_paths (Optional[Union[str, pathlib.Path, Iterable[Union[str, pathlib.Path]]]]) – Optional path(s) to TOML schema file(s) used for label resolution and endpoint validation. An empty schema is used when None.

• event (Optional[Dict[str, Any]]) – Parsed event artifact dict.

• kg_context (Optional[Dict[str, Any]]) – Parsed kg_context artifact dict (components, failure modes, past events).

• telemetry_summary (Optional[Dict[str, Any]]) – Parsed telemetry_summary artifact dict.

• evidence_bundle (Optional[Dict[str, Any]]) – Parsed evidence_bundle artifact dict.

• causality_candidates (Optional[Dict[str, Any]]) – Parsed causality_candidates artifact dict.

• rca_card (Optional[Dict[str, Any]]) – Parsed rca_card artifact dict.

• operational_context (Optional[Dict[str, Any]]) – Parsed operational_context artifact dict.

• pm_compliance (Optional[Dict[str, Any]]) – Parsed pm_compliance artifact dict.

• documents (Optional[Sequence[Dict[str, Any]]]) – List of document descriptor dicts.

• processed_text_records (Optional[Sequence[Dict[str, Any]]]) – List of processed_text_record dicts.

Returns:

A two-tuple (nodes, edges) — lists of dicts ready for bulk ingestion via ingest_graph_toml().

Return type:

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

src.dackar.knowledge_graph.kg_schema_builder_workflow.build_primary_key_map(schema)

Build a {node_label: primary_key} map from a (merged) TOML schema.

Resolves each node’s merge key via _node_primary_key(), so an explicit primary_key declaration is honoured and "id" is used as the default. Suitable for passing as the primary_keys argument to Py2Neo.upsert_nodes_batch() / Py2Neo.upsert_edges_batch().

Parameters:

schema (Dict[str, Any]) – A schema dict (or merged schema) with a "node" mapping.

Returns:

Mapping of node label to the property name used as its MERGE key.

Return type:

Dict[str, str]

src.dackar.knowledge_graph.kg_schema_builder_workflow.ingest_graph_toml(client, nodes, edges, database=None, primary_keys=None)

Bulk-upsert a node list and edge list into Neo4j.

A thin wrapper that calls Py2Neo.upsert_nodes_batch() followed by Py2Neo.upsert_edges_batch(), skipping each call when the corresponding list is empty.

Parameters:

• client (dackar.knowledge_graph.py2neo.Py2Neo) – Active Py2Neo connection.

• nodes (List[Dict[str, Any]]) – List of node dicts as returned by build_graph_from_workflow_artifacts().

• edges (List[Dict[str, Any]]) – List of edge dicts as returned by build_graph_from_workflow_artifacts().

• database (Optional[str]) – Target database name; uses the driver default when None.

• primary_keys (Optional[Dict[str, str]]) – Optional {label: key_name} merge-key map (e.g. from build_primary_key_map()); defaults to "id" per label.

Return type:

None