Upload make_data_grid.py

make_data_grid.py

@@ -0,0 +1,259 @@
+# make_data_grid.py (clean + robust paths)
+
+from __future__ import annotations
+from decimal import Decimal
+from pathlib import Path
+from typing import Dict, List, Tuple, Union
+import itertools
+import warnings
+import os
+
+import numpy as np
+import pandas as pd
+
+warnings.filterwarnings("ignore", category=FutureWarning)
+
+# =============================
+# Grid & helpers
+# =============================
+def dseq(start: float, stop: float, step: float, q: str = "0.001") -> List[float]:
+    s, e, st = map(lambda x: Decimal(str(x)), [start, stop, step])
+    vals, cur = [], s
+    while cur <= e + Decimal("1e-12"):
+        vals.append(float(cur.quantize(Decimal(q))))
+        cur += st
+    return vals
+
+def bead_to_lr(bead_value: Union[str, None]) -> Tuple[int, int]:
+    mapping = {None:(2,2), "none":(0,0), "right":(0,1), "left":(1,0), "double":(1,1)}
+    key = bead_value.lower() if isinstance(bead_value, str) else bead_value
+    return mapping.get(key, (0,0))
+
+def make_all_combinations(cfg: Dict) -> pd.DataFrame:
+    bead_values = cfg.get("beads") or [None]
+    bead_info = [(b, *bead_to_lr(b)) for b in bead_values]
+
+    materials = cfg["materials"]
+    thickness = dseq(cfg["min_thickness"], cfg["max_thickness"], cfg["thickness_step"])
+    diameter  = [int(x) for x in dseq(cfg["min_diameter"], cfg["max_diameter"], cfg["diameter_step"], q="1")]
+    upper_r   = dseq(cfg["upper_min"], cfg["upper_max"], cfg["upper_step"])
+    lower_r   = dseq(cfg["lower_min"], cfg["lower_max"], cfg["lower_step"])
+    degree    = [int(x) for x in dseq(cfg["min_degree"], cfg["max_degree"], cfg["degree_step"], q="1")]
+
+    grid = itertools.product(materials, thickness, upper_r, lower_r, diameter, degree, bead_info)
+
+    rows = []
+    for mat, th, ur, lr, dia, deg, bead_t in grid:
+        bead_name, lb, rb = bead_t
+        rows.append((str(mat), float(th), float(ur), float(lr), int(dia), int(deg), bead_name, int(lb), int(rb)))
+
+    return pd.DataFrame(
+        rows,
+        columns=["material","thickness","upper_radius","lower_radius",
+                 "diameter","degree","bead","LB","RB"]
+    )
+
+# =============================
+# Sweep-limit helpers
+# =============================
+def build_limit_dicts(df_sweep: pd.DataFrame):
+    t = df_sweep.copy().replace("F", np.nan)
+    if "Sweep" in t.columns:
+        t = t.set_index("Sweep")
+
+    new_cols = []
+    for c in t.columns:
+        try: new_cols.append(int(c))
+        except: new_cols.append(c)
+    t.columns = new_cols
+
+    long = t.stack(dropna=False).reset_index()
+    long.columns = ["row","degree","limit"]
+
+    tmp = long["row"].str.extract(r"(?P<diameter>\d+)_(?P<which>upper|lower)_radius")
+    long = pd.concat([long, tmp], axis=1)
+    long["diameter"] = pd.to_numeric(long["diameter"], errors="coerce")
+    long["degree"]   = pd.to_numeric(long["degree"], errors="coerce")
+    long["limit"]    = pd.to_numeric(long["limit"], errors="coerce")
+
+    upper = long[long["which"]=="upper"].dropna(subset=["diameter","degree"])
+    lower = long[long["which"]=="lower"].dropna(subset=["diameter","degree"])
+
+    upper_dict = {(int(d), int(g)): v for d, g, v in zip(upper["diameter"], upper["degree"], upper["limit"])}
+    lower_dict = {(int(d), int(g)): v for d, g, v in zip(lower["diameter"], lower["degree"], lower["limit"])}
+    return upper_dict, lower_dict
+
+def filter_grid_by_sweep_limits(df_grid: pd.DataFrame, df_sweep: pd.DataFrame) -> pd.DataFrame:
+    upper_dict, lower_dict = build_limit_dicts(df_sweep)
+    key = list(zip(df_grid["diameter"].astype(int), df_grid["degree"].astype(int)))
+
+    df_grid = df_grid.copy()
+    df_grid["limit_upper"] = [upper_dict.get(k, np.nan) for k in key]
+    df_grid["limit_lower"] = [lower_dict.get(k, np.nan) for k in key]
+
+    not_nan = df_grid["limit_upper"].notna() & df_grid["limit_lower"].notna()
+    within  = (df_grid["upper_radius"] <= df_grid["limit_upper"]) & \
+              (df_grid["lower_radius"] <= df_grid["limit_lower"])
+    return df_grid[not_nan & within].reset_index(drop=True)
+
+# =============================
+# Rule-by-bead helpers
+# =============================
+SHEET_BY_BEAD = {"left":"left", "right":"right", "double":"both", "none":"none"}
+
+def _normalize_input_df(df: pd.DataFrame) -> pd.DataFrame:
+    df2 = df.copy()
+    df2.columns = [str(c).strip() for c in df2.columns]
+    rename = {}
+    for c in df2.columns:
+        lc = c.lower().strip()
+        if lc == "diamater": rename[c] = "diameter"
+        elif lc in ("material","diameter","degree","bead"): rename[c] = lc
+    df2 = df2.rename(columns=rename)
+    need = {"material","diameter","degree","bead"}
+    missing = need - set(df2.columns)
+    if missing:
+        raise ValueError(f"입력 데이터프레임에 필요한 컬럼이 없습니다: {missing}")
+    df2["bead"] = df2["bead"].astype(str).str.strip().str.lower()
+    for c in ["material","diameter","degree"]:
+        df2[c] = pd.to_numeric(df2[c], errors="coerce")
+    df2 = df2.dropna(subset=["material","diameter","degree"]).copy()
+    return df2
+
+def _read_rule_sheet(xlsx_path: str, sheet_name: str) -> pd.DataFrame:
+    rule = pd.read_excel(xlsx_path, sheet_name=sheet_name)
+    rule.columns = rule.columns.str.strip().str.lower()
+    rule = rule.rename(columns={"diamater":"diameter"})
+    need_cols = {"material","diameter","min_degree","max_degree"}
+    missing = need_cols - set(rule.columns)
+    if missing:
+        raise ValueError(f"규칙 시트 '{sheet_name}'에 필요한 컬럼이 없습니다: {missing}")
+    for c in need_cols:
+        rule[c] = pd.to_numeric(rule[c], errors="coerce")
+    rule = rule.dropna(subset=list(need_cols)).copy()
+    rule = rule.astype({"material":"int64","diameter":"int64"})
+    return rule[["material","diameter","min_degree","max_degree"]]
+
+def _apply_rules(df_part: pd.DataFrame, rule: pd.DataFrame) -> pd.DataFrame:
+    if df_part.empty:
+        return df_part.copy()
+    df_part = df_part[df_part["material"].isin(rule["material"].unique())].copy()
+    if df_part.empty:
+        return df_part
+    merged = df_part.merge(rule, on=["material","diameter"], how="left")
+    mask = (
+        merged["min_degree"].notna()
+        & merged["max_degree"].notna()
+        & (merged["degree"] >= merged["min_degree"])
+        & (merged["degree"] <= merged["max_degree"])
+    )
+    kept = merged.loc[mask, df_part.columns].reset_index(drop=True)
+    return kept
+
+def filter_all_by_bead(df: pd.DataFrame, rules_xlsx: str) -> pd.DataFrame:
+    base = _normalize_input_df(df)
+    outs = []
+    for bead_value, sheet in SHEET_BY_BEAD.items():
+        part = base[base["bead"] == bead_value].copy()
+        if part.empty:
+            continue
+        rule = _read_rule_sheet(rules_xlsx, sheet)
+        kept = _apply_rules(part, rule)
+        outs.append(kept)
+    if not outs:
+        return base.iloc[0:0].copy()
+    result = pd.concat(outs, axis=0, ignore_index=True)
+    return result.reset_index(drop=True)
+
+# =============================
+# Predictors (blend models)
+# =============================
+DISPLAY_TO_MODEL = {"440":"440.0", "590":"590.0", "780":"780.0"}
+
+def _here() -> Path:
+    try: return Path(__file__).resolve().parent
+    except Exception: return Path.cwd()
+
+def _abs(p: Path | str) -> str:
+    return str(Path(p).resolve())
+
+def _find_art_dir(name: str) -> str:
+    """
+    절대경로 탐색 우선순위:
+      1) 환경변수 FS_MF_ART_DIR / FS_THIN_ART_DIR
+      2) 이 모듈 파일 기준
+      3) 현재 작업 디렉토리
+      4) 모듈 상위 경로
+    """
+    env_map = {
+        "artifacts_blend": os.getenv("FS_MF_ART_DIR"),
+        "artifacts_blend_thinning": os.getenv("FS_THIN_ART_DIR"),
+    }
+    hinted = env_map.get(name)
+    if hinted and Path(hinted).exists():
+        return _abs(hinted)
+
+    for base in (_here(), Path.cwd(), _here().parent):
+        cand = (base / name).resolve()
+        if cand.exists():
+            return _abs(cand)
+    return _abs(name)
+
+_predictor_mf = None
+_predictor_th = None
+
+def _prep_pred_df(df: pd.DataFrame) -> pd.DataFrame:
+    need = ["material","thickness","diameter","degree","upper_radius","lower_radius","LB","RB"]
+    missing = [c for c in need if c not in df.columns]
+    if missing:
+        raise ValueError(f"예측 입력 컬럼 누락: {missing}")
+    x = df.copy()
+    x["material"] = x["material"].astype(str).map(lambda v: DISPLAY_TO_MODEL.get(v, v))
+    x["LB"] = x["LB"].astype(int); x["RB"] = x["RB"].astype(int)
+    for c in ["thickness","diameter","degree","upper_radius","lower_radius"]:
+        x[c] = pd.to_numeric(x[c], errors="coerce")
+    if x[["thickness","diameter","degree","upper_radius","lower_radius"]].isnull().any().any():
+        raise ValueError("숫자 컬럼에 NaN이 있습니다.")
+    return x[need]
+
+def _get_predictor_mf():
+    global _predictor_mf
+    if _predictor_mf is not None:
+        return _predictor_mf
+    from predict_blend import BlendPredictor
+    art_dir = _find_art_dir("artifacts_blend")
+    _predictor_mf = BlendPredictor(art_dir)
+    return _predictor_mf
+
+def _get_predictor_th():
+    global _predictor_th
+    if _predictor_th is not None:
+        return _predictor_th
+    try:
+        from predict_blend_thinning import BlendPredictor as ThinPredictor
+        art_dir = _find_art_dir("artifacts_blend_thinning")
+        _predictor_th = ThinPredictor(art_dir)
+        return _predictor_th
+    except FileNotFoundError:
+        # 폴더가 없을 때 간단 휴리스틱 폴백
+        def _heuristic_thinning(thickness, upperR, lowerR):
+            t = float(thickness); ur = float(upperR); lr = float(lowerR)
+            base = 0.18 + (0.9 - t) * 0.25
+            geom = max(0.0, ur - lr) * 0.01
+            return float(max(0.05, min(0.8, base + geom)))
+        class _HeuristicThinPredictor:
+            def predict_blend(self, df: pd.DataFrame):
+                return np.array([
+                    _heuristic_thinning(r["thickness"], r["upper_radius"], r["lower_radius"])
+                    for _, r in df.iterrows()
+                ], dtype=float)
+        _predictor_th = _HeuristicThinPredictor()
+        return _predictor_th
+
+def predict_max_failure(df: pd.DataFrame) -> np.ndarray:
+    pred_df = _prep_pred_df(df)
+    return _get_predictor_mf().predict_blend(pred_df)
+
+def predict_thinning(df: pd.DataFrame) -> np.ndarray:
+    pred_df = _prep_pred_df(df)
+    return _get_predictor_th().predict_blend(pred_df)