simulatore-induttori/simulate.py

# Motore principale: genera N run randomizzati della fascetta riscaldata da
# sorgenti a induzione in movimento e scrive i CSV del dataset.
#
# Il modello termico è una shell a elementi finiti quadrilateri sull'intera
# superficie cilindrica, con una temperatura per nodo. La pipeline di un run è
# in prepara_stato_termico() e passo_termico(), condivise con i moduli di
# animazione: ogni modifica alla fisica va fatta lì, non duplicata.

import csv
import math
import os
import random
import shutil
from concurrent.futures import ProcessPoolExecutor
from copy import deepcopy
from pathlib import Path

import numpy as np

import assemblaggio
import sensore as sens
import solutore as sol
import sorgente as src
from config import ARIA, FASCETTA, RANDOMIZZAZIONE, SENSORE, SIMULAZIONE, SORGENTE
from materials import MATERIALI
from mesh import costruisci_mesh_cilindrica


MU0 = 4.0 * math.pi * 1e-7


def calcola_skin_depth_m(materiale: dict, frequenza_hz: float) -> float:
    # Skin depth elettromagnetica approssimata:
    # delta = sqrt(2 * rho_e / (omega * mu))
    #
    # Semplificata. Per acciai ferromagnetici il comportamento reale
    # è fortemente non lineare con temperatura e campo magnetico.
    rho_e = materiale["resistivita_elettrica_ohm_m"]
    mu_r = materiale["permeabilita_relativa"]
    omega = 2.0 * math.pi * frequenza_hz
    mu = MU0 * mu_r
    return math.sqrt(2.0 * rho_e / (omega * mu))


def numero_fourier_spessore(
    materiale: dict, spessore_m: float, sorgente: dict
) -> float:
    # Verifica quantitativa dell'ipotesi di parete termicamente sottile:
    #
    #     Fo = alpha * (sigma / |v|) / spessore^2
    #
    # confronta il tempo di diffusione attraverso lo spessore con il tempo di
    # transito della sorgente. Per Fo >> 1 lo spessore si equilibra molto
    # prima che la sorgente sia passata e una shell a una temperatura per
    # nodo è adeguata; scendendo verso Fo ~ 1 servirebbe una formulazione
    # multistrato.
    alpha = materiale["conducibilita_termica_W_mK"] / (
        materiale["densita_kg_m3"] * materiale["calore_specifico_J_kgK"]
    )
    velocita = abs(sorgente["velocita_m_s"])
    if velocita <= 0.0:
        return float("inf")
    tempo_transito = sorgente["sigma_punto_m"] / velocita
    return alpha * tempo_transito / (spessore_m * spessore_m)


def prepara_stato_termico(
    fascetta: dict, aria: dict, sorgente: dict, sensore: dict
) -> dict:
    # Costruisce mesh, matrici globali e solutore fattorizzato per un run:
    # tutto ciò che resta costante durante l'integrazione temporale.
    materiale = MATERIALI[fascetta["materiale"]]

    lunghezza_m = fascetta["lunghezza_mm"] / 1000.0
    spessore_m = fascetta["spessore_mm"] / 1000.0
    raggio_medio_m = fascetta["diametro_medio_mm"] / 2000.0

    mesh = costruisci_mesh_cilindrica(
        lunghezza_m=lunghezza_m,
        raggio_medio_m=raggio_medio_m,
        n_elementi_x=fascetta["n_elementi_x"],
        n_elementi_theta=fascetta["n_elementi_theta"],
    )

    C = assemblaggio.assembla_capacita(mesh, materiale, spessore_m)
    K_cond = assemblaggio.assembla_conduzione(mesh, materiale, spessore_m)
    K_conv, f_ambiente = assemblaggio.assembla_convezione(mesh, aria, spessore_m)

    dt_s = SIMULAZIONE["dt_interno_s"]
    solutore = sol.costruisci_solutore(C, K_cond + K_conv, dt_s)

    assemblatore = assemblaggio.prepara_assemblatore_sorgente(mesh)

    # Il fattore circonferenziale del flusso non dipende dal tempo: le
    # sorgenti traslano solo lungo x.
    fattore_arco = src.fattore_circonferenziale(
        sorgente, raggio_medio_m, assemblatore["arco_gauss_m"]
    )
    attenuazione_al_sensore = float(
        src.fattore_circonferenziale(sorgente, raggio_medio_m, np.zeros(1))[0]
    )

    if sorgente["skin_depth_fissa_m"] is None:
        skin_depth_m = calcola_skin_depth_m(materiale, sorgente["frequenza_hz"])
    else:
        skin_depth_m = float(sorgente["skin_depth_fissa_m"])

    x_sensore_m = sensore["x_mm"] / 1000.0

    return {
        "mesh": mesh,
        "materiale": materiale,
        "spessore_m": spessore_m,
        "raggio_medio_m": raggio_medio_m,
        "dt_s": dt_s,
        "solutore": solutore,
        "assemblatore": assemblatore,
        "f_ambiente": f_ambiente,
        "fattore_arco": fattore_arco,
        "attenuazione_al_sensore": attenuazione_al_sensore,
        "x_sensore_m": x_sensore_m,
        "interpolatore_sensore": sens.prepara_interpolatore(mesh, x_sensore_m, 0.0),
        "sorgente": sorgente,
        "T_ambiente_C": aria["temperatura_ambiente_C"],
        "skin_depth_m": skin_depth_m,
        "numero_fourier_spessore": numero_fourier_spessore(
            materiale, spessore_m, sorgente
        ),
    }


def campo_iniziale(stato: dict) -> np.ndarray:
    return np.full(stato["mesh"]["n_nodi"], stato["T_ambiente_C"], dtype=float)


def passo_termico(stato: dict, T: np.ndarray, t_s: float) -> tuple[np.ndarray, dict]:
    # Avanza il campo di temperatura di un passo dt e restituisce anche lo
    # stato della sorgente in quell'istante.
    sorgente = stato["sorgente"]
    x_riferimento_m, flusso_x = src.profilo_flusso_x_W_m2(
        sorgente, stato["x_sensore_m"], t_s, stato["assemblatore"]["x_gauss_m"]
    )
    f_sorgente = assemblaggio.assembla_sorgente(
        stato["assemblatore"], flusso_x, stato["fattore_arco"]
    )
    T_next = sol.passo_implicito(
        stato["solutore"], T, stato["f_ambiente"], f_sorgente
    )

    _, flusso_al_sensore = src.profilo_flusso_x_W_m2(
        sorgente, stato["x_sensore_m"], t_s, np.array([stato["x_sensore_m"]])
    )
    return T_next, {
        "x_riferimento_m": x_riferimento_m,
        "flusso_al_sensore_W_m2": float(
            flusso_al_sensore[0] * stato["attenuazione_al_sensore"]
        ),
    }


def configurazione_randomizzata(indice_run: int, rng: random.Random) -> dict:
    fascetta = deepcopy(FASCETTA)
    aria = deepcopy(ARIA)
    sorgente = deepcopy(SORGENTE)
    sensore = deepcopy(SENSORE)

    if RANDOMIZZAZIONE.get("abilitata", False):
        def perturba_rel(valore: float, std_rel: float, fattore_min: float = 0.1) -> float:
            fattore = rng.gauss(1.0, std_rel)
            fattore = max(fattore_min, fattore)
            return valore * fattore

        sorgente["velocita_m_s"] = perturba_rel(
            sorgente["velocita_m_s"],
            RANDOMIZZAZIONE["velocita_std_rel"],
        )
        sorgente["flusso_termico_picco_W_m2"] = perturba_rel(
            sorgente["flusso_termico_picco_W_m2"],
            RANDOMIZZAZIONE["flusso_picco_std_rel"],
        )
        sorgente["sigma_punto_m"] = perturba_rel(
            sorgente["sigma_punto_m"],
            RANDOMIZZAZIONE["sigma_punto_std_rel"],
        )
        sorgente["offset_y_percorso_m"] = rng.uniform(
            -RANDOMIZZAZIONE["offset_y_max_assoluto_m"],
            RANDOMIZZAZIONE["offset_y_max_assoluto_m"],
        )
        aria["temperatura_ambiente_C"] += rng.gauss(
            0.0,
            RANDOMIZZAZIONE["temperatura_ambiente_std_C"],
        )
        sensore["rumore_std_C"] = perturba_rel(
            sensore["rumore_std_C"],
            RANDOMIZZAZIONE["rumore_sensore_std_rel"],
            fattore_min=0.0,
        )

    return {
        "id_run": f"run_{indice_run:04d}",
        "fascetta": fascetta,
        "aria": aria,
        "sorgente": sorgente,
        "sensore": sensore,
    }


def simula_singolo(cfg_run: dict, output_csv: Path, rng: random.Random) -> dict:
    fascetta = cfg_run["fascetta"]
    aria = cfg_run["aria"]
    sorgente = cfg_run["sorgente"]
    sensore = cfg_run["sensore"]

    nome_materiale = fascetta["materiale"]

    stato = prepara_stato_termico(fascetta, aria, sorgente, sensore)
    mesh = stato["mesh"]
    interpolatore = stato["interpolatore_sensore"]
    skin_depth = stato["skin_depth_m"]

    dt = stato["dt_s"]
    durata = SIMULAZIONE["durata_s"]
    periodo_campionamento = 1.0 / SIMULAZIONE["frequenza_campionamento_hz"]

    T = campo_iniziale(stato)
    T_letta = sens.temperatura_osservata_C(interpolatore, T)

    prossimo_campione_t = 0.0
    T_vera_max = T_letta
    T_misurata_max = T_letta
    T_fascetta_max = float(T.max())

    output_csv.parent.mkdir(parents=True, exist_ok=True)

    with output_csv.open("w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow([
            "id_run",
            "tempo_s",
            "x_sorgente_m",
            "offset_y_sorgente_m",
            "flusso_termico_sorgente_W_m2",
            "skin_depth_m",
            "T_vera_lato_sensore_C",
            "T_misurata_sensore_C",
            "T_max_fascetta_C",
            "T_media_fascetta_C",
            "T_ambiente_C",
            "velocita_m_s",
            "sigma_punto_m",
            "flusso_picco_W_m2",
            "materiale",
        ])

        t = 0.0
        while t <= durata + 1e-12:
            T, info = passo_termico(stato, T, t)

            # Temperatura vera della superficie interna nel punto osservato
            # dal sensore infrarosso.
            T_vera = sens.temperatura_osservata_C(interpolatore, T)
            T_letta = sens.aggiorna_inerzia_C(
                T_letta, T_vera, dt, sensore["costante_tempo_s"]
            )

            # I massimi delle temperature vere sono aggiornati a ogni passo
            # interno, non solo agli istanti di campionamento: servono come
            # ground truth e il picco è stretto rispetto al periodo di
            # campionamento CSV, che da solo lo taglierebbe di oltre un grado.
            T_max = float(T.max())
            T_vera_max = max(T_vera_max, T_vera)
            T_fascetta_max = max(T_fascetta_max, T_max)

            # Campionamento CSV.
            if t + 1e-12 >= prossimo_campione_t:
                misurata = T_letta + rng.gauss(0.0, sensore["rumore_std_C"])
                misurata = sens.quantizza(misurata, sensore["quantizzazione_C"])

                T_misurata_max = max(T_misurata_max, misurata)

                writer.writerow([
                    cfg_run["id_run"],
                    f"{t:.6f}",
                    f"{info['x_riferimento_m']:.9f}",
                    f"{sorgente['offset_y_percorso_m']:.9f}",
                    f"{info['flusso_al_sensore_W_m2']:.6f}",
                    f"{skin_depth:.9e}",
                    f"{T_vera:.6f}",
                    f"{misurata:.6f}",
                    f"{T_max:.6f}",
                    f"{float(T.mean()):.6f}",
                    f"{aria['temperatura_ambiente_C']:.6f}",
                    f"{sorgente['velocita_m_s']:.9f}",
                    f"{sorgente['sigma_punto_m']:.9f}",
                    f"{sorgente['flusso_termico_picco_W_m2']:.6f}",
                    nome_materiale,
                ])
                prossimo_campione_t += periodo_campionamento

            t += dt

    return {
        "id_run": cfg_run["id_run"],
        "file_csv": str(output_csv.name),
        "materiale": nome_materiale,
        "diametro_medio_m": fascetta["diametro_medio_mm"] / 1000.0,
        "lunghezza_m": fascetta["lunghezza_mm"] / 1000.0,
        "spessore_m": fascetta["spessore_mm"] / 1000.0,
        "n_elementi_x": mesh["n_elementi_x"],
        "n_elementi_theta": mesh["n_elementi_theta"],
        "n_nodi": mesh["n_nodi"],
        "lato_elemento_x_m": mesh["lato_x_m"],
        "lato_elemento_arco_m": mesh["lato_arco_m"],
        "durata_s": durata,
        "frequenza_campionamento_hz": SIMULAZIONE["frequenza_campionamento_hz"],
        "dt_interno_s": dt,
        "temperatura_ambiente_C": aria["temperatura_ambiente_C"],
        "h_esterno_W_m2K": aria["h_esterno_W_m2K"],
        "h_interno_W_m2K": aria["h_interno_W_m2K"],
        "h_bordi_W_m2K": aria["h_bordi_W_m2K"],
        "x_inizio_m": sorgente["x_inizio_m"],
        "x_fine_m": sorgente["x_fine_m"],
        "x_sensore_m": stato["x_sensore_m"],
        "distanza_sensore_parete_m": sensore["distanza_parete_mm"] / 1000.0,
        "offset_y_percorso_m": sorgente["offset_y_percorso_m"],
        "theta_percorso_rad": src.theta_percorso_rad(
            sorgente, stato["raggio_medio_m"]
        ),
        "velocita_m_s": sorgente["velocita_m_s"],
        "numero_sorgenti": sorgente.get("numero_sorgenti", 1),
        "distanza_sorgenti_m": sorgente.get("distanza_sorgenti_m", 0.0),
        "sigma_punto_m": sorgente["sigma_punto_m"],
        "flusso_termico_picco_W_m2": sorgente["flusso_termico_picco_W_m2"],
        "efficienza_riscaldamento": sorgente["efficienza_riscaldamento"],
        "frequenza_hz": sorgente["frequenza_hz"],
        "skin_depth_m": skin_depth,
        "numero_fourier_spessore": stato["numero_fourier_spessore"],
        "costante_tempo_sensore_s": sensore["costante_tempo_s"],
        "rumore_std_sensore_C": sensore["rumore_std_C"],
        "quantizzazione_sensore_C": sensore["quantizzazione_C"],
        "T_vera_max_lato_sensore_C": T_vera_max,
        "T_misurata_max_sensore_C": T_misurata_max,
        "T_max_fascetta_C": T_fascetta_max,
    }


def _esegui_run(indice_e_seme: tuple[int, int]) -> dict:
    # Ogni run riceve un seme indipendente derivato dal seed globale, così
    # l'esecuzione in parallelo resta riproducibile indipendentemente
    # dall'ordine in cui i processi la completano.
    indice, seme = indice_e_seme
    rng = random.Random(seme)
    cfg_run = configurazione_randomizzata(indice, rng)
    cartella_output = Path(SIMULAZIONE["cartella_output"])
    percorso_csv = cartella_output / f"{cfg_run['id_run']}.csv"
    return simula_singolo(cfg_run, percorso_csv, rng)


def main() -> None:
    cartella_output = Path(SIMULAZIONE["cartella_output"])
    if cartella_output.exists():
        shutil.rmtree(cartella_output)
    cartella_output.mkdir(parents=True, exist_ok=True)

    rng_semi = random.Random(SIMULAZIONE["seed"])
    num_run = SIMULAZIONE["num_run"]
    semi = [rng_semi.randrange(2**63) for _ in range(num_run)]

    num_processi = SIMULAZIONE["num_processi"] or os.cpu_count() or 1
    with ProcessPoolExecutor(max_workers=num_processi) as executor:
        righe_metadata = list(
            executor.map(_esegui_run, enumerate(semi, start=1))
        )

    percorso_metadata = cartella_output / "metadata.csv"
    with percorso_metadata.open("w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=list(righe_metadata[0].keys()))
        writer.writeheader()
        writer.writerows(righe_metadata)

    print(f"Generati {len(righe_metadata)} run in: {cartella_output.resolve()}")
    print(f"Metadata: {percorso_metadata.resolve()}")


if __name__ == "__main__":
    main()