import traceback import os os.environ["YOLO_VERBOSE"] = "False" import re import io as _io import json import base64 import tempfile from collections import defaultdict import cv2 import numpy as np import gradio as gr import networkx as nx from ultralytics import YOLO from PIL import Image import easyocr import pandas as pd from openai import OpenAI from skimage.morphology import skeletonize from scipy.spatial import KDTree model = YOLO("best.pt") # Lazy-initialize EasyOCR so startup isn't blocked _ocr_reader = None def get_ocr_reader(): global _ocr_reader if _ocr_reader is None: _ocr_reader = easyocr.Reader(['en'], gpu=False) return _ocr_reader DEFAULT_VALUES = { 'resistor': '1kΩ', 'capacitor': '100µF', 'inductor': '10mH', 'led': 'LED', 'diode': '0.7V', 'transistor': 'NPN', 'voltage source': '5V', 'battery': '9V', 'switch': 'SPST', 'switch open': 'OPEN', 'switch closed': 'CLOSED', 'ground': 'GND', 'gnd': 'GND', 'ammeter': '0A', 'voltmeter': '0V', 'galvanometer': '0A', } PREFIX_MAP = { 'resistor': 'R', 'capacitor': 'C', 'inductor': 'L', 'led': 'LED', 'diode': 'D', 'transistor': 'Q', 'voltage source': 'V', 'battery': 'V', 'switch': 'SW', 'switch open': 'SW', 'switch closed': 'SW', 'ground': 'GND', 'gnd': 'GND', 'op-amp': 'U', 'ammeter': 'A', 'voltmeter': 'VM', 'galvanometer': 'G', } SUPPRESS_CLASSES = {'arrow'} DEBUG_LOG = [] def debug_log(label, content): DEBUG_LOG.append(f"───── {label} ─────\n{content}\n") print(f"[{label}] {content}") def clear_debug_log(): DEBUG_LOG.clear() def get_debug_text(): return "\n".join(DEBUG_LOG) if DEBUG_LOG else "No output yet." # ═══════════════════════════════════════════════════════════════════════════ # LLM / VLM # ═══════════════════════════════════════════════════════════════════════════ def get_client(): key = os.environ.get("OPENROUTER_API_KEY") if not key: raise ValueError("OPENROUTER_API_KEY not set") return OpenAI(api_key=key, base_url="https://openrouter.ai/api/v1") def image_to_b64(image): if image.mode != 'RGB': image = image.convert('RGB') buf = _io.BytesIO() image.save(buf, format="JPEG", quality=85) return base64.standard_b64encode(buf.getvalue()).decode("utf-8") import time VLM_FALLBACK_MODELS = [ "google/gemma-3-27b-it:free", "nvidia/nemotron-nano-12b-v2-vl:free", "openrouter/auto", ] _HARD_SKIP = ("404", "not found", "unavailable", "no endpoints", "paid", "model_not_found", "invalid model") _DEAD_MODELS = set() def call_mistral(messages, label="MISTRAL"): last_err = None for attempt, slug in enumerate(VLM_FALLBACK_MODELS): if slug in _DEAD_MODELS: continue try: resp = get_client().chat.completions.create( model=slug, messages=messages, max_tokens=2000, temperature=0.1, timeout=30 ) if not resp.choices: last_err = "empty choices"; continue text = (resp.choices[0].message.content or "").strip() if not text: last_err = "empty text"; continue debug_log(label, f"[✓ {slug}] {text[:400]}") return text except Exception as e: err = f"{type(e).__name__}: {e}" last_err = err el = err.lower() if any(s in el for s in _HARD_SKIP): debug_log(f"{label} DEAD", f"{slug} → {err[:120]}") _DEAD_MODELS.add(slug) continue if "429" in err or "rate" in el: time.sleep(min(2 ** attempt, 16)) debug_log(f"{label} EXHAUSTED", f"All VLMs failed. Last: {last_err}") return "" # ═══════════════════════════════════════════════════════════════════════════ # STAGE 1 – YOLO DETECTION # ═══════════════════════════════════════════════════════════════════════════ def is_fp(cls, x1, y1, x2, y2): w, h = int(x2 - x1), int(y2 - y1) if h == 0 or w == 0: return True WIDE_OK = {'resistor', 'inductor', 'switch', 'switch open', 'switch closed', 'battery', 'voltage source', 'led', 'diode'} if cls.lower() not in WIDE_OK and w > 10 * h: return True if h > 10 * w and cls.lower() not in {'voltage source', 'battery', 'led'}: return True return False def box_iou(a, b): ax2, ay2 = a['x'] + a['w'], a['y'] + a['h'] bx2, by2 = b['x'] + b['w'], b['y'] + b['h'] ix1, iy1 = max(a['x'], b['x']), max(a['y'], b['y']) ix2, iy2 = min(ax2, bx2), min(ay2, by2) iw, ih = max(0, ix2 - ix1), max(0, iy2 - iy1) inter = iw * ih union = a['w'] * a['h'] + b['w'] * b['h'] - inter return inter / union if union > 0 else 0.0 def cluster_overlap(a, b): if a['component'].lower() != b['component'].lower(): return False if box_iou(a, b) > 0.25: return True acx, acy = a['x'] + a['w'] / 2, a['y'] + a['h'] / 2 if b['x'] <= acx <= b['x'] + b['w'] and b['y'] <= acy <= b['y'] + b['h']: return True bcx, bcy = b['x'] + b['w'] / 2, b['y'] + b['h'] / 2 if a['x'] <= bcx <= a['x'] + a['w'] and a['y'] <= bcy <= a['y'] + a['h']: return True return False def deduplicate_detections(details): n = len(details) parent = list(range(n)) def find(i): while parent[i] != i: parent[i] = parent[parent[i]] i = parent[i] return i def union(i, j): ri, rj = find(i), find(j) if ri != rj: parent[ri] = rj for i in range(n): for j in range(i + 1, n): if cluster_overlap(details[i], details[j]): union(i, j) clusters = {} for i in range(n): clusters.setdefault(find(i), []).append(details[i]) return [max(v, key=lambda d: d['confidence']) for v in clusters.values()] def freeze_labels(details): non_gnd = [d for d in details if 'gnd' not in d['component'].lower() and 'ground' not in d['component'].lower()] gnd = [d for d in details if 'gnd' in d['component'].lower() or 'ground' in d['component'].lower()] # FIX: sort by (prefix, x, y) -- x first -- so left-to-right reading order # matches schematic convention and avoids label/value swaps between # vertically-adjacent components in different columns (e.g. R2 vs R3). ds = sorted(non_gnd, key=lambda d: (d['component'].lower(), d['x'], d['y'])) ds += sorted(gnd, key=lambda d: (d['y'], d['x'])) counters = {} for d in ds: prefix = PREFIX_MAP.get(d['component'].lower(), d['component'][0].upper()) counters[prefix] = counters.get(prefix, 0) + 1 d['label'] = f"{prefix}{counters[prefix]}" if not d.get('schematic_label'): d['schematic_label'] = d['label'] return ds def detect_components(image, confidence): img_np = np.array(image) all_details = [] passes = [ (confidence, 0.55, "p1"), (confidence, 0.30, "p2"), (max(0.01, confidence * 0.40), 0.45, "p3"), ] for conf_t, iou_t, tag in passes: results = model.predict(source=image, conf=conf_t, iou=iou_t, imgsz=1280, augment=True, agnostic_nms=True, verbose=False) nb = len(all_details) for box in results[0].boxes: cls = results[0].names[int(box.cls[0])] cv_ = float(box.conf[0]) x1, y1, x2, y2 = map(int, box.xyxy[0]) if cls.lower() in SUPPRESS_CLASSES: continue if is_fp(cls, x1, y1, x2, y2): continue all_details.append({ "component": cls, "value": None, "confidence": round(cv_ * 100, 1), "x": x1, "y": y1, "w": x2 - x1, "h": y2 - y1, "is_default": False, "source": "", "label": "", "schematic_label": "", }) debug_log("YOLO", f"{tag} conf={conf_t:.3f} iou={iou_t} -> +{len(all_details) - nb}") all_details = deduplicate_detections(all_details) all_details = freeze_labels(all_details) debug_log("LABELS", f"{len(all_details)}: {[d['label'] for d in all_details]}") results_plot = model.predict(source=image, conf=confidence, iou=0.55, imgsz=1280, augment=True, agnostic_nms=True, verbose=False) return img_np, results_plot, all_details # ═══════════════════════════════════════════════════════════════════════════ # STAGE 2 – VALUE / LABEL EXTRACTION # ═══════════════════════════════════════════════════════════════════════════ def fix_ocr_units(v, label): if v is None: return v v = re.sub(r'\s+', ' ', str(v).strip()) if label.startswith('C'): v = re.sub(r'(\d+\.?\d*)\s*[pP][fF]?\b', r'\1µF', v, flags=re.IGNORECASE) v = re.sub(r'(\d+\.?\d*)\s*µ\s*[fF]', r'\1µF', v) if label.startswith('L'): v = re.sub(r'(\d+\.?\d*)\s*m\s*[Hh]\b', r'\1mH', v) v = re.sub(r'(\d+\.?\d*)\s*[pP][Hh]\b', r'\1µH', v) v = re.sub(r'(\d+\.?\d*)\s*µ\s*[Hh]\b', r'\1µH', v) if label.startswith('R'): v = re.sub(r'(\d+\.?\d*)\s*k\b(?!Ω)', r'\1kΩ', v) v = re.sub(r'(\d+\.?\d)\s*K\s*[Ωo]', r'\1kΩ', v) return v def extract_values_near_box(ocr_results, box, label=None): x1, y1, w, h = box x2, y2 = x1 + w, y1 + h cx, cy = (x1 + x2) / 2, (y1 + y2) / 2 lbl_pat = re.compile(r'^[A-Za-z]{1,4}\d*$') val_hint = re.compile(r'\d+\.?\d*\s*([kKMmµunpΩFHVA]|ohm)', re.IGNORECASE) def in_zone(tx, ty): return ((x1 - 60 < tx < x2 + 60 and y1 - 30 < ty < y2 + 130) or (x1 - 120 < tx < x2 + 120 and y1 - 60 < ty < y2 + 60)) cands = [] for (coords, text, *_rest) in ocr_results: tx = int(np.mean([c[0] for c in coords])) ty = int(np.mean([c[1] for c in coords])) t = text.strip() if in_zone(tx, ty) and val_hint.search(t) and not lbl_pat.match(t): cands.append((t, tx, ty, ((tx - cx) ** 2 + (ty - cy) ** 2) ** 0.5)) cands.sort(key=lambda x: x[3]) return [(t, tx, ty) for t, tx, ty, _ in cands] def parse_value(items): full = re.compile(r'(\d+\.?\d*)\s*([kKMmµunp])?\s*(Ω|ohm|F|H|V|A|W)\b', re.IGNORECASE) pfx = re.compile(r'(\d+\.?\d*)\s*([kKMmµunp])\b') nohz = re.compile(r'\d+(\.\d+)?\s*Hz', re.IGNORECASE) bare_ohm = re.compile(r'(\d+\.?\d)\s*Ω') pm = {'k': 'k', 'K': 'k', 'M': 'M', 'm': 'm', 'u': 'µ', 'µ': 'µ', 'n': 'n', 'p': 'p'} texts = [t for t, _, _ in items] joined = texts + [' '.join(texts[i:i + 2]) for i in range(len(texts) - 1)] for text in joined: if nohz.search(text): continue m = full.search(text) if m: return f"{m.group(1)}{pm.get(m.group(2) or '', m.group(2) or '')}{m.group(3)}" m2 = pfx.search(text) if m2: return f"{m2.group(1)}{pm.get(m2.group(2), m2.group(2))}" m3 = bare_ohm.search(text) if m3: return f"{m3.group(1)}Ω" return None def preprocess_for_ocr(img_np): """FIX (OCR improvement): schematic label/value text is often small, thin-stroke, and low-contrast against a white background, which EasyOCR's default detector tends to miss or mis-read at native resolution. This upscales the image and boosts local contrast (CLAHE) before OCR, which substantially improves recall on small printed text like '4.7µF' or '10kΩ' without touching detection/wire/graph logic elsewhere in the pipeline.""" gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8)) enhanced = clahe.apply(gray) enhanced_bgr = cv2.cvtColor(enhanced, cv2.COLOR_GRAY2RGB) h, w = enhanced_bgr.shape[:2] scale_factor = 2.0 if max(h, w) < 1600 else 1.5 upscaled = cv2.resize(enhanced_bgr, (int(w * scale_factor), int(h * scale_factor)), interpolation=cv2.INTER_CUBIC) return upscaled, scale_factor def draw_labels_for_llm(img_np, details): out = img_np.copy() for d in details: x1, y1, x2, y2 = d['x'], d['y'], d['x'] + d['w'], d['y'] + d['h'] cv2.rectangle(out, (x1, y1), (x2, y2), (0, 100, 255), 2) txt = d['label'] (tw, th), _ = cv2.getTextSize(txt, cv2.FONT_HERSHEY_SIMPLEX, 0.65, 2) ty = max(th + 6, y1 - 6) cv2.rectangle(out, (x1, ty - th - 4), (x1 + tw + 8, ty + 2), (0, 100, 255), -1) cv2.putText(out, txt, (x1 + 3, ty - 2), cv2.FONT_HERSHEY_SIMPLEX, 0.65, (255, 255, 255), 2, cv2.LINE_AA) return Image.fromarray(out) def get_all_values_with_llm(labeled_pil, details): comp_list = "\n".join([f"{d['label']} ({d['component']}) at ({d['x']+d['w']//2},{d['y']+d['h']//2})" for d in details]) prompt = ( "You are an expert electronics engineer reading a circuit schematic image.\n" "Blue boxes mark each detected component with a TEMPORARY ID.\n\n" f"Detected components:\n{comp_list}\n\n" "For EACH component:\n" "1. Find the SCHEMATIC LABEL printed near it (e.g. R1, C2, L3, S1, S2, LED1)\n" "2. Find the VALUE printed near it\n\n" "STRICT UNIT RULES:\n" "- Resistors R: Ω ONLY — 270Ω, 1kΩ, 2.2kΩ, 3.3kΩ, 4.7kΩ, 10kΩ\n" "- Capacitors C: Farads ONLY — 10µF, 4.7µF, 2.2µF, 1µF\n" "- Inductors L: Henrys ONLY — 10mH, 22mH, 33mH, 47mH\n" "- Voltage/Battery V: Volts — 12V, 5V\n" "- Switches SW: open OR closed\n" "- LEDs: LED\n\n" "Output ONLY this exact format, one line per component:\n" "TEMP_ID | SCHEMATIC_LABEL | VALUE\n\n" "If label unknown repeat TEMP_ID. No skips. No extra text." ) messages = [{ "role": "user", "content": [ {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_to_b64(labeled_pil)}"}}, {"type": "text", "text": prompt}, ], }] text = call_mistral(messages, "LLM-VALUES") value_map, label_map = {}, {} for line in text.split("\n"): line = line.strip() if not line or "|" not in line: continue parts = [p.strip() for p in line.split("|")] if len(parts) < 3: continue det_lbl = parts[0].strip("*_• ").split()[-1] if parts[0].strip() else "" sch_lbl = parts[1].strip("*_• ") val = re.split(r'\s{2,}|\(', parts[2].strip())[0].strip() if not det_lbl: continue if sch_lbl and sch_lbl.lower() not in ("unknown", "", "n/a", "not visible"): label_map[det_lbl] = sch_lbl if val and val.lower() not in ("unknown", "", "n/a", "not visible"): value_map[det_lbl] = val debug_log("VALUES", str(value_map) if value_map else "EMPTY") debug_log("LABELS", str(label_map) if label_map else "EMPTY") return value_map, label_map def prune_invalid_units(details): for d in details: lbl = d['label'] val = str(d.get('value', '')).lower() bad = False if lbl.startswith('R') and re.search(r'\d+\s*(v\b|µf|nf|pf)', val, re.IGNORECASE): bad = True if lbl.startswith('R') and re.search(r'\d+\s*[kKmM]?h\b', val) and 'ohm' not in val: bad = True if lbl.startswith('C') and re.search(r'\d+\s*(mh|µh|\bh\b|ω|ohm)\b', val, re.IGNORECASE): bad = True if lbl.startswith('L') and re.search(r'\d+\s*(µf|nf|pf|\bf\b|ω|ohm)\b', val, re.IGNORECASE): bad = True if bad: d['value'] = DEFAULT_VALUES.get(d['component'].lower(), "unknown") d['is_default'] = True d['source'] = "default(pruned)" return details def fill_values(img_np, details, ocr_results): labeled_pil = draw_labels_for_llm(img_np, details) lv, label_map = get_all_values_with_llm(labeled_pil, details) for d in details: det_lbl = d['label'] if det_lbl in label_map: d['schematic_label'] = label_map[det_lbl] unified_val = {} for yolo_lbl, val in lv.items(): unified_val[yolo_lbl] = val sch = label_map.get(yolo_lbl) if sch: unified_val[sch] = val debug_log("UNIFIED_VAL", str(unified_val)) for d in details: nearby = extract_values_near_box(ocr_results, (d['x'], d['y'], d['w'], d['h']), d['label']) ocr_val = parse_value(nearby) if ocr_val: d['value'] = fix_ocr_units(ocr_val, d['label']) d['is_default'] = False d['source'] = "OCR" for d in details: det_lbl = d['label'] sch = d.get('schematic_label', det_lbl) v = unified_val.get(sch) or unified_val.get(det_lbl) if v: if d['value'] is None or d.get('source') == 'default': d['value'] = fix_ocr_units(v, det_lbl) d['is_default'] = False d['source'] = "LLM" elif d['value'] is None: d['value'] = DEFAULT_VALUES.get(d['component'].lower(), "unknown") d['is_default'] = True d['source'] = "default" details = prune_invalid_units(details) for d in details: if d.get('source') in ('default', 'default(pruned)'): sch = d.get('schematic_label', d['label']) v = unified_val.get(sch) or unified_val.get(d['label']) if v: fixed = fix_ocr_units(v, d['label']) test = prune_invalid_units([{**d, 'value': fixed, 'source': 'LLM'}]) if test[0].get('source') != 'default(pruned)': d['value'] = fixed d['is_default'] = False d['source'] = "LLM" return details, labeled_pil # ═══════════════════════════════════════════════════════════════════════════ # STAGE 3 – WIRE EXTRACTION & CONNECTIVITY # Skeletonize -> junctions -> break at component boxes -> CC -> # union-find merge -> KDTree reconnect -> graph # ═══════════════════════════════════════════════════════════════════════════ def _img_scale(img_np): h, w = img_np.shape[:2] diag = (h * h + w * w) ** 0.5 return diag / 800.0 def extract_wire_layer(img_np): """FIX (mega-net root cause): the previous version OR'd together Otsu threshold, adaptive threshold, AND dilated Canny edges, then ran a morphological CLOSE on top -- four separate fattening operations stacked on each other. On a dense bottom rail this was enough to weld multiple genuinely-separate wire runs into a single thick blob region before skeletonize/connectedComponents ever got a chance to see them as distinct nets (the "mega-net" NET_58 problem). Fix: drop the edge dilation, drop the adaptive-threshold layer (the most prone to flooding dark regions into wide blobs), and drop the CLOSE morph op entirely -- only keep a light OPEN pass for despeckling.""" gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) _, wire_otsu = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU) blurred = cv2.GaussianBlur(gray, (3, 3), 0) edges = cv2.Canny(blurred, 50, 150) # NOTE: no dilation of edges -- dilating before the OR was the single # biggest contributor to bridging parallel rails into one blob. wire = cv2.bitwise_or(wire_otsu, edges) # NOTE: wire_adapt (adaptive threshold) dropped from the mask -- it # tended to flood thicker dark regions (bottom rails) into wide blobs # rather than thin lines. If thin/faint wires start being missed, # reintroduce it but OR'd in AFTER skeletonization candidate testing, # not before. # NOTE: morphological CLOSE removed -- CLOSE actively closes gaps # between nearby-but-distinct wire runs, which is the opposite of # what's needed here. Only despeckle with OPEN. k_open = cv2.getStructuringElement(cv2.MORPH_RECT, (2, 2)) wire = cv2.morphologyEx(wire, cv2.MORPH_OPEN, k_open) debug_log("WIRE", f"nonzero={int(np.count_nonzero(wire))}") return wire def erase_component_bodies(wire_layer, details): """Erase interior of each component, leaving terminal stubs. FIX: ground symbols are now skipped from erasure entirely AND explicitly kept un-dilated/un-broken downstream so the thin GND lead is never severed from the bus before connected components runs.""" out = wire_layer.copy() for d in details: x1, y1 = d['x'], d['y'] x2, y2 = x1 + d['w'], y1 + d['h'] ctype = d.get('component', '').lower() if 'ground' in ctype or 'gnd' in ctype: continue # never erase ground symbol pixels -- keep its lead intact pad_x = max(4, d['w'] // 10) pad_y = max(4, d['h'] // 10) if 'voltage' in ctype or 'battery' in ctype: pad_x = max(3, d['w'] // 12) pad_y = max(3, d['h'] // 12) ix1, iy1 = x1 + pad_x, y1 + pad_y ix2, iy2 = x2 - pad_x, y2 - pad_y if ix2 > ix1 and iy2 > iy1: out[iy1:iy2, ix1:ix2] = 0 return out def skeletonize_wire(wire_clean): """FIX: light erosion before skeletonize to break thin bridges between adjacent-but-distinct rails that may still have survived mask construction. skeletonize alone can't separate a true blob -- it only traces its centerline -- so erosion needs to happen first to actually sever weak connecting bridges. FIX 2: confirmed via RAW-NET debug logging that a single erosion pass was insufficient -- one net was still forming as a single 3968px blob with a 1343px bounding-box diagonal (vs ~250-700px for every other net) BEFORE merge_split_nets even runs, meaning the blob forms at this mask/skeleton stage, not from later net-merging. Bumped to 2 erosion iterations to more reliably sever the bridge. If this starts fragmenting legitimately-thin wires elsewhere, dial back to 1.""" wire_bin = (wire_clean > 0).astype(np.uint8) k_erode = cv2.getStructuringElement(cv2.MORPH_RECT, (2, 2)) wire_bin = cv2.erode(wire_bin, k_erode, iterations=2) skel = skeletonize(wire_bin.astype(bool)).astype(np.uint8) return skel def break_skeleton_at_component_boxes(skel, details, margin=1): """Cut skeleton inside each component's interior box. FIX: ground symbols are excluded here too, so their lead is never cut.""" out = skel.copy() for d in details: ctype = d.get('component', '').lower() if 'ground' in ctype or 'gnd' in ctype: continue x1 = max(0, d['x'] + margin) y1 = max(0, d['y'] + margin) x2 = min(out.shape[1], d['x'] + d['w'] - margin) y2 = min(out.shape[0], d['y'] + d['h'] - margin) if x2 > x1 and y2 > y1: out[y1:y2, x1:x2] = 0 return out def compute_nets(skel): num_cc, cc_map = cv2.connectedComponents(skel, connectivity=8) return num_cc, cc_map def build_net_pixel_index(cc_map, valid_nets): net_pixels = {} net_endpoints = {} for nid in valid_nets: ys, xs = np.where(cc_map == nid) pts = np.column_stack((xs, ys)) net_pixels[nid] = pts if len(pts): mean = pts.mean(axis=0) centered = pts - mean cov = np.cov(centered.T) if len(pts) > 1 else np.eye(2) try: eigvals, eigvecs = np.linalg.eigh(cov) axis = eigvecs[:, np.argmax(eigvals)] except np.linalg.LinAlgError: axis = np.array([1.0, 0.0]) proj = centered @ axis net_endpoints[nid] = (pts[np.argmin(proj)], pts[np.argmax(proj)]) return net_pixels, net_endpoints class UnionFind: def __init__(self, items): self.parent = {x: x for x in items} def find(self, x): while self.parent[x] != x: self.parent[x] = self.parent[self.parent[x]] x = self.parent[x] return x def union(self, a, b): ra, rb = self.find(a), self.find(b) if ra != rb: self.parent[ra] = rb def merge_split_nets(valid_nets, net_endpoints, gap_threshold, net_pixels=None, max_span_factor=3.0): """Union-find merge of nets split by erasure/OCR gaps. FIX (chaining bug): the previous version unioned any pair under gap_threshold with no bound on the resulting group's size. That let a bottom rail get fused transitively -- net A close to B, B close to C, so A and C end up in the same group even though A and C are far apart and were never actually close to each other. This produced the mega-net swallowing C1-C4/GND/L4/R3/V1/V2. Fix: before unioning a pair (a, b), compute what the merged group's pixel bounding-box diagonal WOULD become (group-of-a U group-of-b), and reject the union if that diagonal exceeds max_span_factor times the larger of the two individual original net diagonals. Pairs are also processed closest-distance-first so genuine local merges aren't blocked by farther candidates being considered first.""" uf = UnionFind(valid_nets) orig_diag = {} group_bbox = {} if net_pixels: for nid, pts in net_pixels.items(): if len(pts): x0, y0 = pts[:, 0].min(), pts[:, 1].min() x1, y1 = pts[:, 0].max(), pts[:, 1].max() orig_diag[nid] = float(((x1 - x0) ** 2 + (y1 - y0) ** 2) ** 0.5) group_bbox[nid] = [x0, y0, x1, y1] else: orig_diag[nid] = 0.0 group_bbox[nid] = [0, 0, 0, 0] def bbox_diag(bb): return ((bb[2] - bb[0]) ** 2 + (bb[3] - bb[1]) ** 2) ** 0.5 def union_bbox(b1, b2): return [min(b1[0], b2[0]), min(b1[1], b2[1]), max(b1[2], b2[2]), max(b1[3], b2[3])] ids = list(valid_nets) pairs = [] for i in range(len(ids)): for j in range(i + 1, len(ids)): a, b = ids[i], ids[j] if a not in net_endpoints or b not in net_endpoints: continue best = min( np.linalg.norm(pa.astype(float) - pb.astype(float)) for pa in net_endpoints[a] for pb in net_endpoints[b] ) if best < gap_threshold: pairs.append((best, a, b)) pairs.sort(key=lambda t: t[0]) rejected = 0 for best, a, b in pairs: ra, rb = uf.find(a), uf.find(b) if ra == rb: continue if net_pixels: cap = max_span_factor * max(orig_diag.get(a, 0.0), orig_diag.get(b, 0.0), 1.0) would_be = union_bbox(group_bbox[ra], group_bbox[rb]) if bbox_diag(would_be) > cap: rejected += 1 continue uf.union(ra, rb) group_bbox[uf.find(ra)] = would_be else: uf.union(a, b) if rejected: debug_log("MERGE-CAP", f"rejected {rejected} merge(s) exceeding span cap") merged = {} for nid in valid_nets: root = uf.find(nid) merged.setdefault(root, []).append(nid) return merged, uf def remap_cc_map(cc_map, merged_groups): remap = {} for root, members in merged_groups.items(): new_id = min(members) for m in members: remap[m] = new_id out = cc_map.copy() for old_id, new_id in remap.items(): if old_id != new_id: out[cc_map == old_id] = new_id new_valid = set(remap.values()) return out, new_valid def build_kdtree(cc_map, valid_nets): points, labels = [], [] for nid in valid_nets: ys, xs = np.where(cc_map == nid) for x, y in zip(xs, ys): points.append((x, y)) labels.append(nid) if not points: return None, [] return KDTree(points), labels def terminal_nets_via_kdtree(tree, labels, terminal_xy, max_dist): """FIX: previously used query_ball_point, which returns EVERY net pixel within max_dist -- so a single physical terminal point near multiple unrelated wire fragments would pick up ALL of them as "connected" nets. This was confirmed in debug logs: a 2-terminal resistor (R1) was reporting nets=[37, 58, 81] -- three nets from one component, when a resistor terminal can only be touching one real wire. That spurious extra net was bridging components that aren't actually wired together (e.g. forcing L1/R1/R2 onto a shared net). Fix: query only the SINGLE nearest net pixel within max_dist, so each terminal point contributes at most one net -- matching how a real pin only touches the one wire it's soldered to.""" if tree is None: return set() dist, idx = tree.query(terminal_xy, k=1) if dist <= max_dist: return {labels[idx]} return set() def component_terminal_points(d): x1, y1 = d['x'], d['y'] x2, y2 = x1 + d['w'], y1 + d['h'] cx, cy = (x1 + x2) // 2, (y1 + y2) // 2 horizontal = d['w'] >= d['h'] if horizontal: return [(x1, cy), (x2, cy), (cx, y1), (cx, y2)] return [(cx, y1), (cx, y2), (x1, cy), (x2, cy)] def build_graph_wire_cc(details, img_np): """Full Stage 3 pipeline. FIX: merge gap and KDTree query radius are scale-relative and tightened; GND lead is preserved end-to-end; merge_split_nets now receives net_pixels so it can cap merge span (see merge_split_nets docstring); GND kdtree radius tightened from 18*scale to 10*scale since the looser radius was eagerly grabbing onto unrelated rail segments, compounding the mega-net problem.""" scale = _img_scale(img_np) merge_gap = max(3, int(4 * scale)) kdtree_radius = max(4, int(5 * scale)) gnd_kdtree_radius = max(8, int(10 * scale)) # was 18*scale -- too eager to grab onto unrelated rail segments wire_layer = extract_wire_layer(img_np) wire_clean = erase_component_bodies(wire_layer, details) skel = skeletonize_wire(wire_clean) skel = break_skeleton_at_component_boxes(skel, details) num_cc, cc_map = compute_nets(skel) # DEBUG: raw (pre-merge) net sizes/bboxes -- helps confirm whether a # mega-net originates here (mask/skeleton stage) or in merge_split_nets. for nid in range(1, num_cc): ys, xs = np.where(cc_map == nid) if len(xs) > 200: raw_diag = ((xs.max() - xs.min()) ** 2 + (ys.max() - ys.min()) ** 2) ** 0.5 debug_log("RAW-NET", f"id={nid} px={len(xs)} bbox_diag={raw_diag:.0f}") MIN_NET_PX = max(8, int(15 * (scale ** 0.5))) net_sizes = {nid: int(np.sum(cc_map == nid)) for nid in range(1, num_cc)} valid_nets = {nid for nid, sz in net_sizes.items() if sz >= MIN_NET_PX} debug_log("VALID NETS", f"{len(valid_nets)} >= {MIN_NET_PX}px (scale={scale:.2f})") net_pixels, net_endpoints = build_net_pixel_index(cc_map, valid_nets) merged_groups, _ = merge_split_nets(valid_nets, net_endpoints, gap_threshold=merge_gap, net_pixels=net_pixels) cc_map, valid_nets = remap_cc_map(cc_map, merged_groups) net_pixels, net_endpoints = build_net_pixel_index(cc_map, valid_nets) debug_log("MERGE", f"merged into {len(valid_nets)} nets (gap={merge_gap}px)") tree, labels = build_kdtree(cc_map, valid_nets) comp_nets = {} for d in details: ctype = d.get('component', '').lower() is_gnd = 'ground' in ctype or 'gnd' in ctype radius = gnd_kdtree_radius if is_gnd else kdtree_radius nets_found = set() for tx, ty in component_terminal_points(d): nets_found |= terminal_nets_via_kdtree(tree, labels, (tx, ty), max_dist=radius) # GND: also try its own centroid in case the lead is short/centered if is_gnd and not nets_found: cx, cy = d['x'] + d['w'] // 2, d['y'] + d['h'] // 2 nets_found |= terminal_nets_via_kdtree(tree, labels, (cx, cy), max_dist=gnd_kdtree_radius) comp_nets[d['label']] = nets_found debug_log("TERMINAL", f"{d['label']}({d['component']}) -> nets={sorted(nets_found)} r={radius}") G = nx.Graph() for d in details: G.add_node(d['label'], node_type='component', **d) for nid in valid_nets: G.add_node(f"NET_{nid}", node_type='net', net_id=nid) for comp, nets in comp_nets.items(): for nid in nets: G.add_edge(comp, f"NET_{nid}", edge_type='pin') orphans = [ n for n in list(G.nodes()) if G.nodes[n].get('node_type') == 'net' and not any(G.nodes[nb].get('node_type') == 'component' for nb in G.neighbors(n)) ] G.remove_nodes_from(orphans) debug_log("GRAPH", f"{G.number_of_nodes()} nodes {G.number_of_edges()} edges") return G, skel, cc_map, num_cc, valid_nets, net_pixels, net_endpoints, scale # ── GND DETECTION ───────────────────────────────────────────────────────── def detect_gnd_symbols(img_np, details): if any('gnd' in d['component'].lower() or 'ground' in d['component'].lower() for d in details): return details gray = cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY) _, bw = cv2.threshold(gray, 128, 255, cv2.THRESH_BINARY_INV) h_img, w_img = img_np.shape[:2] contours, _ = cv2.findContours(bw, cv2.RETR_LIST, cv2.CHAIN_APPROX_SIMPLE) cands = [] for cnt in contours: x, y, w, h = cv2.boundingRect(cnt) area = cv2.contourArea(cnt) if not (30 < area < 12000): continue if not (y > h_img * 0.2): continue aspect = w / max(h, 1) if not (0.8 < aspect < 10): continue if h > 80: continue roi = bw[y:y + h, x:x + w] if np.sum(np.sum(roi > 0, axis=1) > w * 0.3) < 2: continue ccx, ccy = x + w // 2, y + h // 2 if any(d['x'] <= ccx <= d['x'] + d['w'] and d['y'] <= ccy <= d['y'] + d['h'] for d in details): continue cands.append({ "component": "ground", "value": "GND", "confidence": 50.0, "x": x, "y": y, "w": w, "h": h, "is_default": False, "source": "shape", "label": "", "schematic_label": "GND", "_area": area, }) if cands: cands.sort(key=lambda d: -(d['y'] + d['h'] // 2)) best = cands[:3] for i, g in enumerate(best, 1): g.pop('_area', None) g['label'] = f"GND{i}" g['schematic_label'] = "GND" details = details + best debug_log("GND", f"Injected {len(best)}: {[g['label'] for g in best]}") return details def get_gnd_nodes(G): return [n for n, d in G.nodes(data=True) if d.get('node_type', 'component') == 'component' and ('gnd' in d.get('component', '').lower() or 'ground' in d.get('component', '').lower())] def prune_orphan_nets(G, label="ORPHAN-SWEEP"): """FIX: net nodes can lose their last component neighbor after GND-merge, proximity-repair, or VLM edges get added/rewired -- e.g. a net that used to anchor one GND gets that GND's edge redirected elsewhere, leaving the net node behind with zero component neighbors. build_graph_wire_cc only sweeps orphans ONCE, before any of those later stages run, so such leftovers (e.g. a stray "NET_121" with no real connections) silently persist and show up as a fake extra "isolated section" in the final report/netlist. Call this after any stage that can change net<->component edges.""" orphans = [ n for n in list(G.nodes()) if G.nodes[n].get('node_type') == 'net' and not any(G.nodes[nb].get('node_type') == 'component' for nb in G.neighbors(n)) ] if orphans: G.remove_nodes_from(orphans) debug_log(label, f"removed {len(orphans)} orphan net(s): {orphans}") return G def wire_gnd_to_nearest_net(G, details, cc_map, valid_nets, scale): """FIX (isolated-ground-cluster bug): the old guard only fired when a GND node had degree==0. But GND symbols clustered near each other can each find a real *local* net via the initial terminal-kdtree pass (e.g. GND1->NET_123, GND2->NET_122, GND3->NET_121/122) -- so they have degree>0 and this function silently did nothing, even though that local net is itself a tiny isolated fragment, disconnected from the main circuit net (e.g. NET_75). repair_isolated_nodes then just kept re-confirming the same already-attached tiny net every pass (visible as repeated identical REPAIR-DONE lines), since "nearest net" isn't the same question as "nearest net that's part of the main circuit". Fix: identify the MAIN net cluster (the net belonging to the largest component-connected-component in G, e.g. NET_75) and explicitly check, for every GND, whether it is already wired (via any path) to that main cluster. If not, snap directly to the nearest pixel belonging to nets that ARE part of the main cluster, regardless of the GND's current degree.""" h, w = cc_map.shape snap_dist = int(0.25 * (h * h + w * w) ** 0.5) gnd_details = [ d for d in details if ('gnd' in d.get('component', '').lower() or 'ground' in d.get('component', '').lower()) and d['label'] in G ] if not gnd_details: return G # Find the main component cluster (largest set of component nodes in # one connected piece of G) and the net node IDs that belong to it. net_node_ids = {n for n, dd in G.nodes(data=True) if dd.get('node_type') == 'net'} comp_node_ids = {n for n in G.nodes() if n not in net_node_ids} best_cc, best_comp_count = None, -1 for cc_set in nx.connected_components(G): comp_count = len([n for n in cc_set if n in comp_node_ids]) if comp_count > best_comp_count: best_comp_count = comp_count best_cc = cc_set main_net_ids = set() if best_cc: for n in best_cc: if n in net_node_ids: nid = G.nodes[n].get('net_id') if isinstance(nid, int): main_net_ids.add(nid) for d in gnd_details: lbl = d['label'] already_main = best_cc is not None and lbl in best_cc if already_main: continue # already wired into the main circuit, nothing to do cx = d['x'] + d['w'] // 2 ty = d['y'] best_nid, best_dist = None, float('inf') search_space = main_net_ids if main_net_ids else valid_nets for nid in search_space: nys, nxs = np.where(cc_map == nid) if not len(nxs): continue dm = float(np.min((nxs - cx) ** 2 + (nys - ty) ** 2)) if dm < best_dist: best_dist = dm best_nid = nid if best_nid is not None and best_dist < snap_dist ** 2: nn = f"NET_{best_nid}" if not G.has_node(nn): G.add_node(nn, node_type='net', net_id=best_nid) G.add_edge(lbl, nn, edge_type='gnd-snap') debug_log("GND-SNAP", f"{lbl}->NET_{best_nid} ({best_dist**0.5:.0f}px) [main-net]") else: debug_log("GND-SNAP", f"{lbl} no main-circuit net within {snap_dist}px (best={best_dist**0.5:.0f}px)") # Union all ground nodes that ended up wired (directly or via the # original terminal pass) onto a shared GND rail, as before. snapped_nets = set() for d in gnd_details: if d['label'] in G: for nb in G.neighbors(d['label']): if G.nodes[nb].get('node_type') == 'net': snapped_nets.add(nb) if len(snapped_nets) > 1: # FIX: canonical net used to be picked via sorted(snapped_nets)[0], # which sorts net NAMES as plain strings -- "NET_121" sorts before # "NET_75" lexicographically ('1' < '7' as a character), even # though NET_75 is the real net with all the actual component # connections and NET_121 is a tiny orphan fragment. That could # make the merge pick the wrong net as canonical and effectively # rewire the real circuit's net identity onto a near-empty one. # Fix: pick whichever net currently has the most component # neighbors as canonical (ties broken by name for determinism). def _comp_neighbor_count(net): return sum(1 for nb in G.neighbors(net) if G.nodes[nb].get('node_type') != 'net') canonical = max(snapped_nets, key=lambda n: (_comp_neighbor_count(n), n)) for other in snapped_nets: if other == canonical: continue for nb in list(G.neighbors(other)): if nb != canonical: G.add_edge(nb, canonical, edge_type='gnd-merge') G.remove_node(other) debug_log("GND-MERGE", f"merged {snapped_nets} -> {canonical}") return G # ── ISOLATION REPAIR ──────────────────────────────────────────────────────── def repair_isolated_nodes(G, details, img_np, cc_map, valid_nets): h, w = img_np.shape[:2] max_dist = int(0.10 * (h * h + w * w) ** 0.5) debug_log("REPAIR", f"max snap dist={max_dist}px") net_nodes = {n for n, d in G.nodes(data=True) if d.get('node_type') == 'net'} comp_nodes = {n for n in G.nodes() if n not in net_nodes} net_px_cache = {} for nid in valid_nets: ys, xs = np.where(cc_map == nid) if len(xs): net_px_cache[nid] = (xs, ys) def _main_comps(): best = set() for cc_set in nx.connected_components(G): cm = {n for n in cc_set if n in comp_nodes} if len(cm) > len(best): best = cm return best repaired = [] skipped = [] details_map = {d2['label']: d2 for d2 in details} for _ in range(4): main = _main_comps() changed = False for d in details: lbl = d['label'] if lbl not in G or lbl in main: continue cx = d['x'] + d['w'] // 2 cy = d['y'] + d['h'] // 2 best_nid, best_dist = None, float('inf') for nid, (xs, ys) in net_px_cache.items(): dists = (xs - cx) ** 2 + (ys - cy) ** 2 d_min = float(np.min(dists)) if d_min < best_dist: best_dist = d_min best_nid = nid if best_nid is not None and best_dist <= max_dist ** 2: nn = f"NET_{best_nid}" if not G.has_node(nn): G.add_node(nn, node_type='net', net_id=best_nid) G.add_edge(lbl, nn, edge_type='proximity-repair') repaired.append(f"{lbl}->NET_{best_nid} ({best_dist**0.5:.0f}px)") main.add(lbl) changed = True else: best_clbl, best_cdist = None, float('inf') for other in main: od = details_map.get(other) if od is None: continue dist = ((cx - od['x'] - od['w'] // 2) ** 2 + (cy - od['y'] - od['h'] // 2) ** 2) ** 0.5 if dist < best_cdist: best_cdist = dist best_clbl = other if best_clbl and best_cdist <= max_dist: nid_str = f"NET_REPAIR_{lbl}" G.add_node(nid_str, node_type='net', net_id=nid_str) G.add_edge(lbl, nid_str, edge_type='proximity-repair') G.add_edge(best_clbl, nid_str, edge_type='proximity-repair') repaired.append(f"{lbl}->{best_clbl} [centroid-fallback] ({best_cdist:.0f}px)") main.add(lbl) changed = True else: skipped.append(f"{lbl} min_wire={best_dist**0.5:.0f}px>{max_dist}px") if not changed: break if repaired: debug_log("REPAIR-DONE", "\n".join(repaired)) if skipped: debug_log("REPAIR-SKIP", "\n".join(skipped)) return G # ═══════════════════════════════════════════════════════════════════════════ # STAGE 4 – VLM TOPOLOGY VERIFICATION # ═══════════════════════════════════════════════════════════════════════════ def verify_connections_with_llm(image_pil, details, G): debug_log("VLM", "Starting topology verification...") img_np = np.array(image_pil) if hasattr(image_pil, 'convert') else image_pil _, buf = cv2.imencode(".png", img_np) b64 = "data:image/png;base64," + base64.b64encode(buf.tobytes()).decode() comp_list = "\n".join([f" {d.get('schematic_label', d['label'])} ({d['component']})" for d in details]) messages = [{ "role": "user", "content": [ {"type": "text", "text": ( "Analyze this circuit schematic carefully.\n\n" f"Known components:\n{comp_list}\n\n" "List every direct wire connection between two components.\n" 'Return ONLY valid JSON: {"connections": [["A","B"],...]}\n\n' "Rules: use exact labels above; only directly wired pairs; no extra text." )}, {"type": "image_url", "image_url": {"url": b64}}, ], }] text = call_mistral(messages, "VLM-TOPO") if not text: return G text = re.sub(r'```json|```', '', text).strip() try: data = json.loads(text) except Exception as e: debug_log("VLM-JSON-ERR", str(e)) return G label_to_node = {} for node, attrs in G.nodes(data=True): if attrs.get("node_type") != "net": label_to_node[node.strip().upper()] = node sch = attrs.get("schematic_label", node) if sch: label_to_node[sch.strip().upper()] = node comp_to_real_nets = {} for node, attrs in G.nodes(data=True): if attrs.get("node_type") == "net": continue comp_to_real_nets[node] = set() for nb in G.neighbors(node): na = G.nodes[nb] if na.get("node_type") == "net" and isinstance(na.get("net_id"), int): comp_to_real_nets[node].add(na["net_id"]) added = hall = dup = 0 for conn in data.get("connections", []): if len(conn) != 2: continue ak = str(conn[0]).strip().upper() bk = str(conn[1]).strip().upper() if ak not in label_to_node or bk not in label_to_node: continue an = label_to_node[ak] bn = label_to_node[bk] if an == bn: continue if set(G.neighbors(an)) & set(G.neighbors(bn)): dup += 1 continue shared = comp_to_real_nets.get(an, set()) & comp_to_real_nets.get(bn, set()) both_have_nets = (len(comp_to_real_nets.get(an, set())) > 0 and len(comp_to_real_nets.get(bn, set())) > 0) if both_have_nets and not shared: debug_log("VLM-HALL", f"{conn[0]}--{conn[1]} rejected (wired comps, no shared net)") hall += 1 continue nn = f"NET_VLM_{conn[0]}_{conn[1]}" if not G.has_node(nn): G.add_node(nn, node_type="net") G.add_edge(an, nn, edge_type="vlm") G.add_edge(bn, nn, edge_type="vlm") added += 1 debug_log("VLM", f"added={added} hall={hall} dup={dup}") return G # ═══════════════════════════════════════════════════════════════════════════ # STAGE 5 – TOPOLOGY ANALYSIS # ═══════════════════════════════════════════════════════════════════════════ def build_component_graph(G): """Simple Graph (not MultiGraph); dedupe edges per net pair.""" net_nodes = {n for n, d in G.nodes(data=True) if d.get('node_type') == 'net'} comp_nodes = {n for n in G.nodes() if n not in net_nodes} CG = nx.Graph() for n in comp_nodes: CG.add_node(n, **G.nodes[n]) wire_pairs = set() for net in net_nodes: inferred = "REPAIR" in str(net) or "VLM" in str(net) nbs = [nb for nb in G.neighbors(net) if nb in comp_nodes] for i in range(len(nbs)): for j in range(i + 1, len(nbs)): a, b = nbs[i], nbs[j] if not inferred: wire_pairs.add(tuple(sorted([a, b]))) if not CG.has_edge(a, b): CG.add_edge(a, b, net=net, inferred=False) else: pair = tuple(sorted([a, b])) if pair not in wire_pairs and not CG.has_edge(a, b): CG.add_edge(a, b, net=net, inferred=True) return CG def circuit_sections(G): CG = build_component_graph(G) return list(nx.connected_components(CG)) def detect_series_parallel(CG, details_map): res = {'series_chains': [], 'parallel_groups': [], 'junctions': [], 'isolated': []} res['junctions'] = [n for n in CG.nodes() if CG.degree(n) > 2] res['isolated'] = [n for n in CG.nodes() if CG.degree(n) == 0] endpoint_types = set(res['junctions']) | {n for n in CG.nodes() if CG.degree(n) == 1} visited_edges = set() for start in endpoint_types: for nb in CG.neighbors(start): ek = tuple(sorted([start, nb])) if ek in visited_edges: continue visited_edges.add(ek) chain = [start, nb] cur = nb prev = start while cur not in endpoint_types: nbs2 = [n for n in CG.neighbors(cur) if n != prev] if not nbs2: break nxt = nbs2[0] visited_edges.add(tuple(sorted([cur, nxt]))) chain.append(nxt) prev, cur = cur, nxt if len(chain) >= 3: res['series_chains'].append(chain) return res def get_source_nodes(G): return [n for n, d in G.nodes(data=True) if d.get('node_type', 'component') == 'component' and any(k in d.get('component', '').lower() for k in ['voltage', 'battery', 'source'])] HUB_MIN_COMPONENTS = 5 def get_hub_nets(G, min_components=HUB_MIN_COMPONENTS): """A 'hub' net is one with many component neighbors (a shared rail, e.g. a common ground/power bus). FIX (path explosion): all_simple_paths has no concept of 'this is the same physical branch traversed in a different visiting order' -- a single high-degree net acts as a combinatorial fan-out point, producing one distinct 'path' per ordering of which hub members get visited and in what sequence, even though electrically they're all just '...reaches the shared rail'. Surfacing hub nets explicitly lets path-finding stop AT the rail instead of enumerating every walk through it, and lets the report describe the rail once as a fact rather than burying it in dozens of near-duplicate paths.""" hubs = [] for n, d in G.nodes(data=True): if d.get('node_type') != 'net': continue comps = [nb for nb in G.neighbors(n) if G.nodes[nb].get('node_type') != 'net'] if len(comps) >= min_components: hubs.append((n, d.get('net_id', n), sorted(comps))) return hubs def fmt_comp(label, details_map): d = details_map.get(label, {}) sch = d.get('schematic_label', label) val = d.get('value', '') return f"{sch}({val})" if val else sch def find_all_circuit_paths(G, details_map, max_paths=200): """FIX: previously ran all_simple_paths on build_component_graph(G), which collapses every net into a clique of direct component-component edges, letting the search hop between ANY two components on a shared net in any order. Fixed by searching the bipartite component/net graph G directly. FIX 2: relabeling GND1/2/3 to a common key post-hoc didn't catch duplicates that inserted an extra GND->net->GND detour mid-path (a longer tuple than the direct version even after relabeling). Real fix: contract all physical GND nodes into ONE node before the search runs, so hopping between grounds as a detour is structurally impossible. FIX 3 (path explosion on shared rails): a high-degree hub net (e.g. a common rail touching 9 components) made all_simple_paths enumerate one 'path' per ordering of which hub members get visited, producing dozens of paths that are really the same '...reaches the rail' fact restated with different component orderings. Now, once a raw path crosses into a hub net, the component path is truncated right after the first hub member reached -- the path says 'X -> ... -> reaches the rail via Y', not 'X -> ... -> Y -> Z -> W -> ... (every other rail member in some order)'. The hub nets themselves are reported separately (see get_hub_nets / SHARED RAILS section).""" sources = get_source_nodes(G) gnd_nodes = list(get_gnd_nodes(G)) hub_net_ids = {nid for _, nid, _ in get_hub_nets(G) if isinstance(nid, int)} SG = G if len(gnd_nodes) > 1: SG = G.copy() canonical_gnd = gnd_nodes[0] for other in gnd_nodes[1:]: if other not in SG or canonical_gnd not in SG: continue for nb in list(SG.neighbors(other)): if nb != canonical_gnd: SG.add_edge(canonical_gnd, nb) SG.remove_node(other) gnd_set = {gnd_nodes[0]} if gnd_nodes else set() all_paths = [] def to_truncated_comp_path(raw_path): """Walk raw_path (alternating component/net nodes); once a hub net is crossed, include the first component reached via it, then stop -- collapsing the rest of the combinatorial fan-out into that single 'reaches the rail' step.""" out = [] hit_hub = False for node in raw_path: ndata = SG.nodes[node] if ndata.get('node_type') == 'net': if ndata.get('net_id') in hub_net_ids: hit_hub = True continue out.append(node) if hit_hub: break return out for src in sources: if src not in SG: continue src_fmt = fmt_comp(src, details_map) seen_keys = set() comp_paths = [] targets = ([g for g in gnd_set if g in SG and nx.has_path(SG, src, g)] if gnd_set else [n for n in SG.nodes() if SG.nodes[n].get('node_type') != 'net' and SG.degree(n) == 1 and n != src]) if not targets: targets = [n for n in SG.nodes() if n != src and SG.nodes[n].get('node_type') != 'net'] for tgt in targets: try: for raw_path in nx.all_simple_paths(SG, src, tgt, cutoff=30): cpath = to_truncated_comp_path(raw_path) if len(cpath) < 2: continue key = tuple(cpath) if key in seen_keys: continue seen_keys.add(key) comp_paths.append(cpath) if len(comp_paths) >= max_paths: break except (nx.NetworkXNoPath, nx.NodeNotFound): pass if len(comp_paths) >= max_paths: break if comp_paths: comp_paths.sort(key=lambda p: len(p)) def is_strict_subpath(short, long_): if len(short) >= len(long_): return False ls = len(short) for i in range(len(long_) - ls + 1): if long_[i:i + ls] == short: return True return False filt = [] for path in comp_paths: if not any(is_strict_subpath(path, longer) for longer in comp_paths if longer != path): filt.append(path) if filt: all_paths.append((src_fmt, filt[:15])) return all_paths def find_loops(CG, details_map, max_loops=20): loops = [] try: for cycle in nx.cycle_basis(CG)[:max_loops]: loops.append([fmt_comp(n, details_map) for n in cycle]) except Exception: pass return loops def export_spice_netlist(G, details, title='Extracted Circuit'): lines = [f'* {title}', '* Auto-generated by Circuit Detector', ''] net_nodes = {n for n, d in G.nodes(data=True) if d.get('node_type') == 'net'} comp_nodes = {n for n in G.nodes() if n not in net_nodes} all_nets = sorted(net_nodes, key=str) net_map = {} counter = [1] for net in all_nets: if ('GND' in str(net).upper() or any('gnd' in str(G.nodes[nb].get('component', '')).lower() for nb in G.neighbors(net) if G.nodes[nb].get('node_type') == 'component')): net_map[net] = '0' else: net_map[net] = str(counter[0]) counter[0] += 1 dm = {d['label']: d for d in details} spice_prefix = { 'resistor': 'R', 'capacitor': 'C', 'inductor': 'L', 'voltage source': 'V', 'battery': 'V', 'led': 'D', 'diode': 'D', 'switch open': 'S', 'switch closed': 'S', 'switch': 'S', 'transistor': 'Q', 'op-amp': 'U', } for comp in sorted(comp_nodes): d = dm.get(comp, {}) ctype = d.get('component', '').lower() val = d.get('value', '?') lbl = d.get('schematic_label', comp) prefix = next((v for k, v in spice_prefix.items() if k in ctype), 'X') nets_of_comp = [net_map.get(nb, '?') for nb in G.neighbors(comp) if nb in net_map] if len(nets_of_comp) < 2: nets_of_comp += ['?'] * (2 - len(nets_of_comp)) n1, n2 = nets_of_comp[0], nets_of_comp[1] if 'switch' in ctype: lines.append(f'{prefix}{lbl} {n1} {n2} SW_{lbl} ; {"OFF" if "open" in ctype else "ON"}') elif ctype in ('led', 'diode'): lines.append(f'{prefix}{lbl} {n1} {n2} D1N4148') else: spice_val = re.sub(r'µ', 'u', val) lines.append(f'{prefix}{lbl} {n1} {n2} {spice_val}') lines += ['', '.tran 1m 10m', '.end'] return '\n'.join(lines) def describe_connectivity(G, details=None): dm = {d['label']: d for d in details} if details else {} for d in (details or []): sch = d.get('schematic_label', d['label']) if sch != d['label']: dm[sch] = d def fmt(l): return fmt_comp(l, dm) if G.number_of_edges() == 0: return "No connections detected." gnd_nodes = get_gnd_nodes(G) net_nodes = {n for n, d in G.nodes(data=True) if d.get('node_type') == 'net'} comp_nodes = {n for n in G.nodes() if n not in net_nodes} CG = build_component_graph(G) lines = ["=== CIRCUIT ANALYSIS REPORT ===", ""] lines += ["--- DETECTED COMPONENTS ---"] for d in sorted(details or [], key=lambda x: x['label']): sch = d.get('schematic_label', d['label']) val = d.get('value', '') src = d.get('source', '?') lines.append(f" {sch:8s} {d['component']:20s} {val:10s} [{src}]") lines.append("") lines += ["--- WIRE CONNECTIONS ---"] net_to_comps = defaultdict(set) for net in net_nodes: for nb in G.neighbors(net): if nb in comp_nodes: net_to_comps[net].add(nb) seen_pairs = set() net_edges_by_net = defaultdict(list) for net, comps in net_to_comps.items(): comp_list_sorted = sorted(comps) for i in range(len(comp_list_sorted)): for j in range(i + 1, len(comp_list_sorted)): a, b = comp_list_sorted[i], comp_list_sorted[j] pair = tuple(sorted([a, b])) if pair not in seen_pairs: seen_pairs.add(pair) net_edges_by_net[net].append((a, b)) for net in sorted(net_edges_by_net.keys(), key=str): tag = " [inferred]" if ("REPAIR" in str(net) or "VLM" in str(net)) else "" for a, b in sorted(net_edges_by_net[net]): lines.append(f" {fmt(a):18s} ---- {fmt(b):18s} [{net}]{tag}") lines.append("") lines += ["--- TOPOLOGY ANALYSIS ---"] topo = detect_series_parallel(CG, dm) if topo['junctions']: lines.append(" Junction nodes (degree > 2):") for n in sorted(topo['junctions']): neighbors_fmt = ', '.join(fmt(nb) for nb in sorted(CG.neighbors(n))) lines.append(f" {fmt(n):14s} -> {neighbors_fmt}") if topo['isolated']: lines.append(f"\n WARNING Isolated: {', '.join(fmt(n) for n in topo['isolated'])}") lines.append("") lines += ["--- SHARED RAILS ---"] hub_nets = get_hub_nets(G) if not hub_nets: lines.append(" None (no net touches 5+ components)") else: for net, nid, comps in hub_nets: tag = " [GND rail]" if any('gnd' in str(c).lower() or 'gnd' in str(dm.get(c, {}).get('component', '')).lower() for c in comps) else "" lines.append(f" {net}{tag}: {', '.join(fmt(c) for c in comps)}") lines.append(" (paths below stop at the rail rather than enumerating every order") lines.append(" through it -- see rails above for full membership)") lines.append("") has_gnd_path = any(nx.has_path(G, s, g) for s in get_source_nodes(G) if s in G for g in gnd_nodes if g in G) lines += ["--- CIRCUIT PATHS ---"] if not gnd_nodes: lines.append(" WARNING No GND detected.") elif not has_gnd_path: lines.append(" WARNING GND not reachable from any source.") pg = find_all_circuit_paths(G, dm) if not pg: lines.append(" No paths found.") else: total = 0 for src_fmt, paths in pg: lines.append(f"\n From {src_fmt}") for i, path in enumerate(paths, 1): lines.append(f" Path {i}: {' -> '.join(fmt(n) for n in path)}") total += 1 lines.append(f" -- {len(paths)} path(s)") lines.append(f"\n Total: {total} path(s)") lines.append("") lines += ["--- SPICE NETLIST ---"] try: netlist = export_spice_netlist(G, details or []) lines.append(netlist) except Exception as e: lines.append(f" (netlist error: {e})") lines.append("") lines += ["--- CONNECTIVITY ---"] real_secs = circuit_sections(G) if len(real_secs) == 1: lines.append(" OK Fully connected circuit") else: lines.append(f" WARNING {len(real_secs)} separate sections:") for i, sec in enumerate(real_secs, 1): cn = [fmt(n) for n in sorted(sec)] lines.append(f" Section {i}: {', '.join(cn)}") lines.append("") n_conns = len(seen_pairs) n_nets = len([n for n in G.nodes() if G.nodes[n].get('node_type') == 'net']) lines.append(f" Components: {len(details or [])} Connections: {n_conns} Nets: {n_nets}") return "\n".join(lines) # ═══════════════════════════════════════════════════════════════════════════ # VISUALIZATION # ═══════════════════════════════════════════════════════════════════════════ COMP_COLORS = { 'resistor': (100, 180, 255), 'capacitor': (100, 255, 180), 'inductor': (255, 200, 100), 'led': (255, 100, 255), 'diode': (200, 100, 255), 'transistor': (255, 80, 80), 'voltage source': (255, 255, 80), 'battery': (255, 255, 80), 'switch': (80, 255, 255), 'ground': (160, 160, 160), 'gnd': (160, 160, 160), } def get_comp_color(ct): ct = ct.lower() for k, v in COMP_COLORS.items(): if k in ct: return v return (200, 200, 200) def get_center(d): return (int(d['x'] + d['w'] / 2), int(d['y'] + d['h'] / 2)) def draw_graph_overlay(base_img_np, details, G): out = base_img_np.copy() centers = {d['label']: get_center(d) for d in details} dm = {d['label']: d for d in details} for node, attrs in G.nodes(data=True): if attrs.get('node_type') == 'component' and node not in centers: cx = attrs.get('x', 0) + attrs.get('w', 0) // 2 cy = attrs.get('y', 0) + attrs.get('h', 0) // 2 centers[node] = (cx, cy) dm[node] = attrs net_nodes = {n for n, d in G.nodes(data=True) if d.get('node_type') == 'net'} comp_nodes = {n for n in G.nodes() if n not in net_nodes} # FIX (visual clutter): a hub net (shared rail, e.g. a common GND/power # bus touching many components) used to be drawn as a full pairwise # clique -- N components means N*(N-1)/2 crossing lines, which for a # 9-component rail is 36 overlapping cyan lines burying every other # connection in the diagram. Hub nets are now drawn as a dim star: one # short line per component out to a small rail marker, instead of a # line between every pair. Non-hub (normal, small) nets are unaffected # and still drawn as direct point-to-point wires. hub_net_ids = {nid for _, nid, _ in get_hub_nets(G)} drawn = set() for net in net_nodes: is_repair = "REPAIR" in str(net) is_vlm = "VLM" in str(net) is_hub = G.nodes[net].get('net_id') in hub_net_ids color = (255, 80, 80) if is_repair else (255, 165, 50) if is_vlm else (80, 220, 255) thickness = 1 if (is_repair or is_vlm) else 2 nbs = [nb for nb in G.neighbors(net) if nb in comp_nodes and nb in centers] if is_hub and not is_repair and not is_vlm: if len(nbs) < 2: continue pts = np.array([centers[nb] for nb in nbs]) cx, cy = int(pts[:, 0].mean()), int(pts[:, 1].mean()) rail_color = (130, 130, 150) for nb in nbs: cv2.line(out, centers[nb], (cx, cy), rail_color, 1, cv2.LINE_AA) cv2.circle(out, (cx, cy), 4, rail_color, -1) cv2.putText(out, f"RAIL {net}", (cx + 6, cy - 6), cv2.FONT_HERSHEY_SIMPLEX, 0.32, rail_color, 1, cv2.LINE_AA) continue for i in range(len(nbs)): for j in range(i + 1, len(nbs)): a, b = nbs[i], nbs[j] pair = tuple(sorted([a, b])) if pair in drawn: continue pa, pb = centers[a], centers[b] if is_repair: pts = np.linspace(np.array(pa), np.array(pb), 20).astype(int) for k in range(0, len(pts) - 1, 2): cv2.line(out, tuple(pts[k]), tuple(pts[k + 1]), color, 1, cv2.LINE_AA) else: cv2.line(out, pa, pb, color, thickness, cv2.LINE_AA) drawn.add(pair) for lbl, c in centers.items(): d = dm.get(lbl, {}) nc = get_comp_color(d.get('component', '')) deg = G.degree(lbl) if lbl in G else 0 cv2.circle(out, c, 8, nc, -1) cv2.circle(out, c, 8, (255, 255, 255), 1) if deg == 0: cv2.circle(out, c, 11, (200, 50, 50), 2) sch = d.get('schematic_label', lbl) val = d.get('value', '') txt = f"{sch}={val}" if val else sch (tw, th), _ = cv2.getTextSize(txt, cv2.FONT_HERSHEY_SIMPLEX, 0.38, 1) tx, ty = c[0] + 10, c[1] - 4 cv2.rectangle(out, (tx - 2, ty - th - 2), (tx + tw + 2, ty + 2), (20, 20, 40), -1) cv2.putText(out, txt, (tx, ty), cv2.FONT_HERSHEY_SIMPLEX, 0.38, nc, 1, cv2.LINE_AA) for lc, lt, ly in [((80, 220, 255), "Wire-traced", 20), ((255, 165, 50), "VLM-verified", 36), ((255, 80, 80), "Proximity guess", 52), ((130, 130, 150), "Shared rail (hub net)", 68)]: cv2.circle(out, (16, ly), 5, lc, -1) cv2.putText(out, lt, (24, ly + 4), cv2.FONT_HERSHEY_SIMPLEX, 0.32, lc, 1, cv2.LINE_AA) return out def draw_net_overlay(img_np, cc_map, num_cc): out = img_np.copy() rng = np.random.default_rng(42) colors = rng.integers(80, 255, size=(max(num_cc + 1, 2), 3)).tolist() for nid in range(1, num_cc): mask = (cc_map == nid) if int(np.sum(mask)) >= 8: out[mask] = colors[nid] return out def draw_component_graph(details, G, img_shape): h, w = img_shape[:2] canvas = np.zeros((h, w, 3), dtype=np.uint8) canvas[:] = (20, 22, 30) net_nodes = {n for n, d in G.nodes(data=True) if d.get('node_type') == 'net'} comp_nodes = {n for n in G.nodes() if n not in net_nodes} centers = {d['label']: get_center(d) for d in details} dm = {d['label']: d for d in details} for node, attrs in G.nodes(data=True): if attrs.get('node_type') == 'component' and node not in centers: centers[node] = (attrs.get('x', 0) + attrs.get('w', 0) // 2, attrs.get('y', 0) + attrs.get('h', 0) // 2) dm[node] = attrs drawn = set() hub_net_ids = {nid for _, nid, _ in get_hub_nets(G)} for net in net_nodes: inferred = "REPAIR" in str(net) or "VLM" in str(net) is_hub = G.nodes[net].get('net_id') in hub_net_ids color = (60, 100, 60) if inferred else (60, 180, 60) nbs = [nb for nb in G.neighbors(net) if nb in comp_nodes and nb in centers] if is_hub and not inferred: if len(nbs) < 2: continue pts_all = np.array([centers[nb] for nb in nbs]) cx, cy = int(pts_all[:, 0].mean()), int(pts_all[:, 1].mean()) rail_color = (70, 70, 90) for nb in nbs: cv2.line(canvas, centers[nb], (cx, cy), rail_color, 1, cv2.LINE_AA) cv2.circle(canvas, (cx, cy), 3, rail_color, -1) continue for i in range(len(nbs)): for j in range(i + 1, len(nbs)): a, b = nbs[i], nbs[j] pair = tuple(sorted([a, b])) if pair in drawn: continue drawn.add(pair) pa, pb = np.array(centers[a]), np.array(centers[b]) mx = (pa[0] + pb[0]) // 2 pts = [tuple(pa), (mx, pa[1]), (mx, pb[1]), tuple(pb)] for k in range(len(pts) - 1): cv2.line(canvas, pts[k], pts[k + 1], color, 1 if inferred else 2, cv2.LINE_AA) for lbl, c in centers.items(): d = dm.get(lbl, {}) nc = get_comp_color(d.get('component', '')) cv2.circle(canvas, c, 10, nc, -1) cv2.circle(canvas, c, 10, (255, 255, 255), 1) sch = d.get('schematic_label', lbl) val = d.get('value', '') for ti, lt in enumerate([sch, val] if val else [sch]): (tw, th), _ = cv2.getTextSize(lt, cv2.FONT_HERSHEY_SIMPLEX, 0.40, 1) tx = c[0] - tw // 2 ty = c[1] - 14 + ti * 14 cv2.putText(canvas, lt, (tx, ty), cv2.FONT_HERSHEY_SIMPLEX, 0.40, nc, 1, cv2.LINE_AA) return canvas # ═══════════════════════════════════════════════════════════════════════════ # PIPELINE # ═══════════════════════════════════════════════════════════════════════════ def run_detection(image, confidence): clear_debug_log() debug_log("PIPELINE", "Step 1: YOLO three-pass detection") img_np, results, details = detect_components(image, confidence) debug_log("PIPELINE", "Step 2: OCR + LLM value/label extraction") ocr_img, ocr_scale = preprocess_for_ocr(img_np) raw_ocr_results = get_ocr_reader().readtext( ocr_img, text_threshold=0.5, # FIX: lowered from EasyOCR default 0.7 -- small # schematic text often has low confidence low_text=0.3, # FIX: lowered from default 0.4 -- catches # faint/thin-stroke characters contrast_ths=0.05, # FIX: lowered from default 0.1 -- text on # light schematic backgrounds is low-contrast adjust_contrast=0.7, mag_ratio=1.0, # already upscaled in preprocess_for_ocr ) # rescale coordinates back to original image space since OCR ran on # the upscaled image ocr_results = [ ([[c[0] / ocr_scale, c[1] / ocr_scale] for c in coords], text, *rest) for (coords, text, *rest) in raw_ocr_results ] debug_log("OCR", f"{len(ocr_results)} text regions (scale={ocr_scale}x)") details, labeled_pil = fill_values(img_np, details, ocr_results) debug_log("PIPELINE", "Step 2b: GND symbol detection") details = detect_gnd_symbols(img_np, details) gnd_counter = 1 for d in details: if not d['label']: d['label'] = f"GND{gnd_counter}" d['schematic_label'] = "GND" gnd_counter += 1 debug_log("PIPELINE", "Step 3: Wire-CC graph") G, skel, cc_map, num_cc, valid_nets, net_pixels, net_endpoints, scale = build_graph_wire_cc(details, img_np) debug_log("PIPELINE", f"Wire-CC: {G.number_of_nodes()}N {G.number_of_edges()}E gnd={get_gnd_nodes(G)}") debug_log("PIPELINE", "Step 3b: GND snap") G = wire_gnd_to_nearest_net(G, details, cc_map, valid_nets, scale) G = prune_orphan_nets(G, label="ORPHAN-SWEEP(gnd)") debug_log("PIPELINE", "Step 3c: Isolation repair") G = repair_isolated_nodes(G, details, img_np, cc_map, valid_nets) G = prune_orphan_nets(G, label="ORPHAN-SWEEP(repair)") debug_log("PIPELINE", "Step 4: VLM topology verification") G = verify_connections_with_llm(labeled_pil, details, G) G = prune_orphan_nets(G, label="ORPHAN-SWEEP(vlm)") debug_log("PIPELINE", f"Final: {G.number_of_nodes()}N {G.number_of_edges()}E " f"sections={len(circuit_sections(G))}") base_plot = results[0].plot() annotated = draw_graph_overlay(base_plot, details, G) net_vis = draw_net_overlay(img_np.copy(), cc_map, num_cc) comp_g = draw_component_graph(details, G, img_np.shape) return Image.fromarray(annotated), Image.fromarray(net_vis), Image.fromarray(comp_g), details, G # ═══════════════════════════════════════════════════════════════════════════ # HANDLERS # ═══════════════════════════════════════════════════════════════════════════ def handle_image1(image, confidence): if image is None: return None, None, None, {}, "Upload an image to begin.", "", "", get_debug_text() try: ann, nv, cg, details, G = run_detection(image, confidence) total = len(details) det_cnt = sum(1 for d in details if not d.get("is_default")) def_cnt = total - det_cnt summary = f"### {total} component{'s' if total != 1 else ''} detected\n\n" summary += f"OCR/LLM: {det_cnt} | Default: {def_cnt}\n\n" summary += "| Label | Schematic | Component | Value | Src |\n|---|---|---|---|---|\n" for d in sorted(details, key=lambda x: x['label']): src = d.get("source") or "default" sch = d.get("schematic_label", d['label']) summary += f"| {d['label']} | {sch} | {d['component']} | {d['value'] or '-'} | {src} |\n" CG = build_component_graph(G) dm = {d['label']: d for d in details} loops = find_loops(CG, dm) summary += f"\n**Edges:** {G.number_of_edges()} " if loops: summary += f"**Loops:** {len(loops)}" summary += "\n\n_Base saved._" report = describe_connectivity(G, details=details) return ann, nv, cg, details, summary, report, "", get_debug_text() except Exception as e: traceback.print_exc() debug_log("FATAL", f"{type(e).__name__}: {e}\n{traceback.format_exc()}") return None, None, None, {}, "Error during detection", str(e), "", get_debug_text() def parse_base_value_to_float(value_str): if value_str is None: return None, None pm = {'k': 1e3, 'K': 1e3, 'M': 1e6, 'm': 1e-3, 'µ': 1e-6, 'u': 1e-6, 'n': 1e-9, 'p': 1e-12} m = re.compile(r'([\d.]+)\s*([kKMmµunp])?\s*(Ω|ohm|F|H|V|A|W)?', re.IGNORECASE).search(str(value_str).strip()) if not m: return None, None return float(m.group(1)) * pm.get(m.group(2) or '', 1.0), m.group(3) or '' def values_match(base_str, csv_val): try: bf, _ = parse_base_value_to_float(str(base_str)) if bf is None: return str(base_str).strip().upper() == str(csv_val).strip().upper() return abs(bf - float(str(csv_val).strip())) < 1e-9 except Exception: return str(base_str).strip().upper() == str(csv_val).strip().upper() def handle_csv(csv_file, base_details): if csv_file is None: return "Upload CSV.", "", None if not base_details: return "No base circuit saved.", "", None try: csv_path = csv_file if isinstance(csv_file, str) else getattr(csv_file, 'name', None) if csv_path is None: return "Could not read CSV file path.", "", None df = pd.read_csv(csv_path) df.columns = [c.strip() for c in df.columns] sch_to_lbl = {} base_map = {d['label']: d for d in base_details} for d in base_details: sch = d.get('schematic_label', d['label']) sch_to_lbl[sch] = d['label'] sch_to_lbl[d['label']] = d['label'] matched = [c for c in df.columns if c in sch_to_lbl] col_to_lbl = {c: sch_to_lbl[c] for c in matched} if not matched: return (f"No matching labels.\nCSV:{', '.join(df.columns)}\nBase:{', '.join(base_map)}", "", None) total = len(df) wrong_rows = {c: [] for c in matched} ok_rows = {} for col in matched: lbl = col_to_lbl[col] bv = base_map[lbl]['value'] ok = 0 for i, rv in enumerate(df[col]): if values_match(bv, rv): ok += 1 else: wrong_rows[col].append((i + 1, rv, bv)) ok_rows[col] = ok issues = sum(len(v) for v in wrong_rows.values()) status = "ISSUES FOUND" if issues else "ALL OK" summary = f"### {status}\n\nRows: {total} | Checked: {len(matched)}\n\n" unm_base = [l for l in base_map if l not in col_to_lbl.values()] unm_csv = [c for c in df.columns if c not in matched] if unm_base: summary += f"> Not in CSV: {', '.join(unm_base)}\n\n" if unm_csv: summary += f"> CSV extra: {', '.join(unm_csv)}\n\n" summary += "| Component | Base | OK | Wrong | Status |\n|---|---|---|---|---|\n" for col in matched: lbl = col_to_lbl[col] bv = base_map[lbl]['value'] ok = ok_rows[col] wr = len(wrong_rows[col]) comp = base_map[lbl]['component'] summary += f"| {col} ({comp}) | {bv} | {ok} | {wr} | {'OK' if wr == 0 else f'WRONG({wr})'} |\n" for col in matched: if not wrong_rows[col]: continue lbl = col_to_lbl[col] bv = base_map[lbl]['value'] comp = base_map[lbl]['component'] summary += f"\n### {col} ({comp})\n| Row | Got | Expected |\n|---|---|---|\n" for rn, cv_, b in wrong_rows[col]: summary += f"| {rn} | {cv_} | {b} |\n" report = f"CSV COMPARISON\n{'='*38}\nStatus:{status}\nRows:{total} Checked:{len(matched)} Issues:{issues}\n\n" for col in matched: lbl = col_to_lbl[col] bv = base_map[lbl]['value'] ok = ok_rows[col] wr = len(wrong_rows[col]) report += f"{col}({base_map[lbl]['component']}) expected={bv} ok={ok} wrong={wr}\n" for rn, cv_, b in wrong_rows[col]: report += f" row{rn}: got={cv_} expected={b}\n" df_out = df.copy() for col in matched: lbl = col_to_lbl[col] bv = base_map[lbl]['value'] df_out[col + "_status"] = df_out[col].apply(lambda v: "OK" if values_match(bv, v) else "WRONG") fd, out_csv_path = tempfile.mkstemp(suffix=".csv") os.close(fd) df_out.to_csv(out_csv_path, index=False) return summary, report, out_csv_path except Exception as e: traceback.print_exc() return f"Error: {e}", "", None def reset_all(): clear_debug_log() return (None, None, None, None, None, "Upload image to set base circuit.", "", "Upload CSV to compare.", "", [], None, get_debug_text()) # ═══════════════════════════════════════════════════════════════════════════ # UI # ═══════════════════════════════════════════════════════════════════════════ css = """ html,body{margin:0;padding:0} .gradio-container{max-width:100%!important;width:100%!important;margin:0!important;padding:0!important;background:#0f1117!important;font-family:'DM Sans',sans-serif!important} .app-header{background:linear-gradient(135deg,#0f1117,#1a1f2e);border-bottom:1px solid #2a2f3e;padding:12px 28px;display:flex;align-items:center;gap:10px;flex-wrap:wrap} .app-header h1{font-size:1.2rem!important;font-weight:700!important;color:#e2e8f0!important;margin:0!important} .badge{font-size:.70rem;padding:2px 9px;border-radius:20px} .b1{color:#4ade80;background:#0d2818;border:1px solid #166534} .b2{color:#818cf8;background:#1e1b4b;border:1px solid #3730a3} .b3{color:#fb923c;background:#1c0f05;border:1px solid #7c2d12} .b4{color:#f43f5e;background:#1c0509;border:1px solid #881337} .panel-label{font-size:.7rem!important;font-weight:600!important;text-transform:uppercase!important;letter-spacing:1px!important;color:#4b5563!important;margin-bottom:4px!important} .image-box{border-radius:10px!important;overflow:hidden!important;background:#1a1f2e!important;border:1px solid #1e2330!important} .action-btn{background:linear-gradient(135deg,#3b82f6,#6366f1)!important;border:none!important;color:#fff!important;font-weight:600!important;border-radius:8px!important;padding:10px!important;cursor:pointer!important;width:100%!important} .csv-btn{background:linear-gradient(135deg,#059669,#047857)!important;border:none!important;color:#fff!important;font-weight:600!important;border-radius:8px!important;padding:10px!important;cursor:pointer!important;width:100%!important} .clear-btn{background:#1e2330!important;border:1px solid #2a2f3e!important;color:#9ca3af!important;font-weight:500!important;border-radius:8px!important;padding:10px!important;cursor:pointer!important;width:100%!important} .summary-box{overflow-y:auto!important;background:#1a1f2e!important;border-radius:10px!important;padding:12px!important;border:1px solid #1e2330!important;font-size:.82rem!important;color:#cbd5e1!important} .report-box textarea{font-family:'JetBrains Mono',monospace!important;font-size:.70rem!important;background:#1a1f2e!important;color:#94a3b8!important;border:1px solid #1e2330!important;border-radius:10px!important} .debug-box textarea{font-family:'JetBrains Mono',monospace!important;font-size:.68rem!important;background:#120d05!important;color:#facc15!important;border:1px solid #422006!important;border-radius:10px!important} ::-webkit-scrollbar{width:4px} ::-webkit-scrollbar-thumb{background:#2a2f3e;border-radius:4px} footer{display:none!important} """ with gr.Blocks( theme=gr.themes.Base(primary_hue="blue", neutral_hue="slate", font=[gr.themes.GoogleFont("DM Sans"), "sans-serif"], font_mono=[gr.themes.GoogleFont("JetBrains Mono"), "monospace"]), css=css, title="Circuit Detector v24" ) as app: base_state = gr.State([]) gr.HTML("""

Circuit Component Detector

YOLOv8 3-pass OCR + LLM Skeleton + Union-Find + KDTree v24 - terminal nets now nearest-only (kdtree query, not ball_point) -- fixes spurious multi-net terminals
""") with gr.Row(): with gr.Column(scale=1, min_width=260): gr.HTML('
Confidence Threshold
') confidence = gr.Slider(0.01, 1.0, value=0.03, step=0.01, label="", info="Lower=more detections") gr.HTML('
Step 1 - Base Circuit
') image1 = gr.Image(type="pil", label="Upload Circuit Image", elem_classes="image-box", height=200) analyze_btn = gr.Button("Analyze and Save as Base", elem_classes="action-btn") gr.HTML('
Step 2 - CSV Log
') csv_input = gr.File(label="Upload CSV Log File", file_types=[".csv"]) compare_btn = gr.Button("Compare Log vs Base", elem_classes="csv-btn") gr.HTML('
') clear_btn = gr.Button("Reset Everything", elem_classes="clear-btn") with gr.Column(scale=2, min_width=300): gr.HTML('
Connection Graph Overlay
') out_image_graph = gr.Image(label="", elem_classes="image-box", show_label=False, height=210) gr.HTML('
Wire Net Map
') out_image_nets = gr.Image(label="", elem_classes="image-box", show_label=False, height=210) gr.HTML('
Component Graph
') out_comp_graph = gr.Image(label="", elem_classes="image-box", show_label=False, height=210) with gr.Column(scale=2, min_width=300): gr.HTML('
Base Summary
') out_summary1 = gr.Markdown(value="_Upload circuit image to begin._", elem_classes="summary-box", height=200) gr.HTML('
CSV Comparison
') out_summary2 = gr.Markdown(value="_Upload CSV to compare._", elem_classes="summary-box", height=240) gr.HTML('
Download Result CSV
') out_csv_download = gr.File(label="", show_label=False) with gr.Row(): with gr.Column(scale=1): gr.HTML('
Circuit Analysis Report
') out_report1 = gr.Textbox(label="", lines=24, show_label=False, placeholder="Full report: topology, paths, loops, SPICE netlist...", elem_classes="report-box") with gr.Column(scale=1): gr.HTML('
CSV Comparison Report
') out_report2 = gr.Textbox(label="", lines=24, show_label=False, placeholder="CSV comparison report...", elem_classes="report-box") with gr.Row(): with gr.Column(scale=1): gr.HTML('
Debug Log
') out_debug = gr.Textbox(label="", lines=14, show_label=False, elem_classes="debug-box") analyze_btn.click(fn=handle_image1, inputs=[image1, confidence], outputs=[out_image_graph, out_image_nets, out_comp_graph, base_state, out_summary1, out_report1, out_summary2, out_debug]) compare_btn.click(fn=handle_csv, inputs=[csv_input, base_state], outputs=[out_summary2, out_report2, out_csv_download]) clear_btn.click(fn=reset_all, inputs=[], outputs=[image1, csv_input, out_image_graph, out_image_nets, out_comp_graph, out_summary1, out_report1, out_summary2, out_report2, base_state, out_csv_download, out_debug]) app.launch(server_name="0.0.0.0", server_port=7860)