Spaces:
Sleeping
Sleeping
| """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, | |
| } | |