"""Prediction Engine for PDC Construction Recommendation. Client feasibility workflow: Step 1: GSM back-calculation → dynamic warp×weft count-pair cases (expand until matches) Step 2: Per count pair — 5-case cascade search with EPI/PPI range probing Step 3: Rank articles by weighted distance; return top 2–3 exact historical matches Step 4: Reed validation on primary match greige EPI Weightage: user-tunable weights for count, EPI/PPI, GSM — used for ranking only. """ from __future__ import annotations import re from typing import Any import pandas as pd from .count_converter import nearest_standard_counts, primary_standard_counts from .finish_spec_refiner import FINISH_EPI_BAND, extract_archive_rows, merge_archive_rows, refine_finish_specs from .reed_calculator import recommend_reed GSM_CONSTANT = 24.5 DEFAULT_WEIGHTS = { "count": 40, "epi_ppi": 35, "gsm": 25, } CASE_MIN_ROWS = 3 MAX_MATCHES_RETURN = 3 MAX_EXPANSION_LEVEL = 6 INITIAL_NEAREST_N = 2 EPI_PPI_WINDOW = 5 MISSING_GSM_PENALTY = 50.0 def _clean_text(value: Any) -> str: if value is None: return "" return str(value).strip() def _safe_float(value: Any) -> float | None: if value is None: return None text = str(value).strip().replace(",", "") if text == "" or text.lower() in {"nan", "none"}: return None try: return float(text) except (ValueError, TypeError): return None def _article_family_key(master: str) -> str: m = re.match(r"^(\d+)", master or "") return m.group(1) if m else master def _construction_fingerprint(row: pd.Series) -> str: parts: list[str] = [] for col in ( "Greige EPI", "Greige PPI", "FINISH EPI", "FINISH PPI", "FINISH GSM", "Reed Count", "Ends per dent", ): val = _safe_float(row.get(col)) parts.append(str(round(val, 1) if val is not None else "")) return "|".join(parts) def _normalize_weave(value: Any) -> str: text = _clean_text(value).upper() text = text.replace(",", " ") text = re.sub(r"\s+", " ", text) text = text.replace("TWL", "TWILL") text = re.sub(r"(\d+/\d+)\s+TWILL\b", r"\1 S TWILL", text) return text.strip() def _normalize_blend(value: Any) -> str: text = _clean_text(value).upper() text = re.sub(r"\bPER\b", "%", text) text = text.replace("COTTON", "CO") text = text.replace("POLYESTER", "PES") text = re.sub(r"[\s_\-]+", "", text) return text def _weave_family(weave: str) -> str: w = weave.upper() if "TWILL" in w: return "TWILL" if "SATIN" in w or "SATEEN" in w: return "SATIN" if "DOBBY" in w: return "DOBBY" if "PLAIN" in w or w == "PLAIN": return "PLAIN" if "RIBSTOP" in w: return "RIBSTOP" return w def _blend_family(blend: str) -> str: b = blend.upper() if "CO" in b and ("PES" in b or "POLY" in b): return "PES/CO" if "CO" in b: return "CO" if "PES" in b or "POLY" in b: return "PES" return b def _dominant_fiber_pct(blend: str) -> tuple[str | None, float | None]: m = re.match(r"(\d+(?:\.\d+)?)%(\w+)", blend) if m: return (m.group(2), float(m.group(1))) return (None, None) def _blend_matches_strict(row_blend: str, target_blend: str) -> bool: return _normalize_blend(row_blend) == _normalize_blend(target_blend) def _blend_matches_family(row_blend: str, target_blend: str) -> bool: return _blend_family(_normalize_blend(row_blend)) == _blend_family(_normalize_blend(target_blend)) def _compute_gsm(epi: float, ppi: float, warp_count: float, weft_count: float) -> float: return ((epi / max(warp_count, 0.001)) + (ppi / max(weft_count, 0.001))) * GSM_CONSTANT def _confidence(match_count: int, best_score: float) -> str: if match_count >= 10 and best_score <= 10: return "high" if match_count >= 5 and best_score <= 20: return "medium" if match_count >= CASE_MIN_ROWS: return "low" return "very_low" def _pairs_equal(a: float, b: float, tol: float = 0.01) -> bool: return abs(a - b) <= tol def _case_status_from_counts(raw_count: int, match_count: int) -> str: if raw_count == 0: return "no_data" if match_count >= CASE_MIN_ROWS: return "searched" return "insufficient" def _mark_selected_count_case( count_cases: list[dict], warp: float, weft: float, ) -> dict | None: selected: dict | None = None for case in count_cases: if case.get("status") == "selected": case["status"] = _case_status_from_counts( int(case.get("raw_count_in_archive", 0)), int(case.get("match_count", 0)), ) if _pairs_equal(case["warp_count"], warp) and _pairs_equal(case["weft_count"], weft): case["status"] = "selected" selected = case return selected def _format_epi_ppi_range_label(finish_epi: float, finish_ppi: float) -> str: return ( f"EPI {round(finish_epi - EPI_PPI_WINDOW)}-{round(finish_epi + EPI_PPI_WINDOW)}" f" / PPI {round(finish_ppi - EPI_PPI_WINDOW)}-{round(finish_ppi + EPI_PPI_WINDOW)}" ) def generate_count_pair_cases( warp_in: float, weft_in: float, finish_epi: float, finish_ppi: float, target_gsm: float | None, expansion_level: int = INITIAL_NEAREST_N, primary_only: bool = False, ) -> list[dict]: """Build warp×weft count-pair cases for a given expansion level.""" if primary_only or expansion_level == INITIAL_NEAREST_N: warp_candidates = primary_standard_counts(warp_in) weft_candidates = primary_standard_counts(weft_in) else: warp_candidates = nearest_standard_counts(warp_in, n=expansion_level) weft_candidates = nearest_standard_counts(weft_in, n=expansion_level) cases: list[dict] = [] seen: set[tuple[float, float]] = set() warp_order = sorted(warp_candidates, reverse=True) weft_order = sorted(weft_candidates) low_count = max(warp_candidates + weft_candidates) < 30 if primary_only or expansion_level == INITIAL_NEAREST_N: if low_count: warp_order = sorted(warp_candidates, reverse=True) else: warp_order = sorted(warp_candidates) weft_order = sorted(weft_candidates) for wi, wc in enumerate(warp_order): if primary_only or expansion_level == INITIAL_NEAREST_N: wefts = sorted(weft_candidates) if low_count or wi == 0 else sorted( weft_candidates, reverse=True, ) else: wefts = sorted(weft_candidates) if wi == 0 else sorted(weft_candidates, reverse=True) for wtc in wefts: key = (wc, wtc) if key in seen: continue seen.add(key) gsm = round(_compute_gsm(finish_epi, finish_ppi, wc, wtc), 1) gsm_delta = abs(gsm - target_gsm) if target_gsm is not None else None cases.append({ "warp_count": wc, "weft_count": wtc, "finish_epi": finish_epi, "finish_ppi": finish_ppi, "gsm": gsm, "gsm_delta": gsm_delta, "match_count": 0, "status": "pending", }) if target_gsm is not None and not primary_only and expansion_level > INITIAL_NEAREST_N: cases.sort(key=lambda c: (c["gsm_delta"] if c["gsm_delta"] is not None else 9999, c["gsm"])) return cases class PredictionEngine: """Feasibility engine: count-pair search + exact article recommendations.""" def __init__(self, df: pd.DataFrame): self.df = df def _filter_by_count_pair( self, df: pd.DataFrame, norm_warp: float, norm_weft: float, ) -> pd.DataFrame: tol = 0.5 return df[ df["warp_count"].between(norm_warp - tol, norm_warp + tol) & df["weft_count"].between(norm_weft - tol, norm_weft + tol) ] def _search_case( self, df: pd.DataFrame, count_filter: dict | None, weave_filter: str | None, blend_filter: str | None, epi_range: tuple[float, float] | None, ppi_range: tuple[float, float] | None, ) -> pd.DataFrame: work = df.copy() if count_filter: wc = count_filter.get("warp_count") wt = count_filter.get("weft_count") exact_tolerance = count_filter.get("exact_tolerance", 5.0) similar_tolerance = count_filter.get("similar_tolerance", 12.0) similar = count_filter.get("similar", False) if wc is not None: delta = similar_tolerance if similar else exact_tolerance tolerance_pct = 0.12 if similar else 0.075 lo = min(wc - delta, wc * (1 - tolerance_pct)) hi = max(wc + delta, wc * (1 + tolerance_pct)) work = work[work["warp_count"].between(lo, hi)] if wt is not None: delta = similar_tolerance if similar else exact_tolerance tolerance_pct = 0.12 if similar else 0.075 lo = min(wt - delta, wt * (1 - tolerance_pct)) hi = max(wt + delta, wt * (1 + tolerance_pct)) work = work[work["weft_count"].between(lo, hi)] if weave_filter: if weave_filter.startswith("SIMILAR:"): family = weave_filter.split(":", 1)[1] work = work[work["weave"].map(_weave_family) == family] else: work = work[work["weave"] == weave_filter] if blend_filter: if blend_filter.startswith("SIMILAR:"): parts = blend_filter.split(":") family = parts[1] work = work[work["blend"].map(_blend_family) == family] if len(parts) > 2 and parts[2]: target_pct = float(parts[2]) lo = target_pct * 0.85 hi = min(target_pct * 1.15, 100.0) row_pcts = work["blend"].apply( lambda b: _dominant_fiber_pct(b)[1] or 0 ) work = work[row_pcts.between(lo, hi)] else: work = work[work["blend"] == blend_filter] if epi_range and "FINISH EPI" in work.columns: work = work[work["FINISH EPI"].between(epi_range[0], epi_range[1])] if ppi_range and "FINISH PPI" in work.columns: work = work[work["FINISH PPI"].between(ppi_range[0], ppi_range[1])] return work def _filter_rankable_rows(self, df: pd.DataFrame) -> pd.DataFrame: """Keep rows suitable for construction recommendation ranking.""" if df.empty: return df work = df.copy() finish_epi = pd.to_numeric(work.get("FINISH EPI"), errors="coerce") finish_ppi = pd.to_numeric(work.get("FINISH PPI"), errors="coerce") finish_gsm = pd.to_numeric(work.get("FINISH GSM"), errors="coerce") greige_epi = pd.to_numeric(work.get("Greige EPI"), errors="coerce") greige_ppi = pd.to_numeric(work.get("Greige PPI"), errors="coerce") has_finish = finish_epi.notna() | finish_ppi.notna() | finish_gsm.notna() has_greige = greige_epi.notna() & greige_ppi.notna() work = work[has_finish & has_greige] return work def _epi_ppi_range_label( self, finish_epi: float | None, finish_ppi: float | None, ) -> str: if finish_epi is None or finish_ppi is None: return "no-epi-ppi-filter" return _format_epi_ppi_range_label(finish_epi, finish_ppi) def _cascade_search( self, df: pd.DataFrame, norm_wc: float, norm_wtc: float, weave: str, blend: str, ) -> tuple[pd.DataFrame, int, str]: cases = [ { "count": {"warp_count": norm_wc, "weft_count": norm_wtc, "similar": False}, "weave": weave, "blend": blend, "label": "Case-1: Exact Count + Exact Weave + Exact Blend", }, { "count": {"warp_count": norm_wc, "weft_count": norm_wtc, "similar": False}, "weave": f"SIMILAR:{_weave_family(weave)}", "blend": blend, "label": "Case-2: Exact Count + Similar Weave + Exact Blend", }, { "count": {"warp_count": norm_wc, "weft_count": norm_wtc, "similar": True}, "weave": weave, "blend": blend, "label": "Case-3: Similar Count + Exact Weave + Exact Blend", }, { "count": {"warp_count": norm_wc, "weft_count": norm_wtc, "similar": False}, "weave": weave, "blend": f"SIMILAR:{_blend_family(blend)}:{_dominant_fiber_pct(blend)[1] or ''}", "label": "Case-4: Exact Count + Exact Weave + Similar Blend", }, { "count": {"warp_count": norm_wc, "weft_count": norm_wtc, "similar": True}, "weave": f"SIMILAR:{_weave_family(weave)}", "blend": f"SIMILAR:{_blend_family(blend)}:{_dominant_fiber_pct(blend)[1] or ''}", "label": "Case-5: Similar Count + Similar Weave + Similar Blend", }, ] for case_num, case_config in enumerate(cases, start=1): matched = self._search_case( df, case_config["count"], case_config["weave"], case_config["blend"], None, None, ) if len(matched) >= CASE_MIN_ROWS: return matched, case_num, case_config["label"] exact_blend = df[df["blend"].map(lambda b: _blend_matches_strict(str(b), blend))] if len(exact_blend) >= CASE_MIN_ROWS: return exact_blend, 6, "Fallback: Exact Blend Only" family_blend = df[df["blend"].map(lambda b: _blend_matches_family(str(b), blend))] if len(family_blend) >= CASE_MIN_ROWS: return family_blend, 7, "Fallback: Blend Family Only" return pd.DataFrame(columns=df.columns), 8, "Fallback: No strict blend matches" def search_by_count_pair( self, df: pd.DataFrame, norm_warp: float, norm_weft: float, weave: str, blend: str, finish_epi: float, finish_ppi: float, ) -> tuple[pd.DataFrame, int, str, str, int]: """Search archive for one count pair. Returns (matches, case_num, case_label, range_label, raw_count).""" pair_df = self._filter_by_count_pair(df, norm_warp, norm_weft) raw_count = len(pair_df) if raw_count == 0: return pair_df, 0, "", "no-data", 0 base_matches, case_num, case_label = self._cascade_search( pair_df, norm_warp, norm_weft, weave, blend, ) rankable = self._filter_rankable_rows(base_matches) range_label = self._epi_ppi_range_label(finish_epi, finish_ppi) return rankable, case_num, case_label, range_label, raw_count def _absolute_article_score( self, row: pd.Series, target: dict, weights: dict, ) -> float: """Weighted absolute delta score for article ranking (count weight ignored).""" w_epi_ppi = weights.get("epi_ppi", DEFAULT_WEIGHTS["epi_ppi"]) w_gsm = weights.get("gsm", DEFAULT_WEIGHTS["gsm"]) w_epi = w_epi_ppi / 2 w_ppi = w_epi_ppi / 2 score = 0.0 target_epi = target.get("finish_epi") target_ppi = target.get("finish_ppi") target_gsm = target.get("target_gsm") rv_epi = _safe_float(row.get("FINISH EPI")) rv_ppi = _safe_float(row.get("FINISH PPI")) rv_gsm = _safe_float(row.get("FINISH GSM")) if target_epi is not None: if rv_epi is None: score += w_epi * EPI_PPI_WINDOW else: score += w_epi * abs(target_epi - rv_epi) if target_ppi is not None: if rv_ppi is None: score += w_ppi * EPI_PPI_WINDOW else: score += w_ppi * abs(target_ppi - rv_ppi) if target_gsm is not None: if rv_gsm is None: score += w_gsm * MISSING_GSM_PENALTY else: score += w_gsm * abs(target_gsm - rv_gsm) return score def _distance_score(self, row: pd.Series, target: dict, weights: dict) -> float: score = 0.0 total_weight = 0.0 w_count = weights.get("count", DEFAULT_WEIGHTS["count"]) w_epi_ppi = weights.get("epi_ppi", DEFAULT_WEIGHTS["epi_ppi"]) w_gsm = weights.get("gsm", DEFAULT_WEIGHTS["gsm"]) for key, col, w in [ ("norm_warp", "warp_count", w_count / 2), ("norm_weft", "weft_count", w_count / 2), ("finish_epi", "FINISH EPI", w_epi_ppi / 2), ("finish_ppi", "FINISH PPI", w_epi_ppi / 2), ("target_gsm", "FINISH GSM", w_gsm), ]: tv = target.get(key) if tv is None: continue rv = _safe_float(row.get(col)) if rv is None: score += w * 0.5 else: denom = max(abs(tv), 1.0) score += w * abs(tv - rv) / denom total_weight += w if total_weight == 0: return 0.0 return score / total_weight * 100 def _build_construction(self, row: pd.Series) -> dict: def g(col: str): return _safe_float(row.get(col)) return { "warp_count": g("warp_count"), "weft_count": g("weft_count"), "greige_epi": g("Greige EPI"), "greige_ppi": g("Greige PPI"), "reed_count": g("Reed Count"), "ends_per_dent": g("Ends per dent"), "reed_space": g("Reed space"), "greige_width": g("Greige Width in INCH"), "finish_epi": g("FINISH EPI"), "finish_ppi": g("FINISH PPI"), "finish_gsm": g("FINISH GSM"), "finish_width": g("FINISH WIDTH"), "on_loom_epi": g("ON LOOM EPI"), "on_loom_ppi": g("ON LOOM PPI"), "loom_type": _clean_text(row.get("loom_type")), "weave": _clean_text(row.get("weave")), "blend": _clean_text(row.get("blend")), } def _build_article_match( self, row: pd.Series, rank: int, score: float, ) -> dict: long_desc = _clean_text(row.get("article")) or _clean_text(row.get("variant")) return { "rank": rank, "master_article": _clean_text(row.get("master_article")), "article": _clean_text(row.get("article")), "long_description": long_desc, "score": round(score, 3), "construction": self._build_construction(row), } def rank_articles( self, matches: pd.DataFrame, target: dict, weights: dict, top_n: int = MAX_MATCHES_RETURN, ) -> list[dict]: if matches.empty: return [] scored = matches.copy() scored["_score"] = scored.apply( lambda r: self._absolute_article_score(r, target, weights), axis=1, ) best_by_construction: dict[str, tuple[pd.Series, float]] = {} for _, row in scored.iterrows(): master = _clean_text(row.get("master_article")) if not master: continue fp = _construction_fingerprint(row) score = float(row["_score"]) if fp not in best_by_construction or score < best_by_construction[fp][1]: best_by_construction[fp] = (row, score) construction_deduped = sorted(best_by_construction.values(), key=lambda x: x[1]) best_by_family: dict[str, tuple[pd.Series, float]] = {} for row, score in construction_deduped: family = _article_family_key(_clean_text(row.get("master_article"))) if family not in best_by_family or score < best_by_family[family][1]: best_by_family[family] = (row, score) deduped = sorted(best_by_family.values(), key=lambda x: x[1])[:top_n] results = [] for i, (row, score) in enumerate(deduped, start=1): results.append(self._build_article_match(row, i, score)) return results def rank_dataset( self, matches: pd.DataFrame, target: dict, weights: dict, top_n: int = MAX_MATCHES_RETURN, blend: str | None = None, ) -> tuple[list[dict], list[dict]]: """Return top matches and the full scored dataset for the active count pair.""" if matches.empty: return [], [] scored = matches.copy() if blend: exact_mask = scored["blend"].map(lambda b: _blend_matches_strict(str(b), blend)) if exact_mask.any(): scored = scored[exact_mask] else: family_mask = scored["blend"].map(lambda b: _blend_matches_family(str(b), blend)) if family_mask.any(): scored = scored[family_mask] if scored.empty: return [], [] scored["_score"] = scored.apply( lambda r: self._absolute_article_score(r, target, weights), axis=1, ) scored = scored.sort_values("_score", ascending=True) top_matches = self.rank_articles(scored, target, weights, top_n=top_n) recommended_fingerprints: set[str] = set() for m in top_matches: mask = scored["master_article"].astype(str).str.strip() == m["master_article"] if mask.any(): recommended_fingerprints.add( _construction_fingerprint(scored.loc[mask].iloc[0]), ) dataset: list[dict] = [] seen_masters: set[str] = set() for _, row in scored.iterrows(): master = _clean_text(row.get("master_article")) if not master or master in seen_masters: continue seen_masters.add(master) score = float(row["_score"]) entry = self._build_article_match(row, 0, score) entry["recommended"] = _construction_fingerprint(row) in recommended_fingerprints dataset.append(entry) for i, m in enumerate(top_matches, start=1): m["rank"] = i return top_matches, dataset def _apply_article_selection( self, matches_list: list[dict], dataset_articles: list[dict], selected_master: str, top_n: int = MAX_MATCHES_RETURN, ) -> tuple[list[dict], list[dict], bool]: """Promote a user-chosen archive article to the primary match.""" selected_master = _clean_text(selected_master) if not selected_master or not dataset_articles: return matches_list, dataset_articles, False selected_entry = next( (a for a in dataset_articles if a.get("master_article") == selected_master), None, ) if not selected_entry: return matches_list, dataset_articles, False others = [m for m in matches_list if m.get("master_article") != selected_master] new_matches: list[dict] = [{**selected_entry, "rank": 1}] for i, m in enumerate(others[: max(top_n - 1, 0)], start=2): new_matches.append({**m, "rank": i}) new_dataset: list[dict] = [] for row in dataset_articles: entry = {**row} entry["selected"] = row.get("master_article") == selected_master new_dataset.append(entry) return new_matches, new_dataset, True def _expand_and_search( self, df: pd.DataFrame, warp_in: float, weft_in: float, finish_epi: float, finish_ppi: float, target_gsm: float | None, weave: str, blend: str, ) -> tuple[list[dict], dict | None, pd.DataFrame, dict]: """Expand count-pair cases; pick GSM-closest pair that has sufficient archive matches.""" all_cases: list[dict] = [] evaluated: set[tuple[float, float]] = set() best: tuple[dict, pd.DataFrame, dict] | None = None search_meta: dict = {"expansion_level": INITIAL_NEAREST_N, "stopped_reason": "no_matches"} sr_counter = 0 def _gsm_delta(case: dict) -> float: d = case.get("gsm_delta") return d if d is not None else 9999.0 for level in range(INITIAL_NEAREST_N, MAX_EXPANSION_LEVEL + 1): expansion_tier = level - INITIAL_NEAREST_N + 1 batch = generate_count_pair_cases( warp_in, weft_in, finish_epi, finish_ppi, target_gsm, level, ) search_meta["expansion_level"] = level pending = [ c for c in batch if (c["warp_count"], c["weft_count"]) not in evaluated ] if not pending and level > INITIAL_NEAREST_N: break for case in pending: key = (case["warp_count"], case["weft_count"]) evaluated.add(key) sr_counter += 1 case["sr"] = sr_counter case["expansion_tier"] = expansion_tier case["is_primary"] = expansion_tier == 1 matches, case_num, case_label, range_label, raw_count = ( self.search_by_count_pair( df, case["warp_count"], case["weft_count"], weave, blend, finish_epi, finish_ppi, ) ) usable = len(matches) >= CASE_MIN_ROWS case["match_count"] = int(len(matches)) case["raw_count_in_archive"] = raw_count case["status"] = ( "no_data" if raw_count == 0 else ("searched" if usable else "insufficient") ) case["cascade_case"] = case_label if usable else None all_cases.append(case) if usable: meta = { "case_number": case_num, "cascade_case": case_label, "range": range_label, "matches_found": int(len(matches)), } case["case_number"] = case_num case["_matches"] = matches case["_meta"] = meta if best is None or _gsm_delta(case) < _gsm_delta(best[0]): best = (case, matches, meta) if best is not None: next_level = level + 1 if next_level <= MAX_EXPANSION_LEVEL: next_batch = generate_count_pair_cases( warp_in, weft_in, finish_epi, finish_ppi, target_gsm, next_level, ) unseen = [ c for c in next_batch if (c["warp_count"], c["weft_count"]) not in evaluated ] if unseen and min(_gsm_delta(c) for c in unseen) < _gsm_delta(best[0]): continue search_meta["stopped_reason"] = "sufficient_matches" break if level >= MAX_EXPANSION_LEVEL: search_meta["stopped_reason"] = "max_expansion" if best else "no_matches" break if best is None: return all_cases, None, pd.DataFrame(), search_meta usable = [c for c in all_cases if c.get("match_count", 0) >= CASE_MIN_ROWS] primary_usable = [c for c in usable if c.get("is_primary")] pool = primary_usable if primary_usable else usable if pool: best_delta = min(_gsm_delta(c) for c in pool) gsm_tol = max(5.0, (target_gsm or 0) * 0.05) band = [c for c in pool if _gsm_delta(c) <= best_delta + gsm_tol] chosen = min( band, key=lambda c: ( c.get("case_number", 999), c.get("gsm_delta") if c.get("gsm_delta") is not None else 9999, -c.get("match_count", 0), c.get("sr", 0), ), ) best = (chosen, chosen["_matches"], chosen["_meta"]) for c in all_cases: c.pop("_matches", None) c.pop("_meta", None) best_case, best_matches, meta = best for c in all_cases: if c["status"] == "selected": c["status"] = "searched" best_case["status"] = "selected" best_result = { "warp_count": best_case["warp_count"], "weft_count": best_case["weft_count"], "gsm": best_case["gsm"], **meta, } all_cases.sort( key=lambda c: ( 0 if c.get("is_primary") else 1, c.get("sr", 0) if c.get("is_primary") else 0, c.get("gsm_delta") if c.get("gsm_delta") is not None else 9999, c.get("sr", 0), ) ) return all_cases, best_result, best_matches.copy(), search_meta def _activate_count_pair( self, df: pd.DataFrame, count_cases: list[dict], warp: float, weft: float, finish_epi: float, finish_ppi: float, target_gsm: float | None, weave: str, blend: str, ) -> tuple[dict | None, pd.DataFrame]: """Apply a user-selected warp×weft count pair and return active metadata + matches.""" case = _mark_selected_count_case(count_cases, warp, weft) gsm = case["gsm"] if case else round(_compute_gsm(finish_epi, finish_ppi, warp, weft), 1) matches, case_num, case_label, range_label, raw_count = self.search_by_count_pair( df, warp, weft, weave, blend, finish_epi, finish_ppi, ) if case: case["match_count"] = int(len(matches)) case["raw_count_in_archive"] = raw_count case["cascade_case"] = case_label or case.get("cascade_case") active_pair = { "warp_count": warp, "weft_count": weft, "gsm": gsm, "case_number": case_num, "cascade_case": case_label, "range": range_label, "matches_found": int(len(matches)), "user_selected": True, } return active_pair, matches def _preview_archive_matches( self, df: pd.DataFrame, weave: str, blend: str, finish_epi: float, finish_ppi: float, warp_in: float, weft_in: float, limit: int = 25, epi_band: float | None = None, ) -> list[dict]: """Lightweight archive sample for finish-spec refinement before GSM matrix.""" band = float(epi_band if epi_band is not None else 15.0) base = df.copy() base = base[base["weave"].map(_weave_family) == _weave_family(weave)] base = base[base["blend"].map(_blend_family) == _blend_family(blend)] base = base[ base["warp_count"].between(warp_in - 15, warp_in + 15) & base["weft_count"].between(weft_in - 15, weft_in + 15) ] if base.empty: base = df[ df["warp_count"].between(warp_in - 15, warp_in + 15) & df["weft_count"].between(weft_in - 15, weft_in + 15) ] if base.empty: return [] near_epi = base[base["FINISH EPI"].between(finish_epi - band, finish_epi + band)] if near_epi.empty: near_epi = base scored = near_epi.copy() scored["_score"] = scored["FINISH EPI"].apply( lambda v: abs(finish_epi - (_safe_float(v) or finish_epi)), ) if epi_band is None: scored["_score"] = scored.apply( lambda r: abs(finish_epi - (_safe_float(r.get("FINISH EPI")) or finish_epi)) + abs(finish_ppi - (_safe_float(r.get("FINISH PPI")) or finish_ppi)), axis=1, ) scored = scored.sort_values("_score") selected_idx: set[int] = set() preview: list[dict] = [] def append_rows(frame: pd.DataFrame, cap: int) -> None: for idx, row in frame.head(cap).iterrows(): if idx in selected_idx: continue selected_idx.add(idx) score = row["_score"] if "_score" in row.index else 0.0 preview.append({ "master_article": _clean_text(row.get("master_article")), "score": float(score), "construction": self._build_construction(row), }) append_rows(scored, max(limit // 2, 10)) below_cluster = base[ base["FINISH EPI"].between(finish_epi - 8, finish_epi - 4) ].copy() if not below_cluster.empty: below_cluster["_score"] = abs(below_cluster["FINISH EPI"] - (finish_epi - 6)) below_cluster = below_cluster.sort_values("FINISH EPI", ascending=False) append_rows(below_cluster, min(8, limit)) ppi_leaders = near_epi.sort_values("FINISH PPI", ascending=False).drop_duplicates( subset=["FINISH PPI"], keep="first", ) if not ppi_leaders.empty and "_score" not in ppi_leaders.columns: ppi_leaders = ppi_leaders.copy() ppi_leaders["_score"] = ppi_leaders["FINISH EPI"].apply( lambda v: abs(finish_epi - (_safe_float(v) or finish_epi)), ) append_rows(ppi_leaders, min(8, limit)) if len(preview) < limit: append_rows(scored, limit - len(preview)) return preview[:limit] def predict(self, payload: dict) -> dict: warp_in = _safe_float(payload.get("warp_count")) weft_in = _safe_float(payload.get("weft_count")) finish_epi = _safe_float(payload.get("finish_epi")) finish_ppi = _safe_float(payload.get("finish_ppi")) target_gsm = _safe_float(payload.get("target_gsm")) weave_in = _normalize_weave(payload.get("weave", "")) blend_in = _normalize_blend(payload.get("blend", "")) dataset = _clean_text(payload.get("dataset", "all")).lower() loom_type_in = _clean_text(payload.get("loom_type", "")) weights = payload.get("weights", dict(**DEFAULT_WEIGHTS)) weave_type = _clean_text(payload.get("weave_type", "standard")) if not weave_in or not blend_in: return {"error": "weave and blend are required", "required_fields": ["weave", "blend"]} if finish_epi is None or finish_ppi is None: return {"error": "finish_epi and finish_ppi are required"} if warp_in is None or weft_in is None: return {"error": "warp_count and weft_count are required"} df = self.df.copy() if dataset in {"working", "piece_dyed"}: df = df[df["dataset"] == dataset] if loom_type_in: lt_mask = df["loom_type"].str.lower() == loom_type_in.lower() if lt_mask.sum() >= 30: df = df[lt_mask] preview_matches = self._preview_archive_matches( df, weave_in, blend_in, finish_epi, finish_ppi, warp_in, weft_in, epi_band=FINISH_EPI_BAND, ) preview_rows = extract_archive_rows(preview_matches, max_samples=25) count_cases, active_pair, best_matches, expansion_meta = self._expand_and_search( df, warp_in, weft_in, finish_epi, finish_ppi, target_gsm, weave_in, blend_in, ) sel_warp = _safe_float(payload.get("selected_warp_count")) sel_weft = _safe_float(payload.get("selected_weft_count")) user_selected = False if sel_warp is not None and sel_weft is not None: active_pair, best_matches = self._activate_count_pair( df, count_cases, sel_warp, sel_weft, finish_epi, finish_ppi, target_gsm, weave_in, blend_in, ) user_selected = True if active_pair: active_pair["user_selected"] = True act_warp = float(active_pair["warp_count"]) if active_pair else warp_in act_weft = float(active_pair["weft_count"]) if active_pair else weft_in primary_pairs = [ (float(c["warp_count"]), float(c["weft_count"])) for c in count_cases if c.get("is_primary") ][:4] target = { "norm_warp": act_warp, "norm_weft": act_weft, "finish_epi": finish_epi, "finish_ppi": finish_ppi, "target_gsm": target_gsm, } matches_list, dataset_articles = self.rank_dataset( best_matches, target, weights, blend=blend_in, ) merged_rows = merge_archive_rows( preview_rows, extract_archive_rows(matches_list, dataset_articles, max_samples=25), ) refined = refine_finish_specs( user_finish_epi=finish_epi, user_finish_ppi=finish_ppi, target_gsm=target_gsm, warp_count=act_warp, weft_count=act_weft, archive_rows=merged_rows, weave=weave_in, primary_count_pairs=primary_pairs, ) gsm_finish_epi = refined["finish_epi"] gsm_finish_ppi = refined["finish_ppi"] for c in count_cases: c["finish_epi"] = gsm_finish_epi c["finish_ppi"] = gsm_finish_ppi c["gsm"] = round( _compute_gsm(gsm_finish_epi, gsm_finish_ppi, c["warp_count"], c["weft_count"]), 1, ) if target_gsm is not None: c["gsm_delta"] = abs(c["gsm"] - target_gsm) if active_pair: active_pair["finish_epi"] = gsm_finish_epi active_pair["finish_ppi"] = gsm_finish_ppi active_pair["gsm"] = round( _compute_gsm(gsm_finish_epi, gsm_finish_ppi, act_warp, act_weft), 1, ) if target_gsm is not None: active_pair["gsm_delta"] = abs(active_pair["gsm"] - target_gsm) target["finish_epi"] = gsm_finish_epi target["finish_ppi"] = gsm_finish_ppi matches_list, dataset_articles = self.rank_dataset( best_matches, target, weights, blend=blend_in, ) auto_primary = matches_list[0]["master_article"] if matches_list else None selected_master = _clean_text(payload.get("selected_master_article")) user_selected_article = False if selected_master: matches_list, dataset_articles, user_selected_article = self._apply_article_selection( matches_list, dataset_articles, selected_master, ) ai_suggestion = None if active_pair: from .construction_suggester import suggest_construction ai_matches = matches_list if not ai_matches and merged_rows: ai_matches = [ { "master_article": row["master_article"], "score": 50.0, "construction": { "finish_epi": row["finish_epi"], "finish_ppi": row["finish_ppi"], "greige_epi": row["greige_epi"], "greige_ppi": row["greige_ppi"], "on_loom_epi": row.get("on_loom_epi"), "on_loom_ppi": row.get("on_loom_ppi"), "reed_count": row.get("reed_count"), "ends_per_dent": row.get("ends_per_dent"), }, } for row in merged_rows ] if ai_matches: ai_suggestion = suggest_construction( user_finish_epi=finish_epi, user_finish_ppi=finish_ppi, target_gsm=target_gsm, weave=weave_in, warp_count=float(active_pair["warp_count"]), weft_count=float(active_pair["weft_count"]), matches=ai_matches, dataset_articles=dataset_articles, df=self.df, loom_type=loom_type_in or None, refined_finish_epi=gsm_finish_epi, refined_finish_ppi=gsm_finish_ppi, finish_spec_adjustments=refined.get("adjustments"), ) primary = matches_list[0] if matches_list else None recommendation = primary["construction"] if primary else {} if recommendation: recommendation = {**recommendation} recommendation["finish_epi"] = finish_epi recommendation["finish_ppi"] = finish_ppi reed_result = None greige_epi = recommendation.get("greige_epi") if recommendation else None if greige_epi and greige_epi > 0: reed_result = recommend_reed(float(greige_epi), self.df, loom_type_in) best_score = primary["score"] if primary else 999.0 rankable_count = len(best_matches) confidence = _confidence(rankable_count, best_score) search_instructions = "" if active_pair: nw, nt = active_pair["warp_count"], active_pair["weft_count"] search_instructions = ( f"Search: {nw}COM×{nt}COM, " f"EPI({round(finish_epi) - 5}-{round(finish_epi) + 5})" f"×PPI({round(finish_ppi) - 5}-{round(finish_ppi) + 5}), " f"{weave_in}, {blend_in}" ) ref_cols = [ "master_article", "article", "weave", "blend", "Greige EPI", "Greige PPI", "FINISH EPI", "FINISH PPI", "FINISH GSM", "Reed Count", "Reed space", "Ends per dent", "FINISH WIDTH", "Greige Width in INCH", ] reference_articles = [] for m in matches_list: ref = {"master_article": m["master_article"], "_score": m["score"]} for k, col in [ ("weave", "weave"), ("blend", "blend"), ("Greige EPI", "greige_epi"), ("Greige PPI", "greige_ppi"), ("FINISH EPI", "finish_epi"), ("FINISH PPI", "finish_ppi"), ("FINISH GSM", "finish_gsm"), ("Reed Count", "reed_count"), ("Reed space", "reed_space"), ("Ends per dent", "ends_per_dent"), ("FINISH WIDTH", "finish_width"), ]: ref[col if col in ref_cols else k] = m["construction"].get(k.replace(" ", "_").lower().replace("finish_", "finish_").replace("greige_", "greige_")) c = m["construction"] reference_articles.append({ "master_article": m["master_article"], "article": m["article"], "weave": c.get("weave"), "blend": c.get("blend"), "Greige EPI": c.get("greige_epi"), "Greige PPI": c.get("greige_ppi"), "FINISH EPI": c.get("finish_epi"), "FINISH PPI": c.get("finish_ppi"), "FINISH GSM": c.get("finish_gsm"), "Reed Count": c.get("reed_count"), "Reed space": c.get("reed_space"), "Ends per dent": c.get("ends_per_dent"), "FINISH WIDTH": c.get("finish_width"), "Greige Width in INCH": c.get("greige_width"), "_score": m["score"], }) return { "input": { "raw_warp_count": warp_in, "raw_weft_count": weft_in, "finish_epi": finish_epi, "finish_ppi": finish_ppi, "weave": weave_in, "blend": blend_in, "target_gsm": target_gsm, "loom_type": loom_type_in, "dataset": dataset, "weave_type": weave_type, }, "count_cases": count_cases, "primary_count_cases": [c for c in count_cases if c.get("is_primary")], "expanded_count_cases": [c for c in count_cases if not c.get("is_primary")], "expansion_level": expansion_meta.get("expansion_level", INITIAL_NEAREST_N), "active_count_pair": active_pair, "user_selected_pair": user_selected, "search_path": { "active_case": active_pair.get("cascade_case", "") if active_pair else "", "case_number": active_pair.get("case_number", 0) if active_pair else 0, "range": active_pair.get("range", "") if active_pair else "", "matches_found": active_pair.get("matches_found", 0) if active_pair else 0, "search_instructions": search_instructions, "stopped_reason": expansion_meta.get("stopped_reason", ""), }, "matches": matches_list, "dataset_articles": dataset_articles, "auto_primary_article": auto_primary, "selected_master_article": ( matches_list[0]["master_article"] if user_selected_article and matches_list else None ), "user_selected_article": user_selected_article, "recommendation": recommendation, "reed_recommendation": reed_result, "data_quality": { "matching_source": active_pair.get("cascade_case", "") if active_pair else "none", "matches_analyzed": len(matches_list), "confidence": confidence, }, "reference_articles": reference_articles, "gsm_analysis": { "count_cases": count_cases, "target_gsm": target_gsm, }, "refined_finish_specs": { "finish_epi": gsm_finish_epi, "finish_ppi": gsm_finish_ppi, "adjustments": refined.get("adjustments", []), }, "ai_suggestion": ai_suggestion, }