feat(scenario): scenario runner, extensions and working copy
scenario.py runs all five simulators of the core from a JSON scenario file, with sweeps, and reproduces 20 of the note's data files (14 byte for byte, 6 column-wise). Extensions live outside the core and must reproduce it exactly in a limit case; paadrag_tidsserie is the worked example. arbeidskopi.py gives the user a copy of the model they can change. Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>
This commit is contained in:
parent
464ad38b49
commit
03151cfb3b
SSH key fingerprint:
SHA256:JakMjO6FTBBzN0Bhfj9saOoEjaFxlSdYuZQQpM/lF9Q
36 changed files with 3580 additions and 5 deletions
304
tests/test_scenario.py
Normal file
304
tests/test_scenario.py
Normal file
|
|
@ -0,0 +1,304 @@
|
|||
# Copyright (c) 2026 Kjell Tore Guttormsen. MIT License, see LICENSE.
|
||||
"""scenario.py: gjenskaping av notatets datafiler, validering, sveip, utdata og determinisme."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
import scenario
|
||||
from conftest import EKSEMPLER, FASIT, kjor
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
NOTATET = EKSEMPLER / "scenarioer" / "notatet"
|
||||
NYE = EKSEMPLER / "scenarioer"
|
||||
EKSAKT_RASKE = ("p3", "p6", "p11", "p11d", "kjede_s1", "kjede_s2", "kjede_s3", "kjede_s4", "kjede_s5",
|
||||
"endo_e5", "endo_des", "endo_desp", "endo_ko")
|
||||
|
||||
# Kolonnene i notatets fil: (scenariofil, kjøring, kolonne i kjøringens fil).
|
||||
KOLONNEVIS: dict[str, dict[str, tuple[str, str, str]]] = {
|
||||
"sat_B": {"t": ("sat_B", "ksat=inf", "t"), "K": ("sat_B", "ksat=inf", "K"),
|
||||
"Ks3": ("sat_B", "ksat=3", "K"), "Ks15": ("sat_B", "ksat=1.5", "K")},
|
||||
"sat_C": {"t": ("sat_C", "ksat=inf", "t"), "K": ("sat_C", "ksat=inf", "K"),
|
||||
"Ks3": ("sat_C", "ksat=3", "K"), "Ks15": ("sat_C", "ksat=1.5", "K")},
|
||||
"vippe": {"t": ("vippe", "regulator=fast_forsinkelse=3", "t"),
|
||||
"fast": ("vippe", "regulator=fast_forsinkelse=3", "K"),
|
||||
"p3": ("vippe", "regulator=P_forsinkelse=3", "K"), "p6": ("vippe", "regulator=P_forsinkelse=6", "K"),
|
||||
"p3cap": ("vippe_tak", "forsinkelse=3", "K"), "p6cap": ("vippe_tak", "forsinkelse=6", "K")},
|
||||
}
|
||||
for _s in ("prio_s2", "prio_s3", "prio_s5"):
|
||||
KOLONNEVIS[_s] = {
|
||||
**{c: (_s, "prio=true", c) for c in ("t", "K", "KA", "KB", "tauA", "tauB")},
|
||||
"tauF": (_s, "prio=false", "tauA"), "KF": (_s, "prio=false", "K"), "KAF": (_s, "prio=false", "KA"),
|
||||
}
|
||||
|
||||
|
||||
def kjor_fil(fil: Path, ut: Path, capsys: pytest.CaptureFixture[str]) -> tuple[int, str, str]:
|
||||
"""Kjør scenario.main i denne prosessen."""
|
||||
kode = scenario.main([str(fil), "--ut", str(ut)])
|
||||
fanget = capsys.readouterr()
|
||||
return kode, fanget.out, fanget.err
|
||||
|
||||
|
||||
def skriv(tmp_path: Path, data: object, navn: str = "s.json") -> Path:
|
||||
"""Skriv en scenariofil."""
|
||||
fil = tmp_path / navn
|
||||
fil.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
||||
return fil
|
||||
|
||||
|
||||
def kolonner(fil: Path) -> dict[str, list[str]]:
|
||||
"""Kolonnene i en fil i notatets format, som tekst."""
|
||||
linjer = fil.read_text(encoding="utf-8").splitlines()
|
||||
hode = linjer[0].split()
|
||||
rader = [linje.split(" ") for linje in linjer[1:]]
|
||||
return {h: [r[i] for r in rader] for i, h in enumerate(hode)}
|
||||
|
||||
|
||||
# ---------------- notatets datafiler ----------------
|
||||
|
||||
@pytest.mark.parametrize("navn", EKSAKT_RASKE)
|
||||
def test_gjenskaper_notatets_fil_eksakt(navn: str, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
kode, _, err = kjor_fil(NOTATET / f"{navn}.json", tmp_path, capsys)
|
||||
assert kode == 0, err
|
||||
assert (tmp_path / f"{navn}.dat").read_bytes() == (FASIT / "data" / f"{navn}.dat").read_bytes()
|
||||
|
||||
|
||||
@pytest.mark.treg
|
||||
def test_gjenskaper_mpc11_eksakt(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
kode, _, err = kjor_fil(NOTATET / "mpc11.json", tmp_path, capsys)
|
||||
assert kode == 0, err
|
||||
assert (tmp_path / "mpc11.dat").read_bytes() == (FASIT / "data" / "mpc11.dat").read_bytes()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("navn", sorted(KOLONNEVIS))
|
||||
def test_gjenskaper_notatets_fil_kolonnevis(navn: str, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
kjort: dict[str, Path] = {}
|
||||
for scenariofil, _, _ in KOLONNEVIS[navn].values():
|
||||
if scenariofil not in kjort:
|
||||
kode, _, err = kjor_fil(NOTATET / f"{scenariofil}.json", tmp_path / scenariofil, capsys)
|
||||
assert kode == 0, err
|
||||
kjort[scenariofil] = tmp_path / scenariofil
|
||||
fasit = kolonner(FASIT / "data" / f"{navn}.dat")
|
||||
assert set(fasit) == set(KOLONNEVIS[navn])
|
||||
for kolonne, (scenariofil, kjoring, kilde) in KOLONNEVIS[navn].items():
|
||||
egen = kolonner(kjort[scenariofil] / f"{scenariofil}__{kjoring}.dat")
|
||||
assert egen[kilde] == fasit[kolonne], f"{navn}: kolonne {kolonne}"
|
||||
|
||||
|
||||
def test_alle_notatets_simuleringer_har_en_scenariofil() -> None:
|
||||
simulerte = {f.stem for f in (FASIT / "data").glob("*.dat")} - {"stab", "thumb"}
|
||||
dekket = set(EKSAKT_RASKE) | {"mpc11"} | set(KOLONNEVIS)
|
||||
assert simulerte == dekket
|
||||
|
||||
|
||||
# ---------------- utdata ----------------
|
||||
|
||||
def test_sammendrag_csv_og_resultatjson(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
fil = skriv(tmp_path, {"navn": "test", "modell": "simulate_chain",
|
||||
"tiltak": {"kapasitet": {"faktor": 1.34}}})
|
||||
kode, ut, _ = kjor_fil(fil, tmp_path / "ut", capsys)
|
||||
assert kode == 0
|
||||
mappe = tmp_path / "ut"
|
||||
assert sorted(p.name for p in mappe.iterdir()) == ["resultat.json", "sammendrag.md", "test.csv"]
|
||||
assert ut.startswith((mappe / "sammendrag.md").read_text(encoding="utf-8"))
|
||||
assert f"Resultatene ligger i {mappe}" in ut
|
||||
assert "ikke kalibrert på norske data" in ut
|
||||
assert "ikke prognoser" in ut
|
||||
assert "tiltak.kapasitet.faktor" in ut
|
||||
assert "Slik leses tallene: 1,00 er nivået" in ut
|
||||
csv = (mappe / "test.csv").read_text(encoding="utf-8").splitlines()
|
||||
assert csv[0] == "t,K,U,L,T,Q,tauD,JD,RD,xi,Z,Z_rel"
|
||||
assert len(csv) == 1 + 121 # én rad per måned, 0-120
|
||||
assert csv[13].split(",")[0] == "12"
|
||||
data = json.loads((mappe / "resultat.json").read_text(encoding="utf-8"))
|
||||
assert set(data["versjoner"]) == {"python", "numpy", "scipy"}
|
||||
assert set(data["kjerne_sha256"]) == {"sim.py", "diagnose.py", "forlop.py"}
|
||||
brukt = data["kjoringer"][0]["brukt"]
|
||||
assert brukt["parametere"]["P"]["ksat"] == "inf"
|
||||
assert brukt["tiltak"]["kapasitet"] == {"faktor": 1.34, "fra": 24.0, "over": 24.0}
|
||||
assert data["kjoringer"][0]["nokkeltall"]["Z_slutt"] == pytest.approx(1.34)
|
||||
|
||||
|
||||
def test_standard_mappe_er_resultater_under_arbeidsmappa(
|
||||
tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
fil = skriv(tmp_path, {"navn": "Første forsøk", "modell": "simulate"})
|
||||
monkeypatch.chdir(tmp_path)
|
||||
assert scenario.main([str(fil)]) == 0
|
||||
capsys.readouterr()
|
||||
assert (tmp_path / "resultater" / "Forste-forsok" / "Forste-forsok.csv").is_file()
|
||||
|
||||
|
||||
def test_samme_inndata_gir_samme_bytes(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
fil = NYE / "utvidelse_bolge.json"
|
||||
for mappe in ("a", "b"):
|
||||
assert kjor_fil(fil, tmp_path / mappe, capsys)[0] == 0
|
||||
a = sorted((tmp_path / "a").iterdir())
|
||||
assert [p.name for p in a] == sorted(p.name for p in (tmp_path / "b").iterdir())
|
||||
for p in a:
|
||||
assert p.read_bytes() == (tmp_path / "b" / p.name).read_bytes(), p.name
|
||||
|
||||
|
||||
def test_ny_kjoring_i_samme_mappe_rydder_bare_egne_filer(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
sveip = skriv(tmp_path, {"navn": "x", "modell": "simulate",
|
||||
"sveip": {"parameter": "styring.forsinkelse", "verdier": [3, 6]}})
|
||||
enkel = skriv(tmp_path, {"navn": "x", "modell": "simulate"}, "enkel.json")
|
||||
ut = tmp_path / "ut"
|
||||
assert kjor_fil(sveip, ut, capsys)[0] == 0
|
||||
(ut / "notat.txt").write_text("min egen fil", encoding="utf-8")
|
||||
assert kjor_fil(enkel, ut, capsys)[0] == 0
|
||||
assert sorted(p.name for p in ut.iterdir()) == ["notat.txt", "resultat.json", "sammendrag.md", "x.csv"]
|
||||
|
||||
|
||||
def test_nekter_mappe_med_andre_filer(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
fil = skriv(tmp_path, {"modell": "simulate"})
|
||||
ut = tmp_path / "ut"
|
||||
ut.mkdir()
|
||||
(ut / "viktig.txt").write_text("x", encoding="utf-8")
|
||||
kode, _, err = kjor_fil(fil, ut, capsys)
|
||||
assert kode == 2
|
||||
assert "ikke er et tidligere resultat" in err
|
||||
assert sorted(p.name for p in ut.iterdir()) == ["viktig.txt"]
|
||||
|
||||
|
||||
def test_valgte_kolonner_og_hvert(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
fil = skriv(tmp_path, {"navn": "k", "modell": "simulate_endo", "regel": "koord", "slutt": 36,
|
||||
"utdata": {"kolonner": ["t", "Z_rel", "xi"], "hvert": 240}})
|
||||
assert kjor_fil(fil, tmp_path / "ut", capsys)[0] == 0
|
||||
linjer = (tmp_path / "ut" / "k.csv").read_text(encoding="utf-8").splitlines()
|
||||
assert linjer[0] == "t,Z_rel,xi"
|
||||
assert [linje.split(",")[0] for linje in linjer[1:]] == ["0", "12", "24", "36"]
|
||||
assert linjer[1] == "0,1,1"
|
||||
|
||||
|
||||
# ---------------- sveip ----------------
|
||||
|
||||
def test_sveip_over_avviklingstid(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
fil = skriv(tmp_path, {"navn": "ko", "modell": "simulate_endo", "regel": "koord", "slutt": 180,
|
||||
"sveip": {"parameter": "parametere.ENDO.T_clear", "verdier": [12, None, "inf"]}})
|
||||
kode, ut, _ = kjor_fil(fil, tmp_path / "ut", capsys)
|
||||
assert kode == 0
|
||||
data = json.loads((tmp_path / "ut" / "resultat.json").read_text(encoding="utf-8"))
|
||||
brukt = [k["brukt"]["parametere"]["ENDO"]["T_clear"] for k in data["kjoringer"]]
|
||||
assert brukt == [12.0, 24.0, "inf"] # null = notatets verdi
|
||||
assert [k["fil"] for k in data["kjoringer"]] == [
|
||||
"ko__T_clear=12.csv", "ko__T_clear=standard.csv", "ko__T_clear=inf.csv"]
|
||||
standard = data["kjoringer"][1]
|
||||
assert standard["vedtak"] == ["mnd 24: bestill +19 % (F_pred=0.134 Q_exc=0.27)",
|
||||
"mnd 36: bestill +12 % (F_pred=0.134 Q_exc=0.56)"]
|
||||
assert round(standard["nokkeltall"]["kapasitetsaar"], 1) == 3.5 # notatets tabell i avsnitt 8.5
|
||||
assert round(standard["nokkeltall"]["frafall"], 1) == 8.3
|
||||
assert "T_clear=standard" in ut
|
||||
|
||||
|
||||
def test_rutenett_gir_alle_kombinasjoner_og_matrise(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
fil = skriv(tmp_path, {"navn": "g", "modell": "simulate", "slutt": 150,
|
||||
"sveip": [{"parameter": "styring.reaktivitet", "verdier": [1, 3]},
|
||||
{"parameter": "styring.forsinkelse", "verdier": [3, 11]}]})
|
||||
kode, ut, _ = kjor_fil(fil, tmp_path / "ut", capsys)
|
||||
assert kode == 0
|
||||
data = json.loads((tmp_path / "ut" / "resultat.json").read_text(encoding="utf-8"))
|
||||
par = [(k["brukt"]["styring"]["reaktivitet"], k["brukt"]["styring"]["forsinkelse"]) for k in data["kjoringer"]]
|
||||
assert par == [(1.0, 3.0), (1.0, 11.0), (3.0, 3.0), (3.0, 11.0)]
|
||||
spenn = {p: k["nokkeltall"]["spenn_K_sist"] for p, k in zip(par, data["kjoringer"])}
|
||||
assert spenn[3.0, 11.0] > 0.3 > spenn[3.0, 3.0] # regime C svinger, regime A ikke
|
||||
assert "rader: styring.reaktivitet, kolonner: styring.forsinkelse" in ut
|
||||
|
||||
|
||||
def test_sveip_gjennom_kjor_py(tmp_path: Path) -> None:
|
||||
prosess = kjor("scenario", str(NOTATET / "sat_B.json"), "--ut", str(tmp_path))
|
||||
assert prosess.returncode == 0, prosess.stderr
|
||||
assert "ksat=1,5" in prosess.stdout
|
||||
assert len(list(tmp_path.glob("*.dat"))) == 3
|
||||
|
||||
|
||||
# ---------------- validering ----------------
|
||||
|
||||
@pytest.mark.parametrize(("data", "tekster"), [
|
||||
({"modell": "simulate", "styrng": {}}, ["styrng: ukjent felt", 'Mente du "styring"?', "Gyldige felt her:"]),
|
||||
({"modell": "simulat"}, ["modell: må være ett av", 'Mente du "simulate"?']),
|
||||
({"modell": "simulate", "parametere": {"P": {"gama": 0.1}}},
|
||||
["parametere.P.gama: ukjent parameter i P", 'Mente du "gamma"?', "Gyldige: gamma, eta"]),
|
||||
({"modell": "simulate", "parametere": {"CH0": {"Z": 1}}},
|
||||
["parametere.CH0: brukes ikke av modellen simulate", "bruker: P"]),
|
||||
({"modell": "simulate_chain", "parametere": {"CH0": {"Zfun": 1}}}, ["settes med tiltak.kapasitet"]),
|
||||
({"modell": "simulate", "styring": {"forsinkelse": "tre"}},
|
||||
["styring.forsinkelse: må være et tall minst 0", '"tre"']),
|
||||
({"modell": "simulate", "parametere": {"P": {"Lam0": 1.2}}}, ["parametere.P.Lam0", "mindre enn 1"]),
|
||||
({"modell": "simulate", "parametere": {"P": {"gamma": "inf"}}}, ["uendelig er ikke tillatt her"]),
|
||||
({"modell": "simulate_chain2", "parametere": {"CH2": {"c": 2.5}}}, ["parametere.CH2.c: må være et heltall"]),
|
||||
({"modell": "simulate_endo", "regel": "koord", "parametere": {"ENDO": {"qloop": 1}}},
|
||||
["parametere.ENDO.qloop: må være true eller false"]),
|
||||
({"modell": "simulate_endo"}, ["regel: mangler", "sched, desentral, desentral_prog, koord"]),
|
||||
({"modell": "simulate", "regel": "koord"}, ["regel: gjelder bare simulate_endo"]),
|
||||
({"modell": "simulate_chain2", "styring": {"regulator": "MPC"}}, ["styring.regulator", '"P"']),
|
||||
({"modell": "simulate", "sjokk": {"til": 36}}, ["sjokk.til: gjelder bare simulate_soc", "paadrag_tidsserie"]),
|
||||
({"modell": "simulate", "tiltak": {"kapasitet": {"faktor": 1.2}}},
|
||||
["tiltak.kapasitet: brukes ikke av modellen simulate", "simulate_chain, simulate_chain2, simulate_endo"]),
|
||||
({"modell": "simulate_endo", "regel": "koord", "tiltak": {"kapasitet": {"faktor": 1.2}}},
|
||||
['bare regel "sched"']),
|
||||
({"modell": "simulate_chain", "parametere": {"CH0": {"Z": "inf"}}, "tiltak": {"kapasitet": {"faktor": 1.2}}},
|
||||
["tiltak.kapasitet: har ingen mening"]),
|
||||
({"modell": "simulate", "sjokk": {"faktor": 1.2}, "parametere": {"P": {"shock": 1.2}}},
|
||||
["sjokk.faktor: er oppgitt to ganger"]),
|
||||
({"modell": "simulate", "sveip": {"parameter": "styring.reaktivtet", "verdier": [1, 2]}},
|
||||
["sveip[0].parameter", 'ukjent felt "reaktivtet" under styring', 'Mente du "reaktivitet"?']),
|
||||
({"modell": "simulate", "sveip": {"parameter": "parametere.P", "verdier": [1]}},
|
||||
["er en gruppe, ikke én verdi", "parametere.P.gamma"]),
|
||||
({"modell": "simulate", "sveip": {"parameter": "navn", "verdier": ["a"]}},
|
||||
["begynner ikke med et felt som kan sveipes"]),
|
||||
({"modell": "simulate", "sveip": {"parameter": "styring.forsinkelse", "verdier": [3, -1]}},
|
||||
["styring.forsinkelse: må være et tall minst 0", "(i sveipet: styring.forsinkelse = -1)"]),
|
||||
({"modell": "simulate", "sveip": {"parameter": "styring.forsinkelse", "verdier": [3, 3.0]}},
|
||||
["har samme verdi to ganger"]),
|
||||
({"modell": "simulate", "utdata": {"kolonner": ["t", "Q"]}}, ['utdata.kolonner: "Q" finnes ikke', "t, K, U, L, T"]),
|
||||
({"modell": "simulate", "utdata": {"overskrift": "t K"}}, ['gjelder bare "format": "notat"']),
|
||||
({"modell": "utvidelse:finnes_ikke"}, ['fant ikke utvidelsen "finnes_ikke"', "utvidelse:paadrag_tidsserie"]),
|
||||
])
|
||||
def test_feil_i_scenariofila_gir_kode_2_og_navngir_feltet(
|
||||
data: object, tekster: list[str], tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
fil = skriv(tmp_path, data)
|
||||
kode, ut, err = kjor_fil(fil, tmp_path / "ut", capsys)
|
||||
assert kode == 2
|
||||
assert ut == ""
|
||||
assert err.startswith(f"Feil i scenariofila {fil}: ")
|
||||
for t in tekster:
|
||||
assert t in err, err
|
||||
assert not (tmp_path / "ut").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("innhold", "tekst"), [
|
||||
('{"modell": "simulate",\n "slutt": 120,,}', "ikke gyldig JSON: feil på linje 2, kolonne 15"),
|
||||
('{"modell": "simulate", "parametere": {"P": {"ksat": Infinity}}}', 'Skriv uendelig som "inf"'),
|
||||
('{"modell": "simulate", "slutt": 120, "slutt": 60}', "slutt: feltet står to ganger"),
|
||||
("[1, 2]", "må være ett JSON-objekt"),
|
||||
])
|
||||
def test_feil_json(innhold: str, tekst: str, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
fil = tmp_path / "s.json"
|
||||
fil.write_text(innhold, encoding="utf-8")
|
||||
kode, _, err = kjor_fil(fil, tmp_path / "ut", capsys)
|
||||
assert kode == 2
|
||||
assert tekst in err, err
|
||||
|
||||
|
||||
def test_fil_som_mangler_og_feil_bruk(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
assert kjor_fil(tmp_path / "finnes_ikke.json", tmp_path / "ut", capsys)[2].startswith("Fant ikke fila")
|
||||
assert scenario.main([]) == 2
|
||||
assert "Mangler scenariofil" in capsys.readouterr().err
|
||||
assert scenario.main(["a.json", "--tull"]) == 2
|
||||
assert "Ukjent valg --tull" in capsys.readouterr().err
|
||||
assert scenario.main(["--hjelp"]) == 0
|
||||
assert capsys.readouterr().out.startswith("Bruk:")
|
||||
|
||||
|
||||
def test_alle_eksempelscenarioer_er_gyldige() -> None:
|
||||
filer = sorted(NYE.glob("*.json")) + sorted(NOTATET.glob("*.json"))
|
||||
assert len(filer) >= 26
|
||||
for fil in filer:
|
||||
sc = scenario.les_scenario(fil)
|
||||
assert sc.kjoringer, fil.name
|
||||
if fil.parent == NYE:
|
||||
assert sc.beskrivelse, f"{fil.name} mangler beskrivelse"
|
||||
Loading…
Add table
Add a link
Reference in a new issue