utils/calculations.py

"""Domain calculations for crash material intelligence."""

from __future__ import annotations

import numpy as np
import pandas as pd

from utils.data_generator import MATERIAL_CARD_MAP, MATERIAL_FAMILIES


WEIGHTS_DEFAULT = {
    "crash": 0.30,
    "weight": 0.20,
    "cost": 0.20,
    "sustainability": 0.15,
    "failure": 0.15,
}


def multi_objective_score(
    df: pd.DataFrame,
    weights: dict[str, float] | None = None,
) -> pd.Series:
    """Compute weighted multi-objective ranking score."""
    w = weights or WEIGHTS_DEFAULT
    total = sum(w.values()) or 1.0
    w = {k: v / total for k, v in w.items()}
    score = (
        w["crash"] * df["crashworthiness_index"]
        + w["weight"] * df["lightweighting_score"]
        + w["cost"] * np.clip(df["cost_performance_score"] * 1.2, 0, 100)
        + w["sustainability"] * df["sustainability_score"]
        + w["failure"] * (100 * (1.0 - df["failure_risk"]))
    )
    return score.round(2)


def rank_materials(
    materials: pd.DataFrame,
    families: list[str] | None = None,
    max_cost: float | None = None,
    min_uts: float | None = None,
    max_density: float | None = None,
    weights: dict[str, float] | None = None,
    top_n: int = 5,
) -> pd.DataFrame:
    """Filter and rank materials for crash applications."""
    df = materials.copy()
    if families:
        df = df[df["family"].isin(families)]
    if max_cost is not None:
        df = df[df["cost_usd_kg"] <= max_cost]
    if min_uts is not None:
        df = df[df["uts_mpa"] >= min_uts]
    if max_density is not None:
        df = df[df["density_g_cm3"] <= max_density]
    if df.empty:
        return df
    df = df.copy()
    df["mo_score"] = multi_objective_score(df, weights)
    return df.sort_values("mo_score", ascending=False).head(top_n)


def recommend_for_scenario(
    materials: pd.DataFrame,
    recommendations: pd.DataFrame,
    scenario: str,
    component: str,
    families: list[str] | None = None,
    top_n: int = 5,
) -> pd.DataFrame:
    """Recommend materials for a crash scenario + component pair."""
    rec = recommendations[
        (recommendations["crash_scenario"] == scenario)
        & (recommendations["component"] == component)
    ].copy()
    if families:
        rec = rec[rec["family"].isin(families)]
    if rec.empty:
        ranked = rank_materials(materials, families=families, top_n=top_n)
        ranked = ranked.copy()
        ranked["crash_scenario"] = scenario
        ranked["component"] = component
        ranked["crash_score"] = ranked["crashworthiness_index"]
        ranked["thickness_mm"] = 2.0
        ranked["joining_method"] = "Hybrid Weld-Bond"
        ranked["simulation_risk"] = (ranked["failure_risk"] * 100).round(2)
        return ranked

    agg_cols = [
        "crash_score",
        "energy_absorption_kj",
        "intrusion_mm",
        "peak_force_kn",
        "crush_force_efficiency",
        "specific_energy_absorption",
        "weight_reduction_pct",
        "cost_score",
        "sustainability_score",
        "lightweighting_score",
        "simulation_risk",
        "thickness_mm",
    ]
    grouped = (
        rec.groupby(["material_id", "material_name", "family", "joining_method"], as_index=False)[
            agg_cols
        ]
        .mean(numeric_only=True)
        .sort_values("crash_score", ascending=False)
        .head(top_n)
    )
    return grouped


def generate_material_card(material: pd.Series, solver: str = "LS-DYNA") -> dict:
    """Build a draft CAE-ready material card payload."""
    family = material["family"]
    card_type = MATERIAL_CARD_MAP.get(family, "MAT_024")
    curve_pts = 10
    strains = np.linspace(0.0, float(material["failure_strain"]), curve_pts)
    ys = float(material["yield_strength_mpa"])
    uts = float(material["uts_mpa"])
    stresses = []
    for eps in strains:
        if eps <= 0:
            stresses.append(ys)
        else:
            t = min(eps / max(material["failure_strain"], 1e-6), 1.0)
            stresses.append(ys + (uts - ys) * t)

    card = {
        "solver": solver,
        "card_type": card_type,
        "material_name": material["material_name"],
        "family": family,
        "density_g_cm3": float(material["density_g_cm3"]),
        "youngs_modulus_gpa": float(material["youngs_modulus_gpa"]),
        "poisson_ratio": 0.30 if "Aluminum" in family or family == "Magnesium" else 0.29,
        "yield_strength_mpa": ys,
        "uts_mpa": uts,
        "failure_strain": float(material["failure_strain"]),
        "strain_rate_sensitivity": float(material["strain_rate_sensitivity"]),
        "plastic_curve_strain": [round(float(s), 5) for s in strains],
        "plastic_curve_stress_mpa": [round(float(s), 2) for s in stresses],
        "damage_evolution": "Linear softening to zero stress at failure strain",
        "temperature_dependency": "Room-temperature card; scale factors TBD",
        "validation_status": "Draft — public-data prototype",
        "confidence_score": float(material["confidence_score"]),
    }
    return card


def card_to_text(card: dict) -> str:
    """Serialize a material card to a readable text block."""
    lines = [
        f"*KEYWORD  ({card['solver']} draft)",
        f"$ Material: {card['material_name']} ({card['family']})",
        f"$ Card type: {card['card_type']}",
        f"$ Confidence: {card['confidence_score']:.2f}",
        f"$ Validation: {card['validation_status']}",
        "*MAT_PIECEWISE_LINEAR_PLASTICITY",
        f"$ RO (g/cm3) = {card['density_g_cm3']}",
        f"$ E (GPa) = {card['youngs_modulus_gpa']}",
        f"$ PR = {card['poisson_ratio']}",
        f"$ SIGY (MPa) = {card['yield_strength_mpa']}",
        f"$ FAIL = {card['failure_strain']}",
        f"$ C (strain-rate) = {card['strain_rate_sensitivity']}",
        "$ Plastic curve (strain, stress MPa):",
    ]
    for eps, sig in zip(card["plastic_curve_strain"], card["plastic_curve_stress_mpa"]):
        lines.append(f"$   {eps:.5f}, {sig:.2f}")
    lines.append(f"$ Damage: {card['damage_evolution']}")
    lines.append(f"$ Temperature: {card['temperature_dependency']}")
    lines.append("*END")
    return "\n".join(lines)


def family_summary(materials: pd.DataFrame) -> pd.DataFrame:
    """Aggregate key metrics by material family."""
    cols = [
        "density_g_cm3",
        "uts_mpa",
        "crashworthiness_index",
        "energy_absorption_potential",
        "cost_usd_kg",
        "sustainability_score",
        "lightweighting_score",
        "failure_risk",
    ]
    return (
        materials.groupby("family")[cols]
        .mean(numeric_only=True)
        .reset_index()
        .sort_values("crashworthiness_index", ascending=False)
    )


def scenario_kpi(recommendations: pd.DataFrame) -> pd.DataFrame:
    """KPI rollup by crash scenario."""
    return (
        recommendations.groupby("crash_scenario")
        .agg(
            avg_crash_score=("crash_score", "mean"),
            avg_energy=("energy_absorption_kj", "mean"),
            avg_intrusion=("intrusion_mm", "mean"),
            avg_weight_reduction=("weight_reduction_pct", "mean"),
            n_cases=("rec_id", "count"),
        )
        .reset_index()
        .sort_values("avg_crash_score", ascending=False)
    )


def available_families() -> list[str]:
    return list(MATERIAL_FAMILIES.keys())