simulatore-induttori/simulate.py

import csv
import math
import random
import shutil
from copy import deepcopy
from pathlib import Path

import numpy as np
import scipy.sparse as sp
from scipy.sparse.linalg import splu

from config import ARIA, FASCETTA, RANDOMIZZAZIONE, SENSORE, SIMULAZIONE, SORGENTE
from materials import MATERIALI


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 _spread_sorgenti_m(sorgente: dict) -> float:
    # Distanza lungo x tra la prima e l'ultima sorgente del gruppo.
    numero_sorgenti = sorgente.get("numero_sorgenti", 1)
    distanza = sorgente.get("distanza_sorgenti_m", 0.0)
    return (numero_sorgenti - 1) * distanza


def _x_riferimento_iniziale_m(sorgente: dict, x_sensore_m: float) -> float:
    # Posizione a t=0 della sorgente di indice 0. Le sorgenti i sono a
    # x_i = riferimento + i * distanza, quindi per v >= 0 l'indice 0 è la
    # più arretrata nel verso di marcia, per v < 0 è la più avanzata.
    # x_inizio_m è la distanza dal sensore della sorgente più avanzata
    # (quella che lo raggiunge per prima).
    spread = _spread_sorgenti_m(sorgente)
    x_inizio = sorgente["x_inizio_m"]
    if sorgente["velocita_m_s"] >= 0:
        return (x_sensore_m - x_inizio) - spread
    return x_sensore_m + x_inizio


def _x_riferimento_finale_m(sorgente: dict, x_sensore_m: float) -> float:
    # Posizione di fine corsa della sorgente di indice 0. x_fine_m è la
    # distanza dal sensore della sorgente più arretrata nel verso di marcia
    # (quella che lo supera per ultima).
    spread = _spread_sorgenti_m(sorgente)
    x_fine = sorgente["x_fine_m"]
    if sorgente["velocita_m_s"] >= 0:
        return x_sensore_m + x_fine
    return (x_sensore_m - x_fine) - spread


def _intervallo_attivo(inizio: float, fine: float, v: float, x_m: float) -> bool:
    if v >= 0:
        return inizio <= x_m <= fine
    return fine <= x_m <= inizio


def profilo_flusso_incidente_W_m2(
    sorgente: dict,
    x_sensore_m: float,
    t_s: float,
    x_centri_m: np.ndarray,
) -> tuple[float, np.ndarray]:
    # Restituisce x_sorgente_m (posizione della sorgente di riferimento) e
    # il profilo di flusso termico efficace q(x) [W/m²] sul lato esterno,
    # somma dei contributi di tutte le sorgenti attive.
    #
    # Ogni sorgente in moto ha un'impronta gaussiana lungo x, valutata sui
    # centri cella della sezione. L'offset circonferenziale y non è risolto
    # spazialmente: entra come attenuazione gaussiana del flusso.
    x_rif_iniziale = _x_riferimento_iniziale_m(sorgente, x_sensore_m)
    x_rif_finale = _x_riferimento_finale_m(sorgente, x_sensore_m)
    x_riferimento = x_rif_iniziale + sorgente["velocita_m_s"] * t_s

    numero_sorgenti = sorgente.get("numero_sorgenti", 1)
    distanza = sorgente.get("distanza_sorgenti_m", 0.0)
    v = sorgente["velocita_m_s"]
    zero_dopo_fine = sorgente.get("zero_dopo_fine", True)
    dy = sorgente["offset_y_percorso_m"]
    sigma = sorgente["sigma_punto_m"]

    q_picco = sorgente["flusso_termico_picco_W_m2"] * sorgente["efficienza_riscaldamento"]
    attenuazione_y = math.exp(-0.5 * (dy * dy) / (sigma * sigma))

    q_x = np.zeros_like(x_centri_m)
    for i in range(numero_sorgenti):
        x_i = x_riferimento + i * distanza

        if zero_dopo_fine:
            inizio_i = x_rif_iniziale + i * distanza
            fine_i = x_rif_finale + i * distanza
            if not _intervallo_attivo(inizio_i, fine_i, v, x_i):
                continue

        dx = x_centri_m - x_i
        q_x += q_picco * attenuazione_y * np.exp(-0.5 * (dx * dx) / (sigma * sigma))

    return x_riferimento, q_x


def profilo_deposizione_z_1_m(
    z_centri_m: np.ndarray,
    spessore_m: float,
    skin_depth_m: float,
) -> np.ndarray:
    # Profilo di deposizione volumetrica del flusso superficiale [1/m]:
    #
    # p(z) = exp(-z / delta) / (delta * (1 - exp(-spessore / delta)))
    #
    # Normalizzato in modo che integrale_0^spessore p(z) dz = 1, così che
    # q_vol(x, z) = q(x) * p(z) conservi il flusso superficiale.
    delta = max(skin_depth_m, 1e-9)
    normalizzazione = delta * (1.0 - math.exp(-spessore_m / delta))
    return np.exp(-z_centri_m / delta) / normalizzazione


def _laplaciano_1d(n: int) -> sp.spmatrix:
    # Operatore alle differenze -T'' su n celle con bordi adiabatici (Neumann).
    diagonale = np.full(n, 2.0)
    diagonale[0] = 1.0
    diagonale[-1] = 1.0
    fuori = -np.ones(n - 1)
    return sp.diags([fuori, diagonale, fuori], [-1, 0, 1])


def costruisci_solutore_implicito_2d(
    n_x: int,
    n_z: int,
    dt_s: float,
    dx_m: float,
    dz_m: float,
    materiale: dict,
    h_esterno_W_m2K: float,
    h_interno_W_m2K: float,
):
    # Costruisce e fattorizza (LU sparsa) la matrice A per Eulero implicito 2D:
    # A * T_next = rhs
    #
    # Le incognite sono i centri cella T[i, j] con i lungo x e j lungo z,
    # appiattiti in ordine C (indice = i * n_z + j). I bordi x = 0 e
    # x = lunghezza sono adiabatici; i bordi z = 0 (esterno) e z = spessore
    # (interno) includono la convezione verso l'ambiente.
    k = materiale["conducibilita_termica_W_mK"]
    rho = materiale["densita_kg_m3"]
    cp = materiale["calore_specifico_J_kgK"]
    alpha = k / (rho * cp)

    r_x = alpha * dt_s / (dx_m * dx_m)
    r_z = alpha * dt_s / (dz_m * dz_m)
    b_esterno = h_esterno_W_m2K * dt_s / (rho * cp * dz_m)
    b_interno = h_interno_W_m2K * dt_s / (rho * cp * dz_m)

    n = n_x * n_z
    convezione = np.zeros(n)
    convezione[0::n_z] = b_esterno
    convezione[n_z - 1::n_z] = b_interno

    A = (
        sp.identity(n)
        + r_x * sp.kron(_laplaciano_1d(n_x), sp.identity(n_z))
        + r_z * sp.kron(sp.identity(n_x), _laplaciano_1d(n_z))
        + sp.diags(convezione)
    )
    return splu(sp.csc_matrix(A))


def quantizza(valore: float, passo: float) -> float:
    if passo <= 0.0:
        return valore
    return round(valore / passo) * passo


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"]
    materiale = MATERIALI[nome_materiale]

    lunghezza = fascetta["lunghezza_mm"] / 1000.0
    spessore = fascetta["spessore_mm"] / 1000.0
    n_x = fascetta["n_nodi_x"]
    n_z = fascetta["n_nodi_z"]
    dx = lunghezza / n_x
    dz = spessore / n_z
    x_centri = (np.arange(n_x) + 0.5) * dx
    z_centri = (np.arange(n_z) + 0.5) * dz

    x_sensore = sensore["x_mm"] / 1000.0
    i_sensore = min(n_x - 1, max(0, int(x_sensore / dx)))

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

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

    solutore = costruisci_solutore_implicito_2d(
        n_x=n_x,
        n_z=n_z,
        dt_s=dt,
        dx_m=dx,
        dz_m=dz,
        materiale=materiale,
        h_esterno_W_m2K=aria["h_esterno_W_m2K"],
        h_interno_W_m2K=aria["h_interno_W_m2K"],
    )

    rho = materiale["densita_kg_m3"]
    cp = materiale["calore_specifico_J_kgK"]

    b_esterno = aria["h_esterno_W_m2K"] * dt / (rho * cp * dz)
    b_interno = aria["h_interno_W_m2K"] * dt / (rho * cp * dz)

    profilo_z = profilo_deposizione_z_1_m(z_centri, spessore, skin_depth)

    T = np.full((n_x, n_z), fascetta["temperatura_iniziale_C"], dtype=float)
    T_sensore = T[i_sensore, -1]

    prossimo_campione_t = 0.0
    T_vera_max = T[i_sensore, -1]
    T_misurata_max = T_sensore

    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_lato_caldo_C",
            "T_ambiente_C",
            "velocita_m_s",
            "sigma_punto_m",
            "flusso_picco_W_m2",
            "materiale",
        ])

        t = 0.0
        while t <= durata + 1e-12:
            x_sorgente, q_x = profilo_flusso_incidente_W_m2(
                sorgente, x_sensore, t, x_centri
            )

            rhs = T + (dt / (rho * cp)) * q_x[:, None] * profilo_z[None, :]
            rhs[:, 0] += b_esterno * aria["temperatura_ambiente_C"]
            rhs[:, -1] += b_interno * aria["temperatura_ambiente_C"]

            T = solutore.solve(rhs.ravel()).reshape(n_x, n_z)

            # Temperatura vera della superficie interna nel punto osservato
            # dal sensore infrarosso.
            T_vera_lato_sensore = T[i_sensore, -1]

            # Inerzia del sensore del primo ordine.
            tau_sensore = max(sensore["costante_tempo_s"], 1e-9)
            T_sensore += (T_vera_lato_sensore - T_sensore) * dt / tau_sensore

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

                T_vera_max = max(T_vera_max, T_vera_lato_sensore)
                T_misurata_max = max(T_misurata_max, misurata)

                writer.writerow([
                    cfg_run["id_run"],
                    f"{t:.6f}",
                    f"{x_sorgente:.9f}",
                    f"{sorgente['offset_y_percorso_m']:.9f}",
                    f"{q_x[i_sensore]:.6f}",
                    f"{skin_depth:.9e}",
                    f"{T_vera_lato_sensore:.6f}",
                    f"{misurata:.6f}",
                    f"{T[i_sensore, 0]:.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_m": fascetta["diametro_mm"] / 1000.0,
        "lunghezza_m": lunghezza,
        "spessore_m": spessore,
        "n_nodi_x": n_x,
        "n_nodi_z": n_z,
        "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"],
        "x_inizio_m": sorgente["x_inizio_m"],
        "x_fine_m": sorgente["x_fine_m"],
        "x_sensore_m": x_sensore,
        "distanza_sensore_parete_m": sensore["distanza_parete_mm"] / 1000.0,
        "offset_y_percorso_m": sorgente["offset_y_percorso_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,
        "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,
    }


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 = random.Random(SIMULAZIONE["seed"])

    righe_metadata = []
    for i in range(1, SIMULAZIONE["num_run"] + 1):
        cfg_run = configurazione_randomizzata(i, rng)
        percorso_csv = cartella_output / f"{cfg_run['id_run']}.csv"
        righe_metadata.append(simula_singolo(cfg_run, percorso_csv, rng))

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