EquationMiner/src/fitseg/pipeline.py

"""Fit dei modelli di ogni tipo di dominio su ogni (dominio, serie) e scelta con la regola automatica."""

import numpy as np

from .domini import sigma_rumore
from .fit import adatta
from .modelli import MODELLI
from .selezione import scegli


def _analizza_serie(ty, yy, sigma, tipo, fissi, cfg, rng):
    c, f = cfg["domini"], cfg["fit"]
    voce = {"sigma": float(sigma), "fit": [], "scelta": None}
    if len(yy) < c["n_min_punti"]:
        voce["motivo"] = f"indeterminato: {len(yy)} campioni < n_min_punti"
        return voce
    for m in f["modelli_per_tipo"][tipo]:
        if len(MODELLI[m][1]) > len(yy) - 2:
            continue
        voce["fit"].append(adatta(ty, yy, m, sigma, cfg, rng, fissi if tipo == "riscaldamento" else None))
    scelto, motivo, adeguato = scegli(voce["fit"], sigma, cfg)
    voce["scelta"] = scelto["modello"] if scelto else None
    voce["motivo"], voce["adeguato"] = motivo, adeguato
    return voce


def analizza_domini(t, serie, domini, cfg):
    """Ritorna un risultato per dominio. I domini di raffreddamento si fittano per primi: i loro parametri
    (`riscaldamento_fissa_da_raffreddamento`) vengono fissati nei domini di riscaldamento (stesse costanti fisiche)."""
    c, f = cfg["domini"], cfg["fit"]
    rng = np.random.default_rng(f["seed"])
    risultati = [None] * len(domini)
    fissi = {nome: {} for nome in serie}  # nome serie -> parametri del raffreddamento
    ordine = sorted(range(len(domini)), key=lambda i: domini[i]["tipo"] != "raffreddamento")
    for idx in ordine:
        d = domini[idx]
        i0 = d["i0"] + (c["scarta_dopo_salto_n"] if d["tipo"] != "riposo" else 0)
        i1, t0 = d["i1"], t[d["i0"]]
        voce = {"k": idx + 1, "tipo": d["tipo"], "t0_s": float(t0), "t1_s": float(t[i1 - 1]),
                "n": int(i1 - d["i0"]), "n_fit": int(i1 - i0), "serie": {}}
        for nome, y in serie.items():
            yy = y[i0:i1]
            vs = _analizza_serie(t[i0:i1] - t0, yy, sigma_rumore(yy), d["tipo"], fissi[nome], cfg, rng)
            voce["serie"][nome] = vs
            if d["tipo"] == "raffreddamento" and vs["scelta"] and not fissi[nome]:
                par = next(x for x in vs["fit"] if x["modello"] == vs["scelta"])["parametri"]
                fissi[nome] = {k: par[k] for k in f["riscaldamento_fissa_da_raffreddamento"] if k in par}
        risultati[idx] = voce
    return risultati