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 <noreply@anthropic.com>
This commit is contained in:
davideandClaude Sonnet 5.5 committed 2026-10-29 10:13:01 +01:00
1 parent b8c8aeb60e
commit d92bcd198b
22 files changed
+678 -683

No files matched your search

+18 -22
View File
@@ -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_<seed>.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.
+39 -90
View File
@@ -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<k>_<serie>.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_<serie>.png` | panoramica: dati, fit a tratti (un colore per tipo di dominio) e residui |
| `out/dominio<k>_<serie>.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`.
-2
View File
@@ -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"]
+40 -101
View File
@@ -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_<serie>.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<k>_<serie>.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_<serie>.png` | panoramica: dati, fit a tratti (colore per tipo di dominio) e residui |
| `out/dominio<k>_<serie>.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_<seed>.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`.
-27
View File
@@ -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_<seed>.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`.
+63
View File
@@ -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
+28 -63
View File
@@ -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)
+2 -2
View File
@@ -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"],
}
+1 -3
View File
@@ -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"
+33 -82
View File
@@ -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),
}
+5 -21
View File
@@ -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),
}
-49
View File
@@ -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
-141
View File
@@ -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)
-25
View File
@@ -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)
+97
View File
@@ -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<k>_<serie>.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)
+43
View File
@@ -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
+46
View File
@@ -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)
+68
View File
@@ -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)
+94
View File
@@ -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
+60
View File
@@ -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
-55
View File
@@ -1,55 +0,0 @@
"""Verifiche sui CSV generati da gen_data (dati/gen_<seed>.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
+41
View File
@@ -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