src.dackar.RCA.equipment_similarity.equipment_similarity_resolver

equipment_similarity_resolver — EquipmentSimilarityResolver.

Identifies sister equipment using two complementary tiers:

Tier 2 — Failure mode overlap

Derived from kg_context.failure_modes[]. Components that share ≥ fm_overlap_min_shared failure modes with the target are flagged. No Chroma, no KG re-query — purely from the already-built artifact.

Tier 3 — Spec embedding similarity

Queries the EquipmentSpecStore (Chroma equipment_specs collection). Query text is built from kg_context fields — no additional KG call. Skipped silently if spec_store is None or unpopulated. Ranked and thresholded on the raw dense-vector distance (_vector_score), not the fused RRF _score that ChromaRecordStore’s hybrid query overwrites onto each hit — the RRF rank score is not a distance and would break the embedding_min_score gate.

Results from both tiers are merged, deduplicated (same component_id in multiple tiers → combined match_type), and returned as a ranked list of SisterComponent objects.

Attributes

logger
JsonDict
NON_EMBEDDING_DISTANCE

Classes

SisterComponent	A single sister equipment candidate.
EquipmentSimilarityConfig	Configuration for EquipmentSimilarityResolver.
EquipmentSimilarityResolver	Resolves sister equipment using failure mode overlap and spec embeddings.

Module Contents

src.dackar.RCA.equipment_similarity.equipment_similarity_resolver.logger

src.dackar.RCA.equipment_similarity.equipment_similarity_resolver.JsonDict

src.dackar.RCA.equipment_similarity.equipment_similarity_resolver.NON_EMBEDDING_DISTANCE = 1.0

class src.dackar.RCA.equipment_similarity.equipment_similarity_resolver.SisterComponent

A single sister equipment candidate.

component_id

KG element_usage node ID.

component_label

Human-readable name, if available from kg_context.

match_type

How this component was identified. Possible values: "failure_mode_overlap", "spec_embedding", "fm_overlap+spec_embedding".

shared_fm_count

Number of shared failure modes (Tier 2). 0 for Tier 3-only matches.

embedding_score

Chroma similarity score (Tier 3). NON_EMBEDDING_DISTANCE for Tier 2-only matches. Lower score = more similar (Chroma uses distance by default).

component_id: str

component_label: str | None = None

match_type: str = 'spec_embedding'

shared_fm_count: int = 0

embedding_score: float = 1.0

to_dict()

Serialize this candidate to a plain dict for the CMMSContextBuilder boundary.

Returns:

Mapping with keys component_id, component_label, match_type, shared_fm_count and embedding_score.

Return type:

dict

Notes

For failure_mode_overlap matches (Tier 2 — no embedding distance was ever computed) embedding_score is forced to NON_EMBEDDING_DISTANCE (1.0) rather than left at the dataclass default, so a Tier-2-only sister is never ranked as more similar than a genuine embedding hit downstream. For spec_embedding and combined matches it carries the raw Chroma vector distance (lower = closer).

class src.dackar.RCA.equipment_similarity.equipment_similarity_resolver.EquipmentSimilarityConfig

Configuration for EquipmentSimilarityResolver.

Parameters:

• fm_overlap_min_shared – Minimum number of shared failure modes for Tier 2 inclusion. Default: 2. Set to 1 for more inclusive matching.

• embedding_top_k – Number of candidates to request from Chroma (Tier 3). After exclude_ids filtering, up to this many are returned.

• embedding_min_score – Maximum acceptable Chroma distance score. Chroma returns L2 distances (lower = more similar); results above this threshold are filtered out. Default: 0.8 (fairly permissive). Reduce to 0.4–0.6 for stricter matching.

• include_fm_overlap – Enable Tier 2 (failure mode overlap).

• include_spec_embedding – Enable Tier 3 (spec embedding). Has no effect if spec_store is None.

fm_overlap_min_shared: int = 2

embedding_top_k: int = 10

embedding_min_score: float = 0.8

include_fm_overlap: bool = True

include_spec_embedding: bool = True

class src.dackar.RCA.equipment_similarity.equipment_similarity_resolver.EquipmentSimilarityResolver(spec_store=None, config=None)

Resolves sister equipment using failure mode overlap and spec embeddings.

Parameters:

• spec_store (Optional[Any]) – EquipmentSpecStore instance (or None to disable Tier 3).

• config (Optional[EquipmentSimilarityConfig]) – EquipmentSimilarityConfig — defaults are conservative.

spec_store = None

config

resolve_similar(target_component_ids, kg_context)

Return sister equipment candidates for the given target components.

Parameters:

• target_component_ids (List[str]) – KG component IDs of the primary asset’s components. These are excluded from results.

• kg_context (JsonDict) – KG context artifact from Stage 5A. Provides failure modes and component labels for query text construction.

Returns:

Ranked by (embedding_score ASC, shared_fm_count DESC). Empty list if no sisters found or both tiers are disabled.

Return type:

List[SisterComponent]

_resolve_by_fm_overlap(target_set, kg_context)

Find components that share ≥ fm_overlap_min_shared failure modes with any target component.

Parameters:

• target_set (Set[str])

• kg_context (JsonDict)

Return type:

List[SisterComponent]

_resolve_by_embedding(query_text, target_set)

Query EquipmentSpecStore and return candidates above the score threshold.

Parameters:

• query_text (str)

• target_set (Set[str])

Return type:

List[SisterComponent]

_build_query_text(target_component_ids, kg_context)

Build a query string from kg_context for the target components.

Uses component label/type from kg_context.components[] and failure mode names from kg_context.failure_modes[]. No KG re-query required.

Parameters:

• target_component_ids (List[str])

• kg_context (JsonDict)

Return type:

str

_build_label_map(kg_context)

Build component_id → component_label from kg_context.components[].

Parameters:

kg_context (JsonDict)

Return type:

Dict[str, Optional[str]]