""" supersession_chain_builder.py ============================= Builds part supersession lineage chains from a flat part master. Given a table of parts with (optionally sparse and inconsistent) predecessor / successor declarations, this module: 1. Reconciles predecessor and successor columns into a single edge set. 2. Cleans the graph (self-loops, duplicates, cycles, dangling references). 3. Optionally breaks the chain at NON_INTERCHANGEABLE links, because those parts must be planned independently and demand history does not roll forward. 4. Assigns a deterministic, stable CHAIN_ID to every connected family. 5. Assigns CHAIN_SEQUENCE (generation position of the part inside its chain) and CHAIN_LENGTH / CHAIN_DEPTH so a planner can filter, e.g., "show me every chain seven generations deep". 6. Emits a diagnostics table so data-quality problems are visible instead of silently absorbed. Complexity is O(N + E). It is designed for hundreds of thousands of parts. Author: Arun License: MIT """ from __future__ import annotations from collections import Counter, defaultdict from dataclasses import dataclass from typing import Dict, Iterable, List, Optional, Set, Tuple import numpy as np import pandas as pd # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- BACKWARDS_COMPATIBLE = "backwards_compatible" FULLY_INTERCHANGEABLE = "fully_interchangeable" NON_INTERCHANGEABLE = "non_interchangeable" VALID_LINEAGE_TYPES = { BACKWARDS_COMPATIBLE, FULLY_INTERCHANGEABLE, NON_INTERCHANGEABLE, } _NULL_TOKENS = {"", "-", "na", "n/a", "nan", "none", "null", "nil", "#n/a"} # Values a planner will recognise, mapped onto the canonical form above. _LINEAGE_ALIASES = { "bc": BACKWARDS_COMPATIBLE, "backward_compatible": BACKWARDS_COMPATIBLE, "backwards_compat": BACKWARDS_COMPATIBLE, "one_way": BACKWARDS_COMPATIBLE, "fi": FULLY_INTERCHANGEABLE, "interchangeable": FULLY_INTERCHANGEABLE, "two_way": FULLY_INTERCHANGEABLE, "ni": NON_INTERCHANGEABLE, "not_interchangeable": NON_INTERCHANGEABLE, "no": NON_INTERCHANGEABLE, } # --------------------------------------------------------------------------- # Result container # --------------------------------------------------------------------------- @dataclass class ChainResult: """Everything the builder produces.""" parts: pd.DataFrame # one row per part, enriched with chain columns chains: pd.DataFrame # one row per chain, with a planning recommendation diagnostics: pd.DataFrame # one row per data-quality finding edges: pd.DataFrame # the reconciled graph actually used def summary(self) -> str: n_parts = len(self.parts) n_chains = len(self.chains) singletons = int((self.chains["chain_length"] == 1).sum()) if n_chains else 0 longest = int(self.chains["chain_length"].max()) if n_chains else 0 errs = int((self.diagnostics["severity"] == "ERROR").sum()) if len(self.diagnostics) else 0 warns = int((self.diagnostics["severity"] == "WARNING").sum()) if len(self.diagnostics) else 0 return ( f"{n_parts:,} parts -> {n_chains:,} chains " f"({singletons:,} standalone, longest = {longest} parts) | " f"{errs} errors, {warns} warnings" ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _clean_key(value) -> Optional[str]: """Normalise a part-number-like value to a comparable string, or None.""" if value is None: return None if isinstance(value, float) and np.isnan(value): return None if value is pd.NaT: return None text = str(value).strip() if text.lower() in _NULL_TOKENS: return None return text.upper() def _is_null(value) -> bool: """True for None, NaN, NaT and empty strings, without tripping on arrays.""" if value is None or value is pd.NaT: return True try: if bool(pd.isna(value)): return True except (TypeError, ValueError): return False return isinstance(value, str) and value.strip() == "" def _distinct_non_null(series: pd.Series) -> List: """Distinct non-null values, in first-seen order.""" out: List = [] for value in series.tolist(): if _is_null(value): continue if value not in out: out.append(value) return out def _clean_lineage(value) -> Optional[str]: key = _clean_key(value) if key is None: return None token = key.lower().replace(" ", "_").replace("-", "_") token = _LINEAGE_ALIASES.get(token, token) return token if token in VALID_LINEAGE_TYPES else f"__invalid__{key}" class _UnionFind: """Iterative union-find with path compression and union by size.""" def __init__(self) -> None: self.parent: Dict[str, str] = {} self.size: Dict[str, int] = {} def add(self, node: str) -> None: if node not in self.parent: self.parent[node] = node self.size[node] = 1 def find(self, node: str) -> str: root = node while self.parent[root] != root: root = self.parent[root] while self.parent[node] != root: # path compression self.parent[node], node = root, self.parent[node] return root def union(self, a: str, b: str) -> None: ra, rb = self.find(a), self.find(b) if ra == rb: return if self.size[ra] < self.size[rb]: ra, rb = rb, ra self.parent[rb] = ra self.size[ra] += self.size[rb] # --------------------------------------------------------------------------- # Main entry point # --------------------------------------------------------------------------- def build_supersession_chains( df: pd.DataFrame, *, part_col: str = "part_number", predecessor_col: str = "predecessor_part", successor_col: str = "successor_part", release_col: str = "release_date", eol_col: str = "end_of_life_date", lineage_col: str = "lineage_type", break_on_non_interchangeable: bool = True, include_dangling_refs: bool = False, chain_id_prefix: str = "SSC", previous_chain_ids: Optional[pd.DataFrame] = None, ) -> ChainResult: """Build supersession chains from a part master table. Parameters ---------- df Part master. Only ``part_col`` is strictly required; everything else is used when present. break_on_non_interchangeable When True (default), a NON_INTERCHANGEABLE link does not join the two parts into one chain. They stay adjacent in the data (the declared relationship is preserved on the part row) but are planned separately, which matches the rule that non-interchangeable demand history does not roll forward. include_dangling_refs When False (default), a predecessor/successor that is referenced but absent from the part master is excluded from the graph entirely — the edge is dropped and logged as a WARNING, and the chain is built from only the parts that actually exist in the master. When True, the missing part is instead added as a placeholder row with ``in_master_data = False`` so the chain isn't shortened by data that simply hasn't loaded yet. chain_id_prefix Prefix for the generated chain identifier, e.g. ``SSC-ABC-100``. previous_chain_ids Optional two-column table (``part_number``, ``chain_id``) from a prior run. When supplied, a chain that shares members with a previously seen chain keeps that chain's ID even if its origin part changes (e.g. an earlier predecessor is discovered later). Pass ``result.parts[["part_number", "chain_id"]]`` from the last run to keep IDs stable over time. Splits and merges relative to the prior mapping are logged as INFO diagnostics. """ if part_col not in df.columns: raise KeyError(f"required column {part_col!r} not found") diagnostics: List[dict] = [] def log(severity: str, issue: str, part: Optional[str], detail: str) -> None: diagnostics.append( {"severity": severity, "issue": issue, "part_number": part, "detail": detail} ) # -- 1. normalise the input ------------------------------------------------ work = df.copy() work["_part"] = work[part_col].map(_clean_key) missing_pn = work["_part"].isna() if missing_pn.any(): for idx in work.index[missing_pn]: log("ERROR", "missing_part_number", None, f"row {idx} dropped: blank part number") work = work[~missing_pn] work["_pred"] = work[predecessor_col].map(_clean_key) if predecessor_col in work else None work["_succ"] = work[successor_col].map(_clean_key) if successor_col in work else None work["_lineage"] = work[lineage_col].map(_clean_lineage) if lineage_col in work else None for date_key, col in (("_release", release_col), ("_eol", eol_col)): if col in work.columns: parsed = pd.to_datetime(work[col], errors="coerce") bad = parsed.isna() & work[col].notna() & (work[col].map(_clean_key).notna()) for pn in work.loc[bad, "_part"]: log("WARNING", "unparseable_date", pn, f"{col} could not be parsed") work[date_key] = parsed else: work[date_key] = pd.NaT # invalid lineage labels bad_lineage = work["_lineage"].astype("string").str.startswith("__invalid__", na=False) for pn, val in zip(work.loc[bad_lineage, "_part"], work.loc[bad_lineage, "_lineage"]): log("ERROR", "invalid_lineage_type", pn, f"unrecognised value {val[11:]!r}; treated as unknown") work.loc[bad_lineage, "_lineage"] = None # -- 2. duplicate part numbers ------------------------------------------- # Naively keeping the first row loses information: extracts frequently # carry the same part twice, once with the predecessor populated and once # without. Coalesce field by field, and only call it a conflict when two # rows assert two *different* non-null values. merge_fields = ["_pred", "_succ", "_lineage", "_release", "_eol"] dup_mask = work["_part"].duplicated(keep=False) if dup_mask.any(): frames = [work[~dup_mask]] for pn, grp in work[dup_mask].groupby("_part", sort=False): row = grp.iloc[[0]].copy() conflicts: List[str] = [] filled: List[str] = [] for col in merge_fields: distinct = _distinct_non_null(grp[col]) if len(distinct) > 1: conflicts.append(col.lstrip("_")) if distinct: if _is_null(row.iat[0, row.columns.get_loc(col)]): filled.append(col.lstrip("_")) row.iat[0, row.columns.get_loc(col)] = distinct[0] if conflicts: log("ERROR", "duplicate_part_conflict", pn, f"{len(grp)} rows disagree on {', '.join(conflicts)}; " "first populated value kept") elif filled: log("WARNING", "duplicate_part_merged", pn, f"{len(grp)} rows merged; {', '.join(filled)} recovered from a later row") else: log("WARNING", "duplicate_part_row", pn, f"{len(grp)} identical rows collapsed to one") frames.append(row) work = pd.concat(frames) work = work.set_index("_part", drop=False) known: Set[str] = set(work.index) # -- 3. dangling references ---------------------------------------------- referenced: Set[str] = set() for col in ("_pred", "_succ"): referenced |= set(work[col].dropna()) dangling = sorted(referenced - known) for pn in dangling: log("WARNING", "dangling_reference", pn, "referenced as predecessor/successor but absent from part master" + ("; added as placeholder" if include_dangling_refs else "; excluded")) nodes: Set[str] = set(known) if include_dangling_refs: nodes |= set(dangling) # -- 4. reconcile predecessor / successor into one edge set --------------- # An edge is (parent -> child). ``lineage_type`` describes the child's # relationship to its predecessor, so the edge carries the CHILD's label. declared: Dict[Tuple[str, str], Set[str]] = defaultdict(set) for pn, pred, succ in zip(work["_part"], work["_pred"], work["_succ"]): if pred is not None: if pred == pn: log("ERROR", "self_reference", pn, "part lists itself as predecessor; edge dropped") elif pred in nodes: declared[(pred, pn)].add("predecessor_col") if succ is not None: if succ == pn: log("ERROR", "self_reference", pn, "part lists itself as successor; edge dropped") elif succ in nodes: declared[(pn, succ)].add("successor_col") # pandas hands back NaN (a float) rather than None for empty object cells, # which poisons every downstream set/sort. Normalise once, here. lineage_of = { pn: (val if isinstance(val, str) else None) for pn, val in work["_lineage"].to_dict().items() } release_of = work["_release"].to_dict() eol_of = work["_eol"].to_dict() edge_rows: List[dict] = [] retained: List[Tuple[str, str]] = [] for (parent, child), sources in declared.items(): lineage = lineage_of.get(child) one_sided = len(sources) == 1 and parent in known and child in known if one_sided: log("WARNING", "one_sided_relationship", child, f"{parent} -> {child} declared only via {next(iter(sources))}; accepted") if lineage is None: log("WARNING", "missing_lineage_type", child, f"no lineage type for link {parent} -> {child}; chain kept, planning rule unknown") breaks = break_on_non_interchangeable and lineage == NON_INTERCHANGEABLE if breaks: log("INFO", "chain_break_non_interchangeable", child, f"{parent} -> {child} not chained: non-interchangeable, planned independently") else: retained.append((parent, child)) edge_rows.append( { "parent_part": parent, "child_part": child, "lineage_type": lineage, "declared_via": "+".join(sorted(sources)), "used_for_chaining": not breaks, } ) # -- 5. break cycles deterministically ----------------------------------- retained = _break_cycles(retained, nodes, release_of, log) children: Dict[str, List[str]] = defaultdict(list) parents: Dict[str, List[str]] = defaultdict(list) for parent, child in retained: children[parent].append(child) parents[child].append(parent) for node, ps in parents.items(): if len(ps) > 1: log("WARNING", "merge_point", node, f"{len(ps)} predecessors converge here ({', '.join(sorted(ps))})") for node, cs in children.items(): if len(cs) > 1: log("WARNING", "branch_point", node, f"{len(cs)} successors branch from here ({', '.join(sorted(cs))})") # -- 6. components -> chains --------------------------------------------- uf = _UnionFind() for node in nodes: uf.add(node) for parent, child in retained: uf.union(parent, child) components: Dict[str, List[str]] = defaultdict(list) for node in nodes: components[uf.find(node)].append(node) # -- 7. sequence (generation depth) via topological order ---------------- sequence = _generation_sequence(nodes, children, parents) # -- 8. assemble per-part output ----------------------------------------- def head_key(node: str): rel = release_of.get(node, pd.NaT) return (pd.Timestamp.max if pd.isna(rel) else rel, node) chain_id_of: Dict[str, str] = {} chain_len_of: Dict[str, int] = {} chain_depth_of: Dict[str, int] = {} chain_head_of: Dict[str, str] = {} chain_tail_of: Dict[str, str] = {} chain_rows: List[dict] = [] lineage_by_child = {row["child_part"]: row["lineage_type"] for row in edge_rows} # -- persistent chain IDs: reconcile against a prior run's mapping -------- # A component keeps its previous chain_id if a majority of its members # were previously assigned to that ID, even if the chain's origin part has # since changed. If a previous chain's members split across more than one # new component, only the component with the largest overlap keeps the # ID; the rest get a fresh origin-derived ID and the split is logged. If # multiple previous chains merge into one new component, the most # strongly represented previous ID is kept and the merge is logged. prev_map: Dict[str, str] = {} if previous_chain_ids is not None and len(previous_chain_ids): prev_map = dict( zip(previous_chain_ids["part_number"].map(_clean_key), previous_chain_ids["chain_id"]) ) resolved_id_for_root: Dict[str, str] = {} if prev_map: counts_by_root: Dict[str, Counter] = {} claims: Dict[str, List[Tuple[str, int]]] = defaultdict(list) for root, members in components.items(): counts = Counter(prev_map[m] for m in members if m in prev_map) if counts: counts_by_root[root] = counts best_id, best_count = counts.most_common(1)[0] claims[best_id].append((root, best_count)) for prev_id, claimants in claims.items(): claimants.sort(key=lambda rc: (-rc[1], rc[0])) winner_root = claimants[0][0] resolved_id_for_root[winner_root] = prev_id if len(claimants) > 1: losers = ", ".join(r for r, _ in claimants[1:]) log("INFO", "chain_id_split", None, f"chain {prev_id} split into {len(claimants)} components; " f"ID retained by the component with the largest overlap, " f"new IDs assigned to component(s) rooted at: {losers}") for root, counts in counts_by_root.items(): if len(counts) > 1: merged_from = sorted(counts) kept = resolved_id_for_root.get(root) if kept: log("INFO", "chain_ids_merged", None, f"components previously under {merged_from} merged into " f"one chain; retained {kept}") for root, members in components.items(): roots = [n for n in members if not parents.get(n)] head = min(roots or members, key=head_key) tails = sorted((n for n in members if not children.get(n)), key=head_key) chain_id = resolved_id_for_root.get(root) or f"{chain_id_prefix}-{head}" depth = max(sequence[n] for n in members) for node in members: chain_id_of[node] = chain_id chain_len_of[node] = len(members) chain_depth_of[node] = depth chain_head_of[node] = head chain_tail_of[node] = tails[-1] if tails else head link_types = { lineage_by_child[n] for n in members if n != head and isinstance(lineage_by_child.get(n), str) } chain_rows.append( { "chain_id": chain_id, "chain_length": len(members), "chain_depth": depth, "chain_head": head, "chain_tail": tails[-1] if tails else head, "n_branch_points": sum(1 for n in members if len(children.get(n, [])) > 1), "link_types": ", ".join(sorted(link_types)) or "n/a", "planning_strategy": _planning_strategy(link_types, len(members)), "head_release_date": release_of.get(head, pd.NaT), "tail_eol_date": eol_of.get(tails[-1] if tails else head, pd.NaT), } ) ordered_nodes = sorted(nodes, key=lambda n: (chain_id_of[n], sequence[n], n)) out = pd.DataFrame({"part_number": ordered_nodes}) out["chain_id"] = out["part_number"].map(chain_id_of) out["chain_sequence"] = out["part_number"].map(sequence) out["chain_length"] = out["part_number"].map(chain_len_of) out["chain_depth"] = out["part_number"].map(chain_depth_of) out["chain_head"] = out["part_number"].map(chain_head_of) out["chain_tail"] = out["part_number"].map(chain_tail_of) out["predecessor_part"] = out["part_number"].map( lambda n: ", ".join(sorted(parents.get(n, []))) or None ) out["successor_part"] = out["part_number"].map( lambda n: ", ".join(sorted(children.get(n, []))) or None ) out["lineage_type"] = out["part_number"].map(lineage_of.get) out["release_date"] = out["part_number"].map(release_of.get) out["end_of_life_date"] = out["part_number"].map(eol_of.get) out["is_chain_head"] = out["part_number"].eq(out["chain_head"]) out["is_chain_tail"] = out["part_number"].map(lambda n: not children.get(n)) out["is_branch_point"] = out["part_number"].map(lambda n: len(children.get(n, [])) > 1) out["is_merge_point"] = out["part_number"].map(lambda n: len(parents.get(n, [])) > 1) out["in_master_data"] = out["part_number"].isin(known) out["plan_this_part"] = out["is_chain_tail"] & out["in_master_data"] chains = pd.DataFrame(chain_rows).sort_values( ["chain_length", "chain_id"], ascending=[False, True] ).reset_index(drop=True) diag = pd.DataFrame( diagnostics, columns=["severity", "issue", "part_number", "detail"] ) if len(diag): order = {"ERROR": 0, "WARNING": 1, "INFO": 2} diag = diag.sort_values( by=["severity", "issue", "part_number"], key=lambda s: s.map(order) if s.name == "severity" else s, ).reset_index(drop=True) edges = pd.DataFrame( edge_rows, columns=["parent_part", "child_part", "lineage_type", "declared_via", "used_for_chaining"], ) return ChainResult(parts=out, chains=chains, diagnostics=diag, edges=edges) # --------------------------------------------------------------------------- # Internals # --------------------------------------------------------------------------- def _break_cycles( edges: List[Tuple[str, str]], nodes: Iterable[str], release_of: Dict[str, pd.Timestamp], log, ) -> List[Tuple[str, str]]: """Remove the minimum obvious edges needed to make the graph acyclic. Real part masters do contain circular supersessions (usually a data-entry error, occasionally a genuine A/B swap). Rather than fail, we pick the earliest-released part in each cycle, declare it the origin, and drop the edges flowing into it. """ edge_set = set(edges) while True: children: Dict[str, List[str]] = defaultdict(list) indeg: Dict[str, int] = {n: 0 for n in nodes} for parent, child in edge_set: children[parent].append(child) indeg[child] += 1 stack = [n for n, d in indeg.items() if d == 0] seen = 0 while stack: node = stack.pop() seen += 1 for child in children.get(node, []): indeg[child] -= 1 if indeg[child] == 0: stack.append(child) if seen == len(indeg): return sorted(edge_set) stuck = [n for n, d in indeg.items() if d > 0] def key(node: str): rel = release_of.get(node, pd.NaT) return (pd.Timestamp.max if pd.isna(rel) else rel, node) origin = min(stuck, key=key) removed = [(p, c) for (p, c) in edge_set if c == origin] for parent, child in removed: edge_set.discard((parent, child)) log("ERROR", "circular_supersession", child, f"cycle detected; dropped {parent} -> {child} and treated {origin} as chain origin") def _generation_sequence( nodes: Iterable[str], children: Dict[str, List[str]], parents: Dict[str, List[str]], ) -> Dict[str, int]: """Longest-path depth from any chain origin. Origin parts get sequence 1.""" indeg = {n: len(parents.get(n, [])) for n in nodes} seq = {n: 1 for n in nodes} stack = [n for n, d in indeg.items() if d == 0] while stack: node = stack.pop() for child in children.get(node, []): if seq[node] + 1 > seq[child]: seq[child] = seq[node] + 1 indeg[child] -= 1 if indeg[child] == 0: stack.append(child) return seq def _planning_strategy(link_types: Set[str], n_members: int) -> str: if n_members == 1: return "Standalone part - plan on its own demand signal" if link_types == {FULLY_INTERCHANGEABLE}: return "Plan the chain as a single item; net requirements across all generations" if BACKWARDS_COMPATIBLE in link_types: return "One-way substitution - roll demand forward to the newest part; sell down predecessors, never buy them" if not link_types: return "Lineage type unknown - review before rolling demand forward" return "Mixed lineage - roll forward only across interchangeable links" __all__ = [ "build_supersession_chains", "ChainResult", "BACKWARDS_COMPATIBLE", "FULLY_INTERCHANGEABLE", "NON_INTERCHANGEABLE", ]