simulatore-induttori/esporta_csv.py

# Export CSV della serie temporale calcolata dal solutore FEM (fem.py).
#
# Per ogni analisi viene scritto un file run_XXXX.csv con la temperatura vera
# della parete nel punto osservato dal sensore e la lettura del sensore reale
# (inerzia, rumore, quantizzazione), più i parametri effettivi dell'analisi.
# `metadata.csv` riassume un'analisi per riga.
#
# Le righe sono campionate a FEM["frequenza_campionamento_hz"], indipendente
# dal passo di integrazione FEM["dt_s"]. I campi nodali non vengono accumulati:
# servono solo le serie scalari.

import csv
import random
import shutil
from pathlib import Path

from config import FEM, USCITA
from fem import configurazione_randomizzata, simula_campo_fem
from mesh import genera_mesh, riepilogo_mesh

INTESTAZIONE = [
    "id_run",
    "tempo_s",
    "x_sorgente_m",
    "T_vera_sensore_C",
    "T_misurata_sensore_C",
    "T_massima_fascetta_C",
    "T_ambiente_C",
    "offset_y_sorgente_m",
    "velocita_m_s",
    "sigma_punto_m",
    "flusso_picco_W_m2",
    "skin_depth_m",
    "materiale",
]


def esporta_run(
    id_run: str,
    percorso_csv: Path,
    mesh_dati: dict,
    rng: random.Random,
) -> dict:
    """Integra un'analisi FEM, ne scrive il CSV e restituisce i suoi metadati."""
    cfg = configurazione_randomizzata(rng)
    dati = simula_campo_fem(
        dt_frame_s=1.0 / FEM["frequenza_campionamento_hz"],
        cfg=cfg,
        mesh_dati=mesh_dati,
        rng=rng,
        salva_campi=False,
    )

    fascetta = cfg["fascetta"]
    aria = cfg["aria"]
    sorgente = cfg["sorgente"]
    sensore = cfg["sensore"]
    skin_depth = dati["skin_depth_m"]

    percorso_csv.parent.mkdir(parents=True, exist_ok=True)
    with percorso_csv.open("w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(INTESTAZIONE)
        for k, t in enumerate(dati["tempi"]):
            writer.writerow([
                id_run,
                f"{t:.6f}",
                f"{dati['x_riferimenti'][k]:.9f}",
                f"{dati['T_vere'][k]:.6f}",
                f"{dati['T_lette'][k]:.6f}",
                f"{dati['T_massime'][k]:.6f}",
                f"{aria['temperatura_ambiente_C']:.6f}",
                f"{sorgente['offset_y_percorso_m']:.9f}",
                f"{sorgente['velocita_m_s']:.9f}",
                f"{sorgente['sigma_punto_m']:.9f}",
                f"{sorgente['flusso_termico_picco_W_m2']:.6f}",
                f"{skin_depth:.9e}",
                fascetta["materiale"],
            ])

    return {
        "id_run": id_run,
        "file_csv": percorso_csv.name,
        "materiale": fascetta["materiale"],
        "diametro_m": fascetta["diametro_mm"] / 1000.0,
        "lunghezza_m": fascetta["lunghezza_mm"] / 1000.0,
        "spessore_m": fascetta["spessore_mm"] / 1000.0,
        "n_elementi_x": mesh_dati["n_elementi_x"],
        "n_elementi_circonferenza": mesh_dati["n_elementi_circonferenza"],
        "durata_s": FEM["durata_s"],
        "dt_s": FEM["dt_s"],
        "frequenza_campionamento_hz": FEM["frequenza_campionamento_hz"],
        "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": dati["stato"]["x_sensore_m"],
        "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_sensore_C": float(dati["T_vere"].max()),
        "T_misurata_max_sensore_C": float(dati["T_lette"].max()),
        "T_massima_fascetta_C": float(dati["T_massime"].max()),
    }


def main() -> None:
    mesh_dati = genera_mesh()
    print(riepilogo_mesh(mesh_dati))

    # La cartella dei CSV è ricreata da zero a ogni esecuzione, così non
    # restano run di esecuzioni precedenti con parametri diversi.
    cartella = Path(USCITA["cartella"]) / "csv"
    if cartella.exists():
        shutil.rmtree(cartella)
    cartella.mkdir(parents=True, exist_ok=True)

    rng = random.Random(FEM["seed"])
    righe_metadata = []
    for indice in range(1, FEM["num_run"] + 1):
        id_run = f"run_{indice:04d}"
        print(f"Integrazione FEM di {id_run} in corso...")
        riga = esporta_run(id_run, cartella / f"{id_run}.csv", mesh_dati, rng)
        print(
            f"  T vera max al sensore: {riga['T_vera_max_sensore_C']:.1f} °C — "
            f"T misurata max: {riga['T_misurata_max_sensore_C']:.1f} °C"
        )
        righe_metadata.append(riga)

    percorso_metadata = cartella / "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"Esportate {len(righe_metadata)} analisi in: {cartella.resolve()}")
    print(f"Metadata: {percorso_metadata.resolve()}")


if __name__ == "__main__":
    main()