from functools import lru_cache import os import re from typing import Any import numpy as np import pandas as pd from .count_converter import parse_count_string as _parse_count_string from .prediction_engine import PredictionEngine _DATA_DIR = os.path.join( os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data" ) WORKING_PATH = os.path.join(_DATA_DIR, "Working 24 Jan.xlsx") PIECE_PATH = os.path.join(_DATA_DIR, "Piece Dyed Artcle details.xlsx") 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 Exception: return None def _extract_count(value: Any) -> float | None: parsed = _parse_count_string(str(value)) if parsed is None: return None return round(float(parsed.get("ne", 0)), 3) 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) text = text.strip() return text 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 _count_band(value: float | None) -> str: if value is None: return "unknown" return "below_40" if value < 40 else "40_and_above" class DataService: def __init__(self) -> None: self.df: pd.DataFrame | None = None self.is_loaded = False self.last_error = "" self.raw_working_rows = 0 self.raw_piece_rows = 0 self.used_working_rows = 0 self.used_piece_rows = 0 self._prediction_engine: PredictionEngine | None = None @property def prediction_engine(self) -> PredictionEngine: if self._prediction_engine is None: self._prediction_engine = PredictionEngine(self._ensure()) return self._prediction_engine def load_data(self) -> None: try: working = pd.read_excel(WORKING_PATH) piece = pd.read_excel(PIECE_PATH) self.raw_working_rows = int(len(working)) self.raw_piece_rows = int(len(piece)) working["dataset"] = "working" piece["dataset"] = "piece_dyed" if "Loom Type" not in working.columns: working["Loom Type"] = "" cols = sorted(set(working.columns).union(set(piece.columns))) working = working.reindex(columns=cols) piece = piece.reindex(columns=cols) df = pd.concat([working, piece], ignore_index=True) # Canonical fields df["article"] = df["Article"].astype(str) df["master_article"] = df["Master Article"].astype(str) df["variant"] = df["Options/Variants"].astype(str) df["loom_type"] = df["Loom Type"].fillna("").astype(str) df["weave_raw"] = df["WEAVE"].fillna("").astype(str) df["weave"] = df["weave_raw"].map(_normalize_weave) df["blend_raw"] = df["Blend Ratio"].fillna("").astype(str) df["blend"] = df["blend_raw"].map(_normalize_blend) # Remove garbage template/placeholder rows (Model Input/Output) garbage_mask = df["blend_raw"].str.contains("model", case=False, na=False) garbage_mask |= df["blend_raw"].str.contains("intput", case=False, na=False) if garbage_mask.any(): df = df[~garbage_mask] for col in [ "Reed Count", "Ends per dent", "Reed space", "Greige EPI", "Greige PPI", "Greige Width in INCH", "FINISH EPI", "FINISH PPI", "FINISH GSM", "FINISH WIDTH", "ON LOOM EPI", "ON LOOM PPI", ]: df[col] = pd.to_numeric(df[col], errors="coerce") for col in [ "FINISH EPI", "FINISH PPI", "FINISH GSM", "FINISH WIDTH", "Reed space", "Greige Width in INCH", "ON LOOM EPI", "ON LOOM PPI", ]: df.loc[df[col] == 0, col] = np.nan df["warp_count"] = df[" Warp Code 1"].map(_extract_count) df["weft_count"] = df[" Weft Code 1"].map(_extract_count) df["count_ref"] = df[["warp_count", "weft_count"]].max(axis=1) df["count_band"] = df["count_ref"].map(_count_band) df["epi_change_pct"] = np.where( df["Greige EPI"] > 0, ((df["FINISH EPI"] - df["Greige EPI"]) / df["Greige EPI"]) * 100, np.nan, ) df["ppi_change_pct"] = np.where( df["Greige PPI"] > 0, ((df["FINISH PPI"] - df["Greige PPI"]) / df["Greige PPI"]) * 100, np.nan, ) df["width_change_pct"] = np.where( df["Greige Width in INCH"] > 0, ( (df["Greige Width in INCH"] - df["FINISH WIDTH"]) / df["Greige Width in INCH"] ) * 100, np.nan, ) # Keep valid rows with essential fields df = df[(df["weave"].str.len() > 0) & (df["blend"].str.len() > 0)] self.used_working_rows = int((df["dataset"] == "working").sum()) self.used_piece_rows = int((df["dataset"] == "piece_dyed").sum()) self.df = df self.is_loaded = True self.last_error = "" except Exception as exc: self.last_error = str(exc) self.is_loaded = False raise def _ensure(self) -> pd.DataFrame: if not self.is_loaded or self.df is None: self.load_data() return self.df def get_health(self) -> dict: df = self._ensure() return { "status": "healthy" if self.is_loaded else "degraded", "rows": int(len(df)), "error": self.last_error, } @lru_cache(maxsize=1) def get_dashboard_summary(self) -> dict: df = self._ensure() top_weaves = ( df["weave"] .value_counts() .head(10) .rename_axis("weave") .reset_index(name="count") ) top_blends = ( df["blend"] .value_counts() .head(10) .rename_axis("blend") .reset_index(name="count") ) return { "totals": { "rows": int(len(df)), "unique_articles": int(df["article"].nunique()), "unique_master_articles": int(df["master_article"].nunique()), "unique_weaves": int(df["weave"].nunique()), "unique_blends": int(df["blend"].nunique()), }, "quality": { "finish_ppi_missing_pct": round( float(df["FINISH PPI"].isna().mean() * 100), 2 ), "loom_type_missing_pct": round( float(df["loom_type"].eq("").mean() * 100), 2 ), }, "relativity": { "median_epi_change_pct": round( float(df["epi_change_pct"].median(skipna=True)), 3 ), "median_ppi_change_pct": round( float(df["ppi_change_pct"].median(skipna=True)), 3 ), "median_width_change_pct": round( float(df["width_change_pct"].median(skipna=True)), 3 ), }, "top_weaves": top_weaves.to_dict(orient="records"), "top_blends": top_blends.to_dict(orient="records"), } @lru_cache(maxsize=1) def get_filters(self) -> dict: df = self._ensure() return { "weaves": sorted(df["weave"].dropna().unique().tolist())[:500], "blends": sorted(df["blend"].dropna().unique().tolist())[:500], "loom_types": sorted( [ x for x in df["loom_type"].dropna().unique().tolist() if str(x).strip() ] ), "count_bands": ["unknown", "below_40", "40_and_above"], "datasets": ["all", "working", "piece_dyed"], } def get_articles( self, page: int, limit: int, search: str, weave: str, blend: str, loom_type: str, dataset: str, count_band: str, ) -> dict: df = self._ensure() if search: s = search.lower().strip() df = df[ df["master_article"].str.lower().str.contains(s, na=False) | df["article"].str.lower().str.contains(s, na=False) | df["variant"].str.lower().str.contains(s, na=False) | df["weave"].str.lower().str.contains(s, na=False) | df["blend"].str.lower().str.contains(s, na=False) ] if weave: df = df[df["weave"] == weave] if blend: df = df[df["blend"] == blend] if loom_type: df = df[df["loom_type"] == loom_type] if dataset in {"working", "piece_dyed"}: df = df[df["dataset"] == dataset] if count_band in {"unknown", "below_40", "40_and_above"}: df = df[df["count_band"] == count_band] grouped = ( df.groupby("master_article", dropna=False) .agg( article=("article", "first"), variant=("variant", "first"), weave=("weave", "first"), blend=("blend", "first"), warp_count=("warp_count", "max"), weft_count=("weft_count", "max"), loom_type=("loom_type", "first"), rows=("master_article", "count"), reed_count=("Reed Count", "median"), ends_per_dent=("Ends per dent", "median"), reed_space=("Reed space", "median"), finish_gsm_median=("FINISH GSM", "median"), finish_epi_median=("FINISH EPI", "median"), finish_ppi_median=("FINISH PPI", "median"), greige_epi_median=("Greige EPI", "median"), greige_ppi_median=("Greige PPI", "median"), ) .reset_index() .sort_values("rows", ascending=False) ) total = int(len(grouped)) start = (page - 1) * limit end = start + limit chunk = grouped.iloc[start:end].copy() for col in [ "finish_gsm_median", "finish_epi_median", "finish_ppi_median", "greige_epi_median", "greige_ppi_median", ]: chunk[col] = chunk[col].round(3) return { "data": chunk.to_dict(orient="records"), "page": page, "limit": limit, "total": total, "total_pages": int(np.ceil(total / limit)) if total else 0, } def get_article_detail(self, master_article: str) -> dict: df = self._ensure() view = df[df["master_article"].astype(str) == str(master_article)].copy() if view.empty: return {"error": "Article not found"} head = view.iloc[0] stats = { "rows": int(len(view)), "finish_epi": round(float(view["FINISH EPI"].median(skipna=True)), 3), "finish_ppi": round(float(view["FINISH PPI"].median(skipna=True)), 3), "finish_gsm": round(float(view["FINISH GSM"].median(skipna=True)), 3), "greige_epi": round(float(view["Greige EPI"].median(skipna=True)), 3), "greige_ppi": round(float(view["Greige PPI"].median(skipna=True)), 3), "epi_change_pct": round( float(view["epi_change_pct"].median(skipna=True)), 3 ), "ppi_change_pct": round( float(view["ppi_change_pct"].median(skipna=True)), 3 ), "width_change_pct": round( float(view["width_change_pct"].median(skipna=True)), 3 ), } samples = view[ [ "dataset", "article", "variant", "weave", "blend", "Greige EPI", "Greige PPI", "FINISH EPI", "FINISH PPI", "FINISH GSM", "FINISH WIDTH", ] ].head(20) return { "master_article": str(master_article), "profile": { "article": str(head.get("article", "")), "variant": str(head.get("variant", "")), "weave": str(head.get("weave", "")), "blend": str(head.get("blend", "")), "loom_type": str(head.get("loom_type", "")), }, "stats": stats, "samples": samples.to_dict(orient="records"), } def _safe_cv(self, series: pd.Series) -> float: series = pd.to_numeric(series, errors="coerce").dropna() if len(series) < 2: return 0.0 mean = float(series.mean().item()) if abs(mean) < 1e-9: return 0.0 std = float(series.std(ddof=1).item()) return float(abs(std / mean) * 100) def _field_stats(self, series: pd.Series) -> dict: s = pd.to_numeric(series, errors="coerce").dropna() if s.empty: return { "p25": None, "median": None, "p75": None, "mean": None, "std": None, "cv": None, "used_count": 0, "raw_count": 0, } raw_count = len(s) if len(s) >= 8: q1 = s.quantile(0.25).item() q3 = s.quantile(0.75).item() iqr = q3 - q1 lower = q1 - (1.5 * iqr) upper = q3 + (1.5 * iqr) s = s[(s >= lower) & (s <= upper)] if s.empty: s = pd.to_numeric(series, errors="coerce").dropna() return { "p25": round(float(s.quantile(0.25).item()), 3), "median": round(float(s.median().item()), 3), "p75": round(float(s.quantile(0.75).item()), 3), "mean": round(float(s.mean().item()), 3), "std": round(float(s.std(ddof=1).item()) if len(s) > 1 else 0.0, 3), "cv": round(float(self._safe_cv(s)), 3), "used_count": int(len(s)), "raw_count": int(raw_count), } def predict_construction(self, payload: dict) -> dict: """Construction prediction using the new prediction engine. Delegates to PredictionEngine which implements: Step 1: GSM count-pair matrix (primary + expanded search) Step 2: Per-pair 5-case cascade search Step 3: Exact article ranking (top 2–3, no median aggregation) Step 4: Reed alternatives via historical frequency Step 5: Optional user count-pair override from GSM table """ return self.prediction_engine.predict(payload) @lru_cache(maxsize=4) def get_validation_report(self, sample_size: int = 250, seed: int = 42) -> dict: df = self._ensure().dropna( subset=[ "weave", "blend", "Greige EPI", "Greige PPI", "FINISH EPI", "FINISH PPI", ] ) if df.empty: return {"error": "No valid rows for validation"} n = int(min(max(30, sample_size), len(df))) sample = df.sample(n=n, random_state=seed) abs_epi = [] abs_ppi = [] conf = {"high": 0, "medium": 0, "low": 0, "very_low": 0} for _, row in sample.iterrows(): pred = self.predict_construction( { "weave": row["weave"], "blend": row["blend"], "warp_count": float(row["warp_count"]) if pd.notna(row.get("warp_count")) else None, "weft_count": float(row["weft_count"]) if pd.notna(row.get("weft_count")) else None, "finish_epi": float(row["FINISH EPI"]), "finish_ppi": float(row["FINISH PPI"]), "target_gsm": float(row["FINISH GSM"]) if pd.notna(row["FINISH GSM"]) else None, "loom_type": row.get("loom_type", ""), } ) rec = pred.get("recommendation", {}) if rec.get("greige_epi") is None or rec.get("greige_ppi") is None: continue abs_epi.append(abs(float(rec["greige_epi"]) - float(row["Greige EPI"]))) abs_ppi.append(abs(float(rec["greige_ppi"]) - float(row["Greige PPI"]))) c = pred.get("data_quality", {}).get("confidence", "very_low") conf[c] = conf.get(c, 0) + 1 if not abs_epi or not abs_ppi: return {"error": "Validation failed to produce comparable predictions"} def _p(values: list[float], q: float) -> float: vals = sorted(values) idx = max(0, min(len(vals) - 1, int(round((len(vals) - 1) * q)))) return float(vals[idx]) return { "sample_size": n, "scored_rows": len(abs_epi), "mae": { "greige_epi": round(float(np.mean(abs_epi)), 3), "greige_ppi": round(float(np.mean(abs_ppi)), 3), }, "percentiles_abs_error": { "greige_epi_p50": round(_p(abs_epi, 0.50), 3), "greige_epi_p90": round(_p(abs_epi, 0.90), 3), "greige_ppi_p50": round(_p(abs_ppi, 0.50), 3), "greige_ppi_p90": round(_p(abs_ppi, 0.90), 3), }, "confidence_distribution": conf, } @lru_cache(maxsize=1) def get_relativity_analytics(self) -> dict: df = self._ensure() weave_stats = ( df.groupby("weave", dropna=False) .agg( rows=("weave", "count"), epi_change_pct=("epi_change_pct", "median"), ppi_change_pct=("ppi_change_pct", "median"), width_change_pct=("width_change_pct", "median"), ) .reset_index() .sort_values("rows", ascending=False) .head(20) ) blend_stats = ( df.groupby("blend", dropna=False) .agg( rows=("blend", "count"), epi_change_pct=("epi_change_pct", "median"), ppi_change_pct=("ppi_change_pct", "median"), ) .reset_index() .sort_values("rows", ascending=False) .head(20) ) corr_cols = [ "Greige EPI", "Greige PPI", "FINISH EPI", "FINISH PPI", "FINISH GSM", "FINISH WIDTH", "ON LOOM EPI", "ON LOOM PPI", ] corr = df[corr_cols].corr(numeric_only=True).fillna(0).round(3) # Scatter data for chart scatter_df = df[["FINISH EPI", "Greige EPI"]].dropna().sample(n=min(500, len(df))) scatter_data = scatter_df.rename(columns={"FINISH EPI": "finish_epi", "Greige EPI": "greige_epi"}).to_dict(orient="records") return { "by_weave": weave_stats.round(3).to_dict(orient="records"), "by_blend": blend_stats.round(3).to_dict(orient="records"), "correlation_matrix": corr.to_dict(), "epi_correlation": scatter_data } def get_process_flow(self) -> dict: return { "steps": [ { "step": 1, "title": "Customer Swatch / Requirement Intake", "description": "Capture fabric request by swatch or written construction targets.", }, { "step": 2, "title": "Initial Technical Analysis", "description": "Analyze weave, blend, count, finish EPI/PPI, GSM and compare with archive.", }, { "step": 3, "title": "Archive Match & Feasibility", "description": "Find exact/nearest historical article and validate manufacturing feasibility.", }, { "step": 4, "title": "Construction Recommendation", "description": "Predict greige construction and technical sheet parameters from historical behavior.", }, { "step": 5, "title": "Planning Hand-off", "description": "Pass recommendations to yarn and gray planning for execution.", }, ] } def get_documentation(self) -> dict: return { "scope": { "phase_1": "Gray fabric engineering and construction recommendation.", "phase_2": "Secondary parameters like elongation/recovery and strength metrics.", }, "input_parameters": [ "warp_count", "weft_count", "finish_epi", "finish_ppi", "weave", "blend", "target_gsm (optional)", ], "output_parameters": [ "reed_count", "ends_per_dent", "reed_space", "greige_epi", "greige_ppi", "finish_epi", "finish_ppi", "finish_width", "target_gsm", "loom_type", "cover_factor", ], "formulas": { "epi_change_pct": "(FINISH EPI - Greige EPI) / Greige EPI * 100", "ppi_change_pct": "(FINISH PPI - Greige PPI) / Greige PPI * 100", "width_change_pct": "(Greige Width - Finish Width) / Greige Width * 100", }, "prediction_method": { "gsm_formula": "GSM = {(EPI/WC) + (PPI/WtC)} × 24.5", "count_pairs": "Dynamic warp×weft count-pair cases; expands until archive yields matches", "cascade": "5-case per count pair: C+W+B → C+simW+B → simC+W+B → C+W+simB → simC+simW+simB", "range_probing": "EPI/PPI ±5% → ±10% → ±15% → ±20% within active cascade case", "weights": "Percentage-based: Count, EPI/PPI, GSM — ranking only", "recommendation": "Top 2–3 exact historical articles; no median aggregation", "reed_alternatives": "Validates primary match greige EPI against historical reed combos", }, } def get_data_source_audit(self) -> dict: df = self._ensure() required_cols = [ "WEAVE", "Blend Ratio", "Greige EPI", "Greige PPI", "FINISH EPI", "FINISH PPI", "FINISH GSM", "FINISH WIDTH", "Reed Count", "Ends per dent", "Reed space", ] coverage = {} for c in required_cols: if c in df.columns: coverage[c] = { "null_pct": round(float(df[c].isna().mean() * 100), 3), "non_null_rows": int(df[c].notna().sum()), } return { "source_files": { "working_24_jan": WORKING_PATH, "piece_dyed_article_details": PIECE_PATH, }, "row_counts": { "raw_working": self.raw_working_rows, "raw_piece_dyed": self.raw_piece_rows, "raw_total": self.raw_working_rows + self.raw_piece_rows, "used_working": self.used_working_rows, "used_piece_dyed": self.used_piece_rows, "used_total": int(len(df)), }, "dataset_distribution": { "working": int((df["dataset"] == "working").sum()), "piece_dyed": int((df["dataset"] == "piece_dyed").sum()), }, "column_coverage": coverage, } def clear_cache(self): self.get_dashboard_summary.cache_clear() self.get_filters.cache_clear() self.get_relativity_analytics.cache_clear() self.get_validation_report.cache_clear() data_service = DataService()