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, spessore_m: float, materiale: dict, h_esterno_W_m2K: float, h_interno_W_m2K: float, h_bordi_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). Tutti e quattro i lati # della sezione includono la convezione verso l'ambiente; su ogni cella # agisce inoltre la conduzione circonferenziale (y) verso il resto della # fascetta, assunto a temperatura ambiente. Il termine è ricavato # considerando l'intero volume della fascia cilindrica (equazione # dell'aletta): il calore conduce lungo y attraverso l'intero spessore # mentre le superfici esterna e interna dell'intero cilindro scambiano # per convezione, dando un sink distribuito uniformemente su ogni cella # q_y = -(h_esterno + h_interno) / spessore * (T - T_amb). 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) b_bordo = h_bordi_W_m2K * dt_s / (rho * cp * dx_m) c_y = (h_esterno_W_m2K + h_interno_W_m2K) * dt_s / (rho * cp * spessore_m) n = n_x * n_z scambio = np.full(n, c_y) scambio[0::n_z] += b_esterno scambio[n_z - 1::n_z] += b_interno scambio[:n_z] += b_bordo scambio[n - n_z:] += b_bordo 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(scambio) ) return splu(sp.csc_matrix(A)) def prepara_stato_termico(fascetta: dict, aria: dict, sorgente: dict) -> dict: # Prepara griglia, coefficienti e solutore fattorizzato per un run: # tutto ciò che resta costante durante l'integrazione temporale. materiale = MATERIALI[fascetta["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 dt = SIMULAZIONE["dt_interno_s"] 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"]) rho = materiale["densita_kg_m3"] cp = materiale["calore_specifico_J_kgK"] alpha = materiale["conducibilita_termica_W_mK"] / (rho * cp) z_centri = (np.arange(n_z) + 0.5) * dz return { "n_x": n_x, "n_z": n_z, "dx_m": dx, "dz_m": dz, "dt_s": dt, "x_centri_m": (np.arange(n_x) + 0.5) * dx, "z_centri_m": z_centri, "skin_depth_m": skin_depth, "rho": rho, "cp": cp, "profilo_z": profilo_deposizione_z_1_m(z_centri, spessore, skin_depth), "b_esterno": aria["h_esterno_W_m2K"] * dt / (rho * cp * dz), "b_interno": aria["h_interno_W_m2K"] * dt / (rho * cp * dz), "b_bordo": aria["h_bordi_W_m2K"] * dt / (rho * cp * dx), "c_y": (aria["h_esterno_W_m2K"] + aria["h_interno_W_m2K"]) * dt / (rho * cp * spessore), "T_ambiente_C": aria["temperatura_ambiente_C"], "solutore": costruisci_solutore_implicito_2d( n_x=n_x, n_z=n_z, dt_s=dt, dx_m=dx, dz_m=dz, spessore_m=spessore, materiale=materiale, 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"], ), } def passo_implicito(stato: dict, T: np.ndarray, q_x: np.ndarray) -> np.ndarray: # Avanza il campo di temperatura di un passo dt: assembla il termine noto # (sorgente volumetrica, convezione sui quattro lati, conduzione # circonferenziale verso l'ambiente) e risolve il sistema implicito. T_amb = stato["T_ambiente_C"] rhs = T + (stato["dt_s"] / (stato["rho"] * stato["cp"])) * ( q_x[:, None] * stato["profilo_z"][None, :] ) rhs += stato["c_y"] * T_amb rhs[:, 0] += stato["b_esterno"] * T_amb rhs[:, -1] += stato["b_interno"] * T_amb rhs[0, :] += stato["b_bordo"] * T_amb rhs[-1, :] += stato["b_bordo"] * T_amb return stato["solutore"].solve(rhs.ravel()).reshape(T.shape) 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"] stato = prepara_stato_termico(fascetta, aria, sorgente) n_x = stato["n_x"] n_z = stato["n_z"] x_centri = stato["x_centri_m"] skin_depth = stato["skin_depth_m"] x_sensore = sensore["x_mm"] / 1000.0 i_sensore = min(n_x - 1, max(0, int(x_sensore / stato["dx_m"]))) dt = stato["dt_s"] durata = SIMULAZIONE["durata_s"] periodo_campionamento = 1.0 / SIMULAZIONE["frequenza_campionamento_hz"] T = np.full((n_x, n_z), aria["temperatura_ambiente_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 ) T = passo_implicito(stato, T, q_x) # 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": fascetta["lunghezza_mm"] / 1000.0, "spessore_m": fascetta["spessore_mm"] / 1000.0, "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"], "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": 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()