engine/sim/rf/utils.py

from __future__ import annotations

import hashlib
from pathlib import Path
from typing import List

import numpy as np
import pandas as pd

from .calibration import EPS


def safe_read_csv(path: Path) -> pd.DataFrame:
    if not path.exists():
        return pd.DataFrame()
    return pd.read_csv(path)


def sigmoid(x):
    return 1.0 / (1.0 + np.exp(-x))


def clip01(x):
    return np.clip(x, 0.0, 1.0)


def stable_softmax(logits: np.ndarray, tau: float) -> np.ndarray:
    z = logits / max(tau, 1e-9)
    z = z - z.max(axis=1, keepdims=True)
    ez = np.exp(z)
    return ez / np.maximum(ez.sum(axis=1, keepdims=True), EPS)


def fixed_scale(series: pd.Series, low: float, high: float, invert: bool = False) -> pd.Series:
    s = series.astype(float).fillna(low)
    out = ((s - low) / max(high - low, EPS)).clip(0.0, 1.0)
    return 1.0 - out if invert else out


def infer_family(type_name: str) -> str:
    if "Lband" in type_name:
        return "Lband"
    if "Cband" in type_name:
        return "Cband"
    if "Xband" in type_name:
        return "Xband"
    if type_name.startswith("SC_"):
        return "SC"
    return "unknown"


def infer_type_from_family(family: str) -> str:
    return {
        "Lband": "NC_pillbox_Lband",
        "Cband": "NC_Cband_TW",
        "Xband": "NC_Xband_SW",
        "SC": "SC_elliptical_1p3GHz",
    }.get(family, "NC_pillbox_Lband")


def family_label(family: str) -> str:
    return {
        "Lband": "L-band",
        "Cband": "C-band",
        "Xband": "X-band",
        "SC": "SC",
    }.get(family, str(family))


def fmt_float(x, nd=3):
    return "—" if pd.isna(x) else f"{float(x):.{nd}f}"


def fmt_sci(x, nd=2):
    return "—" if pd.isna(x) else f"{float(x):.{nd}e}"


def fmt_text(x):
    return "—" if pd.isna(x) or str(x).strip() == "" else str(x)


def split_flags(flag_str: str) -> List[str]:
    if not isinstance(flag_str, str) or not flag_str.strip():
        return []
    return [x.strip() for x in flag_str.split("|") if x.strip()]


def warning_badges(flag_str: str) -> str:
    flags = split_flags(flag_str)
    if not flags:
        return "—"
    badge_map = {
        "pulse_heating_yellow": "⚠ thermal",
        "BDR_yellow": "⚠ BDR",
    }
    return ", ".join([badge_map.get(f, f"⚠ {f}") for f in flags])


def _fmt_gene_value(x) -> str:
    if pd.isna(x):
        return "nan"
    return f"{float(x):.6f}"


def make_candidate_id(row: pd.Series) -> str:
    fields = [
        fmt_text(row.get("type", "")),
        _fmt_gene_value(row.get("freq_GHz")),
        _fmt_gene_value(row.get("R_over_Q_ohm")),
        _fmt_gene_value(row.get("Q0")),
        _fmt_gene_value(row.get("Eacc_MVpm")),
        _fmt_gene_value(row.get("L_m")),
        _fmt_gene_value(row.get("beta")),
        _fmt_gene_value(row.get("pulse_length_ns")),
        _fmt_gene_value(row.get("rep_rate_Hz")),
        _fmt_gene_value(row.get("P_aux_kW")),
        _fmt_gene_value(row.get("source_power_avail_kW")),
        _fmt_gene_value(row.get("cooling_capacity_kW")),
        _fmt_gene_value(row.get("surf_factor")),
        _fmt_gene_value(row.get("geom_factor")),
        _fmt_gene_value(row.get("freq_factor")),
        _fmt_gene_value(row.get("fab_sigma_um")),
        _fmt_gene_value(row.get("delta_allow_um")),
        _fmt_gene_value(row.get("S_f")),
        _fmt_gene_value(row.get("S_phi")),
        _fmt_gene_value(row.get("S_c")),
    ]
    blob = "|".join(fields).encode("utf-8")
    return "cand_" + hashlib.sha256(blob).hexdigest()[:16]