Hugging Face's logo

Spaces:
SupantBT
/
Beltbot
Running

App Files Community
Beltbot / knowledge_graph.py
SupantBT's picture
SupantBT
Modutech + Movex assistant β€” FastAPI + Qdrant + graph
f817912
Raw History Blame Contribute Delete
26.5 kB
"""
knowledge_graph.py
==================
GraphRAG Module ΰΈͺำหรับ Movex Product Chatbot v3
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Persistence: SQLite (movex_kg.db)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
DB ΰΈ‘ΰΈ΅ 3 tables:
TABLE nodes
node_id TEXT PRIMARY KEY e.g. "product:movex_chain_LF880_TAB_K325"
node_type TEXT "Product" | "Series" | "Material" | "ManufacturingProcess"
attrs TEXT (JSON) attributes ΰΈ—ΰΈ±ΰΉ‰ΰΈ‡ΰΈ«ΰΈ‘ΰΈ”ΰΈ‚ΰΈ­ΰΈ‡ node
TABLE edges
src TEXT
dst TEXT
relation TEXT "BELONGS_TO" | "MADE_OF" | "COMPATIBLE_WITH" | "USES_PROCESS"
PRIMARY KEY (src, dst, relation)
TABLE metadata
key TEXT PRIMARY KEY "built_at" | "node_count" | "schema_version" | ...
value TEXT
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Workflow
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Build (manual ΰΉ€ΰΈ—ΰΉˆΰΈ²ΰΈ™ΰΈ±ΰΉ‰ΰΈ™):
python build_graph.py
β†’ ΰΈ­ΰΉˆΰΈ²ΰΈ™ JSON β†’ build NetworkX DiGraph β†’ save ΰΈ₯ΰΈ‡ movex_kg.db
Runtime (ทุก startup ΰΈ‚ΰΈ­ΰΈ‡ chatbot):
kg = KnowledgeGraph.load_from_db("./movex_kg.db")
β†’ SELECT nodes + edges β†’ rebuild NetworkX DiGraph ΰΉƒΰΈ™ memory
β†’ ΰΉ€ΰΈ£ΰΉ‡ΰΈ§ΰΈΰΈ§ΰΉˆΰΈ² build จาก JSON ΰΈ›ΰΈ£ΰΈ°ΰΈ‘ΰΈ²ΰΈ“ 10x
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Graph Structure
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Nodes:
product:{product_id} Product nodes
series:{series_num} Series nodes (820 | 821 | 880 | 882 | 83 | 103)
material:{code} Material nodes (LF | SS | LFN | ...)
process:{name} ManufacturingProcess nodes (Machined | Molded)
Edges:
BELONGS_TO Product β†’ Series
MADE_OF Product β†’ Material
USES_PROCESS Product β†’ ManufacturingProcess
COMPATIBLE_WITH Product ↔ Product (bidirectional, chain ↔ sprocket)
"""
import re
import json
import sqlite3
import time
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import networkx as nx
# ΰΉ€ΰΈžΰΈ΄ΰΉˆΰΈ‘ΰΈ•ΰΈ±ΰΈ§ΰΉ€ΰΈ₯ΰΈ‚ΰΈ™ΰΈ΅ΰΉ‰ΰΈ—ΰΈΈΰΈΰΈ„ΰΈ£ΰΈ±ΰΉ‰ΰΈ‡ΰΈ—ΰΈ΅ΰΉˆΰΉ€ΰΈ›ΰΈ₯ΰΈ΅ΰΉˆΰΈ’ΰΈ™ FIELDS_BY_TYPE ΰΈ«ΰΈ£ΰΈ·ΰΈ­ edge logic
# β†’ build_graph.py ΰΈˆΰΈ°ΰΈšΰΈ±ΰΈ™ΰΈ—ΰΈΆΰΈΰΉ„ΰΈ§ΰΉ‰ΰΉƒΰΈ™ metadata table
SCHEMA_VERSION = "1"
FIELDS_BY_TYPE: dict[str, list[str]] = {
"Product": [
"Ref", "Art_Nr", "series", "product_type",
"usage", "key_features", "constraints",
"Plate_Width_mm", "Max_Working_Load_N", "Weight_kg_m",
"Min_curve_radius_mm", "Radius_min_mm", "Pitch_mm", "Links_m",
"Z_Teeth", "Bore_mm", "PD_mm", "OD_mm", "S_mm",
"compatible_chain", "Sprocket_Type", "manufacturing_process",
"Material", "Pivot_material",
],
"Series": ["name"],
"Material": ["full_name", "description", "best_used_for"],
"ManufacturingProcess": ["full_name", "description", "best_used_for"],
}
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# GraphContext dataclass
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
@dataclass
class GraphContext:
"""Context ΰΈ—ΰΈ΅ΰΉˆ graph expand ออกฑา ΰΈͺำหรับΰΈͺΰΉˆΰΈ‡ΰΈ•ΰΉˆΰΈ­ΰΉƒΰΈ«ΰΉ‰ retriever / LLM"""
seed_products: list[str]
same_series: list[str] = field(default_factory=list)
compatible_products: list[str] = field(default_factory=list)
same_material: list[str] = field(default_factory=list)
material_details: dict = field(default_factory=dict)
process_details: dict = field(default_factory=dict)
all_candidate_ids: list[str] = field(default_factory=list)
def to_summary_str(self) -> str:
return (
f"seed={self.seed_products} | "
f"same_series({len(self.same_series)}) | "
f"compatible({len(self.compatible_products)}) | "
f"same_material({len(self.same_material)})"
)
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# KnowledgeGraph β€” Runtime class
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
class KnowledgeGraph:
"""
NetworkX Knowledge Graph ΰΈͺำหรับ Movex product catalog
ΰΉ„ΰΈ‘ΰΉˆ instantiate ΰΉ‚ΰΈ”ΰΈ’ΰΈ•ΰΈ£ΰΈ‡ β€” ΰΉƒΰΈŠΰΉ‰ classmethod:
KnowledgeGraph.load_from_db(db_path) ← chatbot runtime
KnowledgeGraph.build_and_save(...) ← build_graph.py
"""
SERIES_RE = re.compile(r'\b(820|821|880|882|83|103)\b')
def __init__(self, graph: nx.DiGraph, db_path: str = ""):
self.graph = graph
self.db_path = db_path
# ──────────────────────────────────────────────────────
# ENTRY POINTS
# ──────────────────────────────────────────────────────
@classmethod
def load_from_db(cls, db_path: str) -> "KnowledgeGraph":
"""
ΰΉ‚ΰΈ«ΰΈ₯ΰΈ” graph จาก SQLite database
Parameters
----------
db_path : str
path ΰΉ„ΰΈ›ΰΈ’ΰΈ±ΰΈ‡ movex_kg.db
Raises
------
FileNotFoundError ΰΈ–ΰΉ‰ΰΈ² db_path ΰΉ„ΰΈ‘ΰΉˆΰΈ‘ΰΈ΅
RuntimeError ΰΈ–ΰΉ‰ΰΈ² DB ΰΈ§ΰΉˆΰΈ²ΰΈ‡ΰΉ€ΰΈ›ΰΈ₯่า (ΰΈ’ΰΈ±ΰΈ‡ΰΉ„ΰΈ‘ΰΉˆΰΉ„ΰΈ”ΰΉ‰ build)
"""
import os
if not os.path.isfile(db_path):
raise FileNotFoundError(
f"ΰΉ„ΰΈ‘ΰΉˆΰΈžΰΈš {db_path}\n"
f"กรุณารัน: python build_graph.py ΰΈΰΉˆΰΈ­ΰΈ™ΰΈ„ΰΈ£ΰΈ±ΰΈš"
)
t0 = time.perf_counter()
con = sqlite3.connect(db_path)
con.row_factory = sqlite3.Row
# ΰΉ‚ΰΈ«ΰΈ₯ΰΈ” nodes
g = nx.DiGraph()
node_rows = con.execute(
"SELECT node_id, node_type, attrs FROM nodes"
).fetchall()
if not node_rows:
con.close()
raise RuntimeError(
f"DB ΰΈ§ΰΉˆΰΈ²ΰΈ‡ΰΉ€ΰΈ›ΰΈ₯่า: {db_path}\n"
f"กรุณารัน: python build_graph.py ΰΈΰΉˆΰΈ­ΰΈ™ΰΈ„ΰΈ£ΰΈ±ΰΈš"
)
for row in node_rows:
attrs = json.loads(row["attrs"])
attrs["node_type"] = row["node_type"]
g.add_node(row["node_id"], **attrs)
# ΰΉ‚ΰΈ«ΰΈ₯ΰΈ” edges
for row in con.execute("SELECT src, dst, relation FROM edges"):
g.add_edge(row["src"], row["dst"], relation=row["relation"])
# metadata ΰΈͺำหรับแΰΈͺΰΈ”ΰΈ‡ info
meta = dict(con.execute("SELECT key, value FROM metadata").fetchall())
con.close()
elapsed = (time.perf_counter() - t0) * 1000
print(
f"⚑ KnowledgeGraph loaded from SQLite in {elapsed:.0f}ms\n"
f" πŸ“ {db_path}\n"
f" nodes={g.number_of_nodes()} "
f"edges={g.number_of_edges()} "
f"built={meta.get('built_at', 'unknown')}"
)
return cls(graph=g, db_path=db_path)
@classmethod
def build_and_save(
cls,
product_db: dict,
material_db: dict,
process_db: dict,
db_path: str,
) -> "KnowledgeGraph":
"""
Build graph จาก JSON dicts แΰΈ₯ΰΉ‰ΰΈ§ save ΰΈ₯ΰΈ‡ SQLite
ΰΉƒΰΈŠΰΉ‰ΰΉ‚ΰΈ”ΰΈ’ build_graph.py ΰΉ€ΰΈ—ΰΉˆΰΈ²ΰΈ™ΰΈ±ΰΉ‰ΰΈ™
"""
builder = _GraphBuilder(product_db, material_db, process_db)
graph = builder.build()
instance = cls(graph=graph, db_path=db_path)
instance._save_to_db(db_path)
return instance
# ──────────────────────────────────────────────────────
# SQLITE SAVE
# ──────────────────────────────────────────────────────
def _save_to_db(self, db_path: str):
"""ΰΈšΰΈ±ΰΈ™ΰΈ—ΰΈΆΰΈ NetworkX graph ΰΈ₯ΰΈ‡ SQLite (เรมฒกใน build_and_save)"""
import os
os.makedirs(os.path.dirname(os.path.abspath(db_path)), exist_ok=True)
t0 = time.perf_counter()
con = sqlite3.connect(db_path)
cur = con.cursor()
# ΰΈͺΰΈ£ΰΉ‰ΰΈ²ΰΈ‡ / ΰΈ₯ΰΉ‰ΰΈ²ΰΈ‡ tables
cur.executescript("""
DROP TABLE IF EXISTS nodes;
DROP TABLE IF EXISTS edges;
DROP TABLE IF EXISTS metadata;
CREATE TABLE nodes (
node_id TEXT PRIMARY KEY,
node_type TEXT NOT NULL,
attrs TEXT NOT NULL
);
CREATE TABLE edges (
src TEXT NOT NULL,
dst TEXT NOT NULL,
relation TEXT NOT NULL,
PRIMARY KEY (src, dst, relation)
);
CREATE INDEX IF NOT EXISTS idx_edges_src ON edges(src);
CREATE INDEX IF NOT EXISTS idx_edges_dst ON edges(dst);
CREATE INDEX IF NOT EXISTS idx_nodes_type ON nodes(node_type);
CREATE TABLE metadata (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
""")
# Insert nodes
cur.executemany(
"INSERT INTO nodes (node_id, node_type, attrs) VALUES (?,?,?)",
[
(
nid,
data.get("node_type", "Unknown"),
json.dumps(
{k: v for k, v in data.items() if k != "node_type"},
ensure_ascii=False,
),
)
for nid, data in self.graph.nodes(data=True)
],
)
# Insert edges
cur.executemany(
"INSERT OR IGNORE INTO edges (src, dst, relation) VALUES (?,?,?)",
[
(src, dst, data.get("relation", ""))
for src, dst, data in self.graph.edges(data=True)
],
)
# ΰΈ™ΰΈ±ΰΈš stats ΰΈͺำหรับ metadata
type_counts: dict[str, int] = {}
for _, data in self.graph.nodes(data=True):
t = data.get("node_type", "Unknown")
type_counts[t] = type_counts.get(t, 0) + 1
rel_counts: dict[str, int] = {}
for _, _, data in self.graph.edges(data=True):
r = data.get("relation", "")
rel_counts[r] = rel_counts.get(r, 0) + 1
# Insert metadata
cur.executemany(
"INSERT OR REPLACE INTO metadata (key, value) VALUES (?,?)",
[
("built_at", datetime.now().isoformat(timespec="seconds")),
("schema_version", SCHEMA_VERSION),
("node_count", str(self.graph.number_of_nodes())),
("edge_count", str(self.graph.number_of_edges())),
("node_types", json.dumps(type_counts)),
("edge_relations", json.dumps(rel_counts)),
],
)
con.commit()
con.close()
elapsed = (time.perf_counter() - t0) * 1000
print(
f"πŸ’Ύ Saved to SQLite in {elapsed:.0f}ms β†’ {db_path}\n"
f" node_types : {type_counts}\n"
f" edge_relations: {rel_counts}"
)
# ──────────────────────────────────────────────────────
# DB INFO
# ──────────────────────────────────────────────────────
def db_info(self) -> dict:
"""ΰΈ­ΰΉˆΰΈ²ΰΈ™ metadata จาก DB (built_at, schema_version, counts, ...)"""
if not self.db_path:
return {"status": "no_db_path"}
try:
con = sqlite3.connect(self.db_path)
info = dict(con.execute("SELECT key, value FROM metadata").fetchall())
con.close()
info["status"] = "ok"
for k in ("node_types", "edge_relations"):
if k in info:
info[k] = json.loads(info[k])
return info
except Exception as e:
return {"status": "error", "error": str(e)}
# ──────────────────────────────────────────────────────
# GRAPH QUERY API (ΰΉ„ΰΈ‘ΰΉˆΰΉ€ΰΈ›ΰΈ₯ΰΈ΅ΰΉˆΰΈ’ΰΈ™ΰΈˆΰΈ²ΰΈ v ΰΉ€ΰΈ”ΰΈ΄ΰΈ‘)
# ──────────────────────────────────────────────────────
def expand_products(
self,
seed_product_ids: list[str],
include_same_series: bool = True,
include_compatible: bool = True,
include_same_material: bool = False,
) -> GraphContext:
"""Main entry ΰΈͺำหรับ retriever β€” expand seed products ΰΈœΰΉˆΰΈ²ΰΈ™ graph"""
ctx = GraphContext(seed_products=list(seed_product_ids))
visited_series: set[str] = set()
visited_material: set[str] = set()
for pid in seed_product_ids:
pnode = f"product:{pid}"
if not self.graph.has_node(pnode):
continue
for nb in self.graph.successors(pnode):
edata = self.graph.edges[pnode, nb]
relation = edata.get("relation", "")
ndata = self.graph.nodes[nb]
ntype = ndata.get("node_type", "")
if relation == "BELONGS_TO" and ntype == "Series":
if include_same_series and nb not in visited_series:
visited_series.add(nb)
for pred in self.graph.predecessors(nb):
if self.graph.edges.get((pred, nb), {}).get("relation") == "BELONGS_TO":
pp = self.graph.nodes[pred].get("product_id", "")
if pp and pp not in seed_product_ids and pp not in ctx.same_series:
ctx.same_series.append(pp)
elif relation == "COMPATIBLE_WITH" and ntype == "Product":
if include_compatible:
np_ = ndata.get("product_id", "")
if np_ and np_ not in seed_product_ids and np_ not in ctx.compatible_products:
ctx.compatible_products.append(np_)
elif relation == "MADE_OF" and ntype == "Material":
mat = ndata.get("material_code", "")
if mat and mat not in ctx.material_details:
ctx.material_details[mat] = self._node_fields(nb, "Material")
if include_same_material and nb not in visited_material:
visited_material.add(nb)
for pred in self.graph.predecessors(nb):
if self.graph.edges.get((pred, nb), {}).get("relation") == "MADE_OF":
pp = self.graph.nodes[pred].get("product_id", "")
if pp and pp not in seed_product_ids and pp not in ctx.same_material:
ctx.same_material.append(pp)
elif relation == "USES_PROCESS" and ntype == "ManufacturingProcess":
proc = ndata.get("process_name", "")
if proc and proc not in ctx.process_details:
ctx.process_details[proc] = self._node_fields(nb, "ManufacturingProcess")
seen: set[str] = set(seed_product_ids)
ctx.all_candidate_ids = list(seed_product_ids)
for p in ctx.compatible_products + ctx.same_series + ctx.same_material:
if p not in seen:
seen.add(p)
ctx.all_candidate_ids.append(p)
return ctx
def get_compatible_products(self, product_id: str) -> list[str]:
pnode = f"product:{product_id}"
if not self.graph.has_node(pnode):
return []
return [
self.graph.nodes[nb].get("product_id", "")
for nb in self.graph.successors(pnode)
if self.graph.edges[pnode, nb].get("relation") == "COMPATIBLE_WITH"
and self.graph.nodes[nb].get("product_id")
]
def get_same_series(self, product_id: str) -> list[str]:
pnode = f"product:{product_id}"
if not self.graph.has_node(pnode):
return []
result: list[str] = []
for nb in self.graph.successors(pnode):
if self.graph.edges[pnode, nb].get("relation") == "BELONGS_TO":
for pred in self.graph.predecessors(nb):
if self.graph.edges.get((pred, nb), {}).get("relation") == "BELONGS_TO":
pid = self.graph.nodes[pred].get("product_id", "")
if pid and pid != product_id and pid not in result:
result.append(pid)
return result
def get_series_num(self, product_id: str) -> Optional[str]:
pnode = f"product:{product_id}"
if not self.graph.has_node(pnode):
return None
for nb in self.graph.successors(pnode):
if self.graph.edges[pnode, nb].get("relation") == "BELONGS_TO":
return self.graph.nodes[nb].get("series_num")
return None
def get_material_details(self, product_id: str) -> Optional[dict]:
pnode = f"product:{product_id}"
if not self.graph.has_node(pnode):
return None
for nb in self.graph.successors(pnode):
if self.graph.edges[pnode, nb].get("relation") == "MADE_OF":
return self._node_fields(nb, "Material")
return None
def get_process_details(self, product_id: str) -> Optional[dict]:
pnode = f"product:{product_id}"
if not self.graph.has_node(pnode):
return None
for nb in self.graph.successors(pnode):
if self.graph.edges[pnode, nb].get("relation") == "USES_PROCESS":
return self._node_fields(nb, "ManufacturingProcess")
return None
def get_products_by_series(self, series_num: str) -> list[str]:
snode = f"series:{series_num}"
if not self.graph.has_node(snode):
return []
return [
self.graph.nodes[pred].get("product_id", "")
for pred in self.graph.predecessors(snode)
if self.graph.edges.get((pred, snode), {}).get("relation") == "BELONGS_TO"
and self.graph.nodes[pred].get("product_id")
]
def get_products_by_material(self, material_code: str) -> list[str]:
mnode = f"material:{material_code}"
if not self.graph.has_node(mnode):
return []
return [
self.graph.nodes[pred].get("product_id", "")
for pred in self.graph.predecessors(mnode)
if self.graph.edges.get((pred, mnode), {}).get("relation") == "MADE_OF"
and self.graph.nodes[pred].get("product_id")
]
def node_count(self) -> dict[str, int]:
counts: dict[str, int] = {}
for _, data in self.graph.nodes(data=True):
t = data.get("node_type", "Unknown")
counts[t] = counts.get(t, 0) + 1
return counts
def _extract_series_num(self, text: str) -> Optional[str]:
m = self.SERIES_RE.search(text or "")
return m.group(1) if m else None
def _node_fields(self, node_id: str, node_type: str) -> dict:
data = self.graph.nodes.get(node_id, {})
return {k: data[k] for k in FIELDS_BY_TYPE.get(node_type, []) if k in data}
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
# _GraphBuilder (internal β€” ΰΉƒΰΈŠΰΉ‰ΰΉ‚ΰΈ”ΰΈ’ build_and_save ΰΉ€ΰΈ—ΰΉˆΰΈ²ΰΈ™ΰΈ±ΰΉ‰ΰΈ™)
# ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
class _GraphBuilder:
"""Build NetworkX DiGraph จาก product/material/process dicts"""
SERIES_RE = re.compile(r'\b(820|821|880|882|83|103)\b')
def __init__(self, product_db: dict, material_db: dict, process_db: dict):
self.product_db = product_db
self.material_db = material_db
self.process_db = process_db
def build(self) -> nx.DiGraph:
t0 = time.perf_counter()
print("⏳ Building Knowledge Graph from JSON...")
g = nx.DiGraph()
self._add_product_nodes(g)
self._add_attribute_nodes_and_edges(g)
compat_count = self._add_compatibility_edges(g)
type_counts: dict[str, int] = {}
for _, d in g.nodes(data=True):
t = d.get("node_type", "?")
type_counts[t] = type_counts.get(t, 0) + 1
elapsed = (time.perf_counter() - t0) * 1000
print(
f"βœ… Built in {elapsed:.0f}ms: "
f"{g.number_of_nodes()} nodes {type_counts} | "
f"{g.number_of_edges()} edges | "
f"COMPATIBLE_WITH={compat_count}"
)
return g
def _add_product_nodes(self, g: nx.DiGraph):
for pid, p in self.product_db.items():
attrs = {"node_type": "Product", "product_id": pid}
for k in FIELDS_BY_TYPE["Product"]:
v = p.get(k)
if v is not None:
attrs[k] = v
g.add_node(f"product:{pid}", **attrs)
def _add_attribute_nodes_and_edges(self, g: nx.DiGraph):
for pid, p in self.product_db.items():
pnode = f"product:{pid}"
# Series
series_raw = str(p.get("series", "")).strip()
snum = self._snum(series_raw) or self._snum(pid)
if snum:
snode = f"series:{snum}"
if not g.has_node(snode):
g.add_node(snode, node_type="Series",
name=series_raw or f"{snum} Series",
series_num=snum)
g.add_edge(pnode, snode, relation="BELONGS_TO")
# Material
mat = str(p.get("Material", "")).strip()
if mat and mat.lower() not in ("nan", "none", ""):
mnode = f"material:{mat}"
if not g.has_node(mnode):
mi = self.material_db.get(mat, {})
attrs = {"node_type": "Material", "material_code": mat}
for k in FIELDS_BY_TYPE["Material"]:
if mi.get(k) is not None:
attrs[k] = mi[k]
g.add_node(mnode, **attrs)
g.add_edge(pnode, mnode, relation="MADE_OF")
# ManufacturingProcess
proc = str(p.get("manufacturing_process", "")).strip()
if proc and proc.lower() not in ("nan", "none", ""):
prnode = f"process:{proc}"
if not g.has_node(prnode):
pi = self.process_db.get(proc, {})
attrs = {"node_type": "ManufacturingProcess", "process_name": proc}
for k in FIELDS_BY_TYPE["ManufacturingProcess"]:
if pi.get(k) is not None:
attrs[k] = pi[k]
g.add_node(prnode, **attrs)
g.add_edge(pnode, prnode, relation="USES_PROCESS")
def _add_compatibility_edges(self, g: nx.DiGraph) -> int:
"""COMPATIBLE_WITH (chain ↔ sprocket, bidirectional)"""
chains_by_series: dict[str, list[str]] = {}
for pid, p in self.product_db.items():
ptype = str(p.get("product_type", "")).lower()
if "sprocket" not in ptype and "sprocket" not in pid.lower():
snum = self._snum(str(p.get("series", ""))) or self._snum(pid)
if snum:
chains_by_series.setdefault(snum, []).append(pid)
count = 0
for pid, p in self.product_db.items():
ptype = str(p.get("product_type", "")).lower()
if "sprocket" not in ptype and "sprocket" not in pid.lower():
continue
spnode = f"product:{pid}"
compat: set[str] = set()
for s in self.SERIES_RE.findall(str(p.get("compatible_chain", ""))):
compat.add(s)
own = self._snum(str(p.get("series", ""))) or self._snum(pid)
if own:
compat.add(own)
for cs in compat:
for chain_pid in chains_by_series.get(cs, []):
cnode = f"product:{chain_pid}"
if not g.has_edge(spnode, cnode):
g.add_edge(spnode, cnode, relation="COMPATIBLE_WITH")
count += 1
if not g.has_edge(cnode, spnode):
g.add_edge(cnode, spnode, relation="COMPATIBLE_WITH")
count += 1
return count
def _snum(self, text: str) -> Optional[str]:
m = self.SERIES_RE.search(text or "")
return m.group(1) if m else None