From d92bcd198b9365e6bc729099f04f6bb8b863cee8 Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Thu, 29 Oct 2026 10:13:01 +0100 Subject: [PATCH] Semplifica: CLI a due comandi, moduli piatti, solo modelli usati; test unitari su segnale sintetico - rimossi scripts/, stadi.py, docs/USO.md, modelli non usati, partenze casuali, output ridondanti - uscita unica (report.md con grafici per dominio, risultati.json, proposta.png) - test: unitari su segnale sintetico, integrazione su dati/gen_*.csv (saltati se mancano) Co-Authored-By: Claude Sonnet 5.5 --- CLAUDE.md | 40 +++++------ README.md | 129 ++++++++++----------------------- config.toml | 2 - docs/SPEC.md | 141 +++++++++++-------------------------- docs/USO.md | 27 ------- src/fitseg/analisi.py | 63 +++++++++++++++++ src/fitseg/cli.py | 91 ++++++++---------------- src/fitseg/config.py | 4 +- src/fitseg/domini.py | 4 +- src/fitseg/fit.py | 115 +++++++++--------------------- src/fitseg/modelli.py | 26 ++----- src/fitseg/pipeline.py | 49 ------------- src/fitseg/riepilogo.py | 141 ------------------------------------- src/fitseg/selezione.py | 25 ------- src/fitseg/uscita.py | 97 +++++++++++++++++++++++++ tests/conftest.py | 43 +++++++++++ tests/test_config_dati.py | 46 ++++++++++++ tests/test_domini.py | 68 ++++++++++++++++++ tests/test_fit.py | 94 +++++++++++++++++++++++++ tests/test_integrazione.py | 60 ++++++++++++++++ tests/test_pipeline.py | 55 --------------- tests/test_selezione.py | 41 +++++++++++ 22 files changed, 678 insertions(+), 683 deletions(-) delete mode 100644 docs/USO.md create mode 100644 src/fitseg/analisi.py delete mode 100644 src/fitseg/pipeline.py delete mode 100644 src/fitseg/riepilogo.py delete mode 100644 src/fitseg/selezione.py create mode 100644 src/fitseg/uscita.py create mode 100644 tests/conftest.py create mode 100644 tests/test_config_dati.py create mode 100644 tests/test_domini.py create mode 100644 tests/test_fit.py create mode 100644 tests/test_integrazione.py delete mode 100644 tests/test_pipeline.py create mode 100644 tests/test_selezione.py diff --git a/CLAUDE.md b/CLAUDE.md index 776b053..bf482c8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -2,41 +2,37 @@ This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. -`fitseg` legge un CSV `t_ms,T_C,dT_C` (temperatura di un sensore IR che si scalda e si raffredda), lo divide in **domini** (riposo / riscaldamento = gradini di flusso in entrata / raffreddamento) e trova per ogni (dominio, serie) l'equazione che meglio lo approssima. Documentazione, commenti e messaggi sono in italiano: mantieni la lingua. +`fitseg` legge un CSV `t_ms,T_C,dT_C` (temperatura di un sensore IR), lo divide in **domini** (riposo / riscaldamento = gradini / raffreddamento) e trova per ogni (dominio, serie) l'equazione che meglio lo approssima. Documentazione, commenti e messaggi sono in italiano: mantieni la lingua. -Fonti di verità (leggile prima di cambiare il comportamento, non duplicarne il contenuto): -- [docs/SPEC.md](docs/SPEC.md) — specifica, stato di implementazione (§1), algoritmi, limiti noti. [docs/USO.md](docs/USO.md) — uso pratico. -- [config.toml](config.toml) — unica fonte dei valori. Nessun default nel codice: ogni chiave nuova va aggiunta a `CHIAVI` in [src/fitseg/config.py](src/fitseg/config.py) (mancante ⇒ exit 3) e descritta in SPEC.md. +Fonti di verità (non duplicarne il contenuto): [docs/SPEC.md](docs/SPEC.md) (algoritmi, limiti), [README.md](README.md) (uso, test), [config.toml](config.toml) (unica fonte dei valori: nessun default nel codice, ogni chiave nuova va in `CHIAVI` di [src/fitseg/config.py](src/fitseg/config.py) e in SPEC). ## Comandi ```bash uv venv .venv && uv pip install -e ".[dev]" -.venv/bin/fitseg domini dati/gen_1.csv # propone i domini → lavoro/domini.toml -.venv/bin/fitseg analizza dati/gen_1.csv [--domini lavoro/domini.toml] # → lavoro/out/ (report.md, modello_finale.json, fit_*.png) -.venv/bin/pytest # tutti i test; singolo: .venv/bin/pytest -k raffreddamento +.venv/bin/fitseg domini dati/gen_1.csv # suddivisione: lavoro/domini.toml + proposta.png, nessun fit +.venv/bin/fitseg analizza dati/gen_1.csv [--domini lavoro/domini.toml] # → lavoro/out/ (report.md, risultati.json, dominio*_*.png) +.venv/bin/pytest # tutti; singolo: -k raffreddamento o tests/test_fit.py +.venv/bin/ruff check src tests ``` -Opzioni globali `--config`/`--lavoro` vanno **prima** del sottocomando. `lavoro/` è ignorato da git. +`--config`/`--lavoro` vanno **prima** del sottocomando. `lavoro/` e `dati/*.csv` sono ignorati da git. -## Dati +## Architettura (`src/fitseg/`, un modulo per responsabilità) -`dati/gen_.csv` sono prodotti dal repository fratello `../gen_data` (`.venv/bin/python genera.py config.toml --seed N -o ...`). Sono i dati dei test: verità nota τ_raffreddamento = 1/k = 15.9 s (k perturbato del 2 % per seed), 3 gradini a ≈15.7/17.8/21.7 s, T ambiente ≈ 23 °C. Il generatore inserisce anche scarti casuali (≤ 4 °C) che rendono i residui autocorrelati: è un limite noto, non un bug del fit. +`cli.py` (due sottocomandi, unico punto d'ingresso) → `config.py`, `dati.py` (lettura) → `domini.py` (gradini ⇒ domini, I/O di `domini.toml`) → `analisi.py` (ordina i domini, sceglie il modello) → `fit.py` (`adatta`) con `modelli.py` (registro `MODELLI`) → `uscita.py` (report, JSON, grafici). -## Architettura (`src/fitseg/`) +- I domini derivano **solo dai gradini positivi** di `serie_riferimento` (`diff > soglia_gradino_c`), con arretramento ai campioni di transizione; `domini.toml` è modificabile e deve essere contiguo. +- `analisi.analizza_domini`: i domini di raffreddamento si fittano **per primi**; i loro `c`/`tau` sono fissati nei domini di riscaldamento (`riscaldamento_fissa_da_raffreddamento`, `adatta(..., fissi=)`). I modelli provati dipendono dal tipo (`[fit.modelli_per_tipo]`: riscaldamento e raffreddamento solo `esp_decad`). +- `fit.adatta`: `least_squares` con perdita robusta; l'esponenziale parte da una griglia su τ. "Non identificabile" = condizione > soglia **o** IC 95 % > |valore| per un parametro diverso da `a`/`c`. σ del rumore **per dominio** da `MAD(d²y)`; tempo traslato a `t' = t − t0`. +- Nuovo modello = voce in `modelli.py` + riga in SPEC §3 + nome in `[fit.modelli_per_tipo]`. Non ci sono partenze casuali: il programma è deterministico. -Flusso: `domini.proponi` → `pipeline.analizza_domini` (per ogni dominio e serie: `fit.adatta` su ogni modello abilitato → `selezione.scegli`) → `riepilogo` (report, JSON, grafici). `cli.py` è l'unico punto di I/O oltre a `riepilogo`/`dati`. +## Test -- [domini.py](src/fitseg/domini.py): i domini derivano **solo dai gradini positivi** della `serie_riferimento` (`diff > soglia_gradino_c`), con arretramento ai campioni di transizione; non c'è più PELT/fusione BIC (presente nelle prime versioni della spec). `domini.toml` è modificabile e rientra con `--domini`; deve essere contiguo. -- [modelli.py](src/fitseg/modelli.py): registro `MODELLI` (funzione, parametri, equazione testuale, indici dei τ). Nuovo modello = voce qui + riga in SPEC §4. -- [fit.py](src/fitseg/fit.py): `least_squares` con perdita robusta; gli esponenziali partono da una griglia su τ (VARPRO). "Non identificabile" = condizione > soglia **o** IC 95 % > |valore| per un parametro diverso dal primo; la `logistica` ha `t_m` limitato a [0, L] altrimenti degenera in esponenziale. -- [pipeline.py](src/fitseg/pipeline.py): i domini di raffreddamento si fittano **per primi** e i loro `c`/`tau` vengono fissati nei domini di riscaldamento (`riscaldamento_fissa_da_raffreddamento`, `adatta(..., fissi=)`); i modelli provati dipendono dal tipo (`[fit.modelli_per_tipo]`). -- [selezione.py](src/fitseg/selezione.py): regola SPEC §5.3 (validi → adeguati se RMSE ≤ soglia·σ → entro ΔAICc vince chi ha meno parametri). -- σ del rumore è stimata **per dominio** da `MAD(d²y)`; il tempo del fit è traslato a `t' = t − t0` del dominio. +Unitari su un segnale sintetico generato in [tests/conftest.py](tests/conftest.py) (τ = 16 s, 3 gradini, nessun disturbo a salto): non servono i CSV. Integrazione su `dati/gen_{1,2,3}.csv` (prodotti da `../gen_data`: `.venv/bin/python genera.py config.toml --seed N -o ...`) e sulla CLI: saltati se i CSV mancano. Verità di `gen_data`: τ = 1/k = 15.9 s (k perturbato del 2 % per seed), 3 gradini a ≈15.7/17.8/21.7 s, ambiente ≈ 23 °C, scarti casuali ≤ 4 °C che rendono i residui autocorrelati (limite noto, non un bug). ## Gotcha -- `scarta_dopo_salto_n` deve coprire i campioni di transizione del sensore (τ_sensore 0.07 s, dt 0.15 s ⇒ ≥ 2–3 campioni); se troppo basso, 1–2 punti a ~100 °C dal fit sporcano la fine del dominio. -- Domini di riscaldamento di 14–26 campioni: τ non identificabile da solo, quindi è fisso al valore del raffreddamento; gli IC di `A` sono condizionati a quei fissi. Se il raffreddamento non è valido, il riscaldamento ricade sul fit libero e può risultare indeterminato. -- Cambiando `config.toml` del generatore (gradini, rumore) ricontrolla `soglia_gradino_c` e i test in [tests/test_pipeline.py](tests/test_pipeline.py), che usano `dati/gen_{1,2,3}.csv`. -- Il parser CSV richiede la colonna tempo e tutte le `serie` in config; il tempo deve essere strettamente crescente (exit 4 altrimenti). +- `scarta_dopo_salto_n` deve coprire i campioni di transizione del sensore (τ_sensore 0.07 s, dt 0.15 s ⇒ ≥ 2–3 campioni); se troppo basso 1–2 punti a ~100 °C sporcano la fine del dominio. +- Domini di riscaldamento di 14–26 campioni: con gli scarti casuali τ non è identificabile da solo (su dati puliti sì), quindi è fisso al valore del raffreddamento; gli IC di `A` sono condizionati. Senza un raffreddamento valido il riscaldamento torna al fit libero e può risultare indeterminato. +- Cambiando la config di `gen_data` (gradini, rumore) ricontrolla `soglia_gradino_c` e i test di integrazione. - Exit code: 0 ok · 2 uso · 3 config · 4 CSV · 6 domini non validi. diff --git a/README.md b/README.md index 5623676..a061f99 100644 --- a/README.md +++ b/README.md @@ -1,35 +1,21 @@ # fitseg -Legge un CSV `t_ms,T_C,dT_C` (temperatura di un sensore IR che si scalda e si raffredda), riconosce quando c'è **riscaldamento** (gradini di flusso in entrata) e quando c'è **raffreddamento**, divide il tempo in **domini** e trova per ogni dominio e ogni serie l'equazione che approssima meglio i dati, con minimi quadrati non lineari. +Legge un CSV `t_ms,T_C,dT_C` (temperatura di un sensore IR che si scalda e si raffredda), riconosce **riposo**, **riscaldamento** (gradini di flusso in entrata) e **raffreddamento**, lo divide in **domini** e trova per ogni dominio e ogni serie l'equazione che lo approssima meglio, con minimi quadrati non lineari. -Risultato per ogni dominio: modello scelto, parametri con intervallo di confidenza al 95 %, metriche (RMSE, R², AICc, BIC, autocorrelazione dei residui), grafici e un report in Markdown. - -## Requisiti - -- Python ≥ 3.11 -- [uv](https://docs.astral.sh/uv/) (consigliato): `curl -LsSf https://astral.sh/uv/install.sh | sh` -- Dipendenze: `numpy`, `scipy`, `matplotlib` (più `pytest` e `ruff` per lo sviluppo) +Risultato per dominio: modello scelto, parametri con intervallo di confidenza al 95 %, metriche (RMSE, R², AICc, BIC, autocorrelazione dei residui), un grafico dati/equazioni e un report Markdown. ## Installazione +Python ≥ 3.11 e [uv](https://docs.astral.sh/uv/) (`curl -LsSf https://astral.sh/uv/install.sh | sh`): + ```bash uv venv .venv -uv pip install -e ".[dev]" +uv pip install -e ".[dev]" # numpy, scipy, matplotlib; pytest e ruff per lo sviluppo ``` -Il programma si lancia con `.venv/bin/fitseg` (oppure `source .venv/bin/activate` e poi `fitseg`). ## Dati di ingresso -Un CSV con intestazione, tempo crescente e una o più serie: - -``` -t_ms,T_C,dT_C -151,22.90,0.20 -301,22.63,0.03 -``` -I nomi delle colonne, l'unità del tempo (`ms` o `s`) e il separatore sono in [config.toml](config.toml), sezione `[dati]`. - -I CSV di prova vengono dal generatore `../gen_data` (repository separato). I test usano i seed 1, 2 e 3 in `dati/` (cartella ignorata da git): +CSV con intestazione, tempo crescente e le serie indicate in [config.toml](config.toml) (`[dati]`: colonne, unità `ms`/`s`, separatore). I CSV di prova vengono dal generatore `../gen_data`; i test di integrazione usano i seed 1–3 in `dati/` (ignorata da git): ```bash mkdir -p dati @@ -41,107 +27,70 @@ done ## Uso ```bash -fitseg domini dati/gen_1.csv # solo i domini: tabella a schermo + lavoro/domini.toml -fitseg analizza dati/gen_1.csv # domini + fit + scelta + report in lavoro/out/ -fitseg modelli # catalogo dei modelli con equazioni e parametri -fitseg valuta lavoro/out/modello_finale.json --t 10 30 60 # valuta il modello a tratti in questi istanti +.venv/bin/fitseg domini dati/gen_1.csv # come verrebbero suddivisi i dati: tabella, lavoro/domini.toml, lavoro/proposta.png (nessun fit) +.venv/bin/fitseg analizza dati/gen_1.csv # domini + regressione + report in lavoro/out/ ``` +`--config PATH` (default `config.toml`) e `--lavoro DIR` (default `lavoro`) vanno **prima** del sottocomando. -Le opzioni globali `--config PATH` (default `config.toml`) e `--lavoro DIR` (default `lavoro`) vanno **prima** del sottocomando: - +**Correggere i domini a mano**: apri `lavoro/domini.toml`, modifica i tempi (i domini devono restare contigui) e rilancia ```bash -fitseg --config altra.toml --lavoro prova1 analizza dati/gen_2.csv +.venv/bin/fitseg analizza dati/gen_1.csv --domini lavoro/domini.toml ``` - -### Correggere i domini a mano - -`fitseg domini` (e `analizza`) scrive la proposta automatica in `lavoro/domini.toml`: - ```toml [[dominio]] tipo = "riscaldamento" # riposo | riscaldamento | raffreddamento t0_s = 15.8 # incluso t1_s = 17.76 # escluso (l'ultimo dominio arriva a fine dati) ``` -Modifica i tempi, aggiungi o togli blocchi `[[dominio]]` (devono restare contigui) e rilancia: -```bash -fitseg analizza dati/gen_1.csv --domini lavoro/domini.toml -``` - -## Cosa produce - -Tutto in `lavoro/` (ignorata da git): +### Cosa produce (in `lavoro/`, ignorata da git) | File | Contenuto | |---|---| -| `domini.toml` | proposta dei domini (o quelli che hai usato) | -| `dati.csv` | copia del CSV analizzato | -| `out/report.md` | con i grafici incorporati (si vede bene in un visualizzatore Markdown); per dominio e serie: modello scelto, parametri ± IC 95 %, metriche, tabella di confronto di tutti i modelli | -| `out/equazioni.md` | solo le equazioni scelte | -| `out/modello_finale.json` | modello e parametri per dominio e serie (input di `valuta`) | +| `domini.toml`, `proposta.png` | domini proposti e grafico dei dati con la suddivisione | +| `out/report.md` | per dominio e serie: grafico, modello scelto con parametri ± IC 95 %, metriche, tabella di tutti i modelli provati | +| `out/dominio_.png` | dati grezzi contro il modello scelto (colore del dominio) e gli altri modelli validi (tratteggiati), più residui; campioni esclusi dal fit in grigio | | `out/risultati.json` | tutti i fit con tutte le metriche | -| `out/fit_.png` | panoramica: dati, fit a tratti (un colore per tipo di dominio) e residui | -| `out/dominio_.png` | un grafico per dominio e serie: dati grezzi contro il modello scelto (rosso) e gli altri modelli validi (tratteggiati), più residui; i campioni esclusi dal fit sono in grigio | -### Come leggere il report - -- **Scelto**: il modello con meno parametri tra quelli equivalenti per AICc (ΔAICc ≤ 2) e validi (convergenti, parametri lontani dai limiti, identificabili). -- **modello non adeguato**: l'RMSE supera 1.3 volte il rumore stimato. Il modello migliore disponibile non spiega tutto: guarda l'ACF1 dei residui. -- **Indeterminato**: nessun modello è valido, oppure il dominio ha meno di `n_min_punti` campioni. -- Nel raffreddamento, `τ` è la costante di tempo e `c` la temperatura asintotica (≈ ambiente per `T_C`, ≈ 0 per `dT_C`). +Come leggere il report: **non adeguato** = RMSE oltre 1.3 volte il rumore stimato (guarda l'ACF1 dei residui); **indeterminato** = nessun modello valido o dominio con meno di `n_min_punti` campioni. Nel raffreddamento `τ` è la costante di tempo e `c` l'asintoto (≈ ambiente per `T_C`, ≈ 0 per `dT_C`). ## Come funziona -1. **Domini**: un salto positivo tra due campioni consecutivi sopra `soglia_gradino_c` è un gradino di riscaldamento. Prima del primo gradino il dominio è `riposo`, tra due gradini `riscaldamento`, dopo l'ultimo `raffreddamento`. -2. **Fit**: in ogni dominio, con il tempo traslato a `t' = t − t0`, si fittano i modelli del suo tipo con `scipy.optimize.least_squares` e perdita robusta. Riscaldamento e raffreddamento sono esponenziali; nel riscaldamento `c` e `τ` sono fissati ai valori del raffreddamento (i tratti sono troppo corti per stimarli) e si stima solo l'ampiezza. -3. **Scelta**: regola automatica (SPEC §5.3), senza intervento umano. +1. **Domini** (`domini.py`): un salto positivo tra due campioni consecutivi sopra `soglia_gradino_c` è un gradino di riscaldamento. Prima del primo gradino: `riposo`; tra due gradini: `riscaldamento`; dopo l'ultimo: `raffreddamento`. +2. **Modelli** (`modelli.py`): `riposo` → `costante`, `lineare`, `esp_decad`; `riscaldamento` e `raffreddamento` → solo `esp_decad` `y = c + A·exp(−t'/τ)`, con `t' = t − t0` del dominio. +3. **Fit** (`fit.py`): `scipy.optimize.least_squares` con perdita robusta; partenza dell'esponenziale da una griglia su τ. I domini di raffreddamento si fittano per primi e nei domini di riscaldamento `c` e `τ` sono **fissati** ai loro valori (stesse costanti fisiche; in 2–4 s di dati con scarti casuali τ non si stima da solo), restando libera solo `A`. Gli IC di `A` sono condizionati ai valori fissi. +4. **Scelta** (`analisi.py`): scarta i modelli non convergenti, con parametri a limite o non identificabili; preferisce i modelli «adeguati» (RMSE ≤ `rmse_vs_rumore_max` × σ rumore); tra quelli entro `soglia_delta_aicc` dal miglior AICc vince chi ha meno parametri. -Dettagli, modelli e limiti: [docs/SPEC.md](docs/SPEC.md). Uso esteso: [docs/USO.md](docs/USO.md). - -## Modelli - -| Id | Equazione | -|---|---| -| `costante` | `y = a` | -| `lineare` | `y = a + b·t'` | -| `quadratico` | `y = a + b·t' + c·t'²` | -| `esp_decad` | `y = c + A·exp(−t'/τ)` | -| `esp_doppio` | `y = c + A1·exp(−t'/τ1) + A2·exp(−t'/τ2)` | -| `logistica` | `y = c + A / (1 + exp(−(t'−t_m)/w))` | - -Per aggiungerne uno: nuova voce in `MODELLI` di [src/fitseg/modelli.py](src/fitseg/modelli.py), nuova riga in SPEC §4 e nome nelle liste di `[fit.modelli_per_tipo]` dei tipi di dominio in cui va provato. +Specifica e limiti: [docs/SPEC.md](docs/SPEC.md). ## Configurazione -Tutti i valori sono in [config.toml](config.toml); non c'è nessun default nel codice (chiave mancante = errore). Quelle che conviene conoscere: +Tutti i valori sono in [config.toml](config.toml), nessun default nel codice (chiave mancante = errore, exit 3). Le principali: `[domini] soglia_gradino_c` (20 °C), `scarta_dopo_salto_n` (campioni di transizione esclusi dal fit), `n_min_punti`; `[fit.modelli_per_tipo]` (modelli per tipo di dominio); `[fit] riscaldamento_fissa_da_raffreddamento` (`["c", "tau"]`, `[]` = parametri liberi), `soglia_delta_aicc`, `rmse_vs_rumore_max`. -| Chiave | Effetto | -|---|---| -| `[domini] soglia_gradino_c` | salto minimo che segna un gradino (20 °C); da adattare se i gradini reali sono più piccoli | -| `[domini] scarta_dopo_salto_n` | campioni esclusi all'inizio di un dominio dopo un gradino (transizione del sensore) | -| `[domini] n_min_punti` | campioni minimi per dominio | -| `[fit.modelli_per_tipo]` | modelli provati per tipo di dominio (riposo: costante/lineare/esponenziale; riscaldamento e raffreddamento: esponenziale) | -| `[fit] riscaldamento_fissa_da_raffreddamento` | parametri dei domini di riscaldamento fissati al valore del raffreddamento (`["c", "tau"]`; `[]` = liberi) | -| `[fit] soglia_delta_aicc` | entro questa differenza vince il modello con meno parametri | -| `[fit] rmse_vs_rumore_max` | soglia di «adeguato» (RMSE ≤ valore × σ rumore) | -| `[fit] seed` | seme delle partenze multiple: stessi dati e stessa config danno gli stessi risultati | +Nuovo modello: voce in `MODELLI` di [src/fitseg/modelli.py](src/fitseg/modelli.py), riga in SPEC §4 e nome in `[fit.modelli_per_tipo]`. ## Test ```bash -.venv/bin/pytest # tutti -.venv/bin/pytest -k raffreddamento # uno solo +.venv/bin/pytest # tutti (≈ 4 s) +.venv/bin/pytest tests/test_fit.py # un file +.venv/bin/pytest -k raffreddamento # per nome +.venv/bin/ruff check src tests ``` -I test verificano, sui seed 1–3, che i domini siano `riposo, riscaldamento, riscaldamento, raffreddamento`, che nel raffreddamento vinca `esp_decad` con τ entro il 10 % di 15.9 s, che nel riscaldamento sia `esp_decad` con `c` e `τ` del raffreddamento e che il riposo sia `costante`. Richiedono `dati/gen_1.csv`, `gen_2.csv`, `gen_3.csv` (vedi sopra). -## Codici di uscita +| File | Verifica | Servono i CSV? | +|---|---|---| +| `test_domini.py` | 3 gradini, nessun gradino, un solo gradino, gradino sotto soglia, confini vicini ai gradini veri, `domini.toml` (andata e ritorno, non contiguo, tipo non valido) | no | +| `test_fit.py` | recupero di τ, `c`, `A` su segnale sintetico (τ = 16 s), parametri fissi, τ non identificabile su tratti corti e rumorosi, analisi completa, dominio troppo corto, riproducibilità | no | +| `test_selezione.py` | regola di scelta | no | +| `test_config_dati.py` | errori di config e CSV | no | +| `test_integrazione.py` | domini e τ su `dati/gen_{1,2,3}.csv`; CLI `domini` e `analizza --domini`; codici di uscita | sì (altrimenti saltati) | -0 ok · 2 uso errato · 3 configurazione non valida · 4 CSV illeggibile o non valido · 6 domini non validi. +I test unitari generano il segnale da soli (`tests/conftest.py`). Prova manuale: `fitseg analizza dati/gen_1.csv` e controlla in `lavoro/out/report.md` 4 domini (riposo, due riscaldamenti, raffreddamento) con τ ≈ 16 s. + +Codici di uscita: 0 ok · 3 configurazione non valida · 4 CSV illeggibile/non valido · 6 `domini.toml` non valido. ## Limiti noti -- Nel raffreddamento i residui sono autocorrelati (ACF1 ≈ 0.7) per scarti casuali e deriva lenta del sensore: i parametri sono buoni, ma gli intervalli di confidenza sono troppo stretti e il dominio risulta «non adeguato». -- I domini di riscaldamento sono corti (2–4 s): τ non è identificabile da solo e viene preso dal raffreddamento; gli IC di `A` sono condizionati a quei valori fissi. -- Ogni dominio è fittato in modo indipendente: nessun vincolo di continuità tra domini. -- Provato su 3 seed; non su file senza gradini o con un solo gradino. +- Nel raffreddamento i residui sono autocorrelati (ACF1 ≈ 0.7) per scarti casuali e deriva lenta del sensore: i parametri sono buoni, ma gli IC sono troppo stretti e il dominio risulta «non adeguato». +- Ogni dominio è fittato in modo indipendente (nessun vincolo di continuità); provato su pochi seed e con la config d'esempio di `gen_data`. diff --git a/config.toml b/config.toml index e3c3eeb..788ffd7 100644 --- a/config.toml +++ b/config.toml @@ -17,7 +17,6 @@ scarta_dopo_salto_n = 3 # campioni esclusi dal fit all'inizio di un d [fit] riscaldamento_fissa_da_raffreddamento = ["c", "tau"] # parametri dei domini di riscaldamento fissati ai valori del fit del raffreddamento ([] = liberi) -n_start = 20 # partenze multiple per i modelli non lineari (logistica) perdita = "soft_l1" # "linear" | "soft_l1" | "huber" (robusta agli outlier) scala_perdita_sigma = 3.0 # scala della perdita robusta, in unità di sigma rumore tau_min_rel = 0.05 # limite inferiore di tau, come frazione della durata del dominio @@ -28,7 +27,6 @@ correggi_autocorrelazione = true # allarga gli intervalli di confidenza se ACF soglia_acf = 0.3 soglia_delta_aicc = 2.0 # entro questa differenza vince il modello con meno parametri rmse_vs_rumore_max = 1.3 # RMSE <= valore × sigma rumore => modello adeguato -seed = 0 # seme delle partenze multiple [fit.modelli_per_tipo] # modelli provati in ogni tipo di dominio (riscaldamento e raffreddamento: esponenziali) riposo = ["costante", "lineare", "esp_decad"] diff --git a/docs/SPEC.md b/docs/SPEC.md index d3be3e6..757ce40 100644 --- a/docs/SPEC.md +++ b/docs/SPEC.md @@ -1,129 +1,68 @@ -# `fitseg` — domini di riscaldamento/raffreddamento e equazione per ciascuno — specifica +# `fitseg` — specifica -**Obiettivo**: leggere un CSV (`t_ms,T_C,dT_C`, p.es. prodotto da `gen_data`), riconoscere quando c'è **riscaldamento** (flusso in -entrata: gradini di temperatura) e quando c'è **raffreddamento**, dividere il tempo in **domini**, e per ogni (dominio, serie) -trovare con minimi quadrati non lineari l'equazione che approssima meglio i dati. Valori configurabili: solo in -[`config.toml`](../config.toml). Uso pratico: [USO.md](USO.md). +**Obiettivo**: leggere un CSV (`t_ms,T_C,dT_C`, p.es. da `gen_data`), riconoscere **riposo**, **riscaldamento** (flusso in entrata: gradini di temperatura) e **raffreddamento**, dividere il tempo in **domini** e trovare per ogni (dominio, serie) l'equazione che lo approssima meglio. Valori configurabili: solo in [`config.toml`](../config.toml). Uso: [README](../README.md). -## 1. Stato di implementazione - -| Parte | Stato | -|---|---| -| Riconoscimento dei domini (§3), fit (§5), regola di scelta (§5.3), report, grafici, `valuta` | **implementato** (`src/fitseg/`) | -| Modifica manuale dei domini (`domini.toml` + `analizza --domini`) | **implementato** (senza editor interattivo) | -| PELT/fusione BIC per cambi di regime non legati ai gradini | **eliminato**: i domini derivano dai gradini di riscaldamento (§3) | - -## 2. Flusso +## 1. Flusso ``` -CSV ─► [1 DOMINI] gradini ⇒ riposo | riscaldamento… | raffreddamento ─► lavoro/domini.toml (modificabile) +CSV ─► [DOMINI] gradini ⇒ riposo | riscaldamento… | raffreddamento ─► lavoro/domini.toml (modificabile), proposta.png ▼ - [2 FIT] per ogni (dominio, serie): tutti i modelli abilitati ─► regola di scelta ─► lavoro/out/ - ▼ - [3 RIEPILOGO] report.md, equazioni.md, modello_finale.json, risultati.json, fit_.png + [FIT] per (dominio, serie): modelli del tipo ─► regola di scelta ─► lavoro/out/report.md, risultati.json, dominio*_*.png ``` -Il **motore Python** fa tutti i calcoli (deterministici, seed in config); la scelta del modello segue una regola automatica (§5.3). +Tutto il calcolo è deterministico (nessuna partenza casuale, nessun LLM). -## 3. Domini (`fitseg domini`) +## 2. Domini (`domini.py`) -Struttura fisica del segnale (vedi `gen_data/docs/SPEC.md`): T parte dall'ambiente, **gradini istantanei positivi** (riscaldamento, -2–4 s uno dall'altro) ognuno seguito da un decadimento esponenziale; dopo l'ultimo gradino resta solo il **raffreddamento** -esponenziale verso l'ambiente. Durante il riscaldamento e il raffreddamento compaiono anche scarti casuali (±4 °C, rientro < 1 s). +Struttura fisica (vedi `gen_data/docs/SPEC.md`): T parte dall'ambiente, **gradini istantanei positivi** (riscaldamento, 2–4 s uno dall'altro) ognuno seguito da un decadimento esponenziale; dopo l'ultimo resta solo il **raffreddamento** esponenziale verso l'ambiente. Compaiono anche scarti casuali (±4 °C, rientro < 1 s). -Algoritmo (nessun LLM), sulla serie `serie_riferimento`: -1. `d = diff(y)`; gradino = `d > soglia_gradino_c`; rilevazioni più vicine di `distanza_min_salti_campioni` sono lo stesso gradino. -2. Il gradino parte dal primo campione già in salita: si arretra finché `d > soglia_bordo_sigma · σ_d` (σ_d = MAD dei salti). Quel - campione apre il nuovo dominio; i primi `scarta_dopo_salto_n` campioni del dominio sono esclusi dal fit (transizione del sensore). -3. Tipi: prima del primo gradino `riposo`; tra due gradini `riscaldamento`; dopo l'ultimo `raffreddamento`. Senza gradini: un solo `riposo`. +1. `d = diff(y)` sulla `serie_riferimento`; gradino = `d > soglia_gradino_c`; rilevazioni più vicine di `distanza_min_salti_campioni` sono lo stesso gradino. +2. Il gradino parte dal primo campione già in salita (si arretra finché `d > soglia_bordo_sigma · σ_d`, σ_d = MAD dei salti). Quel campione apre il nuovo dominio; i primi `scarta_dopo_salto_n` campioni sono esclusi dal fit (transizione del sensore). +3. Prima del primo gradino `riposo`; tra due gradini `riscaldamento`; dopo l'ultimo `raffreddamento`. Senza gradini: un solo `riposo`. -Ipotesi: tempo crescente (H1); rumore circa gaussiano con σ stimato per dominio da `1.4826·MAD(d²y)/√6` (H2); le serie condividono i -domini (H3); i gradini sono salti positivi ≫ rumore e scarti casuali (H4, `soglia_gradino_c` ≫ `salti_max_c` del generatore). +Ipotesi: tempo crescente; rumore circa gaussiano, σ stimato per dominio da `1.4826·MAD(d²y)/√6`; le serie condividono i domini; i gradini sono salti positivi ≫ rumore e scarti casuali (`soglia_gradino_c` ≫ ampiezza degli scarti). -`domini.toml` (scritto in `lavoro/`, modificabile; `analizza --domini FILE` lo usa al posto della proposta): -```toml -[[dominio]] -tipo = "riscaldamento" # riposo | riscaldamento | raffreddamento -t0_s = 15.8 # incluso -t1_s = 17.76 # escluso (ultimo dominio: fino a fine dati) -``` -I domini devono essere contigui; in caso contrario exit 6. +`domini.toml` è modificabile (`tipo`, `t0_s` incluso, `t1_s` escluso; ultimo dominio fino a fine dati); i domini devono essere contigui, altrimenti exit 6. -## 4. Modelli (`modelli.py`, t' = t − t0 del dominio) +## 3. Modelli (`modelli.py`, t' = t − t0 del dominio) -| Id | Equazione | Parametri | -|---|---|---| -| `costante` | `y = a` | a | -| `lineare` | `y = a + b·t'` | a, b | -| `quadratico` | `y = a + b·t' + c·t'²` | a, b, c | -| `esp_decad` | `y = c + A·exp(−t'/τ)` | c, A, τ | -| `esp_doppio` | `y = c + A1·exp(−t'/τ1) + A2·exp(−t'/τ2)`, τ1 < τ2 | c, A1, τ1, A2, τ2 | -| `logistica` | `y = c + A / (1 + exp(−(t'−t_m)/w))`, 0 ≤ t_m ≤ L | c, A, t_m, w | +| Id | Equazione | Parametri | Tipi di dominio (`[fit.modelli_per_tipo]`) | +|---|---|---|---| +| `costante` | `y = a` | a | riposo | +| `lineare` | `y = a + b·t'` | a, b | riposo | +| `esp_decad` | `y = c + A·exp(−t'/τ)` | c, A, τ | riposo, riscaldamento, raffreddamento | -Nuovo modello = nuova voce nel registro `MODELLI` e nuova riga qui. I modelli provati dipendono dal tipo di dominio (`[fit.modelli_per_tipo]`): `riposo` → `costante`, `lineare`, `esp_decad`; `riscaldamento` e `raffreddamento` → solo `esp_decad` (sono decadimenti esponenziali). Nel raffreddamento `c` è la temperatura asintotica (≈ ambiente) e -`τ = 1/k`; nel riposo `costante`/`lineare` descrivono la baseline. +Riscaldamento e raffreddamento sono esponenziali. Nel raffreddamento `c` è l'asintoto (≈ ambiente) e `τ = 1/k`. Nuovo modello = voce in `MODELLI`, riga qui, nome in config. -## 5. Fit +## 4. Fit (`fit.py`) 1. Dati del dominio senza i primi `scarta_dopo_salto_n` campioni (se non `riposo`); tempo traslato. -2. Partenze: lineari per i polinomi; **proiezione delle variabili** per `esp_decad`/`esp_doppio` (griglia log-spaziata su τ tra - `tau_min_rel·L` e `tau_max_rel·L`, L = durata, ampiezze per minimi quadrati lineari); `n_start` partenze casuali per `logistica`. -3. `scipy.optimize.least_squares` con perdita `perdita` (robusta `soft_l1`, scala `scala_perdita_sigma·σ`) e limiti sui parametri. -4. Incertezze: `s²(JᵀJ)⁻¹`, IC 95 % = ±1.96σ_p; con `correggi_autocorrelazione` e ACF1 > `soglia_acf` moltiplicate per `√((1+ρ)/(1−ρ))`. -5. Metriche: RSS, RMSE, R², AICc, BIC, ACF1 dei residui, parametri a limite, numero di condizione (jacobiano normalizzato). - **Non identificabile** se condizione > `soglia_condizione` oppure un parametro (escluso il primo) ha IC 95 % > |valore|. +2. Partenza: polinomi per minimi quadrati; esponenziale con griglia log-spaziata su τ (`tau_min_rel·L` … `tau_max_rel·L`, L = durata) e `c`, `A` lineari. +3. `scipy.optimize.least_squares`, perdita `perdita` (robusta `soft_l1`, scala `scala_perdita_sigma·σ`), limiti su τ. +4. IC 95 % = ±1.96·σ_p da `s²(JᵀJ)⁻¹`, moltiplicati per `√((1+ρ)/(1−ρ))` se ACF1 > `soglia_acf` (`correggi_autocorrelazione`). +5. Metriche: RSS, RMSE, R², AICc, BIC, ACF1, parametri a limite, numero di condizione. **Non identificabile** se condizione > `soglia_condizione` o un parametro (esclusi `a`, `c`) ha IC 95 % > |valore|. -### 5.2 Costanti condivise tra riscaldamento e raffreddamento -Dopo ogni gradino la temperatura decade con la stessa legge del raffreddamento finale (stessa costante di tempo, stesso asintoto ≈ ambiente). -I domini di raffreddamento si fittano per primi; nei domini di riscaldamento i parametri di `riscaldamento_fissa_da_raffreddamento` -(default `c` e `tau`) sono **fissati** ai valori del raffreddamento e resta libera solo l'ampiezza `A`. Motivo: in 2–4 s di dati τ non si -identifica da solo. Gli IC 95 % di `A` sono condizionati ai valori fissi (l'incertezza di `c` e `τ` non si propaga); i parametri fissi sono -marcati "(fisso)" nel report. Se manca un raffreddamento valido, o la lista è vuota, i parametri restano liberi. Con `[]` si torna al fit libero. +### 4.1 Costanti condivise tra riscaldamento e raffreddamento (`analisi.py`) +Dopo ogni gradino la temperatura decade con la stessa legge del raffreddamento finale. I domini di raffreddamento si fittano per primi; nei domini di riscaldamento i parametri di `riscaldamento_fissa_da_raffreddamento` (default `c`, `tau`) sono **fissati** ai loro valori e resta libera solo `A`. Motivo: con gli scarti casuali del sensore, in 2–4 s di dati τ non si identifica da solo (su dati puliti sì). Gli IC di `A` sono condizionati ai valori fissi; i parametri fissi sono marcati «(fisso)» nel report. Se manca un raffreddamento valido, o la lista è vuota, i parametri restano liberi. -### 5.3 Regola di scelta (`selezione.py`) -Scarta non convergenti, con parametri a limite o non identificabili. Se esistono modelli con `RMSE ≤ rmse_vs_rumore_max·σ` ("adeguati") -si sceglie tra questi, altrimenti tra tutti i validi; tra i modelli entro `soglia_delta_aicc` dal miglior AICc vince quello con meno -parametri. Se nessun modello è valido il dominio è **indeterminato**; un dominio con meno di `n_min_punti` campioni non viene -fittato. Un modello scelto ma con RMSE > soglia·σ è segnalato come **non adeguato** nel report. +### 4.2 Regola di scelta (`analisi.scegli`) +Scarta non convergenti, con parametri a limite o non identificabili. Se esistono modelli con `RMSE ≤ rmse_vs_rumore_max·σ` («adeguati») si sceglie tra questi, altrimenti tra tutti i validi; tra i modelli entro `soglia_delta_aicc` dal miglior AICc vince quello con meno parametri. Nessun modello valido ⇒ **indeterminato**; meno di `n_min_punti` campioni ⇒ non fittato. Scelto ma con RMSE > soglia·σ ⇒ «non adeguato» nel report. -## 6. CLI +## 5. CLI e file | Comando | Funzione | |---|---| -| `fitseg domini CSV` | Propone i domini, stampa la tabella, scrive `lavoro/domini.toml` | -| `fitseg analizza CSV [--domini FILE]` | Domini + fit + scelta + report in `lavoro/out/` | -| `fitseg modelli` | Elenca il catalogo | -| `fitseg valuta modello_finale.json --t 12.5 …` | Valuta il modello a tratti agli istanti dati (NaN se indeterminato) | +| `fitseg domini CSV` | propone i domini: tabella, `lavoro/domini.toml`, `lavoro/proposta.png` | +| `fitseg analizza CSV [--domini FILE]` | domini (proposti o da FILE) + fit + `lavoro/out/{report.md, risultati.json, dominio_.png}` | -Opzioni globali (prima del sottocomando): `--config PATH` (default `config.toml`), `--lavoro DIR`. -Exit code: 0 ok · 2 uso errato · 3 config non valida · 4 CSV illeggibile · 6 domini non validi. +Opzioni globali prima del sottocomando: `--config PATH`, `--lavoro DIR`. Exit: 0 ok · 3 config non valida · 4 CSV illeggibile · 6 domini non validi. -## 7. File in `lavoro/` +## 6. Test -| File | Contenuto | -|---|---| -| `domini.toml` | proposta (o input) dei domini | -| `dati.csv` | copia del CSV analizzato | -| `out/risultati.json` | tutti i fit per (dominio, serie), con metriche e scelta | -| `out/modello_finale.json` | per dominio e serie: modello scelto e parametri | -| `out/report.md`, `out/equazioni.md` | tabelle di confronto, equazioni con IC 95 % | -| `out/fit_.png` | panoramica: dati, fit a tratti (colore per tipo di dominio) e residui | -| `out/dominio_.png` | per dominio e serie: dati grezzi vs modello scelto e altri modelli validi, residui; incorporati nel `report.md` | +Unitari su un segnale sintetico (`tests/conftest.py`: τ = 16 s, 3 gradini, nessun disturbo a salto) e di integrazione su `dati/gen_{1,2,3}.csv` (saltati se mancano). Dettaglio: README, «Test». Coperto: domini (3, 0, 1 gradino, sotto soglia, confini), recupero di τ/`c`/`A`, parametri fissi e ereditati dal raffreddamento, regola di scelta, errori di config/CSV/`domini.toml`, CLI. -## 8. Test (`uv run pytest`, dati in `dati/gen_.csv`) +## 7. Limiti noti -| ID | Test | Esito | -|---|---|---| -| T1 | Domini sui seed 1–3 | `riposo, riscaldamento, riscaldamento, raffreddamento` ✔ | -| T2 | Raffreddamento | `esp_decad` scelto per T e ΔT, τ entro il 10 % di 15.9 s ✔ (seed 1: τ = 16.08 ± 0.07 s) | -| T2b | Riscaldamento | `esp_decad` con `c` e `τ` fissi uguali a quelli del raffreddamento, `A` > 0 ✔ | -| T3 | Riposo | `costante` per ΔT ✔ | -| — | da fare | più seed, robustezza a config diverse (nessun gradino, un solo gradino), `domini.toml` manuale, riproducibilità | - -## 9. Limiti noti - -- **Residui autocorrelati nel raffreddamento** (ACF1 ≈ 0.7, RMSE ≈ 0.36 contro σ ≈ 0.15): scarti casuali e deriva lenta non sono - nel modello; i parametri sono buoni ma gli IC 95 % sono troppo stretti (nonostante la correzione) e il dominio risulta "non adeguato". -- **Domini di riscaldamento corti** (14–26 campioni, 2–4 s): τ non è identificabile da solo, per questo è fissato al valore del raffreddamento (§5.2). - Il secondo tratto di riscaldamento ha ACF1 ≈ 0.7 e RMSE ≈ 2 σ (scarti casuali): risulta "non adeguato" anche se la forma è giusta. -- La soglia `soglia_gradino_c` va adattata se i gradini reali sono più piccoli di 20 °C o gli scarti più grandi. -- Il vincolo di continuità tra domini è solo ignorato: ogni dominio è fittato in modo indipendente. +- **Residui autocorrelati nel raffreddamento** (ACF1 ≈ 0.7, RMSE ≈ 0.36 contro σ ≈ 0.15): scarti casuali e deriva lenta non sono nel modello; parametri buoni, IC troppo stretti, dominio «non adeguato». +- **Riscaldamento**: secondo tratto con ACF1 ≈ 0.7 e RMSE ≈ 2σ per gli scarti: «non adeguato» anche se la forma è giusta. +- `soglia_gradino_c` va adattata se i gradini reali sono più piccoli di 20 °C o gli scarti più grandi. +- Nessun vincolo di continuità tra domini. Provato su pochi seed e con la config d'esempio di `gen_data`. diff --git a/docs/USO.md b/docs/USO.md deleted file mode 100644 index e210afb..0000000 --- a/docs/USO.md +++ /dev/null @@ -1,27 +0,0 @@ -# Uso di `fitseg` - -## 1. Ambiente -```bash -uv venv .venv && uv pip install -e ".[dev]" # numpy, scipy, matplotlib, pytest, ruff -``` - -## 2. Generare i dati -I CSV vengono da `../gen_data` (repository separato); `dati/gen_.csv` ne contiene tre già pronti. Per rigenerarli: -```bash -cd ../gen_data && .venv/bin/python genera.py config.toml --seed 4 -o ../EquationMiner/dati/gen_4.csv -``` - -## 3. Analisi -```bash -.venv/bin/fitseg domini dati/gen_1.csv # solo i domini → lavoro/domini.toml -.venv/bin/fitseg analizza dati/gen_1.csv # domini + fit + report in lavoro/out/ -.venv/bin/fitseg analizza dati/gen_1.csv --domini lavoro/domini.toml # con domini corretti a mano -.venv/bin/fitseg valuta lavoro/out/modello_finale.json --t 10 20 50 -.venv/bin/pytest # test (singolo: -k raffreddamento) -``` -Il report è in `lavoro/out/report.md` (con i grafici dati-contro-equazioni incorporati, uno per dominio e serie): per ogni dominio e serie, modello scelto con parametri ± IC 95 %, metriche e tabella di -confronto di tutti i modelli. Leggi sempre la riga "non adeguato" e l'ACF1: indicano residui che il modello non spiega. - -## 4. Configurazione -Tutto in [config.toml](../config.toml) (nessun default nel codice). Chiavi di rilievo: `[domini] soglia_gradino_c`, -`scarta_dopo_salto_n`, `n_min_punti`; `[fit.modelli_per_tipo]`, `riscaldamento_fissa_da_raffreddamento`, `soglia_delta_aicc`, `rmse_vs_rumore_max`, `seed`. diff --git a/src/fitseg/analisi.py b/src/fitseg/analisi.py new file mode 100644 index 0000000..b6a5a2f --- /dev/null +++ b/src/fitseg/analisi.py @@ -0,0 +1,63 @@ +"""Analisi: fit dei modelli di ogni tipo di dominio su ogni (dominio, serie) e scelta con la regola di SPEC §5.3.""" + +import math + +from .domini import sigma_rumore +from .fit import adatta +from .modelli import MODELLI + + +def valido(fit): + return (fit.get("convergenza") and not fit["parametri_a_limite"] and not fit["non_identificabile"] + and math.isfinite(fit["aicc"])) + + +def scegli(fit_list, sigma, cfg): + """(fit scelto, motivo, adeguato); (None, motivo, False) se nessun fit è valido.""" + f = cfg["fit"] + validi = [x for x in fit_list if valido(x)] + if not validi: + return None, "indeterminato: nessun modello convergente, identificabile e lontano dai limiti", False + adeguati = [x for x in validi if x["rmse"] <= f["rmse_vs_rumore_max"] * sigma] + pool = adeguati or validi + miglior = min(x["aicc"] for x in pool) + scelto = min((x for x in pool if x["aicc"] <= miglior + f["soglia_delta_aicc"]), + key=lambda x: (x["n_parametri"], x["aicc"])) + motivo = (f"{len(validi)}/{len(fit_list)} modelli validi; " + + ("adeguato (RMSE ≤ soglia·σ)" if adeguati else "nessun modello adeguato (RMSE > soglia·σ)") + + f"; meno parametri entro ΔAICc ≤ {f['soglia_delta_aicc']}") + return scelto, motivo, bool(adeguati) + + +def _serie(ty, yy, sigma, tipo, fissi, cfg): + voce = {"sigma": float(sigma), "fit": [], "scelta": None} + if len(yy) < cfg["domini"]["n_min_punti"]: + voce["motivo"] = f"indeterminato: {len(yy)} campioni < n_min_punti" + return voce + for m in cfg["fit"]["modelli_per_tipo"][tipo]: + if len(MODELLI[m][1]) <= len(yy) - 2: + voce["fit"].append(adatta(ty, yy, m, sigma, cfg, fissi if tipo == "riscaldamento" else None)) + scelto, voce["motivo"], voce["adeguato"] = scegli(voce["fit"], sigma, cfg) + voce["scelta"] = scelto["modello"] if scelto else None + return voce + + +def analizza_domini(t, serie, domini, cfg): + """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).""" + scarta = cfg["domini"]["scarta_dopo_salto_n"] + risultati = [None] * len(domini) + fissi = {nome: {} for nome in serie} # serie -> parametri del raffreddamento + for idx in sorted(range(len(domini)), key=lambda i: domini[i]["tipo"] != "raffreddamento"): + d = domini[idx] + i0, i1, t0 = d["i0"] + (scarta if d["tipo"] != "riposo" else 0), 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 = voce["serie"][nome] = _serie(t[i0:i1] - t0, yy, sigma_rumore(yy), d["tipo"], fissi[nome], cfg) + 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 cfg["fit"]["riscaldamento_fissa_da_raffreddamento"] if k in par} + risultati[idx] = voce + return risultati diff --git a/src/fitseg/cli.py b/src/fitseg/cli.py index 66279f5..3bc7ce9 100644 --- a/src/fitseg/cli.py +++ b/src/fitseg/cli.py @@ -1,18 +1,14 @@ -"""CLI: fitseg domini | analizza | modelli | valuta.""" +"""CLI: `fitseg domini CSV` (vedi la suddivisione) e `fitseg analizza CSV` (domini + regressione + report).""" import argparse -import json -import shutil import sys from pathlib import Path -import numpy as np +from . import config, dati, uscita +from . import domini as dom +from .analisi import analizza_domini -from . import config, dati, domini as dom, riepilogo -from .modelli import MODELLI -from .pipeline import analizza_domini - -EXIT_USO, EXIT_CONFIG, EXIT_DATI, EXIT_DOMINI = 2, 3, 4, 6 +EXIT_CONFIG, EXIT_DATI, EXIT_DOMINI = 3, 4, 6 def errore(msg, codice): @@ -20,72 +16,46 @@ def errore(msg, codice): sys.exit(codice) -def _carica(args): +def _prepara(args): + """cfg, cartella di lavoro, (t, serie) e domini (da domini.toml se dato, altrimenti la proposta, che viene salvata).""" try: cfg = config.leggi(args.config) + lavoro = Path(args.lavoro or cfg["output"]["cartella_lavoro"]) + t, serie = dati.leggi_csv(args.csv, cfg) except config.ErroreConfig as e: errore(e, EXIT_CONFIG) - lavoro = Path(args.lavoro or cfg["output"]["cartella_lavoro"]) - return cfg, lavoro - - -def _domini(args, cfg, lavoro, t, serie): + except dati.ErroreDati as e: + errore(e, EXIT_DATI) lavoro.mkdir(parents=True, exist_ok=True) - percorso = Path(args.domini) if getattr(args, "domini", None) else None - if percorso: + if getattr(args, "domini", None): try: - return dom.leggi_toml(percorso, t) + return cfg, lavoro, t, serie, dom.leggi_toml(args.domini, t) except (OSError, ValueError, KeyError) as e: - errore(f"{percorso}: {e}", EXIT_DOMINI) - proposta = dom.proponi(t, serie, cfg) - dom.scrivi_toml(lavoro / "domini.toml", proposta, t) - return proposta + errore(f"{args.domini}: {e}", EXIT_DOMINI) + domini = dom.proponi(t, serie, cfg) + dom.scrivi_toml(lavoro / "domini.toml", domini, t) + uscita.grafico_proposta(lavoro / "proposta.png", t, serie, domini) + return cfg, lavoro, t, serie, domini def cmd_domini(args): - cfg, lavoro = _carica(args) - try: - t, serie = dati.leggi_csv(args.csv, cfg) - except dati.ErroreDati as e: - errore(e, EXIT_DATI) - proposta = _domini(argparse.Namespace(domini=None), cfg, lavoro, t, serie) - for k, d in enumerate(proposta, 1): + _, lavoro, t, _, domini = _prepara(args) + for k, d in enumerate(domini, 1): print(f"{k:2d} {d['tipo']:15s} {t[d['i0']]:8.2f} – {t[d['i1'] - 1]:8.2f} s n={d['i1'] - d['i0']}") - print(f"scritto: {lavoro / 'domini.toml'}", file=sys.stderr) + print(f"scritto: {lavoro / 'domini.toml'}, {lavoro / 'proposta.png'}", file=sys.stderr) def cmd_analizza(args): - cfg, lavoro = _carica(args) - try: - t, serie = dati.leggi_csv(args.csv, cfg) - except dati.ErroreDati as e: - errore(e, EXIT_DATI) - domini = _domini(args, cfg, lavoro, t, serie) + cfg, lavoro, t, serie, domini = _prepara(args) out = lavoro / "out" out.mkdir(parents=True, exist_ok=True) - shutil.copy(args.csv, lavoro / "dati.csv") risultati = analizza_domini(t, serie, domini, cfg) - riepilogo.scrivi_json(out, risultati) - riepilogo.scrivi_report(out, risultati, Path(args.csv).name) - riepilogo.grafici(out, t, serie, risultati, domini) - riepilogo.grafici_domini(out, t, serie, risultati, domini, cfg) + uscita.scrivi_report(out, risultati, Path(args.csv).name) + uscita.grafici_domini(out, t, serie, risultati, domini, cfg) for r in risultati: for nome, vs in r["serie"].items(): print(f"dominio {r['k']} {r['tipo']:15s} {nome:5s} -> {vs['scelta'] or 'indeterminato'}") - print(f"scritto: {out}/ (report.md, equazioni.md, modello_finale.json, fit_*.png, dominio*_*.png)", file=sys.stderr) - - -def cmd_modelli(args): - for nome, (_, par, eq, _) in MODELLI.items(): - print(f"{nome:12s} {eq:48s} parametri: {', '.join(par)}") - - -def cmd_valuta(args): - modello = json.loads(Path(args.modello).read_text()) - ts = np.array(args.t, float) - for nome in modello[0]["serie"]: - for t, y in zip(ts, riepilogo.valuta_finale(modello, nome, ts)): - print(f"{nome} t={t:g} s -> {y:.4f}") + print(f"scritto: {out}/ (report.md, risultati.json, dominio*_*.png)", file=sys.stderr) def main(argv=None): @@ -93,18 +63,13 @@ def main(argv=None): ap.add_argument("--config", default="config.toml") ap.add_argument("--lavoro", help="cartella di lavoro (default: [output] cartella_lavoro)") sub = ap.add_subparsers(dest="cmd", required=True) - p = sub.add_parser("domini", help="propone i domini (riposo/riscaldamento/raffreddamento) e scrive domini.toml") + p = sub.add_parser("domini", help="propone i domini: tabella, domini.toml e proposta.png (nessun fit)") p.add_argument("csv") p.set_defaults(f=cmd_domini) - p = sub.add_parser("analizza", help="domini + fit + scelta + report") + p = sub.add_parser("analizza", help="domini + regressione + report in lavoro/out/") p.add_argument("csv") - p.add_argument("--domini", help="domini.toml modificato a mano, al posto della proposta automatica") + p.add_argument("--domini", help="domini.toml corretto a mano, al posto della proposta automatica") p.set_defaults(f=cmd_analizza) - sub.add_parser("modelli", help="elenca il catalogo").set_defaults(f=cmd_modelli) - p = sub.add_parser("valuta", help="valuta modello_finale.json in uno o più istanti") - p.add_argument("modello") - p.add_argument("--t", type=float, nargs="+", required=True) - p.set_defaults(f=cmd_valuta) args = ap.parse_args(argv) args.f(args) diff --git a/src/fitseg/config.py b/src/fitseg/config.py index bf87fc1..4908a3b 100644 --- a/src/fitseg/config.py +++ b/src/fitseg/config.py @@ -6,9 +6,9 @@ CHIAVI = { "dati": ["colonna_tempo", "unita_tempo", "serie", "separatore"], "domini": ["serie_riferimento", "soglia_gradino_c", "soglia_bordo_sigma", "distanza_min_salti_campioni", "n_min_punti", "scarta_dopo_salto_n"], - "fit": ["modelli_per_tipo", "riscaldamento_fissa_da_raffreddamento", "n_start", "perdita", "scala_perdita_sigma", "tau_min_rel", "tau_max_rel", + "fit": ["modelli_per_tipo", "riscaldamento_fissa_da_raffreddamento", "perdita", "scala_perdita_sigma", "tau_min_rel", "tau_max_rel", "n_griglia_tau", "soglia_condizione", "correggi_autocorrelazione", "soglia_acf", - "soglia_delta_aicc", "rmse_vs_rumore_max", "seed"], + "soglia_delta_aicc", "rmse_vs_rumore_max"], "output": ["cartella_lavoro"], } diff --git a/src/fitseg/domini.py b/src/fitseg/domini.py index 7fcc02a..ecc3e4d 100644 --- a/src/fitseg/domini.py +++ b/src/fitseg/domini.py @@ -50,9 +50,7 @@ def proponi(t, serie, cfg): tagli = [0, *inizi, len(y)] domini = [] for k in range(len(tagli) - 1): - if not inizi: - tipo = "riposo" - elif k == 0: + if not inizi or k == 0: tipo = "riposo" elif k == len(tagli) - 2: tipo = "raffreddamento" diff --git a/src/fitseg/fit.py b/src/fitseg/fit.py index 365311c..1c29f54 100644 --- a/src/fitseg/fit.py +++ b/src/fitseg/fit.py @@ -1,4 +1,4 @@ -"""Fit di un modello su un dominio con scipy.optimize.least_squares (VARPRO su tau per gli esponenziali).""" +"""Fit di un modello su un dominio con scipy.optimize.least_squares (esponenziale: partenza su griglia di τ).""" import numpy as np from scipy.optimize import least_squares @@ -12,110 +12,60 @@ def _acf1(r): return float(np.sum(r[1:] * r[:-1]) / den) if den > 0 and len(r) > 2 else 0.0 -def _varpro(t, y, nome, cfg, L): - """Partenza per gli esponenziali: griglia log-spaziata su tau, c e ampiezze per minimi quadrati lineari.""" +def _partenza(t, y, nome, cfg, L): + """Polinomi: minimi quadrati. Esponenziale: griglia log-spaziata su τ, c e A per minimi quadrati lineari.""" + if nome != "esp_decad": + return np.linalg.lstsq(np.vander(t, len(MODELLI[nome][1]), increasing=True), y, rcond=None)[0] f = cfg["fit"] - taus = np.geomspace(f["tau_min_rel"] * L, f["tau_max_rel"] * L, f["n_griglia_tau"]) migliore, rss_min = None, np.inf - - def prova(ts): - nonlocal migliore, rss_min - X = np.column_stack([np.ones_like(t), *[np.exp(-t / s) for s in ts]]) + for tau in np.geomspace(f["tau_min_rel"] * L, f["tau_max_rel"] * L, f["n_griglia_tau"]): + X = np.column_stack([np.ones_like(t), np.exp(-t / tau)]) coef = np.linalg.lstsq(X, y, rcond=None)[0] rss = np.sum((y - X @ coef) ** 2) if rss < rss_min: - rss_min, migliore = rss, (coef, ts) - - if nome == "esp_decad": - for s in taus: - prova([s]) - c, ts = migliore - return [np.array([c[0], c[1], ts[0]])] - grossa = taus[:: max(1, len(taus) // 12)] - for i, s1 in enumerate(grossa): - for s2 in grossa[i + 1:]: - prova([s1, s2]) - c, ts = migliore - return [np.array([c[0], c[1], ts[0], c[2], ts[1]])] + rss_min, migliore = rss, np.array([coef[0], coef[1], tau]) + return migliore -def _partenze(t, y, nome, cfg, L, rng): - if nome in ("esp_decad", "esp_doppio"): - return _varpro(t, y, nome, cfg, L) - if nome in ("costante", "lineare", "quadratico"): - k = len(MODELLI[nome][1]) - return [np.linalg.lstsq(np.vander(t, k, increasing=True), y, rcond=None)[0]] - ampiezza = y[-1] - y[0] if abs(y[-1] - y[0]) > 0 else np.ptp(y) # logistica - return [np.array([y[0], ampiezza, rng.uniform(0.1, 0.9) * L, rng.uniform(0.02, 0.3) * L]) - for _ in range(cfg["fit"]["n_start"])] - - -def _limiti(nome, cfg, L): - f = cfg["fit"] - lo = np.full(len(MODELLI[nome][1]), -np.inf) - hi = np.full_like(lo, np.inf) - for i in MODELLI[nome][3]: - lo[i], hi[i] = f["tau_min_rel"] * L, f["tau_max_rel"] * L - if nome == "logistica": - lo[2], hi[2] = 0.0, L - return lo, hi - - -def adatta(t, y, nome, sigma, cfg, rng, fissi=None): +def adatta(t, y, nome, sigma, cfg, fissi=None): """Fit di `nome` su (t, y). `fissi` = {parametro: valore} tenuti costanti (non contano nei gradi di libertà).""" f = cfg["fit"] - funzione, nomi, equazione, idx_tau = MODELLI[nome] + funzione, nomi, equazione, i_tau = MODELLI[nome] fissi = {k: v for k, v in (fissi or {}).items() if k in nomi} n, k_tot = len(t), len(nomi) libero = [i for i, nm in enumerate(nomi) if nm not in fissi] k = len(libero) L = float(t[-1] - t[0]) or 1.0 - lo, hi = _limiti(nome, cfg, L) + lo, hi = np.full(k_tot, -np.inf), np.full(k_tot, np.inf) + if i_tau is not None: + lo[i_tau], hi[i_tau] = f["tau_min_rel"] * L, f["tau_max_rel"] * L scala = f["scala_perdita_sigma"] * sigma if np.isfinite(sigma) and sigma > 0 else 1.0 def completo(x): - p = np.empty(k_tot) + p = np.array([fissi.get(nm, 0.0) for nm in nomi]) p[libero] = x - for i, nm in enumerate(nomi): - if nm in fissi: - p[i] = fissi[nm] return p - res_fn = lambda x: funzione(t, completo(x)) - y - migliore = None - for p0 in _partenze(t, y, nome, cfg, L, rng): - p0 = completo(np.asarray(p0)[libero]) - if nome == "esp_decad" and libero == [1]: # c e tau fissi: l'ampiezza è lineare - e = np.exp(-t / p0[2]) - p0[1] = np.sum((y - p0[0]) * e) / np.sum(e * e) - x0 = np.clip(p0[libero], lo[libero], hi[libero]) - try: - r = least_squares(res_fn, x0, bounds=(lo[libero], hi[libero]), loss=f["perdita"], f_scale=scala, - x_scale="jac") - except (ValueError, FloatingPointError): - continue - if migliore is None or r.cost < migliore.cost: - migliore = r - if migliore is None: + p0 = _partenza(t, y, nome, cfg, L) + for i, nm in enumerate(nomi): + p0[i] = fissi.get(nm, p0[i]) + if nome == "esp_decad" and libero == [1]: # c e τ fissi: l'ampiezza è lineare + e = np.exp(-t / p0[2]) + p0[1] = np.sum((y - p0[0]) * e) / np.sum(e * e) + try: + r = least_squares(lambda x: funzione(t, completo(x)) - y, np.clip(p0[libero], lo[libero], hi[libero]), + bounds=(lo[libero], hi[libero]), loss=f["perdita"], f_scale=scala, x_scale="jac") + except (ValueError, FloatingPointError): return {"modello": nome, "convergenza": False, "n_parametri": k} - p, J = completo(migliore.x), migliore.jac - if nome == "esp_doppio" and p[2] > p[4] and not fissi: # tau1 < tau2 - perm = [0, 3, 4, 1, 2] - p, J = p[perm], J[:, perm] - r = funzione(t, p) - y - rss = float(np.sum(r**2)) - gdl = max(n - k, 1) + p, J = completo(r.x), r.jac + res = funzione(t, p) - y + rss = float(np.sum(res**2)) Jn = J / np.maximum(np.linalg.norm(J, axis=0), 1e-300) cond = float(np.linalg.cond(Jn)) - acf1 = _acf1(r) + acf1 = _acf1(res) corr = np.sqrt((1 + acf1) / (1 - acf1)) if f["correggi_autocorrelazione"] and acf1 > f["soglia_acf"] else 1.0 - try: - cov = (rss / gdl) * np.linalg.pinv(J.T @ J) - err_liberi = np.sqrt(np.maximum(np.diag(cov), 0)) * corr - except np.linalg.LinAlgError: - err_liberi = np.full(k, np.nan) errori = np.zeros(k_tot) - errori[libero] = err_liberi + errori[libero] = np.sqrt(np.maximum(np.diag((rss / max(n - k, 1)) * np.linalg.pinv(J.T @ J)), 0)) * corr a_limite = [nomi[i] for i in libero if np.isfinite(lo[i]) and (abs(p[i] - lo[i]) < 1e-3 * abs(lo[i]) or abs(p[i] - hi[i]) < 1e-3 * abs(hi[i]))] non_ident = cond > f["soglia_condizione"] or any( @@ -124,9 +74,10 @@ def adatta(t, y, nome, sigma, cfg, rng, fissi=None): aicc = n * np.log(max(rss, 1e-300) / n) + 2 * k + (2 * k * (k + 1) / (n - k - 1) if n - k - 1 > 0 else np.inf) return { "modello": nome, "equazione": equazione, "n_parametri": k, "n": n, "fissi": sorted(fissi), - "parametri": dict(zip(nomi, map(float, p))), "ic95": dict(zip(nomi, map(float, 1.96 * errori))), + "parametri": dict(zip(nomi, map(float, p), strict=True)), + "ic95": dict(zip(nomi, map(float, 1.96 * errori), strict=True)), "rss": rss, "rmse": float(np.sqrt(rss / n)), "r2": 1 - rss / tss if tss > 0 else float("nan"), "aicc": float(aicc), "bic": float(n * np.log(max(rss, 1e-300) / n) + k * np.log(n)), "acf1": acf1, "condizione": cond, "parametri_a_limite": a_limite, - "non_identificabile": bool(non_ident), "convergenza": bool(migliore.success), + "non_identificabile": bool(non_ident), "convergenza": bool(r.success), } diff --git a/src/fitseg/modelli.py b/src/fitseg/modelli.py index f2ad7c9..9cbdfdd 100644 --- a/src/fitseg/modelli.py +++ b/src/fitseg/modelli.py @@ -1,4 +1,4 @@ -"""Registro dei modelli (t' = t − inizio del dominio). Nuovo modello = nuova voce qui e in docs/SPEC.md.""" +"""Registro dei modelli (t' = t − inizio del dominio). Nuovo modello = nuova voce qui, in docs/SPEC.md e in config.toml.""" import numpy as np @@ -11,31 +11,15 @@ def _lineare(t, p): return p[0] + p[1] * t -def _quadratico(t, p): - return p[0] + p[1] * t + p[2] * t**2 - - def _esp_decad(t, p): return p[0] + p[1] * np.exp(-t / p[2]) -def _esp_doppio(t, p): - return p[0] + p[1] * np.exp(-t / p[2]) + p[3] * np.exp(-t / p[4]) - - -def _logistica(t, p): - return p[0] + p[1] / (1 + np.exp(-(t - p[2]) / p[3])) - - -# nome: (funzione, nomi parametri, equazione testuale, indici dei parametri-tempo tau) +# nome: (funzione, nomi parametri, equazione testuale, indice del parametro tau o None) MODELLI = { - "costante": (_costante, ["a"], "y = a", []), - "lineare": (_lineare, ["a", "b"], "y = a + b·t'", []), - "quadratico": (_quadratico, ["a", "b", "c"], "y = a + b·t' + c·t'²", []), - "esp_decad": (_esp_decad, ["c", "A", "tau"], "y = c + A·exp(−t'/τ)", [2]), - "esp_doppio": (_esp_doppio, ["c", "A1", "tau1", "A2", "tau2"], - "y = c + A1·exp(−t'/τ1) + A2·exp(−t'/τ2)", [2, 4]), - "logistica": (_logistica, ["c", "A", "t_m", "w"], "y = c + A / (1 + exp(−(t'−t_m)/w))", [3]), + "costante": (_costante, ["a"], "y = a", None), + "lineare": (_lineare, ["a", "b"], "y = a + b·t'", None), + "esp_decad": (_esp_decad, ["c", "A", "tau"], "y = c + A·exp(−t'/τ)", 2), } diff --git a/src/fitseg/pipeline.py b/src/fitseg/pipeline.py deleted file mode 100644 index d769fc7..0000000 --- a/src/fitseg/pipeline.py +++ /dev/null @@ -1,49 +0,0 @@ -"""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 diff --git a/src/fitseg/riepilogo.py b/src/fitseg/riepilogo.py deleted file mode 100644 index 8f76731..0000000 --- a/src/fitseg/riepilogo.py +++ /dev/null @@ -1,141 +0,0 @@ -"""Report, equazioni, modello finale (JSON), valutazione e grafici.""" - -import json - -import numpy as np - -from .modelli import valuta - - -def _scelto(voce_s): - return next((x for x in voce_s["fit"] if x["modello"] == voce_s["scelta"]), None) - - -def _eq(fit): - p = ", ".join(f"{k} = {v:.4g} (fisso)" if k in fit.get("fissi", []) else f"{k} = {v:.4g} ± {fit['ic95'][k]:.2g}" - for k, v in fit["parametri"].items()) - return f"{fit['equazione']} ({p})" - - -def modello_finale(risultati): - out = [] - for r in risultati: - s = {} - for nome, vs in r["serie"].items(): - fit = _scelto(vs) - s[nome] = ({"modello": fit["modello"], "parametri": fit["parametri"]} if fit - else {"modello": None, "motivo": vs["motivo"]}) - out.append({"tipo": r["tipo"], "t0_s": r["t0_s"], "t1_s": r["t1_s"], "serie": s}) - return out - - -def scrivi_report(cartella, risultati, nome_csv): - righe = [f"# Report fitseg — {nome_csv}\n", - "t' = t − t0 (inizio del dominio). Numeri dal motore; intervalli al 95 %.\n"] - righe += [f"![{nome}](fit_{nome}.png)\n" for nome in risultati[0]["serie"]] - righe += ["| # | tipo | t0 [s] | t1 [s] | n |", "|---|---|---|---|---|"] - righe += [f"| {r['k']} | {r['tipo']} | {r['t0_s']:.2f} | {r['t1_s']:.2f} | {r['n']} |" for r in risultati] - eq = ["# Equazioni\n"] - for r in risultati: - righe.append(f"\n## Dominio {r['k']} — {r['tipo']} ({r['t0_s']:.2f}–{r['t1_s']:.2f} s)\n") - eq.append(f"## Dominio {r['k']} — {r['tipo']} ({r['t0_s']:.2f}–{r['t1_s']:.2f} s)\n") - for nome, vs in r["serie"].items(): - fit = _scelto(vs) - righe.append(f"### {nome} (σ rumore {vs['sigma']:.3g})\n") - righe.append(f"![dominio {r['k']} {nome}](dominio{r['k']}_{nome}.png)\n") - if not fit: - righe.append(f"**Indeterminato** — {vs['motivo']}\n") - eq.append(f"- {nome}: indeterminato\n") - continue - righe.append(f"**Scelto: {fit['modello']}** — {_eq(fit)} \n" - f"RMSE {fit['rmse']:.3g}, R² {fit['r2']:.4f}, ACF1 {fit['acf1']:.2f}" - f"{'' if vs['adeguato'] else ' — **modello non adeguato** (RMSE > soglia·σ)'} \n" - f"Motivo: {vs['motivo']}\n") - righe += ["| modello | k | RMSE | AICc | BIC | ACF1 | valido |", "|---|---|---|---|---|---|---|"] - for x in sorted((x for x in vs["fit"] if "aicc" in x), key=lambda x: x["aicc"]): - ok = not (x["parametri_a_limite"] or x["non_identificabile"] or not x["convergenza"]) - righe.append(f"| {x['modello']} | {x['n_parametri']} | {x['rmse']:.3g} | {x['aicc']:.1f} | " - f"{x['bic']:.1f} | {x['acf1']:.2f} | {'sì' if ok else 'no'} |") - righe.append("") - eq.append(f"- {nome}: `{_eq(fit)}`\n") - (cartella / "report.md").write_text("\n".join(righe), encoding="utf-8") - (cartella / "equazioni.md").write_text("\n".join(eq), encoding="utf-8") - - -def scrivi_json(cartella, risultati): - (cartella / "risultati.json").write_text(json.dumps(risultati, indent=1, ensure_ascii=False), encoding="utf-8") - (cartella / "modello_finale.json").write_text( - json.dumps(modello_finale(risultati), indent=1, ensure_ascii=False), encoding="utf-8") - - -def valuta_finale(modello, nome_serie, t): - """Valuta il modello a tratti in t (array di secondi); NaN fuori dai domini o indeterminati.""" - t = np.asarray(t, float) - y = np.full_like(t, np.nan) - for k, d in enumerate(modello): - ultimo = k == len(modello) - 1 - m = (t >= d["t0_s"]) & ((t <= d["t1_s"]) if ultimo else (t < modello[k + 1]["t0_s"])) - s = d["serie"].get(nome_serie) - if s and s["modello"]: - y[m] = valuta(s["modello"], t[m] - d["t0_s"], list(s["parametri"].values())) - return y - - -def grafici(cartella, t, serie, risultati, domini): - import matplotlib - matplotlib.use("Agg") - import matplotlib.pyplot as plt - colori = {"riposo": "#9aa0a6", "riscaldamento": "#d93025", "raffreddamento": "#1a73e8"} - modello = modello_finale(risultati) - for nome, y in serie.items(): - fig, (a, b) = plt.subplots(2, 1, figsize=(11, 7), sharex=True, height_ratios=[3, 1]) - a.plot(t, y, ".", ms=2, color="#555", label="dati") - yf = valuta_finale(modello, nome, t) - for d in domini: - sl = slice(d["i0"], d["i1"]) - a.axvspan(t[d["i0"]], t[d["i1"] - 1], color=colori[d["tipo"]], alpha=0.12) - a.plot(t[sl], yf[sl], "-", color=colori[d["tipo"]], lw=1.6) - b.plot(t, y - yf, ".", ms=2, color="#555") - b.axhline(0, color="k", lw=0.5) - a.set_ylabel(nome); b.set_ylabel("residuo"); b.set_xlabel("t [s]") - fig.tight_layout() - fig.savefig(cartella / f"fit_{nome}.png", dpi=110) - plt.close(fig) - - -def grafici_domini(cartella, t, serie, risultati, domini, cfg): - """Un PNG per (dominio, serie): dati grezzi contro il modello scelto e gli altri modelli validi, più residui.""" - import matplotlib - matplotlib.use("Agg") - import matplotlib.pyplot as plt - from .selezione import valido - scarta = cfg["domini"]["scarta_dopo_salto_n"] - for r, d in zip(risultati, domini): - i0 = d["i0"] + (scarta if d["tipo"] != "riposo" else 0) - sl_tutti, sl_fit = slice(d["i0"], d["i1"]), slice(i0, d["i1"]) - tt = np.linspace(t[i0], t[d["i1"] - 1], 400) - for nome, vs in r["serie"].items(): - y = serie[nome] - fig, (a, b) = plt.subplots(2, 1, figsize=(9, 6), sharex=True, height_ratios=[3, 1]) - a.plot(t[sl_tutti], y[sl_tutti], ".", ms=4, color="#aaa", label="dati (esclusi dal fit)") - a.plot(t[sl_fit], y[sl_fit], ".", ms=5, color="#222", label="dati grezzi") - scelto = _scelto(vs) - for x in vs["fit"]: - if "parametri" not in x or x is scelto or not valido(x, cfg): - continue - a.plot(tt, valuta(x["modello"], tt - r["t0_s"], list(x["parametri"].values())), "--", lw=1, - alpha=0.8, label=f"{x['modello']} (RMSE {x['rmse']:.3g})") - if scelto: - p = list(scelto["parametri"].values()) - a.plot(tt, valuta(scelto["modello"], tt - r["t0_s"], p), "-", lw=2.2, color="#d93025", - label=f"scelto: {scelto['modello']} (RMSE {scelto['rmse']:.3g})") - b.plot(t[sl_fit], y[sl_fit] - valuta(scelto["modello"], t[sl_fit] - r["t0_s"], p), ".", ms=4, color="#222") - else: - a.set_title("indeterminato: nessun modello valido") - b.axhline(0, color="k", lw=0.5) - a.set_ylabel(nome); b.set_ylabel("residuo"); b.set_xlabel("t [s]") - a.set_title(f"Dominio {r['k']} — {r['tipo']} — {nome}", fontsize=10) - a.legend(fontsize=8) - fig.tight_layout() - fig.savefig(cartella / f"dominio{r['k']}_{nome}.png", dpi=110) - plt.close(fig) diff --git a/src/fitseg/selezione.py b/src/fitseg/selezione.py deleted file mode 100644 index f10c53d..0000000 --- a/src/fitseg/selezione.py +++ /dev/null @@ -1,25 +0,0 @@ -"""Regola di scelta del modello (SPEC §5.3).""" - -import math - - -def valido(fit, cfg): - return (fit.get("convergenza") and not fit["parametri_a_limite"] and not fit["non_identificabile"] - and math.isfinite(fit["aicc"])) - - -def scegli(fit_list, sigma, cfg): - """Ritorna (fit scelto, motivo, adeguato) oppure (None, motivo, False) se nessun fit è valido.""" - f = cfg["fit"] - validi = [x for x in fit_list if valido(x, cfg)] - if not validi: - return None, "indeterminato: nessun modello convergente, identificabile e lontano dai limiti", False - adeguati = [x for x in validi if x["rmse"] <= f["rmse_vs_rumore_max"] * sigma] - pool = adeguati or validi - miglior_aicc = min(x["aicc"] for x in pool) - entro = [x for x in pool if x["aicc"] <= miglior_aicc + f["soglia_delta_aicc"]] - scelto = min(entro, key=lambda x: (x["n_parametri"], x["aicc"])) - motivo = (f"{len(validi)}/{len(fit_list)} modelli validi; " - + ("adeguato (RMSE ≤ soglia·σ)" if adeguati else "nessun modello adeguato (RMSE > soglia·σ)") - + f"; meno parametri entro ΔAICc ≤ {f['soglia_delta_aicc']}") - return scelto, motivo, bool(adeguati) diff --git a/src/fitseg/uscita.py b/src/fitseg/uscita.py new file mode 100644 index 0000000..bbb4a79 --- /dev/null +++ b/src/fitseg/uscita.py @@ -0,0 +1,97 @@ +"""File prodotti: report.md (con i grafici), risultati.json, proposta.png e un grafico dati/modelli per dominio.""" + +import json + +import matplotlib +import numpy as np + +from .analisi import valido +from .modelli import valuta + +matplotlib.use("Agg") +import matplotlib.pyplot as plt + +COLORI = {"riposo": "#9aa0a6", "riscaldamento": "#d93025", "raffreddamento": "#1a73e8"} + + +def _scelto(vs): + return next((x for x in vs["fit"] if x["modello"] == vs["scelta"]), None) + + +def _eq(fit): + p = ", ".join(f"{k} = {v:.4g} (fisso)" if k in fit["fissi"] else f"{k} = {v:.4g} ± {fit['ic95'][k]:.2g}" + for k, v in fit["parametri"].items()) + return f"{fit['equazione']} ({p})" + + +def scrivi_report(cartella, risultati, nome_csv): + """report.md, risultati.json; i grafici dominio_.png sono incorporati nel report.""" + righe = [f"# Report fitseg — {nome_csv}\n", "t' = t − t0 (inizio del dominio). Intervalli al 95 %.\n", + "| # | tipo | t0 [s] | t1 [s] | n |", "|---|---|---|---|---|"] + righe += [f"| {r['k']} | {r['tipo']} | {r['t0_s']:.2f} | {r['t1_s']:.2f} | {r['n']} |" for r in risultati] + for r in risultati: + righe.append(f"\n## Dominio {r['k']} — {r['tipo']} ({r['t0_s']:.2f}–{r['t1_s']:.2f} s)\n") + for nome, vs in r["serie"].items(): + fit = _scelto(vs) + righe += [f"### {nome} (σ rumore {vs['sigma']:.3g})\n", f"![dominio {r['k']} {nome}](dominio{r['k']}_{nome}.png)\n"] + if not fit: + righe.append(f"**Indeterminato** — {vs['motivo']}\n") + continue + righe.append(f"**Scelto: {fit['modello']}** — {_eq(fit)} \n" + f"RMSE {fit['rmse']:.3g}, R² {fit['r2']:.4f}, ACF1 {fit['acf1']:.2f}" + f"{'' if vs['adeguato'] else ' — **modello non adeguato** (RMSE > soglia·σ)'} \n" + f"Motivo: {vs['motivo']}\n") + righe += ["| modello | k | RMSE | AICc | BIC | ACF1 | valido |", "|---|---|---|---|---|---|---|"] + righe += [f"| {x['modello']} | {x['n_parametri']} | {x['rmse']:.3g} | {x['aicc']:.1f} | {x['bic']:.1f} | " + f"{x['acf1']:.2f} | {'sì' if valido(x) else 'no'} |" + for x in sorted((x for x in vs["fit"] if "aicc" in x), key=lambda x: x["aicc"])] + righe.append("") + (cartella / "report.md").write_text("\n".join(righe), encoding="utf-8") + (cartella / "risultati.json").write_text(json.dumps(risultati, indent=1, ensure_ascii=False), encoding="utf-8") + + +def grafico_proposta(percorso, t, serie, domini): + """Dati grezzi con i domini proposti (colore per tipo), senza alcun fit.""" + fig, assi = plt.subplots(len(serie), 1, figsize=(11, 3.6 * len(serie)), sharex=True, squeeze=False) + for ax, (nome, y) in zip(assi[:, 0], serie.items(), strict=True): + ax.plot(t, y, ".", ms=2, color="#333") + for k, d in enumerate(domini, 1): + ax.axvspan(t[d["i0"]], t[d["i1"] - 1], color=COLORI[d["tipo"]], alpha=0.15) + ax.text(t[d["i0"]], ax.get_ylim()[1], f" {k} {d['tipo']}", va="top", fontsize=8, rotation=90) + ax.set_ylabel(nome) + assi[-1, 0].set_xlabel("t [s]") + fig.tight_layout() + fig.savefig(percorso, dpi=110) + plt.close(fig) + + +def grafici_domini(cartella, t, serie, risultati, domini, cfg): + """Un PNG per (dominio, serie): dati grezzi contro il modello scelto e gli altri modelli validi, più residui.""" + scarta = cfg["domini"]["scarta_dopo_salto_n"] + for r, d in zip(risultati, domini, strict=True): + i0 = d["i0"] + (scarta if d["tipo"] != "riposo" else 0) + tutti, usati = slice(d["i0"], d["i1"]), slice(i0, d["i1"]) + tt = np.linspace(t[i0], t[d["i1"] - 1], 400) + for nome, vs in r["serie"].items(): + y, scelto = serie[nome], _scelto(vs) + fig, (a, b) = plt.subplots(2, 1, figsize=(9, 6), sharex=True, height_ratios=[3, 1]) + a.plot(t[tutti], y[tutti], ".", ms=4, color="#aaa", label="dati (esclusi dal fit)") + a.plot(t[usati], y[usati], ".", ms=5, color="#222", label="dati grezzi") + for x in vs["fit"]: + if x is not scelto and valido(x): + a.plot(tt, valuta(x["modello"], tt - r["t0_s"], list(x["parametri"].values())), "--", lw=1, + alpha=0.8, label=f"{x['modello']} (RMSE {x['rmse']:.3g})") + if scelto: + p = list(scelto["parametri"].values()) + a.plot(tt, valuta(scelto["modello"], tt - r["t0_s"], p), "-", lw=2.2, color=COLORI[r["tipo"]], + label=f"scelto: {scelto['modello']} (RMSE {scelto['rmse']:.3g})") + b.plot(t[usati], y[usati] - valuta(scelto["modello"], t[usati] - r["t0_s"], p), ".", ms=4, color="#222") + b.axhline(0, color="k", lw=0.5) + a.set_ylabel(nome) + b.set_ylabel("residuo") + b.set_xlabel("t [s]") + a.set_title(f"Dominio {r['k']} — {r['tipo']} — {nome}" + ("" if scelto else " — indeterminato"), fontsize=10) + a.legend(fontsize=8) + fig.tight_layout() + fig.savefig(cartella / f"dominio{r['k']}_{nome}.png", dpi=110) + plt.close(fig) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..20e02d5 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,43 @@ +"""Fixture comuni: configurazione, segnale sintetico (verità nota) e CSV generati da gen_data (se presenti).""" + +from pathlib import Path + +import numpy as np +import pytest + +from fitseg import config + +RADICE = Path(__file__).parent.parent +SEEDS = [1, 2, 3] + +# verità del segnale sintetico +TAU, T_AMB = 16.0, 23.0 +GRADINI = [(15.0, 80.0), (18.0, 70.0), (22.0, 60.0)] # (istante [s], ampiezza [°C]) + + +@pytest.fixture(scope="session") +def cfg(): + return config.leggi(RADICE / "config.toml") + + +def segnale(rumore=0.1, seed=0, gradini=GRADINI, durata=120.0, dt=0.15): + """T(t) = ambiente + Σ ampiezza·exp(−(t−t_j)/τ) per t ≥ t_j, più rumore gaussiano: nessun disturbo a salto.""" + rng = np.random.default_rng(seed) + t = np.arange(dt, durata, dt) + x = sum(a * np.exp(-(t - tj) / TAU) * (t >= tj) for tj, a in gradini) if gradini else np.zeros_like(t) + T = T_AMB + x + rumore * rng.standard_normal(len(t)) + return t, {"T_C": T, "dT_C": T - T_AMB} + + +@pytest.fixture(scope="session") +def sintetico(): + return segnale() + + +@pytest.fixture(params=SEEDS) +def csv_gen(request): + """CSV prodotto da gen_data per il seed; il test è saltato se il file non c'è (dati/ non è in git).""" + percorso = RADICE / f"dati/gen_{request.param}.csv" + if not percorso.exists(): + pytest.skip(f"{percorso} assente: rigenera con gen_data (vedi README)") + return percorso diff --git a/tests/test_config_dati.py b/tests/test_config_dati.py new file mode 100644 index 0000000..d76f8a9 --- /dev/null +++ b/tests/test_config_dati.py @@ -0,0 +1,46 @@ +"""Lettura di config.toml e del CSV: errori attesi.""" + +import pytest +from conftest import RADICE + +from fitseg import config, dati + + +def test_config_reale_valida(): + assert config.leggi(RADICE / "config.toml")["dati"]["serie"] == ["T_C", "dT_C"] + + +def test_chiave_mancante(tmp_path): + (tmp_path / "c.toml").write_text("[dati]\ncolonna_tempo='t'\n") + with pytest.raises(config.ErroreConfig, match="chiavi mancanti"): + config.leggi(tmp_path / "c.toml") + + +def test_modelli_per_tipo_incompleto(tmp_path): + testo = (RADICE / "config.toml").read_text().replace('raffreddamento = ["esp_decad"]', "raffreddamento = []") + (tmp_path / "c.toml").write_text(testo) + with pytest.raises(config.ErroreConfig, match="raffreddamento"): + config.leggi(tmp_path / "c.toml") + + +def test_config_inesistente(): + with pytest.raises(config.ErroreConfig, match="impossibile leggere"): + config.leggi("/non/esiste.toml") + + +def test_csv_valido(cfg, tmp_path): + (tmp_path / "d.csv").write_text("t_ms,T_C,dT_C\n100,20.0,0.1\n250,20.1,0.2\n") + t, serie = dati.leggi_csv(tmp_path / "d.csv", cfg) + assert t.tolist() == [0.1, 0.25] and serie["T_C"].tolist() == [20.0, 20.1] + + +def test_csv_colonna_mancante(cfg, tmp_path): + (tmp_path / "d.csv").write_text("t_ms,T_C\n100,20.0\n") + with pytest.raises(dati.ErroreDati, match="dT_C"): + dati.leggi_csv(tmp_path / "d.csv", cfg) + + +def test_csv_tempo_non_crescente(cfg, tmp_path): + (tmp_path / "d.csv").write_text("t_ms,T_C,dT_C\n200,20,0\n100,20,0\n") + with pytest.raises(dati.ErroreDati, match="crescente"): + dati.leggi_csv(tmp_path / "d.csv", cfg) diff --git a/tests/test_domini.py b/tests/test_domini.py new file mode 100644 index 0000000..8eb5f46 --- /dev/null +++ b/tests/test_domini.py @@ -0,0 +1,68 @@ +"""Riconoscimento dei domini (riposo / riscaldamento / raffreddamento) su segnali sintetici.""" + +import itertools + +import pytest +from conftest import GRADINI, segnale + +from fitseg import domini + + +def tipi(t, serie, cfg): + return [d["tipo"] for d in domini.proponi(t, serie, cfg)] + + +def test_tre_gradini(sintetico, cfg): + t, serie = sintetico + assert tipi(t, serie, cfg) == ["riposo", "riscaldamento", "riscaldamento", "raffreddamento"] + + +def test_confini_vicini_agli_istanti_dei_gradini(sintetico, cfg): + t, serie = sintetico + ds = domini.proponi(t, serie, cfg) + for d, (tj, _) in zip(ds[1:], GRADINI, strict=True): + assert abs(t[d["i0"]] - tj) < 0.5 + + +def test_domini_contigui_e_completi(sintetico, cfg): + t, serie = sintetico + ds = domini.proponi(t, serie, cfg) + assert ds[0]["i0"] == 0 and ds[-1]["i1"] == len(t) + assert all(a["i1"] == b["i0"] for a, b in itertools.pairwise(ds)) + + +def test_senza_gradini_un_solo_riposo(cfg): + t, serie = segnale(gradini=[]) + assert tipi(t, serie, cfg) == ["riposo"] + + +def test_un_solo_gradino(cfg): + t, serie = segnale(gradini=[(20.0, 80.0)]) + assert tipi(t, serie, cfg) == ["riposo", "raffreddamento"] + + +def test_gradino_sotto_soglia_ignorato(cfg): + t, serie = segnale(gradini=[(20.0, 10.0)]) # 10 °C < soglia_gradino_c + assert tipi(t, serie, cfg) == ["riposo"] + + +def test_domini_toml_andata_e_ritorno(sintetico, cfg, tmp_path): + t, serie = sintetico + ds = domini.proponi(t, serie, cfg) + domini.scrivi_toml(tmp_path / "d.toml", ds, t) + assert domini.leggi_toml(tmp_path / "d.toml", t) == ds + + +def test_domini_toml_non_contigui_rifiutati(sintetico, tmp_path): + t, _ = sintetico + (tmp_path / "d.toml").write_text('[[dominio]]\ntipo="riposo"\nt0_s=0\nt1_s=10\n' + '[[dominio]]\ntipo="raffreddamento"\nt0_s=20\nt1_s=30\n') + with pytest.raises(ValueError, match="contigui"): + domini.leggi_toml(tmp_path / "d.toml", t) + + +def test_domini_toml_tipo_non_valido(sintetico, tmp_path): + t, _ = sintetico + (tmp_path / "d.toml").write_text('[[dominio]]\ntipo="boh"\nt0_s=0\nt1_s=10\n') + with pytest.raises(ValueError, match="non valido"): + domini.leggi_toml(tmp_path / "d.toml", t) diff --git a/tests/test_fit.py b/tests/test_fit.py new file mode 100644 index 0000000..70d4587 --- /dev/null +++ b/tests/test_fit.py @@ -0,0 +1,94 @@ +"""Fit, regola di scelta e analisi dei domini su segnali sintetici (verità nota: τ = 16 s, ambiente 23 °C).""" + +import numpy as np +import pytest +from conftest import GRADINI, TAU, segnale + +from fitseg import domini +from fitseg.analisi import analizza_domini +from fitseg.fit import adatta +from fitseg.modelli import MODELLI, valuta + + +def test_modelli_valutazione(): + t = np.array([0.0, 1.0]) + assert valuta("costante", t, [5.0]).tolist() == [5.0, 5.0] + assert valuta("lineare", t, [1.0, 2.0]).tolist() == [1.0, 3.0] + assert valuta("esp_decad", t, [1.0, 2.0, 1.0])[0] == pytest.approx(3.0) + assert set(MODELLI) == {"costante", "lineare", "esp_decad"} + + +@pytest.mark.parametrize("seed", range(5)) +def test_esp_decad_recupera_parametri(cfg, seed): + rng = np.random.default_rng(seed) + t = np.arange(0, 100, 0.15) + y = 23 + 180 * np.exp(-t / 16) + 0.1 * rng.standard_normal(len(t)) + fit = adatta(t, y, "esp_decad", 0.1, cfg) + p = fit["parametri"] + assert p["tau"] == pytest.approx(16, rel=0.02) + assert p["c"] == pytest.approx(23, abs=0.2) + assert p["A"] == pytest.approx(180, rel=0.02) + assert fit["rmse"] == pytest.approx(0.1, rel=0.2) + assert not fit["non_identificabile"] and not fit["parametri_a_limite"] + + +def test_parametri_fissi_restano_fissi(cfg): + rng = np.random.default_rng(0) + t = np.arange(0, 4, 0.15) + y = 23 + 80 * np.exp(-t / 16) + 0.1 * rng.standard_normal(len(t)) + fit = adatta(t, y, "esp_decad", 0.1, cfg, fissi={"c": 23.0, "tau": 16.0}) + assert fit["fissi"] == ["c", "tau"] + assert fit["parametri"]["c"] == 23.0 and fit["parametri"]["tau"] == 16.0 + assert fit["n_parametri"] == 1 + assert fit["parametri"]["A"] == pytest.approx(80, rel=0.03) + + +@pytest.mark.parametrize("seed", range(5)) +def test_tau_libero_non_identificabile_su_tratto_corto_e_rumoroso(cfg, seed): + # 3 s di dati con disturbi dell'ordine degli scarti casuali del sensore (≈ 2 °C): τ non è stimabile da solo + rng = np.random.default_rng(seed) + t = np.arange(0, 3, 0.15) + y = 23 + 80 * np.exp(-t / 16) + 2.0 * rng.standard_normal(len(t)) + assert adatta(t, y, "esp_decad", 2.0, cfg)["non_identificabile"] + + +def test_analisi_completa_sintetico(sintetico, cfg): + t, serie = sintetico + ris = analizza_domini(t, serie, domini.proponi(t, serie, cfg), cfg) + assert [r["tipo"] for r in ris] == ["riposo", "riscaldamento", "riscaldamento", "raffreddamento"] + for nome in serie: + assert ris[0]["serie"][nome]["scelta"] in ("costante", "lineare") + raff = next(x for x in ris[3]["serie"][nome]["fit"] if x["modello"] == "esp_decad")["parametri"] + assert raff["tau"] == pytest.approx(TAU, rel=0.03) + for r in ris[1:3]: # riscaldamento: c e τ ereditati dal raffreddamento + fit = next(x for x in r["serie"][nome]["fit"] if x["modello"] == "esp_decad") + assert r["serie"][nome]["scelta"] == "esp_decad" and fit["fissi"] == ["c", "tau"] + assert fit["parametri"]["tau"] == raff["tau"] and fit["parametri"]["c"] == raff["c"] + # ampiezza del secondo tratto = somma dei decadimenti dei gradini precedenti + il nuovo, in t'=0 + t1 = ris[1]["t0_s"] + attesa = sum(a * np.exp(-(t1 - tj) / TAU) for tj, a in GRADINI[:1]) + fit1 = next(x for x in ris[1]["serie"]["dT_C"]["fit"] if x["modello"] == "esp_decad") + assert fit1["parametri"]["A"] == pytest.approx(attesa, rel=0.05) + + +def test_senza_raffreddamento_riscaldamento_resta_libero(cfg): + t, serie = segnale(gradini=[(20.0, 80.0)], durata=40.0) + ds = [{"tipo": "riposo", "i0": 0, "i1": 100}, {"tipo": "riscaldamento", "i0": 100, "i1": len(t)}] + ris = analizza_domini(t, serie, ds, cfg) # nessun dominio di raffreddamento: niente parametri da ereditare + for x in ris[1]["serie"]["T_C"]["fit"]: + assert x["fissi"] == [] + + +def test_dominio_troppo_corto_indeterminato(sintetico, cfg): + t, serie = sintetico + ds = [{"tipo": "riposo", "i0": 0, "i1": 5}] + vs = analizza_domini(t, serie, ds, cfg)[0]["serie"]["T_C"] + assert vs["scelta"] is None and "indeterminato" in vs["motivo"] + + +def test_riproducibile(sintetico, cfg): + t, serie = sintetico + ds = domini.proponi(t, serie, cfg) + a = analizza_domini(t, serie, ds, cfg) + b = analizza_domini(t, serie, ds, cfg) + assert a == b diff --git a/tests/test_integrazione.py b/tests/test_integrazione.py new file mode 100644 index 0000000..0d3773e --- /dev/null +++ b/tests/test_integrazione.py @@ -0,0 +1,60 @@ +"""Integrazione sui CSV di gen_data (dati/gen_{1,2,3}.csv, ignorati da git: i test sono saltati se mancano) e sulla CLI.""" + +import pytest +from conftest import RADICE + +from fitseg import cli, dati, domini +from fitseg.analisi import analizza_domini + + +def analizza(csv, cfg): + t, serie = dati.leggi_csv(csv, cfg) + ds = domini.proponi(t, serie, cfg) + return ds, analizza_domini(t, serie, ds, cfg) + + +def test_domini_su_csv_gen(csv_gen, cfg): + ds, _ = analizza(csv_gen, cfg) + assert [d["tipo"] for d in ds] == ["riposo", "riscaldamento", "riscaldamento", "raffreddamento"] + + +def test_raffreddamento_tau_nominale(csv_gen, cfg): + _, ris = analizza(csv_gen, cfg) # τ nominale 1/k = 15.9 s, k perturbato del 2 % per seed + for vs in ris[-1]["serie"].values(): + assert vs["scelta"] == "esp_decad" + fit = next(x for x in vs["fit"] if x["modello"] == "esp_decad") + assert fit["parametri"]["tau"] == pytest.approx(15.9, rel=0.1) + + +def test_riscaldamento_eredita_costanti(csv_gen, cfg): + _, ris = analizza(csv_gen, cfg) + for nome in ris[0]["serie"]: + raff = next(x for x in ris[-1]["serie"][nome]["fit"] if x["modello"] == "esp_decad")["parametri"] + for r in ris[1:3]: + fit = next(x for x in r["serie"][nome]["fit"] if x["modello"] == "esp_decad") + assert fit["fissi"] == ["c", "tau"] and fit["parametri"]["tau"] == raff["tau"] and fit["parametri"]["A"] > 0 + + +def test_cli_domini_e_analizza(tmp_path, capsys): + csv = RADICE / "dati/gen_1.csv" + if not csv.exists(): + pytest.skip(f"{csv} assente") + lavoro = tmp_path / "lavoro" + base = ["--config", str(RADICE / "config.toml"), "--lavoro", str(lavoro)] + cli.main([*base, "domini", str(csv)]) + assert (lavoro / "domini.toml").exists() and (lavoro / "proposta.png").exists() + assert not (lavoro / "out").exists() # `domini` non fa nessun fit + cli.main([*base, "analizza", str(csv), "--domini", str(lavoro / "domini.toml")]) # domini corretti a mano + for f in ("out/report.md", "out/risultati.json", "out/dominio4_T_C.png"): + assert (lavoro / f).exists(), f + assert "dominio 4 raffreddamento" in capsys.readouterr().out + + +def test_cli_errori(tmp_path): + base = ["--config", str(RADICE / "config.toml"), "--lavoro", str(tmp_path)] + with pytest.raises(SystemExit) as e: + cli.main([*base, "domini", str(tmp_path / "non_esiste.csv")]) + assert e.value.code == 4 + with pytest.raises(SystemExit) as e: + cli.main(["--config", "/non/esiste.toml", "domini", "x.csv"]) + assert e.value.code == 3 diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py deleted file mode 100644 index e6e7be8..0000000 --- a/tests/test_pipeline.py +++ /dev/null @@ -1,55 +0,0 @@ -"""Verifiche sui CSV generati da gen_data (dati/gen_.csv, config d'esempio: tau di raffreddamento nominale 15.9 s).""" - -from pathlib import Path - -import pytest - -from fitseg import config, dati, domini -from fitseg.pipeline import analizza_domini - -RADICE = Path(__file__).parent.parent -SEEDS = [1, 2, 3] - - -@pytest.fixture(scope="module") -def cfg(): - return config.leggi(RADICE / "config.toml") - - -@pytest.mark.parametrize("seed", SEEDS) -def test_domini_riposo_gradini_raffreddamento(cfg, seed): - t, serie = dati.leggi_csv(RADICE / f"dati/gen_{seed}.csv", cfg) - tipi = [d["tipo"] for d in domini.proponi(t, serie, cfg)] - assert tipi == ["riposo", "riscaldamento", "riscaldamento", "raffreddamento"] - - -@pytest.mark.parametrize("seed", SEEDS) -def test_raffreddamento_esponenziale(cfg, seed): - t, serie = dati.leggi_csv(RADICE / f"dati/gen_{seed}.csv", cfg) - ris = analizza_domini(t, serie, domini.proponi(t, serie, cfg), cfg) - for nome in serie: - vs = ris[-1]["serie"][nome] - assert vs["scelta"] == "esp_decad" - fit = next(x for x in vs["fit"] if x["modello"] == "esp_decad") - assert fit["parametri"]["tau"] == pytest.approx(15.9, rel=0.1) - - -def test_riposo_costante(cfg): - t, serie = dati.leggi_csv(RADICE / "dati/gen_1.csv", cfg) - ris = analizza_domini(t, serie, domini.proponi(t, serie, cfg), cfg) - assert ris[0]["serie"]["dT_C"]["scelta"] == "costante" - - -@pytest.mark.parametrize("seed", SEEDS) -def test_riscaldamento_esponenziale_con_costanti_del_raffreddamento(cfg, seed): - t, serie = dati.leggi_csv(RADICE / f"dati/gen_{seed}.csv", cfg) - ris = analizza_domini(t, serie, domini.proponi(t, serie, cfg), cfg) - for nome in serie: - raff = next(x for x in ris[-1]["serie"][nome]["fit"] if x["modello"] == "esp_decad")["parametri"] - for r in ris[1:3]: - vs = r["serie"][nome] - assert vs["scelta"] == "esp_decad" - fit = next(x for x in vs["fit"] if x["modello"] == "esp_decad") - assert fit["fissi"] == ["c", "tau"] - assert fit["parametri"]["tau"] == raff["tau"] and fit["parametri"]["c"] == raff["c"] - assert fit["parametri"]["A"] > 0 diff --git a/tests/test_selezione.py b/tests/test_selezione.py new file mode 100644 index 0000000..220304d --- /dev/null +++ b/tests/test_selezione.py @@ -0,0 +1,41 @@ +"""Regola di scelta del modello (SPEC §5.3) su fit finti.""" + +from fitseg.analisi import scegli + + +def fit(nome, k, aicc, rmse=0.1, **extra): + base = {"modello": nome, "n_parametri": k, "aicc": aicc, "rmse": rmse, "convergenza": True, + "parametri_a_limite": [], "non_identificabile": False} + return {**base, **extra} + + +def test_vince_meno_parametri_entro_delta(cfg): + scelto, _, _ = scegli([fit("a", 1, -100.0), fit("b", 3, -101.0)], 0.1, cfg) # ΔAICc = 1 < 2 + assert scelto["modello"] == "a" + + +def test_vince_il_migliore_oltre_delta(cfg): + scelto, _, _ = scegli([fit("a", 1, -100.0), fit("b", 3, -110.0)], 0.1, cfg) + assert scelto["modello"] == "b" + + +def test_scarta_non_validi(cfg): + fits = [fit("a", 1, -200.0, non_identificabile=True), fit("b", 2, -100.0), + fit("c", 2, -300.0, parametri_a_limite=["tau"]), fit("d", 2, -400.0, convergenza=False)] + assert scegli(fits, 0.1, cfg)[0]["modello"] == "b" + + +def test_nessun_valido_indeterminato(cfg): + scelto, motivo, adeguato = scegli([fit("a", 1, -1.0, non_identificabile=True)], 0.1, cfg) + assert scelto is None and not adeguato and "indeterminato" in motivo + + +def test_preferisce_i_modelli_adeguati(cfg): + # b ha AICc molto migliore ma RMSE oltre soglia·σ; a è adeguato: vince tra gli adeguati + scelto, _, adeguato = scegli([fit("a", 1, -50.0, rmse=0.1), fit("b", 3, -500.0, rmse=1.0)], 0.1, cfg) + assert scelto["modello"] == "a" and adeguato + + +def test_nessun_adeguato_lo_segnala(cfg): + scelto, _, adeguato = scegli([fit("a", 1, -50.0, rmse=1.0)], 0.1, cfg) + assert scelto["modello"] == "a" and not adeguato