Aggiunge genera.py, README e gitignore; output in cartella output/

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

genera.py

@@ -0,0 +1,140 @@
+# /// script
+# requires-python = ">=3.11"
+# dependencies = ["numpy"]
+# ///
+"""Genera un CSV con T e ΔT = T − T_amb con rumore realistico (vedi docs/SPEC.md)."""
+
+import argparse
+import secrets
+import sys
+import tomllib
+from pathlib import Path
+
+import numpy as np
+
+CARTELLA_OUTPUT = Path("output")
+GRIGLIA_MS = 5.0  # passo della griglia fine per il filtro del sensore
+
+CHIAVI = [
+    "durata_s", "dt_ms", "jitter_ms",
+    "amb_inf_c", "amb_A_c", "amb_tau_s", "amb_sigma_c", "amb_passo_c",
+    "sigma_c", "sigma_caldo_c", "x_rif_c", "wander_sigma_c", "wander_tau_s", "passo_T_c",
+    "t_impulsi_s", "dT_impulsi_c", "k_raffreddamento_hz", "tau_sensore_s",
+]
+
+
+def errore(msg):
+    print(f"Errore: {msg}", file=sys.stderr)
+    sys.exit(1)
+
+
+def leggi_config(percorso):
+    try:
+        with open(percorso, "rb") as f:
+            cfg = tomllib.load(f)
+    except OSError as e:
+        errore(f"impossibile leggere '{percorso}': {e.strerror}")
+    except tomllib.TOMLDecodeError as e:
+        errore(f"TOML non valido in '{percorso}': {e}")
+    mancanti = [k for k in CHIAVI if k not in cfg]
+    if mancanti:
+        errore(f"chiavi mancanti in '{percorso}': {', '.join(mancanti)}")
+    if len(cfg["t_impulsi_s"]) != len(cfg["dT_impulsi_c"]):
+        errore("'t_impulsi_s' e 'dT_impulsi_c' hanno lunghezze diverse")
+    return cfg
+
+
+def eccesso(t_s, cfg):
+    """Eccesso termico x(t): un gradino per impulso, poi raffreddamento esponenziale."""
+    x = np.zeros_like(t_s)
+    for tj, dj in zip(cfg["t_impulsi_s"], cfg["dT_impulsi_c"]):
+        dopo = t_s >= tj
+        x[dopo] += dj * np.exp(-cfg["k_raffreddamento_hz"] * (t_s[dopo] - tj))
+    return x
+
+
+def ambiente_vero(t_s, cfg):
+    return cfg["amb_inf_c"] + cfg["amb_A_c"] * np.exp(-t_s / cfg["amb_tau_s"])
+
+
+def quantizza(v, passo):
+    return np.round(v / passo) * passo
+
+
+def genera(cfg, rng):
+    # 1. tempi
+    n_max = int(cfg["durata_s"] * 1000 / (cfg["dt_ms"] - cfg["jitter_ms"])) + 2
+    dt = cfg["dt_ms"] + rng.uniform(-cfg["jitter_ms"], cfg["jitter_ms"], n_max)
+    t_ms = np.cumsum(dt)
+    t_ms = t_ms[t_ms <= cfg["durata_s"] * 1000]
+    dt = np.diff(t_ms, prepend=0.0)
+    t_s = t_ms / 1000
+
+    # 2. T vera su griglia fine = ambiente vero + eccesso termico
+    g_s = np.arange(0.0, cfg["durata_s"] + GRIGLIA_MS / 1000, GRIGLIA_MS / 1000)
+    t_vera = ambiente_vero(g_s, cfg) + eccesso(g_s, cfg)
+
+    # 3. risposta del sensore: filtro del 1° ordine, interpolato ai t_i
+    a = np.exp(-GRIGLIA_MS / 1000 / cfg["tau_sensore_s"])
+    y = np.empty_like(t_vera)
+    y[0] = t_vera[0]
+    for n in range(1, len(y)):
+        y[n] = a * y[n - 1] + (1 - a) * t_vera[n]
+    y_i = np.interp(t_s, g_s, y)
+
+    # 4. rumore di T: bianco con σ(x) + deriva Ornstein-Uhlenbeck, poi quantizzazione
+    x_i = eccesso(t_s, cfg)
+    sigma = cfg["sigma_c"] + (cfg["sigma_caldo_c"] - cfg["sigma_c"]) * np.minimum(1.0, x_i / cfg["x_rif_c"])
+    phi = np.exp(-(dt / 1000) / cfg["wander_tau_s"])
+    inn = rng.standard_normal(len(t_s))
+    w = np.empty_like(t_s)
+    w[0] = cfg["wander_sigma_c"] * inn[0]
+    for i in range(1, len(w)):
+        w[i] = phi[i] * w[i - 1] + cfg["wander_sigma_c"] * np.sqrt(1 - phi[i] ** 2) * inn[i]
+    T = quantizza(y_i + w + sigma * rng.standard_normal(len(t_s)), cfg["passo_T_c"])
+
+    # 5. ambiente misurato e ΔT (dopo rumore e quantizzazione di entrambi i canali)
+    amb = quantizza(
+        ambiente_vero(t_s, cfg) + cfg["amb_sigma_c"] * rng.standard_normal(len(t_s)),
+        cfg["amb_passo_c"],
+    )
+    dT = quantizza(T - amb, cfg["passo_T_c"])
+    return np.round(t_ms).astype(int), T, dT
+
+
+def scrivi(f, t_ms, T, dT):
+    f.write("t_ms,T_C,dT_C\n")
+    for t, a, b in zip(t_ms, T, dT):
+        # "+ 0.0" evita "-0.00"
+        f.write(f"{t},{round(a, 2) + 0.0:.2f},{round(b, 2) + 0.0:.2f}\n")
+
+
+def main():
+    ap = argparse.ArgumentParser(description="Genera un CSV con T e ΔT con rumore realistico.")
+    ap.add_argument("config", help="file di configurazione TOML")
+    ap.add_argument("-o", "--output", help="file CSV di uscita ('-' = stdout; default: output/gen_<seed>.csv)")
+    ap.add_argument("--seed", type=int, help="seme del generatore (default: casuale, stampato su stderr)")
+    args = ap.parse_args()
+
+    cfg = leggi_config(args.config)
+    seed = args.seed if args.seed is not None else secrets.randbelow(2**32)
+    if args.seed is None:
+        print(f"seed: {seed}", file=sys.stderr)
+
+    t_ms, T, dT = genera(cfg, np.random.default_rng(seed))
+
+    if args.output == "-":
+        scrivi(sys.stdout, t_ms, T, dT)
+        return
+    percorso = Path(args.output) if args.output else CARTELLA_OUTPUT / f"gen_{seed}.csv"
+    try:
+        percorso.parent.mkdir(parents=True, exist_ok=True)
+        with open(percorso, "w", newline="\n") as f:
+            scrivi(f, t_ms, T, dT)
+    except OSError as e:
+        errore(f"impossibile scrivere '{percorso}': {e.strerror}")
+    print(f"scritto: {percorso}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()