"""Parse client golden examples from PDC data/Examples .xlsx.""" from __future__ import annotations import re from pathlib import Path import pandas as pd EXAMPLES_PATH = Path(__file__).resolve().parents[3] / "PDC data" / "Examples .xlsx" AI_FIELD_MAP = { "Warp Count": "warp_count", "Weft Count": "weft_count", "Reed Count": "reed_count", "Ends Per Dent": "ends_per_dent", "Onloom Epi": "on_loom_epi", "Onloom Ppi": "on_loom_ppi", "Greige Epi": "greige_epi", "Greige Ppi": "greige_ppi", "Finish Epi": "finish_epi", "Finish Ppi": "finish_ppi", } def _parse_analysis(text: str) -> dict: """Parse analysis lines including twill weaves and multi-line blends.""" out: dict = {} cleaned = text.replace("Analysis:", "").strip() lines = [ln.strip() for ln in cleaned.split("\n") if ln.strip()] main = lines[0] if lines else cleaned for ln in lines[1:]: if "%" in ln: out["blend"] = ln.strip() m = re.search( r"([\d.]+)'?s?\*([\d.]+)'?s?-(\d+)\*(\d+)-(.+?)\s+GSM:\s*([\d.]+)", main, re.IGNORECASE, ) if m: out["warp_count"] = float(m.group(1)) out["weft_count"] = float(m.group(2)) out["finish_epi"] = float(m.group(3)) out["finish_ppi"] = float(m.group(4)) out["weave"] = m.group(5).strip().upper() out["target_gsm"] = float(m.group(6)) if "blend" not in out and "cotton" in text.lower(): out["blend"] = "100% COTTON" return out def _parse_key_value_block(df: pd.DataFrame, key_col: int, val_col: int, start_row: int) -> dict: ai: dict = {} for j in range(start_row, min(start_row + 15, len(df))): key = df.iloc[j, key_col] if pd.isna(key): break key_s = str(key).strip() if key_s.lower().startswith("master"): break val = df.iloc[j, val_col] if pd.notna(val): try: ai[key_s] = float(val) except (TypeError, ValueError): ai[key_s] = val return ai def _parse_ai_block(df: pd.DataFrame) -> tuple[dict, str]: ai: dict = {} note = "" for i in range(len(df)): for col in range(len(df.columns)): cell = df.iloc[i, col] if not isinstance(cell, str) or "AI Suggestion" not in cell: continue parts = cell.split(":", 1) if len(parts) > 1 and parts[1].strip(): note = parts[1].strip() val_col = col + 2 key_col = col + 1 if val_col >= len(df.columns): continue ai = _parse_key_value_block(df, key_col, val_col, i) return ai, note for i in range(len(df) - 1, -1, -1): for col in range(len(df.columns) - 1): key = df.iloc[i, col] if pd.isna(key) or str(key).strip() != "Warp Count": continue val = df.iloc[i, col + 1] try: float(val) except (TypeError, ValueError): continue ai = _parse_key_value_block(df, col, col + 1, i) if ai: return ai, note return ai, note def _find_analysis_row(df: pd.DataFrame) -> str: for i in range(len(df)): for col in range(len(df.columns)): cell = df.iloc[i, col] if isinstance(cell, str) and "Analysis:" in cell: return cell.replace("Analysis:", "").strip() return "" def _detect_archive_cols(df: pd.DataFrame) -> dict: header = [str(c).strip().lower() if pd.notna(c) else "" for c in df.iloc[0]] def idx(*names: str) -> int | None: for i, h in enumerate(header): if any(n in h for n in names): return i return None return { "warp": idx("warp code"), "weft": idx("weft code"), "reed": idx("reed count"), "epd": idx("ends per dent"), "onloom_epi": idx("on loom epi"), "onloom_ppi": idx("on loom ppi"), "greige_epi": idx("greige epi"), "greige_ppi": idx("greige ppi"), "finish_epi": idx("finish epi"), "finish_ppi": idx("finish ppi"), } def _parse_gsm_cases(df: pd.DataFrame) -> list[dict]: cases: list[dict] = [] for i in range(len(df)): for col in range(len(df.columns)): if str(df.iloc[i, col]).strip() != "Case 1": continue for j in range(i, min(i + 6, len(df))): label = str(df.iloc[j, col]).strip() if not label.startswith("Case"): continue try: cases.append({ "case": label, "warp_count": float(df.iloc[j, col + 1]), "weft_count": float(df.iloc[j, col + 2]), "finish_epi": float(df.iloc[j, col + 3]), "finish_ppi": float(df.iloc[j, col + 4]), "gsm": float(df.iloc[j, col + 5]), }) except (TypeError, ValueError, IndexError): break return cases return cases def load_client_examples() -> list[dict]: if not EXAMPLES_PATH.exists(): return [] xl = pd.ExcelFile(EXAMPLES_PATH) examples: list[dict] = [] for sheet in xl.sheet_names: df = pd.read_excel(EXAMPLES_PATH, sheet_name=sheet, header=None) analysis_text = _find_analysis_row(df) inputs = _parse_analysis(analysis_text) ai, note = _parse_ai_block(df) if not inputs or not ai: continue cols = _detect_archive_cols(df) archive_matches: list[dict] = [] for r in range(1, 20): if r >= len(df) or pd.isna(df.iloc[r, 0]): break master = str(df.iloc[r, 0]).strip() if not master or master.lower() in {"nan", "sr no"}: break try: archive_matches.append({ "master_article": master, "score": float(r), "construction": { "warp_count": _safe_float_col(df.iloc[r, cols["warp"]]) if cols["warp"] is not None else None, "weft_count": _safe_float_col(df.iloc[r, cols["weft"]]) if cols["weft"] is not None else None, "reed_count": float(df.iloc[r, cols["reed"]]), "ends_per_dent": float(df.iloc[r, cols["epd"]]), "on_loom_epi": float(df.iloc[r, cols["onloom_epi"]]), "on_loom_ppi": float(df.iloc[r, cols["onloom_ppi"]]), "greige_epi": float(df.iloc[r, cols["greige_epi"]]), "greige_ppi": float(df.iloc[r, cols["greige_ppi"]]), "finish_epi": float(df.iloc[r, cols["finish_epi"]]), "finish_ppi": float(df.iloc[r, cols["finish_ppi"]]), }, }) except (TypeError, ValueError, IndexError): break expected = {field: ai.get(label) for label, field in AI_FIELD_MAP.items()} examples.append({ "sheet": sheet, "inputs": inputs, "gsm_cases": _parse_gsm_cases(df), "archive_matches": archive_matches, "expected": expected, "note": note, }) return examples def _safe_float_col(value) -> float | None: if pd.isna(value): return None text = str(value).strip() if "/" in text: try: return float(text.split("/")[-1]) except ValueError: return None try: return float(value) except (TypeError, ValueError): return None