Kjell Tore Guttormsen
03151cfb3b
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>
1508 lines
63 KiB
Python
1508 lines
63 KiB
Python
# Copyright (c) 2026 Kjell Tore Guttormsen. MIT License, see LICENSE.
|
|
"""Scenariokjøreren: kjør modellens simuleringer fra en scenariofil (JSON).
|
|
|
|
Bruk:
|
|
python3 modell/kjor.py scenario <fil.json> [--ut <mappe>]
|
|
python3 modell/scenario.py <fil.json> [--ut <mappe>]
|
|
|
|
Formatet er beskrevet i docs/scenarioformat.md. Kjøreren bruker kjernen i
|
|
modell/kjerne/ (notatets sim.py) uendret: den kaller simulatorene der med
|
|
parametrene fra fila, setter horisonten for simulatorene som bruker den globale
|
|
TEND midlertidig, og skriver sammendrag.md, én CSV per kjøring og resultat.json.
|
|
Utvidelser i modell/utvidelser/ kjøres med "modell": "utvidelse:<navn>".
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import difflib
|
|
import hashlib
|
|
import importlib
|
|
import importlib.util
|
|
import itertools
|
|
import json
|
|
import math
|
|
import re
|
|
import sys
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING, Any, NamedTuple, NoReturn
|
|
|
|
import numpy as np
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable, Iterator
|
|
from types import ModuleType
|
|
|
|
import numpy.typing as npt
|
|
|
|
Arr = npt.NDArray[np.float64]
|
|
|
|
MODELLMAPPE = Path(__file__).resolve().parent
|
|
KJERNE = MODELLMAPPE / "kjerne"
|
|
UTVIDELSER = MODELLMAPPE / "utvidelser"
|
|
KJERNEFILER = ("sim.py", "diagnose.py", "forlop.py")
|
|
|
|
MODELLER = ("simulate", "simulate_chain", "simulate_chain2", "simulate_soc", "simulate_endo")
|
|
UTVIDELSE_PREFIKS = "utvidelse:"
|
|
MODELLTEKST = {
|
|
"simulate": "aggregert modell uten straffesakskjede (avsnitt 8.1, regime A-E)",
|
|
"simulate_chain": "koblet modell med domstolskø (avsnitt 7.3 og 8.2)",
|
|
"simulate_chain2": "domstol med to sakstyper, forrang og flere spor (avsnitt 8.3)",
|
|
"simulate_soc": "rekruttering og vippepunkt med forbigående sjokk (avsnitt 8.4)",
|
|
"simulate_endo": "endogene tiltak: beslutningsregler for andel utenom og kapasitet (avsnitt 8.5)",
|
|
}
|
|
DICTS_FOR_MODELL = {
|
|
"simulate": ("P",),
|
|
"simulate_chain": ("P", "CH0"),
|
|
"simulate_chain2": ("P", "CH2"),
|
|
"simulate_soc": ("P", "SOC"),
|
|
"simulate_endo": ("P", "CH0", "ENDO"),
|
|
}
|
|
ALLE_DICTS = ("P", "CH0", "CH2", "ENDO", "SOC")
|
|
REGULATORER = {
|
|
"simulate": ("P", "MPC"),
|
|
"simulate_chain": ("P", "MPC"),
|
|
"simulate_chain2": ("P",),
|
|
"simulate_soc": ("P", "fast"),
|
|
"simulate_endo": ("P",),
|
|
}
|
|
REGULATORTEKST = {"P": "proporsjonal regel (P)", "MPC": "modellprediktiv styring (MPC)",
|
|
"fast": "fast innsats (T = T0)"}
|
|
REGLER = ("sched", "desentral", "desentral_prog", "koord")
|
|
TILTAK_FOR_MODELL = {
|
|
"simulate": (),
|
|
"simulate_chain": ("andel_utenom", "kapasitet", "trommel_buffer_tau"),
|
|
"simulate_chain2": ("andel_utenom", "kapasitet", "trommel_buffer_tau"),
|
|
"simulate_soc": (),
|
|
"simulate_endo": ("andel_utenom", "kapasitet"),
|
|
}
|
|
# Nøkler i kjernens dicts som ikke er tall, men settes av tiltakene.
|
|
SATT_AV_TILTAK = {
|
|
("CH0", "Zfun"): "tiltak.kapasitet",
|
|
("CH0", "xifun"): "tiltak.andel_utenom",
|
|
("CH0", "dbr"): "tiltak.trommel_buffer_tau",
|
|
("CH2", "Zfun"): "tiltak.kapasitet",
|
|
("CH2", "xiBfun"): "tiltak.andel_utenom",
|
|
}
|
|
|
|
TOPPFELT = ("navn", "beskrivelse", "merknad", "modell", "styring", "parametere", "regel", "sjokk",
|
|
"tiltak", "slutt", "sveip", "utdata", "utvidelse")
|
|
SVEIPBARE = ("styring", "parametere", "regel", "sjokk", "tiltak", "slutt", "utvidelse")
|
|
STYRINGSFELT = ("regulator", "forsinkelse", "reaktivitet", "mpc_kjenner_kjeden")
|
|
SJOKKFELT = ("faktor", "fra", "til")
|
|
TILTAKSFELT = ("andel_utenom", "kapasitet", "trommel_buffer_tau")
|
|
ANDELFELT = ("andel", "fra")
|
|
KAPASITETSFELT = ("faktor", "fra", "over")
|
|
UTDATAFELT = ("format", "kolonner", "hvert", "overskrift")
|
|
SVEIPFELT = ("parameter", "verdier")
|
|
FORMATER = ("csv", "notat")
|
|
|
|
STANDARD_FORSINKELSE = 3.0 # regime A i notatet
|
|
STANDARD_REAKTIVITET = 3.0
|
|
STANDARD_TILTAK_FRA = 24.0 # T_TILTAK i sim.py
|
|
STANDARD_TILTAK_OVER = 24.0 # T_UTV i sim.py
|
|
STANDARD_SOC_SJOKK = (1.6, 12.0, 36.0) # simulate_soc: shock, t_on, t_off
|
|
MAKS_SLUTT = 2400.0
|
|
MAKS_KJORINGER = 400
|
|
RUTENETT = 2 # høyst to parametere i et sveip
|
|
SEN_FASE = 60.0 # spennet i K måles etter denne måneden (som amp(K,t>60) i sim.py)
|
|
SNITT_FRA = 12.0 # summary_endo: gjennomsnitt fra måned 12
|
|
MAKS_KOLONNER_I_TABELL = 6
|
|
CSV_HVERT = 20 # standard: én rad per måned (DT = 0,05)
|
|
NOTAT_HVERT = 10 # write_csv i sim.py
|
|
|
|
|
|
class ScenarioFeil(Exception):
|
|
"""Feil i scenariofila eller i bruken; gir exit-kode 2."""
|
|
|
|
|
|
class FeltFeil(ScenarioFeil):
|
|
"""Feil i et bestemt felt i scenariofila."""
|
|
|
|
|
|
class Grense(NamedTuple):
|
|
"""Gyldig område for et tall."""
|
|
|
|
nedre: float | None = None
|
|
nedre_med: bool = True
|
|
ovre: float | None = None
|
|
ovre_med: bool = True
|
|
inf: bool = False
|
|
|
|
|
|
IKKE_NEG = Grense(0.0)
|
|
POSITIV = Grense(0.0, nedre_med=False)
|
|
ANDEL = Grense(0.0, ovre=1.0)
|
|
ANDEL_POS = Grense(0.0, nedre_med=False, ovre=1.0)
|
|
APEN_ANDEL = Grense(0.0, nedre_med=False, ovre=1.0, ovre_med=False)
|
|
POSITIV_INF = Grense(0.0, nedre_med=False, inf=True)
|
|
GRENSER: dict[tuple[str, str], Grense] = {
|
|
("P", "gamma"): IKKE_NEG, ("P", "eta"): IKKE_NEG, ("P", "delta"): IKKE_NEG,
|
|
("P", "Omega0"): IKKE_NEG, ("P", "shock"): POSITIV, ("P", "t_shock"): IKKE_NEG,
|
|
("P", "T0"): IKKE_NEG, ("P", "Tmax"): POSITIV, ("P", "Kstar"): POSITIV,
|
|
("P", "kU"): IKKE_NEG, ("P", "U0"): ANDEL, ("P", "psiU"): IKKE_NEG, ("P", "kL"): IKKE_NEG,
|
|
("P", "mu1"): IKKE_NEG, ("P", "mu2"): IKKE_NEG, ("P", "Tprop"): IKKE_NEG,
|
|
("P", "Lam0"): APEN_ANDEL, ("P", "ksat"): POSITIV_INF,
|
|
("CH0", "Z"): POSITIV_INF, ("CH0", "m"): IKKE_NEG, ("CH0", "theta"): ANDEL,
|
|
("CH0", "chi"): IKKE_NEG, ("CH0", "mu4"): IKKE_NEG, ("CH0", "tauh"): POSITIV,
|
|
("CH0", "varsigma"): IKKE_NEG, ("CH0", "xi"): ANDEL_POS,
|
|
("CH2", "Z"): POSITIV, ("CH2", "m"): IKKE_NEG, ("CH2", "piA"): APEN_ANDEL,
|
|
("CH2", "thetaA"): ANDEL, ("CH2", "thetaB"): ANDEL, ("CH2", "chi"): IKKE_NEG,
|
|
("CH2", "mu4"): IKKE_NEG, ("CH2", "tauh"): POSITIV, ("CH2", "varsigma"): IKKE_NEG,
|
|
("CH2", "xiA"): ANDEL, ("CH2", "xiB"): ANDEL,
|
|
("ENDO", "lagD"): IKKE_NEG, ("ENDO", "xi_min"): ANDEL_POS, ("ENDO", "tau_thr"): POSITIV,
|
|
("ENDO", "k_xi"): IKKE_NEG, ("ENDO", "tau_xi"): POSITIV, ("ENDO", "ups_star"): POSITIV,
|
|
("ENDO", "ups_hi"): POSITIV, ("ENDO", "ups_lo"): IKKE_NEG, ("ENDO", "kZ"): IKKE_NEG,
|
|
("ENDO", "L1"): IKKE_NEG, ("ENDO", "L2"): POSITIV, ("ENDO", "tau_down"): POSITIV,
|
|
("ENDO", "T_clear"): POSITIV_INF, ("ENDO", "H_pred"): POSITIV, ("ENDO", "period"): POSITIV,
|
|
("ENDO", "mu3"): IKKE_NEG, ("ENDO", "Lam_star"): APEN_ANDEL, ("ENDO", "k_q"): IKKE_NEG,
|
|
("ENDO", "q_max"): IKKE_NEG, ("ENDO", "lag_survey"): IKKE_NEG,
|
|
("SOC", "eps"): IKKE_NEG, ("SOC", "Kh"): POSITIV,
|
|
}
|
|
HELTALLSGRENSER = {("CH2", "c"): (1, 50), ("SOC", "pexp"): (1, 20)}
|
|
|
|
|
|
# --------------------------------------------------------------------------- kjernen
|
|
|
|
|
|
def fra_kjernen(modul: ModuleType | None) -> bool:
|
|
"""Sann hvis modulen er lastet fra kjerne/ i denne modellmappa."""
|
|
return modul is not None and Path(str(getattr(modul, "__file__", ""))).resolve().parent == KJERNE
|
|
|
|
|
|
def last_kjerne() -> ModuleType:
|
|
"""Importer notatets sim.py fra kjerne/ ved siden av denne fila (aldri fra en annen modellmappe)."""
|
|
if fra_kjernen(sys.modules.get("sim")):
|
|
return sys.modules["sim"]
|
|
sys.modules.pop("sim", None)
|
|
if str(KJERNE) in sys.path:
|
|
sys.path.remove(str(KJERNE))
|
|
sys.path.insert(0, str(KJERNE))
|
|
sim = importlib.import_module("sim")
|
|
kilde = Path(str(sim.__file__)).resolve().parent
|
|
if kilde != KJERNE:
|
|
melding = (f"sim.py er allerede lastet fra {kilde}, ikke fra {KJERNE}. Kjør hver modellmappe "
|
|
"i sin egen prosess.")
|
|
raise ScenarioFeil(melding)
|
|
return sim
|
|
|
|
|
|
def last_utvidelse(navn: str) -> ModuleType:
|
|
"""Last modell/utvidelser/<navn>.py fra denne modellmappa og kontroller grensesnittet."""
|
|
tilgjengelige = sorted(p.stem for p in UTVIDELSER.glob("*.py") if not p.stem.startswith("_"))
|
|
if not re.fullmatch(r"[a-z_][a-z0-9_]*", navn) or navn not in tilgjengelige:
|
|
feil("modell", f'fant ikke utvidelsen "{navn}" i {UTVIDELSER}. Tilgjengelige: '
|
|
+ (", ".join(f"utvidelse:{n}" for n in tilgjengelige) or "ingen") + ".")
|
|
last_kjerne()
|
|
fil = UTVIDELSER / f"{navn}.py"
|
|
modulnavn = f"utvidelser.{navn}"
|
|
gammel = sys.modules.get(modulnavn)
|
|
if gammel is not None and Path(str(getattr(gammel, "__file__", ""))).resolve() == fil:
|
|
modul = gammel
|
|
else:
|
|
spec = importlib.util.spec_from_file_location(modulnavn, fil)
|
|
if spec is None or spec.loader is None:
|
|
feil("modell", f"kunne ikke laste {fil}.")
|
|
modul = importlib.util.module_from_spec(spec)
|
|
sys.modules[modulnavn] = modul
|
|
spec.loader.exec_module(modul)
|
|
mangler = [a for a in ("NAVN", "BESKRIVELSE", "PARAMETERE", "simuler", "grensetilfelle")
|
|
if not hasattr(modul, a)]
|
|
if mangler or not isinstance(modul.PARAMETERE, dict):
|
|
feil("modell", f'utvidelsen "{navn}" følger ikke grensesnittet i docs/utvidelser.md: mangler '
|
|
+ (", ".join(mangler) if mangler else "PARAMETERE som objekt") + ".")
|
|
return modul
|
|
|
|
|
|
def sha256(sti: Path) -> str:
|
|
"""Sjekksum for en fil."""
|
|
return hashlib.sha256(sti.read_bytes()).hexdigest()
|
|
|
|
|
|
@contextmanager
|
|
def horisont(sim: ModuleType, slutt: float) -> Iterator[None]:
|
|
"""Sett sim.TEND midlertidig (simulate, simulate_chain, simulate_chain2, simulate_soc)."""
|
|
kjerne: Any = sim # modulattributtet settes; ModuleType tillater bare lesing for mypy
|
|
gammel = kjerne.TEND
|
|
kjerne.TEND = slutt
|
|
try:
|
|
yield
|
|
finally:
|
|
kjerne.TEND = gammel
|
|
|
|
|
|
# --------------------------------------------------------------------------- tall og tekst
|
|
|
|
|
|
def tall(x: float | None, d: int = 2) -> str:
|
|
"""Norsk tallformat (som i kjernen): mellomrom som tusenskille, komma som desimaltegn."""
|
|
if x is None or math.isnan(x):
|
|
return "-"
|
|
return f"{x:,.{d}f}".replace(",", " ").replace(".", ",")
|
|
|
|
|
|
def tall_kort(x: float) -> str:
|
|
"""Kort tall i meldinger og overskrifter."""
|
|
return (str(int(x)) if float(x).is_integer() else repr(x)).replace(".", ",")
|
|
|
|
|
|
def vis(verdi: object) -> str:
|
|
"""Verdien slik den står i JSON (til meldinger)."""
|
|
return json.dumps(verdi, ensure_ascii=False)
|
|
|
|
|
|
def vis_verdi(verdi: object) -> str:
|
|
"""Kort visning av en sveipverdi i tabeller (norsk desimaltegn)."""
|
|
if verdi is None:
|
|
return "standard"
|
|
if isinstance(verdi, bool):
|
|
return "ja" if verdi else "nei"
|
|
if isinstance(verdi, float):
|
|
return repr(verdi).replace(".", ",")
|
|
if isinstance(verdi, (int, str)):
|
|
return str(verdi)
|
|
return vis(verdi)
|
|
|
|
|
|
def filverdi(verdi: object) -> str:
|
|
"""Sveipverdi i et filnavn."""
|
|
if verdi is None:
|
|
tekst_ = "standard"
|
|
elif isinstance(verdi, bool):
|
|
tekst_ = "true" if verdi else "false"
|
|
elif isinstance(verdi, float):
|
|
tekst_ = repr(verdi)
|
|
else:
|
|
tekst_ = vis(verdi).strip('"')
|
|
return re.sub(r"[^A-Za-z0-9.+-]+", "-", tekst_).strip("-") or "x"
|
|
|
|
|
|
def slug(navn: str) -> str:
|
|
"""Filnavn fra et scenarionavn."""
|
|
for a, b in (("æ", "ae"), ("ø", "o"), ("å", "aa"), ("Æ", "AE"), ("Ø", "O"), ("Å", "AA")):
|
|
navn = navn.replace(a, b)
|
|
return re.sub(r"[^A-Za-z0-9._-]+", "-", navn).strip("-.") or "scenario"
|
|
|
|
|
|
def lignende(navn: str, gyldige: list[str] | tuple[str, ...]) -> str:
|
|
"""Forslag ved skrivefeil."""
|
|
treff = difflib.get_close_matches(navn, list(gyldige), n=1, cutoff=0.6)
|
|
return f' Mente du "{treff[0]}"?' if treff else ""
|
|
|
|
|
|
def json_tall(f: float) -> float | str:
|
|
"""Tall i resultat.json; uendelig og nan som tekst."""
|
|
if math.isnan(f):
|
|
return "nan"
|
|
if math.isinf(f):
|
|
return "inf" if f > 0 else "-inf"
|
|
return f
|
|
|
|
|
|
def json_verdi(x: object) -> object:
|
|
"""Gjør en verdi klar for json.dumps (inf og nan som tekst, numpy som Python)."""
|
|
if isinstance(x, dict):
|
|
return {str(k): json_verdi(v) for k, v in x.items()}
|
|
if isinstance(x, (list, tuple)):
|
|
return [json_verdi(v) for v in x]
|
|
if isinstance(x, (bool, np.bool_)):
|
|
return bool(x)
|
|
if isinstance(x, (int, np.integer)):
|
|
return int(x)
|
|
if isinstance(x, (float, np.floating)):
|
|
return json_tall(float(x))
|
|
return x
|
|
|
|
|
|
# --------------------------------------------------------------------------- validering
|
|
|
|
|
|
def feil(sti: str, melding: str) -> NoReturn:
|
|
"""Stopp med en feil som nevner feltet."""
|
|
tekst_ = f"{sti}: {melding}" if sti else melding
|
|
raise FeltFeil(tekst_)
|
|
|
|
|
|
def objekt(verdi: object, sti: str, gyldige: tuple[str, ...] | list[str]) -> dict[str, object]:
|
|
"""Krev et objekt med bare kjente felt."""
|
|
if not isinstance(verdi, dict):
|
|
feil(sti, f"må være et objekt {{ ... }}, ikke {vis(verdi)}.")
|
|
for nokkel in verdi:
|
|
if nokkel not in gyldige:
|
|
hvor = f"{sti}.{nokkel}" if sti else str(nokkel)
|
|
feil(hvor, "ukjent felt." + lignende(str(nokkel), gyldige)
|
|
+ " Gyldige felt her: " + ", ".join(gyldige) + ".")
|
|
return verdi
|
|
|
|
|
|
def beskriv(g: Grense) -> str:
|
|
"""Tekst for et område."""
|
|
deler = []
|
|
if g.nedre is not None:
|
|
deler.append(("minst " if g.nedre_med else "større enn ") + tall_kort(g.nedre))
|
|
if g.ovre is not None:
|
|
deler.append(("høyst " if g.ovre_med else "mindre enn ") + tall_kort(g.ovre))
|
|
tekst_ = "et tall" + (" " + " og ".join(deler) if deler else "")
|
|
return tekst_ + (' (eller "inf" for uendelig)' if g.inf else "")
|
|
|
|
|
|
def tallverdi(verdi: object, sti: str, g: Grense) -> float:
|
|
"""Les et tall, med "inf" der det er tillatt."""
|
|
if isinstance(verdi, str) and verdi.strip().lower() in {"inf", "uendelig"}:
|
|
if not g.inf:
|
|
feil(sti, f"uendelig er ikke tillatt her; oppgi {beskriv(g)}.")
|
|
return math.inf
|
|
if isinstance(verdi, bool) or not isinstance(verdi, (int, float)):
|
|
feil(sti, f"må være {beskriv(g)}, ikke {vis(verdi)}.")
|
|
x = float(verdi)
|
|
ok = math.isfinite(x)
|
|
if ok and g.nedre is not None:
|
|
ok = x >= g.nedre if g.nedre_med else x > g.nedre
|
|
if ok and g.ovre is not None:
|
|
ok = x <= g.ovre if g.ovre_med else x < g.ovre
|
|
if not ok:
|
|
feil(sti, f"må være {beskriv(g)}, ikke {vis(verdi)}.")
|
|
return x
|
|
|
|
|
|
def heltall(verdi: object, sti: str, nedre: int, ovre: int) -> int:
|
|
"""Les et heltall i et område."""
|
|
if isinstance(verdi, bool) or not isinstance(verdi, int) or not nedre <= verdi <= ovre:
|
|
feil(sti, f"må være et heltall fra {nedre} til {ovre}, ikke {vis(verdi)}.")
|
|
return verdi
|
|
|
|
|
|
def sannhet(verdi: object, sti: str) -> bool:
|
|
"""Les true/false."""
|
|
if not isinstance(verdi, bool):
|
|
feil(sti, f"må være true eller false, ikke {vis(verdi)}.")
|
|
return verdi
|
|
|
|
|
|
def tekst(verdi: object, sti: str, valg: tuple[str, ...] | None = None) -> str:
|
|
"""Les tekst, eventuelt ett av gitte valg."""
|
|
if not isinstance(verdi, str) or (valg is not None and verdi not in valg):
|
|
hva = "ett av: " + ", ".join(f'"{v}"' for v in valg) if valg else "tekst i anførselstegn"
|
|
hint = lignende(verdi, valg) if isinstance(verdi, str) and valg else ""
|
|
feil(sti, f"må være {hva}, ikke {vis(verdi)}.{hint}")
|
|
return verdi
|
|
|
|
|
|
def tallfelt(d: dict[str, object], nokkel: str) -> float:
|
|
"""Et tall fra en oppløst dict (verdiene er validert før)."""
|
|
x = d[nokkel]
|
|
if isinstance(x, (int, float)) and not isinstance(x, bool):
|
|
return float(x)
|
|
melding = f"forventet et tall i {nokkel}, fikk {x!r}"
|
|
raise TypeError(melding)
|
|
|
|
|
|
def les_fil(fil: Path) -> dict[str, object]:
|
|
"""Les scenariofila som JSON, med feilmeldinger som i diagnose.les."""
|
|
def ingen_dubletter(par: list[tuple[str, object]]) -> dict[str, object]:
|
|
ut: dict[str, object] = {}
|
|
for nokkel, verdi in par:
|
|
if nokkel in ut:
|
|
feil(nokkel, "feltet står to ganger i samme objekt.")
|
|
ut[nokkel] = verdi
|
|
return ut
|
|
|
|
def konstant(navn: str) -> object:
|
|
melding = f'{navn} er ikke gyldig JSON. Skriv uendelig som "inf" (i anførselstegn).'
|
|
raise ScenarioFeil(melding)
|
|
|
|
try:
|
|
innhold = fil.read_text(encoding="utf-8")
|
|
except FileNotFoundError:
|
|
melding = f"Fant ikke fila {fil}."
|
|
raise ScenarioFeil(melding) from None
|
|
except UnicodeDecodeError:
|
|
melding = f"Fila {fil} er ikke lagret som UTF-8."
|
|
raise ScenarioFeil(melding) from None
|
|
except OSError:
|
|
melding = f"Kunne ikke lese {fil}. Er det en mappe, eller mangler du lesetilgang?"
|
|
raise ScenarioFeil(melding) from None
|
|
try:
|
|
data = json.loads(innhold, object_pairs_hook=ingen_dubletter, parse_constant=konstant)
|
|
except json.JSONDecodeError as e:
|
|
melding = f"Fila {fil} er ikke gyldig JSON: feil på linje {e.lineno}, kolonne {e.colno}."
|
|
raise ScenarioFeil(melding) from None
|
|
if not isinstance(data, dict):
|
|
melding = "Scenariofila må være ett JSON-objekt { ... } med feltene i docs/scenarioformat.md."
|
|
raise ScenarioFeil(melding)
|
|
return data
|
|
|
|
|
|
# --------------------------------------------------------------------------- kjøringer
|
|
|
|
|
|
@dataclass
|
|
class Kjoring:
|
|
"""Én fullt oppløst kjøring: alle verdier slik simulatoren får dem."""
|
|
|
|
etikett: str
|
|
sveipverdier: list[tuple[str, object]]
|
|
modell: str
|
|
regulator: str
|
|
forsinkelse: float
|
|
reaktivitet: float
|
|
mpc_kjenner_kjeden: bool | None
|
|
regel: str | None
|
|
slutt: float
|
|
sjokk: dict[str, float]
|
|
tiltak: dict[str, object]
|
|
dicts: dict[str, dict[str, object]]
|
|
utvidelse: dict[str, object] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class Scenario:
|
|
"""Scenariofila etter validering."""
|
|
|
|
navn: str
|
|
beskrivelse: str
|
|
modell: str
|
|
format: str
|
|
kolonner: list[str] | None
|
|
hvert: int
|
|
overskrift: str | None
|
|
sveip: list[tuple[str, list[object]]]
|
|
valgt: list[str]
|
|
kjoringer: list[Kjoring]
|
|
utvidelse: ModuleType | None = None
|
|
|
|
|
|
class Variant(NamedTuple):
|
|
"""Navnet på en kjøring og verdiene sveipet ga den."""
|
|
|
|
etikett: str
|
|
sveipverdier: list[tuple[str, object]]
|
|
|
|
|
|
def modellnavn(verdi: object) -> str:
|
|
"""Valider "modell"."""
|
|
if isinstance(verdi, str) and (verdi in MODELLER or verdi.startswith(UTVIDELSE_PREFIKS)):
|
|
return verdi
|
|
hint = lignende(verdi, MODELLER) if isinstance(verdi, str) else ""
|
|
feil("modell", f"må være ett av: {', '.join(MODELLER)} eller utvidelse:<navn>, "
|
|
f"ikke {vis(verdi)}.{hint}")
|
|
|
|
|
|
def parameterdicts(sim: ModuleType, modell: str, utv: ModuleType | None) -> tuple[str, ...]:
|
|
"""Hvilke av kjernens dicts modellen bruker."""
|
|
if utv is not None:
|
|
bruker = getattr(utv, "BRUKER", ALLE_DICTS)
|
|
return tuple(d for d in bruker if hasattr(sim, d))
|
|
return DICTS_FOR_MODELL[modell]
|
|
|
|
|
|
def kjernedict(sim: ModuleType, navn: str) -> dict[str, object]:
|
|
"""Kopi av en av kjernens parameterdicts."""
|
|
return dict(getattr(sim, navn))
|
|
|
|
|
|
def overstyrbare(sim: ModuleType, navn: str) -> list[str]:
|
|
"""Nøkler i kjernens dict som kan overstyres under parametere (tall og sannhetsverdier)."""
|
|
return [k for k, v in kjernedict(sim, navn).items()
|
|
if (navn, k) not in SATT_AV_TILTAK and isinstance(v, (bool, int, float))]
|
|
|
|
|
|
def les_parametere(sim: ModuleType, raa: object, bruker: tuple[str, ...],
|
|
modell: str) -> dict[str, dict[str, object]]:
|
|
"""Valider parametere og lag fullt oppløste dicts."""
|
|
oppgitt = objekt(raa if raa is not None else {}, "parametere", list(ALLE_DICTS))
|
|
for navn in oppgitt:
|
|
if navn not in bruker:
|
|
feil(f"parametere.{navn}", f"brukes ikke av modellen {modell}. Denne modellen bruker: "
|
|
f"{', '.join(bruker)}.")
|
|
ut: dict[str, dict[str, object]] = {}
|
|
for navn in bruker:
|
|
d = kjernedict(sim, navn)
|
|
gyldige = overstyrbare(sim, navn)
|
|
endr = oppgitt.get(navn, {})
|
|
if not isinstance(endr, dict):
|
|
feil(f"parametere.{navn}", f"må være et objekt {{ ... }}, ikke {vis(endr)}.")
|
|
for nokkel, verdi in endr.items():
|
|
sti = f"parametere.{navn}.{nokkel}"
|
|
if (navn, nokkel) in SATT_AV_TILTAK:
|
|
feil(sti, f"settes med {SATT_AV_TILTAK[navn, nokkel]}, ikke under parametere.")
|
|
if nokkel not in gyldige:
|
|
feil(sti, f"ukjent parameter i {navn}.{lignende(nokkel, gyldige)} Gyldige: "
|
|
+ ", ".join(gyldige) + " (se docs/modellkart.md).")
|
|
d[nokkel] = parameterverdi(navn, nokkel, d[nokkel], verdi, sti)
|
|
ut[navn] = d
|
|
return ut
|
|
|
|
|
|
def parameterverdi(navn: str, nokkel: str, standard: object, verdi: object, sti: str) -> object:
|
|
"""Valider én parameter etter typen kjernen bruker."""
|
|
if navn == "CH2" and nokkel == "m" and verdi == "kalibrer":
|
|
return "kalibrer"
|
|
if isinstance(standard, bool):
|
|
return sannhet(verdi, sti)
|
|
if isinstance(standard, int):
|
|
nedre, ovre = HELTALLSGRENSER.get((navn, nokkel), (-10**9, 10**9))
|
|
return heltall(verdi, sti, nedre, ovre)
|
|
return tallverdi(verdi, sti, GRENSER.get((navn, nokkel), Grense()))
|
|
|
|
|
|
def les_styring(raa: object, modell: str, slutt: float) -> tuple[str, float, float, bool | None]:
|
|
"""Valider styring."""
|
|
s = objekt(raa if raa is not None else {}, "styring", STYRINGSFELT)
|
|
regulatorer = REGULATORER.get(modell, ("P", "MPC", "fast"))
|
|
regulator = tekst(s.get("regulator", "P"), "styring.regulator", regulatorer)
|
|
forsinkelse = tallverdi(s.get("forsinkelse", STANDARD_FORSINKELSE), "styring.forsinkelse",
|
|
Grense(0.0, ovre=slutt))
|
|
reaktivitet = tallverdi(s.get("reaktivitet", STANDARD_REAKTIVITET), "styring.reaktivitet",
|
|
Grense(0.0, ovre=100.0))
|
|
mpc: bool | None = None
|
|
if "mpc_kjenner_kjeden" in s:
|
|
if modell != "simulate_chain" or regulator != "MPC":
|
|
feil("styring.mpc_kjenner_kjeden", 'gjelder bare modell "simulate_chain" med regulator "MPC".')
|
|
mpc = sannhet(s["mpc_kjenner_kjeden"], "styring.mpc_kjenner_kjeden")
|
|
elif modell == "simulate_chain" and regulator == "MPC":
|
|
mpc = True
|
|
return regulator, forsinkelse, reaktivitet, mpc
|
|
|
|
|
|
def oppgitt_p(parametere: object) -> dict[str, object]:
|
|
"""parametere.P slik det står i fila (tomt hvis ikke oppgitt)."""
|
|
p = parametere.get("P", {}) if isinstance(parametere, dict) else {}
|
|
return p if isinstance(p, dict) else {}
|
|
|
|
|
|
def les_sjokk(raa: object, modell: str, p: dict[str, object], parametere: object) -> dict[str, float]:
|
|
"""Valider sjokk. Utenom simulate_soc settes P.shock og P.t_shock (varig sjokk)."""
|
|
s = objekt(raa if raa is not None else {}, "sjokk", SJOKKFELT)
|
|
if modell == "simulate_soc":
|
|
for nokkel in ("shock", "t_shock"):
|
|
if nokkel in oppgitt_p(parametere):
|
|
feil(f"parametere.P.{nokkel}", "brukes ikke av simulate_soc. Sjokket der er forbigående "
|
|
"og settes med sjokk.faktor, sjokk.fra og sjokk.til.")
|
|
faktor = tallverdi(s.get("faktor", STANDARD_SOC_SJOKK[0]), "sjokk.faktor", POSITIV)
|
|
fra = tallverdi(s.get("fra", STANDARD_SOC_SJOKK[1]), "sjokk.fra", IKKE_NEG)
|
|
til = tallverdi(s.get("til", STANDARD_SOC_SJOKK[2]), "sjokk.til", IKKE_NEG)
|
|
if til <= fra:
|
|
feil("sjokk.til", f"må være senere enn sjokk.fra ({tall_kort(fra)}).")
|
|
return {"faktor": faktor, "fra": fra, "til": til}
|
|
if "til" in s:
|
|
feil("sjokk.til", "gjelder bare simulate_soc. For et sjokk som går over i andre modeller: bruk "
|
|
'"modell": "utvidelse:paadrag_tidsserie" (docs/utvidelser.md).')
|
|
for felt, nokkel in (("faktor", "shock"), ("fra", "t_shock")):
|
|
if felt in s:
|
|
if nokkel in oppgitt_p(parametere):
|
|
feil(f"sjokk.{felt}", f"er oppgitt to ganger: også som parametere.P.{nokkel}. Bruk ett av dem.")
|
|
p[nokkel] = tallverdi(s[felt], f"sjokk.{felt}", GRENSER["P", nokkel])
|
|
return {"faktor": tallfelt(p, "shock"), "fra": tallfelt(p, "t_shock")}
|
|
|
|
|
|
def les_andel(raa: object) -> dict[str, object]:
|
|
"""tiltak.andel_utenom."""
|
|
a = objekt(raa, "tiltak.andel_utenom", ANDELFELT)
|
|
if "andel" not in a:
|
|
feil("tiltak.andel_utenom.andel", "mangler.")
|
|
return {"andel": tallverdi(a["andel"], "tiltak.andel_utenom.andel", Grense(0.0, ovre=1.0, ovre_med=False)),
|
|
"fra": tallverdi(a.get("fra", STANDARD_TILTAK_FRA), "tiltak.andel_utenom.fra", IKKE_NEG)}
|
|
|
|
|
|
def les_kapasitet(raa: object) -> dict[str, object]:
|
|
"""tiltak.kapasitet."""
|
|
k = objekt(raa, "tiltak.kapasitet", KAPASITETSFELT)
|
|
if "faktor" not in k:
|
|
feil("tiltak.kapasitet.faktor", "mangler.")
|
|
return {"faktor": tallverdi(k["faktor"], "tiltak.kapasitet.faktor", POSITIV),
|
|
"fra": tallverdi(k.get("fra", STANDARD_TILTAK_FRA), "tiltak.kapasitet.fra", IKKE_NEG),
|
|
"over": tallverdi(k.get("over", STANDARD_TILTAK_OVER), "tiltak.kapasitet.over", POSITIV)}
|
|
|
|
|
|
def les_tiltak(raa: object, modell: str, regel: str | None) -> dict[str, object]:
|
|
"""Valider tiltak (deklarativt; funksjonene bygges i bygg_tiltak)."""
|
|
t = objekt(raa if raa is not None else {}, "tiltak", TILTAKSFELT)
|
|
if not t:
|
|
return {}
|
|
tillatt = TILTAK_FOR_MODELL.get(modell, TILTAKSFELT)
|
|
for nokkel in t:
|
|
if nokkel not in tillatt:
|
|
hvilke = ", ".join(m for m in MODELLER if nokkel in TILTAK_FOR_MODELL[m])
|
|
feil(f"tiltak.{nokkel}", f"brukes ikke av modellen {modell}; bare av {hvilke}.")
|
|
if modell == "simulate_endo" and regel != "sched":
|
|
feil("tiltak", 'med simulate_endo gjelder faste tiltak bare regel "sched". De andre reglene '
|
|
"bestemmer andel utenom og kapasitet selv.")
|
|
ut: dict[str, object] = {}
|
|
if "andel_utenom" in t:
|
|
ut["andel_utenom"] = les_andel(t["andel_utenom"])
|
|
if "kapasitet" in t:
|
|
ut["kapasitet"] = les_kapasitet(t["kapasitet"])
|
|
if "trommel_buffer_tau" in t:
|
|
ut["trommel_buffer_tau"] = tallverdi(t["trommel_buffer_tau"], "tiltak.trommel_buffer_tau", POSITIV)
|
|
return ut
|
|
|
|
|
|
def xi_trinn(xi_ny: float, fra: float) -> Callable[[float], float]:
|
|
"""Andel til domstolen: 1 før fra, xi_ny fra og med fra (som xi_und i sim.py)."""
|
|
def xi(t: float) -> float:
|
|
return xi_ny if t >= fra else 1.0
|
|
return xi
|
|
|
|
|
|
def z_rampe(z0: float, faktor: float, fra: float, over: float) -> Callable[[float], float]:
|
|
"""Kapasitet økt lineært over en periode (som Z_utv i sim.py, samme regneuttrykk)."""
|
|
def z(t: float) -> float:
|
|
return float(z0 * (1 + (faktor - 1) * np.clip((t - fra) / over, 0, 1)))
|
|
return z
|
|
|
|
|
|
def bygg_tiltak(k: Kjoring, navn: str, d: dict[str, object]) -> dict[str, object]:
|
|
"""Kjernens kjededict med tiltakene som funksjoner (Zfun, xifun/xiBfun, dbr)."""
|
|
ch = dict(d)
|
|
andel = k.tiltak.get("andel_utenom")
|
|
kap = k.tiltak.get("kapasitet")
|
|
if isinstance(andel, dict):
|
|
a, fra = tallfelt(andel, "andel"), tallfelt(andel, "fra")
|
|
if navn == "CH2":
|
|
ch["xiBfun"] = xi_trinn(1.0 - a / (1.0 - tallfelt(ch, "piA")), fra)
|
|
else:
|
|
ch["xifun"] = xi_trinn(1.0 - a, fra)
|
|
if isinstance(kap, dict):
|
|
ch["Zfun"] = z_rampe(tallfelt(ch, "Z"), tallfelt(kap, "faktor"), tallfelt(kap, "fra"),
|
|
tallfelt(kap, "over"))
|
|
if "trommel_buffer_tau" in k.tiltak:
|
|
ch["dbr"] = tallfelt(k.tiltak, "trommel_buffer_tau")
|
|
return ch
|
|
|
|
|
|
def sjekk_ch0(k: Kjoring) -> None:
|
|
"""Tiltak i kjeden krever en endelig kapasitet."""
|
|
if math.isinf(tallfelt(k.dicts["CH0"], "Z")):
|
|
for nokkel in ("kapasitet", "trommel_buffer_tau"):
|
|
if nokkel in k.tiltak:
|
|
feil(f"tiltak.{nokkel}", "har ingen mening når parametere.CH0.Z er inf (ingen flaskehals).")
|
|
|
|
|
|
def sjekk_ch2(k: Kjoring, sim: ModuleType) -> None:
|
|
"""Andel utenom fra klasse B, og kalibrering av m."""
|
|
ch2 = k.dicts["CH2"]
|
|
andel = k.tiltak.get("andel_utenom")
|
|
rom = 1.0 - tallfelt(ch2, "piA")
|
|
if isinstance(andel, dict) and tallfelt(andel, "andel") > rom:
|
|
feil("tiltak.andel_utenom.andel", "i simulate_chain2 ledes sakene utenom fra klasse B; andelen av "
|
|
f"alle saker kan da være høyst 1 - piA = {tall_kort(rom)}.")
|
|
if ch2["m"] == "kalibrer":
|
|
try:
|
|
ch2["m"] = float(sim.calibrate_m(dict(ch2, m=None), k.dicts["P"]))
|
|
except ValueError:
|
|
ch2["m"] = None
|
|
if ch2["m"] is None:
|
|
feil("parametere.CH2.m", "kalibreringen fant ingen m som gir saksbehandlingstid 3 mnd i "
|
|
"utgangslikevekten med disse parametrene.")
|
|
|
|
|
|
def sjekk_endo(k: Kjoring) -> None:
|
|
"""Rekkefølger i ENDO."""
|
|
e = k.dicts["ENDO"]
|
|
if tallfelt(e, "L1") >= tallfelt(e, "L2"):
|
|
feil("parametere.ENDO.L1", "må være mindre enn L2 (kapasitet realiseres fra L1 til L2 mnd etter vedtak).")
|
|
if tallfelt(e, "ups_lo") >= tallfelt(e, "ups_hi"):
|
|
feil("parametere.ENDO.ups_lo", "må være mindre enn ups_hi (dødsonen).")
|
|
|
|
|
|
def sjekk_kombinasjoner(k: Kjoring, sim: ModuleType, utv: ModuleType | None) -> None:
|
|
"""Sammenhenger mellom felt som ikke kan sjekkes ett og ett."""
|
|
if utv is not None:
|
|
if hasattr(utv, "sjekk"):
|
|
try:
|
|
utv.sjekk(til_utvidelse(k))
|
|
except ValueError as e:
|
|
feil("", str(e))
|
|
return
|
|
if "CH0" in k.dicts:
|
|
sjekk_ch0(k)
|
|
if "CH2" in k.dicts:
|
|
sjekk_ch2(k, sim)
|
|
if "ENDO" in k.dicts:
|
|
sjekk_endo(k)
|
|
|
|
|
|
def les_utvidelse(raa: object, utv: ModuleType | None) -> dict[str, object]:
|
|
"""Valider utvidelsens parametere mot PARAMETERE (samme type som standardverdien)."""
|
|
if utv is None:
|
|
if raa is not None:
|
|
feil("utvidelse", 'brukes bare med "modell": "utvidelse:<navn>".')
|
|
return {}
|
|
standard: dict[str, object] = copy.deepcopy(utv.PARAMETERE)
|
|
oppgitt = objekt(raa if raa is not None else {}, "utvidelse", list(standard))
|
|
for nokkel, verdi in oppgitt.items():
|
|
sti = f"utvidelse.{nokkel}"
|
|
std = standard[nokkel]
|
|
if isinstance(std, bool):
|
|
standard[nokkel] = sannhet(verdi, sti)
|
|
elif isinstance(std, int):
|
|
standard[nokkel] = heltall(verdi, sti, -10**9, 10**9)
|
|
elif isinstance(std, float):
|
|
standard[nokkel] = tallverdi(verdi, sti, Grense(inf=True))
|
|
elif isinstance(std, str):
|
|
standard[nokkel] = tekst(verdi, sti)
|
|
elif isinstance(std, list) and not isinstance(verdi, list):
|
|
feil(sti, f"må være en liste [ ... ], ikke {vis(verdi)}.")
|
|
else:
|
|
standard[nokkel] = copy.deepcopy(verdi)
|
|
return standard
|
|
|
|
|
|
def les_regel(raa: dict[str, object], modell: str) -> str | None:
|
|
"""Regelen (bare simulate_endo)."""
|
|
if modell == "simulate_endo":
|
|
if "regel" not in raa:
|
|
feil("regel", "mangler. simulate_endo trenger en regel: " + ", ".join(REGLER) + ".")
|
|
return tekst(raa["regel"], "regel", REGLER)
|
|
if "regel" in raa:
|
|
feil("regel", "gjelder bare simulate_endo.")
|
|
return None
|
|
|
|
|
|
def los_opp(sim: ModuleType, raa: dict[str, object], modell: str, utv: ModuleType | None,
|
|
variant: Variant) -> Kjoring:
|
|
"""Valider én variant av scenariofila og lag en kjøring."""
|
|
objekt(raa, "", TOPPFELT)
|
|
slutt = tallverdi(raa.get("slutt", float(sim.TEND)), "slutt", Grense(0.0, nedre_med=False, ovre=MAKS_SLUTT))
|
|
regel = les_regel(raa, modell)
|
|
regulator, forsinkelse, reaktivitet, mpc = les_styring(raa.get("styring"), modell, slutt)
|
|
dicts = les_parametere(sim, raa.get("parametere"), parameterdicts(sim, modell, utv), modell)
|
|
sjokk: dict[str, float] = {}
|
|
if utv is not None and "sjokk" in raa:
|
|
feil("sjokk", 'brukes ikke med utvidelser. Se utvidelsens egne parametere under "utvidelse" '
|
|
"(docs/utvidelser.md).")
|
|
if utv is None:
|
|
sjokk = les_sjokk(raa.get("sjokk"), modell, dicts["P"], raa.get("parametere"))
|
|
k = Kjoring(etikett=variant.etikett, sveipverdier=variant.sveipverdier, modell=modell,
|
|
regulator=regulator, forsinkelse=forsinkelse, reaktivitet=reaktivitet,
|
|
mpc_kjenner_kjeden=mpc, regel=regel, slutt=slutt, sjokk=sjokk,
|
|
tiltak=les_tiltak(raa.get("tiltak"), modell, regel), dicts=dicts,
|
|
utvidelse=les_utvidelse(raa.get("utvidelse"), utv))
|
|
sjekk_kombinasjoner(k, sim, utv)
|
|
return k
|
|
|
|
|
|
def sveipsti(sti_: object, nr: int, utv: ModuleType | None, sim: ModuleType, modell: str) -> str:
|
|
"""Valider stien til en sveipet parameter."""
|
|
hvor = f"sveip[{nr}].parameter"
|
|
sti = tekst(sti_, hvor)
|
|
deler = sti.split(".")
|
|
if deler[0] not in SVEIPBARE:
|
|
feil(hvor, f'"{sti}" begynner ikke med et felt som kan sveipes.' + lignende(deler[0], SVEIPBARE)
|
|
+ " Gyldige: " + ", ".join(SVEIPBARE) + ".")
|
|
gyldige: dict[str, tuple[str, ...]] = {
|
|
"styring": STYRINGSFELT, "sjokk": SJOKKFELT, "tiltak": TILTAKSFELT,
|
|
"tiltak.andel_utenom": ANDELFELT, "tiltak.kapasitet": KAPASITETSFELT,
|
|
"parametere": parameterdicts(sim, modell, utv),
|
|
}
|
|
for navn in ALLE_DICTS:
|
|
if hasattr(sim, navn):
|
|
gyldige[f"parametere.{navn}"] = tuple(overstyrbare(sim, navn))
|
|
if utv is not None:
|
|
gyldige["utvidelse"] = tuple(utv.PARAMETERE)
|
|
for i in range(1, len(deler)):
|
|
forelder, barn = ".".join(deler[:i]), deler[i]
|
|
valg = gyldige.get(forelder)
|
|
if valg is None:
|
|
feil(hvor, f'"{sti}": {forelder} har ingen underfelt.')
|
|
if barn not in valg:
|
|
feil(hvor, f'"{sti}": ukjent felt "{barn}" under {forelder}.' + lignende(barn, valg)
|
|
+ " Gyldige: " + ", ".join(valg) + ".")
|
|
if sti in gyldige:
|
|
feil(hvor, f'"{sti}" er en gruppe, ikke én verdi. Velg ett av: '
|
|
+ ", ".join(f"{sti}.{v}" for v in gyldige[sti]) + ".")
|
|
return sti
|
|
|
|
|
|
def sett_sti(raa: dict[str, object], sti: str, verdi: object) -> None:
|
|
"""Sett en verdi i scenariofila etter en punktum-sti, og lag objektene som mangler."""
|
|
deler = sti.split(".")
|
|
node = raa
|
|
for d in deler[:-1]:
|
|
neste = node.get(d)
|
|
if not isinstance(neste, dict):
|
|
neste = {}
|
|
node[d] = neste
|
|
node = neste
|
|
if verdi is None: # null i et sveip: kjernens verdi (feltet fjernes)
|
|
node.pop(deler[-1], None)
|
|
else:
|
|
node[deler[-1]] = copy.deepcopy(verdi)
|
|
|
|
|
|
def les_sveip(raa: object, utv: ModuleType | None, sim: ModuleType,
|
|
modell: str) -> list[tuple[str, list[object]]]:
|
|
"""Valider sveip: ett objekt eller en liste med ett eller to."""
|
|
if raa is None:
|
|
return []
|
|
liste = raa if isinstance(raa, list) else [raa]
|
|
if not 1 <= len(liste) <= RUTENETT:
|
|
feil("sveip", f"kan ha én eller to parametere (et rutenett), ikke {len(liste)}.")
|
|
ut: list[tuple[str, list[object]]] = []
|
|
for nr, s in enumerate(liste):
|
|
obj = objekt(s, f"sveip[{nr}]", SVEIPFELT)
|
|
if "parameter" not in obj or "verdier" not in obj:
|
|
feil(f"sveip[{nr}]", 'trenger både "parameter" og "verdier".')
|
|
sti = sveipsti(obj["parameter"], nr, utv, sim, modell)
|
|
verdier = obj["verdier"]
|
|
if not isinstance(verdier, list) or not verdier:
|
|
feil(f"sveip[{nr}].verdier", f"må være en liste med minst én verdi, ikke {vis(verdier)}.")
|
|
etiketter = [repr(float(v)) if isinstance(v, (int, float)) and not isinstance(v, bool) else filverdi(v)
|
|
for v in verdier]
|
|
if len(set(etiketter)) != len(etiketter):
|
|
feil(f"sveip[{nr}].verdier", "har samme verdi to ganger.")
|
|
ut.append((sti, list(verdier)))
|
|
if len(ut) == RUTENETT and ut[0][0] == ut[1][0]:
|
|
feil("sveip[1].parameter", "er samme parameter som sveip[0].")
|
|
antall = math.prod(len(v) for _, v in ut)
|
|
if antall > MAKS_KJORINGER:
|
|
feil("sveip", f"gir {antall} kjøringer; høyst {MAKS_KJORINGER} er tillatt.")
|
|
return ut
|
|
|
|
|
|
def les_utdata(raa: object) -> tuple[str, list[str] | None, int, str | None]:
|
|
"""Valider utdata."""
|
|
u = objekt(raa if raa is not None else {}, "utdata", UTDATAFELT)
|
|
fmt = tekst(u.get("format", "csv"), "utdata.format", FORMATER)
|
|
kolonner: list[str] | None = None
|
|
if "kolonner" in u:
|
|
k = u["kolonner"]
|
|
if not isinstance(k, list) or not k or not all(isinstance(x, str) for x in k):
|
|
feil("utdata.kolonner", f"må være en liste med navn på tidsserier, ikke {vis(k)}.")
|
|
kolonner = [str(x) for x in k]
|
|
if len(set(kolonner)) != len(kolonner):
|
|
feil("utdata.kolonner", "har samme kolonne to ganger.")
|
|
hvert = heltall(u.get("hvert", NOTAT_HVERT if fmt == "notat" else CSV_HVERT), "utdata.hvert", 1, 10**6)
|
|
overskrift: str | None = None
|
|
if "overskrift" in u:
|
|
if fmt != "notat":
|
|
feil("utdata.overskrift", 'gjelder bare "format": "notat" (CSV-fila bruker kolonnenavnene).')
|
|
overskrift = tekst(u["overskrift"], "utdata.overskrift")
|
|
if kolonner is None or len(overskrift.split()) != len(kolonner):
|
|
feil("utdata.overskrift", "må ha like mange ord som utdata.kolonner har navn.")
|
|
return fmt, kolonner, hvert, overskrift
|
|
|
|
|
|
def valgte_stier(raa: dict[str, object], sveip: list[tuple[str, list[object]]]) -> list[str]:
|
|
"""Bladstiene brukeren har satt (til linjen om hva som er overstyrt)."""
|
|
ut: list[str] = []
|
|
|
|
def besok(node: object, sti: str) -> None:
|
|
if isinstance(node, dict) and node:
|
|
for k, v in node.items():
|
|
besok(v, f"{sti}.{k}")
|
|
elif sti not in ut:
|
|
ut.append(sti)
|
|
|
|
for felt in SVEIPBARE:
|
|
if felt in raa:
|
|
besok(raa[felt], felt)
|
|
for sti, _ in sveip:
|
|
if sti not in ut:
|
|
ut.append(sti)
|
|
return ut
|
|
|
|
|
|
def les_scenario(fil: Path) -> Scenario:
|
|
"""Les, valider og løs opp scenariofila til en liste kjøringer."""
|
|
raa = les_fil(fil)
|
|
objekt(raa, "", TOPPFELT)
|
|
if "modell" not in raa:
|
|
feil("modell", "mangler. Gyldige: " + ", ".join(MODELLER) + ", utvidelse:<navn>.")
|
|
modell = modellnavn(raa["modell"])
|
|
sim = last_kjerne()
|
|
utv = last_utvidelse(modell[len(UTVIDELSE_PREFIKS):]) if modell.startswith(UTVIDELSE_PREFIKS) else None
|
|
navn = tekst(raa.get("navn", fil.stem), "navn")
|
|
if not navn.strip():
|
|
feil("navn", "kan ikke være tom.")
|
|
beskrivelse = tekst(raa.get("beskrivelse", ""), "beskrivelse")
|
|
if "merknad" in raa:
|
|
tekst(raa["merknad"], "merknad")
|
|
fmt, kolonner, hvert, overskrift = les_utdata(raa.get("utdata"))
|
|
sveip = les_sveip(raa.get("sveip"), utv, sim, modell)
|
|
grunn = {k: v for k, v in raa.items() if k not in {"sveip", "utdata", "navn", "beskrivelse", "merknad"}}
|
|
kjoringer: list[Kjoring] = []
|
|
if not sveip:
|
|
kjoringer.append(los_opp(sim, grunn, modell, utv, Variant(slug(navn), [])))
|
|
for kombinasjon in itertools.product(*(v for _, v in sveip)) if sveip else ():
|
|
variant = copy.deepcopy(grunn)
|
|
verdier = [(sti, verdi) for (sti, _), verdi in zip(sveip, kombinasjon)]
|
|
for sti, verdi in verdier:
|
|
sett_sti(variant, sti, verdi)
|
|
etikett = "_".join(f"{sti.rsplit('.', 1)[-1]}={filverdi(v)}" for sti, v in verdier)
|
|
kjoringer.append(los_opp_i_sveip(sim, variant, modell, utv, Variant(etikett, verdier)))
|
|
return Scenario(navn=navn, beskrivelse=beskrivelse, modell=modell, format=fmt, kolonner=kolonner,
|
|
hvert=hvert, overskrift=overskrift, sveip=sveip, valgt=valgte_stier(raa, sveip),
|
|
kjoringer=kjoringer, utvidelse=utv)
|
|
|
|
|
|
def los_opp_i_sveip(sim: ModuleType, raa: dict[str, object], modell: str, utv: ModuleType | None,
|
|
variant: Variant) -> Kjoring:
|
|
"""Som los_opp, men feilmeldingen sier hvilken kombinasjon i sveipet som feilet."""
|
|
try:
|
|
return los_opp(sim, raa, modell, utv, variant)
|
|
except FeltFeil as e:
|
|
hvor = ", ".join(f"{sti} = {vis(v)}" for sti, v in variant.sveipverdier)
|
|
melding = f"{e} (i sveipet: {hvor})"
|
|
raise FeltFeil(melding) from None
|
|
|
|
|
|
# --------------------------------------------------------------------------- simulering
|
|
|
|
|
|
@dataclass
|
|
class Resultat:
|
|
"""Tidsserier, skalarer og vedtakslogg fra én kjøring."""
|
|
|
|
serier: dict[str, Arr]
|
|
skalarer: dict[str, float]
|
|
logg: list[str]
|
|
|
|
|
|
def til_utvidelse(k: Kjoring) -> dict[str, object]:
|
|
"""Scenarioet slik en utvidelse får det (docs/utvidelser.md)."""
|
|
dicts: dict[str, object] = {}
|
|
for navn, d in k.dicts.items():
|
|
dicts[navn] = bygg_tiltak(k, navn, d) if navn in {"CH0", "CH2"} else dict(d)
|
|
return {
|
|
"navn": k.etikett,
|
|
"styring": {"regulator": k.regulator, "forsinkelse": k.forsinkelse, "reaktivitet": k.reaktivitet},
|
|
"slutt": k.slutt,
|
|
"tiltak": copy.deepcopy(k.tiltak),
|
|
"parametere": dicts,
|
|
"utvidelse": copy.deepcopy(k.utvidelse),
|
|
}
|
|
|
|
|
|
def fra_tuppel(r: tuple[Arr, Arr, Arr, Arr, Arr]) -> dict[str, object]:
|
|
"""(t, K, U, L, T) fra simulate og simulate_soc."""
|
|
return dict(zip(("t", "K", "U", "L", "T"), r))
|
|
|
|
|
|
def kjor_endo(sim: ModuleType, k: Kjoring) -> dict[str, object]:
|
|
"""simulate_endo; tiltakene blir tidsplanene xi_sched og Z_sched (regel "sched")."""
|
|
ch = bygg_tiltak(k, "CH0", k.dicts["CH0"])
|
|
xs, zs = ch.get("xifun"), ch.get("Zfun")
|
|
ch.update(xifun=None, Zfun=None)
|
|
raa = dict(sim.simulate_endo(k.regel, tau=k.forsinkelse, p=k.dicts["P"], ch=ch, e=k.dicts["ENDO"],
|
|
kp=k.reaktivitet, tend=k.slutt, xi_sched=xs, Z_sched=zs))
|
|
raa["Z_rel"] = raa["Z"] / ch["Z"]
|
|
return raa
|
|
|
|
|
|
def kjor(sim: ModuleType, k: Kjoring, utv: ModuleType | None) -> Resultat:
|
|
"""Kjør én simulering med kjernens funksjoner."""
|
|
p = k.dicts["P"]
|
|
tau, kp = k.forsinkelse, k.reaktivitet
|
|
with horisont(sim, k.slutt), np.errstate(all="ignore"):
|
|
if utv is not None:
|
|
raa: dict[str, object] = dict(utv.simuler(til_utvidelse(k)))
|
|
elif k.modell == "simulate":
|
|
raa = fra_tuppel(sim.simulate(k.regulator, tau, p=p, kp=kp))
|
|
elif k.modell == "simulate_chain":
|
|
ch = bygg_tiltak(k, "CH0", k.dicts["CH0"])
|
|
raa = dict(sim.simulate_chain(k.regulator, tau, ch, p=p, kp=kp,
|
|
mpc_knows_chain=bool(k.mpc_kjenner_kjeden)))
|
|
avledet_kjede(sim, ch, raa)
|
|
elif k.modell == "simulate_chain2":
|
|
raa = dict(sim.simulate_chain2(tau, bygg_tiltak(k, "CH2", k.dicts["CH2"]), p=p, kp=kp))
|
|
elif k.modell == "simulate_soc":
|
|
raa = fra_tuppel(sim.simulate_soc(k.regulator, tau, p, s=k.dicts["SOC"], kp=kp,
|
|
shock=k.sjokk["faktor"], t_on=k.sjokk["fra"],
|
|
t_off=k.sjokk["til"]))
|
|
else:
|
|
raa = kjor_endo(sim, k)
|
|
return del_opp(raa, k)
|
|
|
|
|
|
def avledet_kjede(sim: ModuleType, ch: dict[str, object], raa: dict[str, object]) -> None:
|
|
"""Kapasitet og andel til domstolen over tid i simulate_chain (fra chain_at i kjernen)."""
|
|
t = raa["t"]
|
|
if not isinstance(t, np.ndarray):
|
|
return
|
|
par = [sim.chain_at(ch, float(x)) for x in t]
|
|
z = np.array([float(a) for a, _ in par])
|
|
raa["xi"] = np.array([float(b) for _, b in par])
|
|
if np.all(np.isfinite(z)):
|
|
raa["Z"] = z
|
|
raa["Z_rel"] = z / tallfelt(ch, "Z")
|
|
|
|
|
|
def del_opp(raa: dict[str, object], k: Kjoring) -> Resultat:
|
|
"""Skill tidsserier, skalarer og logg."""
|
|
t = raa.get("t")
|
|
if not isinstance(t, np.ndarray):
|
|
melding = f'Kjøringen {k.etikett} ga ingen tidsserie "t".'
|
|
raise ScenarioFeil(melding)
|
|
serier: dict[str, Arr] = {}
|
|
skalarer: dict[str, float] = {}
|
|
logg: list[str] = []
|
|
for navn, v in raa.items():
|
|
if isinstance(v, np.ndarray) and v.shape == t.shape:
|
|
serier[navn] = v.astype(np.float64, copy=False)
|
|
elif isinstance(v, (int, float, np.floating)) and not isinstance(v, bool):
|
|
skalarer[navn] = float(v)
|
|
elif navn == "log" and isinstance(v, list):
|
|
logg = [str(x) for x in v]
|
|
return Resultat(serier=serier, skalarer=skalarer, logg=logg)
|
|
|
|
|
|
def kjor_trygt(sim: ModuleType, k: Kjoring, utv: ModuleType | None) -> Resultat:
|
|
"""Som kjor, men regnefeil i kjernen blir en norsk melding (exit-kode 2)."""
|
|
try:
|
|
return kjor(sim, k, utv)
|
|
except (ArithmeticError, ValueError) as e:
|
|
hvem = f" ({kolonneoverskrift(k, 1)})" if k.sveipverdier else ""
|
|
melding = (f"Kjøringen{hvem} stoppet i kjernen: {e}. Parameterkombinasjonen gir trolig ingen "
|
|
"likevekt ved start eller en løsning som vokser over alle grenser.")
|
|
raise ScenarioFeil(melding) from None
|
|
|
|
|
|
# --------------------------------------------------------------------------- nøkkeltall
|
|
|
|
|
|
class Tall(NamedTuple):
|
|
"""Ett nøkkeltall."""
|
|
|
|
nokkel: str
|
|
tekst: str
|
|
verdi: float | None
|
|
desimaler: int
|
|
|
|
|
|
class Nokkeltall:
|
|
"""Samler nøkkeltallene til én kjøring."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Tom liste."""
|
|
self.tall: list[Tall] = []
|
|
|
|
def legg(self, nokkel: str, tekst_: str, verdi: float | None, d: int) -> None:
|
|
"""Legg til ett nøkkeltall."""
|
|
self.tall.append(Tall(nokkel, tekst_, None if verdi is None else float(verdi), d))
|
|
|
|
|
|
def tall_krim(n: Nokkeltall, s: dict[str, Arr], endo: bool) -> None: # noqa: FBT001 - intern hjelper
|
|
"""Kriminalitet K (summary, summary_endo)."""
|
|
t, k_ = s["t"], s["K"]
|
|
if endo and np.any(t >= SNITT_FRA):
|
|
n.legg("K_snitt", "Kriminalitet K, gjennomsnitt fra mnd 12", k_[t >= SNITT_FRA].mean(), 3)
|
|
n.legg("K_maks", "Kriminalitet K, høyeste", k_.max(), 3)
|
|
n.legg("K_slutt", "Kriminalitet K, ved slutt", k_[-1], 3)
|
|
sen = t > SEN_FASE
|
|
n.legg("spenn_K", "Spenn i K etter mnd 60 (høyeste minus laveste)",
|
|
k_[sen].max() - k_[sen].min() if np.any(sen) else None, 3)
|
|
if t[-1] > 2 * SEN_FASE:
|
|
sist = t > t[-1] - SEN_FASE
|
|
n.legg("spenn_K_sist", "Spenn i K de siste 60 mnd (vedvarende svingning)",
|
|
k_[sist].max() - k_[sist].min(), 3)
|
|
|
|
|
|
def tall_innsats_tillit(n: Nokkeltall, s: dict[str, Arr]) -> None:
|
|
"""Belastning per klasse, innsats T og tillit L."""
|
|
for navn, tekst_ in (("KA", "Belastning klasse A (alvorlige), ved slutt"),
|
|
("KB", "Belastning klasse B (øvrige), ved slutt")):
|
|
if navn in s:
|
|
n.legg(f"{navn}_slutt", tekst_, s[navn][-1], 3)
|
|
if "T" in s:
|
|
n.legg("T_maks", "Politiets innsats T, høyeste", s["T"].max(), 2)
|
|
n.legg("T_slutt", "Politiets innsats T, ved slutt", s["T"][-1], 2)
|
|
if "L" in s:
|
|
n.legg("L_min", "Tillit, laveste", s["L"].min(), 3)
|
|
n.legg("L_slutt", "Tillit, ved slutt", s["L"][-1], 3)
|
|
|
|
|
|
def tall_kjede(n: Nokkeltall, s: dict[str, Arr], endo: bool) -> None: # noqa: FBT001 - intern hjelper
|
|
"""Saksbehandlingstid, kø og strøm i kjeden (summary, summary2)."""
|
|
for navn, tekst_ in (("tauD", "Saksbehandlingstid i domstolen"), ("tauA", "Saksbehandlingstid klasse A"),
|
|
("tauB", "Saksbehandlingstid klasse B")):
|
|
if navn in s:
|
|
n.legg(f"{navn}_maks", f"{tekst_}, lengste (mnd)", s[navn].max(), 1)
|
|
n.legg(f"{navn}_slutt", f"{tekst_}, ved slutt (mnd)", s[navn][-1], 1)
|
|
for navn in ("Q", "QA", "QB"):
|
|
if navn in s:
|
|
n.legg(f"{navn}_slutt", f"Kø {navn}, ved slutt", s[navn][-1], 2)
|
|
if "JD" in s and "RD" in s and not endo:
|
|
rd = float(s["RD"][-1])
|
|
n.legg("inn_ut_slutt", "Saker inn / avgjort, ved slutt", float(s["JD"][-1]) / rd if rd > 0 else None, 2)
|
|
|
|
|
|
def tall_tiltak(n: Nokkeltall, s: dict[str, Arr]) -> None:
|
|
"""Kapasitet, andel utenom og kvalitet."""
|
|
if "Z_rel" in s and np.ptp(s["Z_rel"]) > 0:
|
|
n.legg("Z_maks", "Kapasitet, høyeste (1,00 = ved start)", s["Z_rel"].max(), 2)
|
|
n.legg("Z_slutt", "Kapasitet, ved slutt (1,00 = ved start)", s["Z_rel"][-1], 2)
|
|
if "xi" in s and np.ptp(s["xi"]) > 0:
|
|
n.legg("utenom_maks_pst", "Andel ledet utenom, høyeste (%)", (1.0 - s["xi"].min()) * 100, 0)
|
|
n.legg("utenom_slutt_pst", "Andel ledet utenom, ved slutt (%)", (1.0 - s["xi"][-1]) * 100, 0)
|
|
if "q" in s and np.any(s["q"] != 0):
|
|
n.legg("q_maks", "Prosessuell kvalitet q, høyeste", s["q"].max(), 3)
|
|
|
|
|
|
def tall_endo(n: Nokkeltall, s: dict[str, Arr], ch: dict[str, object], p: dict[str, object]) -> None:
|
|
"""Frafall og kapasitetsår (summary_endo; frafallet i måneder av innstrømningen, som i notatets tabell)."""
|
|
t = s["t"]
|
|
z0 = tallfelt(ch, "Z")
|
|
j0 = tallfelt(p, "gamma") * tallfelt(p, "Lam0") * tallfelt(p, "T0") # J0 i sim.py
|
|
n.legg("frafall", "Frafall fra køen, i måneder av innstrømningen før sjokket",
|
|
float(np.trapezoid(tallfelt(ch, "varsigma") * s["Q"], t)) / j0 if j0 > 0 else None, 1)
|
|
n.legg("kapasitetsaar", "Kapasitetsår over utgangsnivået",
|
|
float(np.trapezoid(np.maximum(s["Z"] - z0, 0) / z0, t) / 12), 2)
|
|
|
|
|
|
def nokkeltall(r: Resultat, k: Kjoring) -> list[Tall]:
|
|
"""Nøkkeltallene kjernens summary, summary2 og summary_endo skriver ut, for én kjøring."""
|
|
s = r.serier
|
|
n = Nokkeltall()
|
|
endo = k.modell == "simulate_endo"
|
|
if "K" in s:
|
|
tall_krim(n, s, endo)
|
|
tall_innsats_tillit(n, s)
|
|
tall_kjede(n, s, endo)
|
|
tall_tiltak(n, s)
|
|
if endo and "Z" in s and "Q" in s:
|
|
tall_endo(n, s, k.dicts["CH0"], k.dicts["P"])
|
|
return n.tall
|
|
|
|
|
|
# --------------------------------------------------------------------------- utdata
|
|
|
|
|
|
@dataclass
|
|
class Utfall:
|
|
"""Alt som trengs for å skrive sammendraget og resultat.json."""
|
|
|
|
sc: Scenario
|
|
fil: Path
|
|
alle: list[list[Tall]]
|
|
resultater: list[Resultat]
|
|
filer: list[tuple[str, list[str]]]
|
|
dt: float
|
|
|
|
|
|
def kolonner_for(sc: Scenario, r: Resultat) -> list[str]:
|
|
"""Kolonnene som skrives for en kjøring."""
|
|
if sc.kolonner is None:
|
|
return list(r.serier)
|
|
mangler = [c for c in sc.kolonner if c not in r.serier]
|
|
if mangler:
|
|
feil("utdata.kolonner", f'"{mangler[0]}" finnes ikke for denne modellen.'
|
|
+ lignende(mangler[0], list(r.serier)) + " Tilgjengelige: " + ", ".join(r.serier) + ".")
|
|
return sc.kolonner
|
|
|
|
|
|
def skriv_csv(sim: ModuleType, sc: Scenario, r: Resultat, sti: Path) -> list[str]:
|
|
"""Skriv tidsseriene. "notat": kjernens write_csv (mellomrom, %.5f); "csv": komma og .10g."""
|
|
kolonner = kolonner_for(sc, r)
|
|
cols = [r.serier[c] for c in kolonner]
|
|
if sc.format == "notat":
|
|
sim.write_csv(str(sti), cols, sc.overskrift or " ".join(kolonner), every=sc.hvert)
|
|
return kolonner
|
|
tabell_ = np.column_stack(cols)[::sc.hvert]
|
|
linjer = [",".join(kolonner)]
|
|
linjer.extend(",".join(format(float(x), ".10g") for x in rad) for rad in tabell_)
|
|
sti.write_text("\n".join(linjer) + "\n", encoding="utf-8")
|
|
return kolonner
|
|
|
|
|
|
def filnavn(sc: Scenario, k: Kjoring) -> str:
|
|
"""CSV-fila til en kjøring."""
|
|
endelse = ".dat" if sc.format == "notat" else ".csv"
|
|
return (k.etikett if not sc.sveip else f"{slug(sc.navn)}__{k.etikett}") + endelse
|
|
|
|
|
|
def kolonneoverskrift(k: Kjoring, antall_sveip: int) -> str:
|
|
"""Kort navn på en kjøring i tabellen: «navn=verdi» for ett sveip, verdiene for et rutenett."""
|
|
if not k.sveipverdier:
|
|
return "verdi"
|
|
if antall_sveip == 1:
|
|
sti, v = k.sveipverdier[0]
|
|
return f"{sti.rsplit('.', 1)[-1]}={vis_verdi(v)}"
|
|
return " / ".join(vis_verdi(v) for _, v in k.sveipverdier)
|
|
|
|
|
|
def tabell(rader: list[list[str]]) -> list[str]:
|
|
"""Markdown-tabell med kolonner som står rett også i terminalen (tall høyrejustert)."""
|
|
bredde = [max(len(rad[i]) for rad in rader) for i in range(len(rader[0]))]
|
|
ut = []
|
|
for nr, rad in enumerate(rader):
|
|
celler = [c.ljust(bredde[i]) if i == 0 else c.rjust(bredde[i]) for i, c in enumerate(rad)]
|
|
ut.append("| " + " | ".join(celler) + " |")
|
|
if nr == 0:
|
|
ut.append("|" + "|".join("-" * (b + 2) if i == 0 else "-" * (b + 1) + ":"
|
|
for i, b in enumerate(bredde)) + "|")
|
|
return ut
|
|
|
|
|
|
def nokkeltabeller(sc: Scenario, alle: list[list[Tall]]) -> list[str]:
|
|
"""Nøkkeltall: én rad per nøkkeltall, én kolonne per kjøring (høyst seks per tabell)."""
|
|
ut: list[str] = []
|
|
nokler: dict[str, tuple[str, int]] = {}
|
|
for tallene in alle:
|
|
for x in tallene:
|
|
nokler.setdefault(x.nokkel, (x.tekst, x.desimaler))
|
|
oppslag = [{x.nokkel: x for x in tallene} for tallene in alle]
|
|
for start in range(0, len(sc.kjoringer), MAKS_KOLONNER_I_TABELL):
|
|
del_ = range(start, min(start + MAKS_KOLONNER_I_TABELL, len(sc.kjoringer)))
|
|
rader = [["Nøkkeltall", *(kolonneoverskrift(sc.kjoringer[i], len(sc.sveip)) for i in del_)]]
|
|
for nokkel, (tekst_, d) in nokler.items():
|
|
rader.append([tekst_, *(tall(oppslag[i][nokkel].verdi, d) if nokkel in oppslag[i] else "-"
|
|
for i in del_)])
|
|
ut.extend(tabell(rader))
|
|
ut.append("")
|
|
return ut
|
|
|
|
|
|
def matrise(sc: Scenario, alle: list[list[Tall]]) -> list[str]:
|
|
"""For et rutenett: ett nøkkeltall som tabell med den første parameteren i rader."""
|
|
(sti1, v1), (sti2, v2) = sc.sveip
|
|
oppslag = [{x.nokkel: x for x in tallene} for tallene in alle]
|
|
nokkel = next((n for n in ("spenn_K_sist", "spenn_K", "K_slutt") if n in oppslag[0]), "")
|
|
if nokkel not in oppslag[0]:
|
|
return []
|
|
x0 = oppslag[0][nokkel]
|
|
rader = [[f"{sti1.rsplit('.', 1)[-1]} / {sti2.rsplit('.', 1)[-1]}", *(vis_verdi(v) for v in v2)]]
|
|
for i, a in enumerate(v1):
|
|
rad = [vis_verdi(a)]
|
|
for j in range(len(v2)):
|
|
x = oppslag[i * len(v2) + j].get(nokkel)
|
|
rad.append(tall(x.verdi if x else None, x0.desimaler))
|
|
rader.append(rad)
|
|
return [f"{x0.tekst}; rader: {sti1}, kolonner: {sti2}", "", *tabell(rader), ""]
|
|
|
|
|
|
def styringstekst(k: Kjoring) -> str:
|
|
"""Kort tekst om styringen."""
|
|
tekst_ = REGULATORTEKST.get(k.regulator, k.regulator)
|
|
if k.regulator == "fast":
|
|
return tekst_
|
|
tekst_ += f", forsinkelse {tall_kort(k.forsinkelse)} mnd, reaktivitet {tall_kort(k.reaktivitet)}"
|
|
if k.mpc_kjenner_kjeden is not None:
|
|
tekst_ += ", MPC kjenner kjeden" if k.mpc_kjenner_kjeden else ", MPC uten kjedemodell"
|
|
return tekst_
|
|
|
|
|
|
def hode(u: Utfall) -> list[str]:
|
|
"""Overskrift, modell, styring og linjen om hva som er notatets parametere."""
|
|
sc = u.sc
|
|
k0 = sc.kjoringer[0]
|
|
modelltekst = sc.utvidelse.BESKRIVELSE if sc.utvidelse is not None else MODELLTEKST.get(sc.modell, "")
|
|
linjer = [f"# Scenario: {sc.navn}", ""]
|
|
if sc.beskrivelse:
|
|
linjer += [sc.beskrivelse, ""]
|
|
linjer.append(f"- Scenariofil: {u.fil.name}")
|
|
linjer.append(f"- Modell: {sc.modell} - {modelltekst}")
|
|
sveipet = {sti.split(".", 1)[0] for sti, _ in sc.sveip}
|
|
merk = " (første kjøring; se sveipet)"
|
|
if k0.regel:
|
|
linjer.append(f"- Regel: {k0.regel}" + (merk if "regel" in sveipet else ""))
|
|
linjer.append(f"- Styring: {styringstekst(k0)}" + (merk if "styring" in sveipet else ""))
|
|
linjer.append(f"- Horisont: {tall_kort(k0.slutt)} mnd" + (merk if "slutt" in sveipet else ""))
|
|
linjer.extend(f"- Sveip: {sti} = " + "; ".join(vis_verdi(v) for v in verdier) for sti, verdier in sc.sveip)
|
|
if len(sc.sveip) == RUTENETT:
|
|
linjer.append(f"- Kolonnene i tabellen er {sc.sveip[0][0]} / {sc.sveip[1][0]}.")
|
|
valgt = ", ".join(sc.valgt) if sc.valgt else "ingenting"
|
|
linjer += ["", ("Parametrene er notatets stiliserte verdier (tabell 3 i notatet), ikke kalibrert på norske "
|
|
f"data, unntatt det scenariofila setter selv: {valgt}. Tallene viser modellens dynamikk, "
|
|
"ikke prognoser."), ""]
|
|
return linjer
|
|
|
|
|
|
def vedtakslinjer(u: Utfall) -> list[str]:
|
|
"""Budsjettvedtakene fra simulate_endo (loggen i kjernen)."""
|
|
logger = [(k, r.logg) for k, r in zip(u.sc.kjoringer, u.resultater) if r.logg]
|
|
if not logger:
|
|
return []
|
|
linjer = ["", "## Budsjettvedtak (fra regelen i simulate_endo)"]
|
|
for k, logg in logger:
|
|
linjer.append("")
|
|
if u.sc.sveip:
|
|
linjer += [f"{kolonneoverskrift(k, len(u.sc.sveip))}:", ""]
|
|
linjer.extend(f"- {x}" for x in logg)
|
|
return linjer
|
|
|
|
|
|
def sammendrag(u: Utfall) -> str:
|
|
"""Teksten i sammendrag.md (og på skjermen)."""
|
|
linjer = [*hode(u), "## Nøkkeltall", "", *nokkeltabeller(u.sc, u.alle)]
|
|
if len(u.sc.sveip) == RUTENETT:
|
|
linjer += matrise(u.sc, u.alle)
|
|
linjer.append("Slik leses tallene: 1,00 er nivået i referanselikevekten før sjokket for kriminaliteten K "
|
|
"og politiets innsats T; tilliten er 0,80 der. Kapasitet: 1,00 er kapasiteten ved start. "
|
|
"Tider i måneder.")
|
|
linjer += vedtakslinjer(u)
|
|
linjer += ["", "## Filer", ""]
|
|
linjer.extend(f"- {navn}: {', '.join(kolonner)}" for navn, kolonner in u.filer)
|
|
linjer.append(f"- Hvert {u.sc.hvert}. tidssteg ({tall(u.sc.hvert * u.dt, 2)} mnd). Alle parametere slik de "
|
|
"ble brukt, versjoner og sjekksummer står i resultat.json.")
|
|
return "\n".join(linjer) + "\n"
|
|
|
|
|
|
def kjoringsdata(sim: ModuleType, k: Kjoring) -> dict[str, object]:
|
|
"""Alle parametere slik simulatoren fikk dem (til resultat.json)."""
|
|
styring: dict[str, object] = {"regulator": k.regulator, "forsinkelse": k.forsinkelse,
|
|
"reaktivitet": k.reaktivitet}
|
|
if k.mpc_kjenner_kjeden is not None:
|
|
styring["mpc_kjenner_kjeden"] = k.mpc_kjenner_kjeden
|
|
ut: dict[str, object] = {"modell": k.modell, "styring": styring, "slutt": k.slutt, "DT": float(sim.DT)}
|
|
if k.regel is not None:
|
|
ut["regel"] = k.regel
|
|
if k.sjokk:
|
|
ut["sjokk"] = k.sjokk
|
|
if k.tiltak:
|
|
ut["tiltak"] = k.tiltak
|
|
ut["parametere"] = {navn: {n: v for n, v in d.items() if (navn, n) not in SATT_AV_TILTAK}
|
|
for navn, d in k.dicts.items()}
|
|
if k.utvidelse:
|
|
ut["utvidelse"] = k.utvidelse
|
|
return ut
|
|
|
|
|
|
def resultatjson(sim: ModuleType, u: Utfall) -> str:
|
|
"""Innholdet i resultat.json."""
|
|
sc = u.sc
|
|
kjoringer = []
|
|
for k, tallene, r, (navn, kolonner) in zip(sc.kjoringer, u.alle, u.resultater, u.filer):
|
|
post: dict[str, object] = {
|
|
"navn": k.etikett,
|
|
"sveip": dict(k.sveipverdier),
|
|
"fil": navn,
|
|
"kolonner": kolonner,
|
|
"brukt": kjoringsdata(sim, k),
|
|
"nokkeltall": {x.nokkel: x.verdi for x in tallene},
|
|
}
|
|
if r.skalarer:
|
|
post["skalarer"] = r.skalarer
|
|
if r.logg:
|
|
post["vedtak"] = r.logg
|
|
kjoringer.append(post)
|
|
forklaring: dict[str, str] = {}
|
|
for tallene in u.alle:
|
|
for x in tallene:
|
|
forklaring.setdefault(x.nokkel, x.tekst)
|
|
data: dict[str, object] = {
|
|
"scenario": sc.navn,
|
|
"beskrivelse": sc.beskrivelse,
|
|
"scenariofil": u.fil.name,
|
|
"scenariofil_sha256": sha256(u.fil),
|
|
"modell": sc.modell,
|
|
"utdata": {"format": sc.format, "hvert": sc.hvert},
|
|
"versjoner": {"python": sys.version.split()[0], "numpy": np.__version__,
|
|
"scipy": importlib.import_module("scipy").__version__},
|
|
"kjerne_sha256": {n: sha256(KJERNE / n) for n in KJERNEFILER if (KJERNE / n).exists()},
|
|
}
|
|
if sc.utvidelse is not None:
|
|
data["utvidelse_sha256"] = sha256(UTVIDELSER / f"{sc.modell[len(UTVIDELSE_PREFIKS):]}.py")
|
|
data["merknad"] = ("Parametrene er notatets stiliserte verdier, unntatt det scenariofila setter selv. "
|
|
"Tallene viser modellens dynamikk, ikke prognoser.")
|
|
data["nokkeltall_forklaring"] = forklaring
|
|
data["kjoringer"] = kjoringer
|
|
return json.dumps(json_verdi(data), ensure_ascii=False, indent=2) + "\n"
|
|
|
|
|
|
def klargjor_mappe(ut: Path) -> None:
|
|
"""Lag mappa; fjern bare filer et tidligere resultat i samme mappe har laget."""
|
|
if ut.exists() and not ut.is_dir():
|
|
melding = f"--ut: {ut} finnes og er ikke en mappe."
|
|
raise ScenarioFeil(melding)
|
|
if ut.exists() and any(ut.iterdir()):
|
|
gammel = ut / "resultat.json"
|
|
if not gammel.is_file():
|
|
melding = (f"--ut: mappa {ut} finnes og har filer som ikke er et tidligere resultat. Velg en "
|
|
"annen mappe med --ut.")
|
|
raise ScenarioFeil(melding)
|
|
try:
|
|
data = json.loads(gammel.read_text(encoding="utf-8"))
|
|
filer = [str(x["fil"]) for x in data.get("kjoringer", [])]
|
|
except (OSError, ValueError, TypeError, KeyError, AttributeError):
|
|
filer = []
|
|
for navn in [*filer, "sammendrag.md", "resultat.json"]:
|
|
sti = ut / navn
|
|
if sti.parent == ut and sti.is_file():
|
|
sti.unlink()
|
|
ut.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
# --------------------------------------------------------------------------- inngang
|
|
|
|
|
|
BRUK = ("Bruk: python3 modell/kjor.py scenario <fil.json> [--ut <mappe>]\n"
|
|
"Formatet er beskrevet i docs/scenarioformat.md.")
|
|
|
|
|
|
def tolk(argv: list[str]) -> tuple[Path, Path | None] | None:
|
|
"""Les argumentene; None betyr at bruken skal skrives ut."""
|
|
fil: Path | None = None
|
|
ut: Path | None = None
|
|
i = 0
|
|
while i < len(argv):
|
|
a = argv[i]
|
|
if a in {"-h", "--hjelp", "--help"}:
|
|
return None
|
|
if a == "--ut":
|
|
if i + 1 >= len(argv):
|
|
raise ScenarioFeil("--ut trenger en mappe.\n" + BRUK)
|
|
ut = Path(argv[i + 1])
|
|
i += 2
|
|
continue
|
|
if a.startswith("--ut="):
|
|
ut = Path(a[len("--ut="):])
|
|
elif a.startswith("-"):
|
|
raise ScenarioFeil(f"Ukjent valg {a}.\n" + BRUK)
|
|
elif fil is None:
|
|
fil = Path(a)
|
|
else:
|
|
raise ScenarioFeil(f"Bare én scenariofil om gangen (fikk også {a}).\n" + BRUK)
|
|
i += 1
|
|
if fil is None:
|
|
raise ScenarioFeil("Mangler scenariofil.\n" + BRUK)
|
|
return fil, ut
|
|
|
|
|
|
def kjor_scenario(fil: Path, ut: Path | None) -> tuple[str, Path]:
|
|
"""Kjør hele scenariofila og skriv filene. Returnerer sammendraget og mappa."""
|
|
sc = les_scenario(fil)
|
|
sim = last_kjerne()
|
|
mappe = ut if ut is not None else Path.cwd() / "resultater" / slug(sc.navn)
|
|
resultater = [kjor_trygt(sim, k, sc.utvidelse) for k in sc.kjoringer]
|
|
alle = [nokkeltall(r, k) for r, k in zip(resultater, sc.kjoringer)]
|
|
for r in resultater:
|
|
kolonner_for(sc, r)
|
|
klargjor_mappe(mappe)
|
|
filer: list[tuple[str, list[str]]] = []
|
|
for k, r in zip(sc.kjoringer, resultater):
|
|
navn = filnavn(sc, k)
|
|
filer.append((navn, skriv_csv(sim, sc, r, mappe / navn)))
|
|
u = Utfall(sc=sc, fil=fil, alle=alle, resultater=resultater, filer=filer, dt=float(sim.DT))
|
|
tekst_ = sammendrag(u)
|
|
(mappe / "sammendrag.md").write_text(tekst_, encoding="utf-8")
|
|
(mappe / "resultat.json").write_text(resultatjson(sim, u), encoding="utf-8")
|
|
return tekst_, mappe
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
"""Kommandoen scenario. Exit-kode 0 ok, 2 feil i scenariofila eller bruken."""
|
|
fil_tekst = ""
|
|
try:
|
|
tolket = tolk(argv)
|
|
if tolket is None:
|
|
sys.stdout.write(BRUK + "\n")
|
|
return 0
|
|
fil, ut = tolket
|
|
fil_tekst = str(fil)
|
|
tekst_, mappe = kjor_scenario(fil, ut)
|
|
except ScenarioFeil as e:
|
|
hvor = f"Feil i scenariofila {fil_tekst}: " if isinstance(e, FeltFeil) else ""
|
|
sys.stdout.flush()
|
|
sys.stderr.write(f"{hvor}{e}\n")
|
|
return 2
|
|
sys.stdout.write(tekst_)
|
|
sys.stdout.write(f"\nResultatene ligger i {mappe}\n")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|