"""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())