Kjell Tore Guttormsen
19858bbda9
Skills ask before using a working copy found in the user's folder and state its check level first. The drift of the queue models before the shock is stated in the caveats and the README. README shows how to run the check and the tests. The drift line names the month a gradual drive begins. Co-Authored-By: Claude Fable 5.2 <noreply@anthropic.com>
1875 lines
84 KiB
Python
1875 lines
84 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 os
|
|
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", "startlikevekt", "styring", "parametere", "regel",
|
|
"sjokk", "tiltak", "slutt", "sveip", "utdata", "utvidelse")
|
|
STARTVALG = ("krev", "kalibrer", "ubalanse")
|
|
SVEIPBARE = ("styring", "parametere", "regel", "sjokk", "tiltak", "slutt", "utvidelse", "startlikevekt")
|
|
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
|
|
MIN_SLUTT = 1.0 # kjernen trenger minst to tidssteg; én måned er minste meningsfulle horisont
|
|
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)
|
|
GRADVIS_ANDEL = 0.05 # første avvik under denne andelen av det største: pådraget stiger gradvis
|
|
NOTAT_HVERT = 10 # write_csv i sim.py
|
|
LIKEVEKT_TOL = 1e-12 # største tillatte rest i dK/dt ved start (pådraget er om lag 0,12 per måned)
|
|
FLYTPARAMETERE = ("gamma", "Lam0", "T0") # gir strømmen J0 = gamma Lam0 T0 som domstolen er kalibrert for
|
|
ENDO_FAST_STROM = 0.1 # simulate_endo: observert strøm de første lagD månedene (sim.py:588), fast i kjernen
|
|
UTEN_SJOKK_MND = 12.0 # måneden driften vises for når det ikke er noe sjokk (notatets sjokkmåned)
|
|
K_OMRADE = (0.01, 12.0) # merknad utenfor: området soc_equilibria i sim.py leter i (notatets K: 0,83-9,82)
|
|
|
|
|
|
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"): Grense(1.0),
|
|
("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)}
|
|
HELE_MAANEDER = {("ENDO", "period")} # budsjettsyklusen: vedtak når t % period == 0 i hele måneder (sim.py:614)
|
|
|
|
|
|
# --------------------------------------------------------------------------- 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)
|
|
startlikevekt: str = "krev"
|
|
ubalanse: list[str] = field(default_factory=list) # hvorfor starten ikke er en likevekt
|
|
kalibrert: list[tuple[str, float, float]] = field(default_factory=list) # (sti, notatets, ny)
|
|
|
|
|
|
@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)
|
|
x = tallverdi(verdi, sti, GRENSER.get((navn, nokkel), Grense()))
|
|
if (navn, nokkel) in HELE_MAANEDER and not x.is_integer():
|
|
feil(sti, f"må være et helt antall måneder (kjernen gjør vedtak bare i hele måneder), ikke {vis(verdi)}.")
|
|
return x
|
|
|
|
|
|
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", IKKE_NEG)
|
|
if forsinkelse > slutt:
|
|
feil("slutt", f"horisonten ({tall_kort(slutt)} mnd) er kortere enn forsinkelsen i styringen "
|
|
f"(styring.forsinkelse = {tall_kort(forsinkelse)} mnd). Øk slutt eller kort ned forsinkelsen.")
|
|
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)
|
|
|
|
|
|
# --------------------------------------------------------------------------- startlikevekten
|
|
|
|
|
|
def tall_g(x: float) -> str:
|
|
"""Et tall med fire gjeldende sifre, norsk desimaltegn."""
|
|
return f"{x:.4g}".replace(".", ",")
|
|
|
|
|
|
def oppgitte_nokler(parametere: object) -> dict[str, set[str]]:
|
|
"""Nøklene fila setter selv under parametere, per gruppe."""
|
|
ut: dict[str, set[str]] = {}
|
|
if isinstance(parametere, dict):
|
|
for navn, d in parametere.items():
|
|
if isinstance(d, dict):
|
|
ut[str(navn)] = {str(n) for n in d}
|
|
return ut
|
|
|
|
|
|
def paadragsrest(sim: ModuleType, k: Kjoring) -> float:
|
|
"""dK/dt ved start (K = 1, Lambda = Lam0, T = T0, før sjokket), regnet med kjernens funksjoner."""
|
|
p = k.dicts["P"]
|
|
om = tallfelt(p, "Omega0")
|
|
if k.modell == "simulate_soc": # kjernen rekalibrerer Omega0 for rekrutteringsleddet (sim.py:455)
|
|
s = k.dicts["SOC"]
|
|
om = om - tallfelt(s, "eps") * float(sim.hill(1.0, s)) + tallfelt(s, "eps") * float(sim.hill(1.0, s))
|
|
fjernet = float(sim.removal(1.0, tallfelt(p, "Lam0"), tallfelt(p, "T0"), p))
|
|
return om - fjernet - tallfelt(p, "delta") * 1.0
|
|
|
|
|
|
def strom_j0(p: dict[str, object]) -> float:
|
|
"""Strømmen til domstolen ved start slik kjernen kalibrerer den (J0 i sim.py:154, uten eta)."""
|
|
return tallfelt(p, "gamma") * tallfelt(p, "Lam0") * tallfelt(p, "T0") * 1.0
|
|
|
|
|
|
class Startfunn(NamedTuple):
|
|
"""Det som gjør at starten ikke er en likevekt."""
|
|
|
|
kan_kalibreres: list[str]
|
|
kan_ikke: list[str]
|
|
rettinger: list[tuple[str, str, float, float]] # (gruppe, nøkkel, gammel, ny)
|
|
|
|
|
|
def funn_grunnmodell(sim: ModuleType, k: Kjoring, oppgitt: dict[str, set[str]], f: Startfunn) -> None:
|
|
"""Pådrag, mål, tak og proporsjonalitet i grunnmodellen."""
|
|
p = k.dicts["P"]
|
|
rest = paadragsrest(sim, k)
|
|
if abs(rest) > LIKEVEKT_TOL:
|
|
ny = tallfelt(p, "Omega0") - rest
|
|
tekst_ = (f"pådraget Omega0 ({tall_g(tallfelt(p, 'Omega0'))}) er ikke lik det politiets innsats fjerner "
|
|
f"pluss det som faller fra av seg selv ved K = 1 ({tall_g(ny)}), så kriminaliteten ville "
|
|
"drevet bort fra 1,00 også uten sjokk")
|
|
if "Omega0" in oppgitt.get("P", set()):
|
|
f.kan_ikke.append(tekst_ + " (Omega0 er satt i fila)")
|
|
else:
|
|
f.kan_kalibreres.append(tekst_)
|
|
f.rettinger.append(("P", "Omega0", tallfelt(p, "Omega0"), ny))
|
|
t0, kstar = tallfelt(p, "T0"), tallfelt(p, "Kstar")
|
|
regel_virker = k.regulator == "MPC" or (k.regulator == "P" and k.reaktivitet > 0)
|
|
if regel_virker and kstar != 1.0:
|
|
f.kan_ikke.append(f"målet Kstar er {tall_g(kstar)}, ikke 1, så styringen flytter innsatsen bort fra T0 "
|
|
"allerede ved start")
|
|
if k.regulator != "fast" and t0 > tallfelt(p, "Tmax"):
|
|
f.kan_ikke.append(f"T0 ({tall_g(t0)}) er over taket Tmax ({tall_g(tallfelt(p, 'Tmax'))})")
|
|
if t0 > tallfelt(p, "Tprop") * 1.0:
|
|
f.kan_ikke.append(f"T0 ({tall_g(t0)}) er over Tprop ({tall_g(tallfelt(p, 'Tprop'))}) ganger K, så "
|
|
"innsatsen regnes som uforholdsmessig og tilliten faller fra start")
|
|
|
|
|
|
def funn_domstol(sim: ModuleType, k: Kjoring, navn: str, oppgitt: dict[str, set[str]], f: Startfunn) -> None:
|
|
"""Domstolens kapasitet og behandlingstid mot strømmen ved start (CH0 eller CH2)."""
|
|
p, ch = k.dicts["P"], k.dicts[navn]
|
|
if tallfelt(p, "eta") != 1.0:
|
|
f.kan_ikke.append(f"eta er {tall_g(tallfelt(p, 'eta'))}, men kjernen regner strømmen til domstolen ved "
|
|
"start uten eta (init_chain og simulate_chain2 i sim.py), så køen starter ikke i balanse")
|
|
j0 = strom_j0(p)
|
|
if j0 <= 0:
|
|
f.kan_ikke.append("strømmen til domstolen er 0 ved start (gamma, Lam0 eller T0 er 0), så domstolen kan "
|
|
"ikke kalibreres")
|
|
return
|
|
satt = oppgitt.get(navn, set())
|
|
flyt_endret = any(n in oppgitt.get("P", set()) and p[n] != sim.P[n] for n in FLYTPARAMETERE)
|
|
z = tallfelt(ch, "Z")
|
|
if flyt_endret and not {"Z", "m"} <= satt:
|
|
z_ny = z if "Z" in satt else j0 / float(sim.UPS0)
|
|
f.kan_kalibreres.append(
|
|
f"domstolens kapasitet og minste behandlingstid ({navn}.Z, {navn}.m) er notatets, regnet for en "
|
|
f"strøm på {tall_g(strom_j0(sim.P))} saker per måned, men strømmen er nå {tall_g(j0)}, så "
|
|
"kapasitet og saksbehandlingstid passer ikke lenger til strømmen")
|
|
if "Z" not in satt:
|
|
f.rettinger.append((navn, "Z", z, z_ny))
|
|
if "m" not in satt:
|
|
f.rettinger.append((navn, "m", tallfelt(ch, "m") if ch["m"] != "kalibrer" else math.nan, math.nan))
|
|
z = z_ny
|
|
andel = (tallfelt(ch, "xi") if navn == "CH0"
|
|
else tallfelt(ch, "piA") * tallfelt(ch, "xiA") + (1 - tallfelt(ch, "piA")) * tallfelt(ch, "xiB"))
|
|
if andel * j0 >= z:
|
|
f.kan_ikke.append(f"domstolen får {tall_g(andel * j0)} saker per måned ved start, minst like mye som "
|
|
f"kapasiteten {navn}.Z = {tall_g(z)}, så køen kan ikke stå i ro")
|
|
|
|
|
|
def kalibrert_m(sim: ModuleType, k: Kjoring, navn: str) -> float:
|
|
"""Minste behandlingstid m som gir saksbehandlingstid TAU_REF ved start.
|
|
|
|
Som MD i sim.py:155, eller calibrate_m for flere spor.
|
|
"""
|
|
ch, p = k.dicts[navn], k.dicts["P"]
|
|
j0, z, tau = strom_j0(p), tallfelt(ch, "Z"), float(sim.TAU_REF)
|
|
if navn == "CH2" and ch["c"] != 1:
|
|
return float(sim.calibrate_m(dict(ch, m=None), p))
|
|
return tau - j0 / (z * (1.0 / tau + tallfelt(ch, "varsigma")))
|
|
|
|
|
|
def startmelding(f: Startfunn, kan_kalibrere: bool) -> str: # noqa: FBT001 - intern hjelper
|
|
"""Melding når starten ikke er en likevekt og fila ikke har valgt hva som skal skje."""
|
|
grunner = "; ".join([*f.kan_kalibreres, *f.kan_ikke])
|
|
ut = (f"startpunktet er ikke en likevekt: {grunner}. Tallene ville da vist en drift som ikke kommer av "
|
|
"sjokket eller tiltakene. To veier: ")
|
|
if kan_kalibrere:
|
|
ut += ('"startlikevekt": "kalibrer" regner referansetilstanden om (' + ", ".join(
|
|
f"{g}.{n}" for g, n, _, _ in f.rettinger) + ") slik at startpunktet blir notatets referansetilstand; ")
|
|
else:
|
|
ut += '"startlikevekt": "kalibrer" kan ikke rette dette (verdiene er satt i fila eller ligger fast); '
|
|
return ut + ('"startlikevekt": "ubalanse" kjører fra ubalansen med vilje og viser i sammendraget hvor '
|
|
"langt modellen driver før sjokket. Se docs/scenarioformat.md.")
|
|
|
|
|
|
def vurder_start(sim: ModuleType, k: Kjoring, parametere: object) -> None:
|
|
"""Stopp, kalibrer eller merk kjøringen når startpunktet ikke er en likevekt ("startlikevekt")."""
|
|
oppgitt = oppgitte_nokler(parametere)
|
|
f = Startfunn([], [], [])
|
|
funn_grunnmodell(sim, k, oppgitt, f)
|
|
kjede = next((n for n in ("CH0", "CH2") if n in k.dicts), None)
|
|
if kjede is not None and not math.isinf(tallfelt(k.dicts[kjede], "Z")):
|
|
funn_domstol(sim, k, kjede, oppgitt, f)
|
|
if k.modell == "simulate_endo":
|
|
f0 = float(sim.removal(1.0, tallfelt(k.dicts["P"], "Lam0"), tallfelt(k.dicts["P"], "T0"), k.dicts["P"]))
|
|
if abs(f0 - ENDO_FAST_STROM) > LIKEVEKT_TOL:
|
|
feil("parametere.P", f"simulate_endo i kjernen bruker {tall_g(ENDO_FAST_STROM)} som observert strøm de "
|
|
f"første månedene (sim.py:588), men med disse parametrene er strømmen ved start {tall_g(f0)}. Det "
|
|
"endrer hva den koordinerte regelen gjør, og kan ikke rettes fra scenariofila. Å endre gamma, "
|
|
"Lam0, T0, eta eller ksat i simulate_endo er derfor en endring i kjernen (arbeidskopi og "
|
|
"ferdigheten forbedre).")
|
|
if not f.kan_kalibreres and not f.kan_ikke:
|
|
return
|
|
if k.startlikevekt == "krev":
|
|
feil("startlikevekt", startmelding(f, kan_kalibrere=not f.kan_ikke))
|
|
if k.startlikevekt == "ubalanse":
|
|
k.ubalanse = [*f.kan_kalibreres, *f.kan_ikke]
|
|
return
|
|
if f.kan_ikke:
|
|
feil("startlikevekt", '"kalibrer" kan ikke gjøre starten til en likevekt: ' + "; ".join(f.kan_ikke)
|
|
+ '. Endre verdiene, eller bruk "startlikevekt": "ubalanse".')
|
|
for gruppe, nokkel, gammel, ny in f.rettinger:
|
|
verdi = kalibrert_m(sim, k, gruppe) if (gruppe, nokkel) in {("CH0", "m"), ("CH2", "m")} else ny
|
|
k.dicts[gruppe][nokkel] = verdi
|
|
k.kalibrert.append((f"{gruppe}.{nokkel}", gammel, verdi))
|
|
if abs(paadragsrest(sim, k)) > LIKEVEKT_TOL:
|
|
feil("startlikevekt", "kalibreringen ga ikke en likevekt; dette er en feil i scenariokjøreren.")
|
|
|
|
|
|
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(MIN_SLUTT, 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))
|
|
k.startlikevekt = tekst(raa.get("startlikevekt", "krev"), "startlikevekt", STARTVALG)
|
|
vurder_start(sim, k, raa.get("parametere"))
|
|
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":
|
|
ch2 = bygg_tiltak(k, "CH2", k.dicts["CH2"])
|
|
raa = dict(sim.simulate_chain2(tau, ch2, p=p, kp=kp))
|
|
avledet_kjede2(ch2, raa)
|
|
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 avledet_kjede2(ch2: dict[str, object], raa: dict[str, object]) -> None:
|
|
"""Kapasitet og andelen av alle saker som går til domstolen over tid i simulate_chain2 (som court2)."""
|
|
t = raa["t"]
|
|
if not isinstance(t, np.ndarray):
|
|
return
|
|
zfun, xibfun = ch2.get("Zfun"), ch2.get("xiBfun")
|
|
z = np.array([float(zfun(float(x))) if callable(zfun) else tallfelt(ch2, "Z") for x in t])
|
|
xib = np.array([float(xibfun(float(x))) if callable(xibfun) else tallfelt(ch2, "xiB") for x in t])
|
|
pia = tallfelt(ch2, "piA")
|
|
raa["Z"] = z
|
|
raa["Z_rel"] = z / tallfelt(ch2, "Z")
|
|
raa["xi"] = pia * tallfelt(ch2, "xiA") + (1.0 - pia) * xib
|
|
|
|
|
|
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)."""
|
|
hvem = f" ({kolonneoverskrift(k, 1)})" if k.sveipverdier else ""
|
|
try:
|
|
r = kjor(sim, k, utv)
|
|
except (ArithmeticError, ValueError) as e:
|
|
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. Ingenting er skrevet.")
|
|
raise ScenarioFeil(melding) from None
|
|
ikke_tall = [navn for navn, v in r.serier.items() if not np.all(np.isfinite(v))]
|
|
if ikke_tall:
|
|
melding = (f"Kjøringen{hvem} ga verdier som ikke er endelige tall (uendelig eller NaN) i "
|
|
f"{', '.join(ikke_tall)}: løsningen har vokst over alle grenser. Ingenting er skrevet.")
|
|
raise ScenarioFeil(melding)
|
|
return r
|
|
|
|
|
|
# --------------------------------------------------------------------------- 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, j0: float) -> None: # noqa: FBT001 - intern hjelper
|
|
"""Saksbehandlingstid, kø og strøm i kjeden (summary, summary2). Køen i måneder av strømmen J0."""
|
|
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, hva in (("Q", "Kø i domstolen"), ("QA", "Kø klasse A"), ("QB", "Kø klasse B")):
|
|
if navn in s:
|
|
n.legg(f"{navn}_slutt_mnd", f"{hva} ved slutt (mnd innstrømning før sjokket)",
|
|
float(s[navn][-1]) / j0 if j0 > 0 else None, 1)
|
|
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 = strom_j0(p)
|
|
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, strom_j0(k.dicts["P"]))
|
|
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."), ""]
|
|
kalibrert = [(k, k.kalibrert) for k in sc.kjoringer if k.kalibrert]
|
|
if kalibrert:
|
|
linjer += [('Referansetilstanden er kalibrert på nytt ("startlikevekt": "kalibrer"): pådraget og domstolen '
|
|
"er regnet om som i notatets kalibrering, for de nye verdiene:"), ""]
|
|
for k, endr in kalibrert:
|
|
hvem = f"{kolonneoverskrift(k, len(sc.sveip))}: " if sc.sveip else ""
|
|
linjer.append("- " + hvem + "; ".join(
|
|
f"{sti} {tall_g(ny)}" + ("" if math.isnan(gammel) else f" (var {tall_g(gammel)})")
|
|
for sti, gammel, ny in endr))
|
|
linjer.append("")
|
|
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 sjokkstart(k: Kjoring, r: Resultat) -> float | None:
|
|
"""Måneden sjokket (eller pådraget fra utvidelsen) begynner; None uten sjokk."""
|
|
s = r.serier
|
|
if k.modell == "simulate_soc":
|
|
return k.sjokk["fra"] if k.sjokk["faktor"] != 1.0 else None
|
|
if "Omega_faktor" in s:
|
|
avvik = np.abs(s["Omega_faktor"] - 1.0)
|
|
endret = np.flatnonzero(avvik != 0.0)
|
|
if not endret.size:
|
|
return None
|
|
i0 = int(endret[0])
|
|
# Et pådrag som stiger gradvis, begynner i tidssteget før det første avviket (måned 12, ikke 12,05).
|
|
gradvis = i0 > 0 and float(avvik[i0]) < GRADVIS_ANDEL * float(avvik.max())
|
|
return float(s["t"][i0 - 1 if gradvis else i0])
|
|
p = k.dicts["P"]
|
|
return tallfelt(p, "t_shock") if tallfelt(p, "shock") != 1.0 else None
|
|
|
|
|
|
def drift_maaned(k: Kjoring, r: Resultat) -> float:
|
|
"""Måneden driften før sjokket vises for: sjokkets start, eller måned 12 uten sjokk (høyst slutt)."""
|
|
m = sjokkstart(k, r)
|
|
return min(UTEN_SJOKK_MND if m is None else m, k.slutt)
|
|
|
|
|
|
def har_ko(r: Resultat) -> bool:
|
|
"""Sann for modeller med kø i domstolen (simulate_chain, simulate_chain2, simulate_endo, utvidelse med kjede)."""
|
|
return "Q" in r.serier or "QA" in r.serier
|
|
|
|
|
|
def driftlinje(k: Kjoring, r: Resultat, dt: float, antall_sveip: int, *, med_grunner: bool) -> str:
|
|
"""Hvor langt en kjøring har drevet fra start til sjokket (ubalanse, eller kø i domstolen)."""
|
|
s = r.serier
|
|
m = drift_maaned(k, r)
|
|
i = min(round(m / dt), len(s["t"]) - 1)
|
|
deler = [f"{hva} fra {tall(float(s[navn][0]), d)} til {tall(float(s[navn][i]), d)}"
|
|
for navn, hva, d in (("K", "K", 3), ("T", "innsatsen T", 2), ("L", "tilliten", 3),
|
|
("tauD", "saksbehandlingstiden (mnd)", 1), ("tauA", "tiden i klasse A (mnd)", 1),
|
|
("tauB", "tiden i klasse B (mnd)", 1))
|
|
if navn in s]
|
|
hvem = f"{kolonneoverskrift(k, antall_sveip)}: " if antall_sveip else ""
|
|
grunner = f"{'; '.join(k.ubalanse)}. " if med_grunner else ""
|
|
deler[0] = deler[0].replace(" fra ", " gått fra ", 1)
|
|
return f"- {hvem}{grunner}Fram til måned {tall_kort(m)} har " + ", ".join(deler) + "."
|
|
|
|
|
|
KO_DRIFT = ("I modeller med kø i domstolen er dette nesten, men ikke helt, en likevekt: det som faller fra køen, "
|
|
"blir aldri fjernet fra kriminaliteten, så K og saksbehandlingstiden kryper sakte oppover også uten sjokk.")
|
|
|
|
|
|
def ko_driftlinjer(u: Utfall, kjoringer: list[tuple[Kjoring, Resultat]]) -> list[str]:
|
|
"""Driften før sjokket i kjøringer med kø som starter i notatets referansetilstand."""
|
|
med_sjokk = [(k, r) for k, r in kjoringer if (sjokkstart(k, r) or 0.0) > 0.0]
|
|
uten = len(med_sjokk) < len(kjoringer)
|
|
linjer: list[str] = []
|
|
if med_sjokk:
|
|
linjer += ["", "Slik har modellen drevet før sjokket:", ""]
|
|
linjer += [driftlinje(k, r, u.dt, len(u.sc.sveip), med_grunner=False) for k, r in med_sjokk]
|
|
if uten:
|
|
linjer += ["", ("Uten sjokk (eller med sjokk fra måned 0) kryper modellen med kø sakte oppover av seg selv; en "
|
|
'kjøring uten sjokk ("sjokk": {"faktor": 1}) viser hvor mye.')]
|
|
return linjer
|
|
|
|
|
|
def forklaring(u: Utfall) -> list[str]:
|
|
"""Hvordan tallene leses; byttes ut når starten ikke er en likevekt."""
|
|
k0 = u.sc.kjoringer[0]
|
|
lam0 = {tallfelt(k.dicts["P"], "Lam0") for k in u.sc.kjoringer}
|
|
tillit = f"tilliten er {tall(lam0.pop(), 2)} der" if len(lam0) == 1 else "tilliten er Lam0 der"
|
|
ko = any(har_ko(r) for r in u.resultater)
|
|
resten = ("Kapasitet: 1,00 er kapasiteten ved start. Tider i måneder."
|
|
+ (f" Køen er målt i måneder av innstrømningen til domstolen før sjokket (J0 = "
|
|
f"{tall_g(strom_j0(k0.dicts['P']))} i modellens enheter); i datafilene står køen i modellens "
|
|
"enheter." if ko else ""))
|
|
alle = list(zip(u.sc.kjoringer, u.resultater))
|
|
skjeve = [(k, r) for k, r in alle if k.ubalanse]
|
|
rette_med_ko = [(k, r) for k, r in alle if not k.ubalanse and har_ko(r)]
|
|
if not skjeve and not ko:
|
|
return [(f"Slik leses tallene: 1,00 er nivået i referanselikevekten før sjokket for kriminaliteten K og "
|
|
f"politiets innsats T; {tillit}. {resten}")]
|
|
if not skjeve:
|
|
return [(f"Slik leses tallene: 1,00 er nivået ved start for kriminaliteten K og politiets innsats T "
|
|
f"(notatets referansetilstand); {tillit}. {KO_DRIFT} {resten}"), *ko_driftlinjer(u, rette_med_ko)]
|
|
like = len({tuple(k.ubalanse) for k, _ in skjeve}) == 1
|
|
linjer = [('Slik leses tallene: startpunktet er ikke en likevekt ("startlikevekt": "ubalanse")'
|
|
+ (": " + "; ".join(skjeve[0][0].ubalanse) if like else "") + ". Tallene er målt fra "
|
|
"startverdiene (K og T er 1,00 ved start), men modellen ville drevet bort fra dem også uten sjokk, "
|
|
f"så 1,00 er ikke et nivå den holder seg på. {resten} Slik har modellen drevet før sjokket:"), ""]
|
|
linjer += [driftlinje(k, r, u.dt, len(u.sc.sveip), med_grunner=not like) for k, r in skjeve]
|
|
if len(skjeve) < len(u.sc.kjoringer):
|
|
linjer += ["", ("Kjøringene som ikke er nevnt, starter i notatets referansetilstand." + (f" {KO_DRIFT}" if
|
|
rette_med_ko else ""))]
|
|
linjer += ko_driftlinjer(u, rette_med_ko) if rette_med_ko else []
|
|
return linjer
|
|
|
|
|
|
def merknader(u: Utfall) -> list[str]:
|
|
"""Merknader (ikke stopp): K langt utenfor modellens område, og sjokk utenfor horisonten."""
|
|
ut: list[str] = []
|
|
for k, r in zip(u.sc.kjoringer, u.resultater):
|
|
hvem = f" ({kolonneoverskrift(k, len(u.sc.sveip))})" if u.sc.sveip else ""
|
|
if "K" in r.serier:
|
|
kmax, kmin = float(r.serier["K"].max()), float(r.serier["K"].min())
|
|
if kmax > K_OMRADE[1] or kmin < K_OMRADE[0]:
|
|
ut.append(f"Kriminaliteten{hvem} går fra {tall(kmin, 3)} til {tall(kmax, 3)}, utenfor området "
|
|
f"kjernen selv regner på (K mellom {tall(K_OMRADE[0], 2)} og {tall(K_OMRADE[1], 0)}, der "
|
|
"soc_equilibria i sim.py leter etter likevekter; notatets egne kjøringer ligger mellom "
|
|
"0,83 og 9,82). Tallene er regnet slik modellen sier, men ligningene er ikke laget for "
|
|
"så store utslag.")
|
|
if k.modell == "simulate_soc":
|
|
if k.sjokk["fra"] >= k.slutt:
|
|
ut.append(f"Sjokket{hvem} begynner i måned {tall_kort(k.sjokk['fra'])}, ikke før slutt "
|
|
f"({tall_kort(k.slutt)}): kjøringen viser ingen virkning av det.")
|
|
elif k.sjokk["til"] > k.slutt:
|
|
ut.append(f"Sjokket{hvem} varer til måned {tall_kort(k.sjokk['til'])}, etter slutt "
|
|
f"({tall_kort(k.slutt)}): kjøringen viser ikke at det går over.")
|
|
elif k.sjokk and k.sjokk["faktor"] != 1.0 and k.sjokk["fra"] >= k.slutt:
|
|
ut.append(f"Sjokket{hvem} kommer i måned {tall_kort(k.sjokk['fra'])}, ikke før slutt "
|
|
f"({tall_kort(k.slutt)}): kjøringen viser ingen virkning av det.")
|
|
return list(dict.fromkeys(ut))
|
|
|
|
|
|
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 += forklaring(u)
|
|
merk = merknader(u)
|
|
if merk:
|
|
linjer += ["", "## Merknader (kjøringen er ikke stoppet)", "", *(f"- {m}" for m in merk)]
|
|
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
|
|
post["startlikevekt"] = {
|
|
"valg": k.startlikevekt,
|
|
"kalibrert": {sti: {"for": gammel, "etter": ny} for sti, gammel, ny in k.kalibrert},
|
|
"ubalanse": k.ubalanse,
|
|
}
|
|
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"
|
|
|
|
|
|
EGNE_NOKLER = ("scenario", "scenariofil_sha256", "kjerne_sha256", "kjoringer") # i resultat.json
|
|
EGEN_DATAFIL = re.compile(r"[^/\\]+\.(csv|dat)")
|
|
|
|
|
|
def egne_filer(gammel: Path) -> list[str] | None:
|
|
"""Filene et tidligere resultat laget, eller None hvis resultat.json ikke er kjørerens."""
|
|
try:
|
|
data = json.loads(gammel.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return None
|
|
if not isinstance(data, dict) or not all(n in data for n in EGNE_NOKLER):
|
|
return None
|
|
kjoringer = data["kjoringer"]
|
|
if not isinstance(kjoringer, list):
|
|
return None
|
|
filer = [str(x.get("fil", "")) for x in kjoringer if isinstance(x, dict)]
|
|
return [f for f in filer if EGEN_DATAFIL.fullmatch(f) and f not in {".csv", ".dat"}]
|
|
|
|
|
|
def klargjor_mappe(ut: Path) -> None:
|
|
"""Lag mappa; fjern bare datafilene, sammendraget og resultat.json et tidligere resultat 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()):
|
|
filer = egne_filer(ut / "resultat.json")
|
|
if filer is None:
|
|
melding = (f"--ut: mappa {ut} finnes og har filer som ikke er et tidligere resultat fra "
|
|
"scenariokjøreren. Velg en annen mappe med --ut.")
|
|
raise ScenarioFeil(melding)
|
|
for navn in [*filer, "sammendrag.md", "resultat.json"]:
|
|
sti = ut / navn
|
|
if sti.parent == ut and sti.is_file() and not sti.is_symlink():
|
|
sti.unlink()
|
|
ut.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def skriv_alt(sim: ModuleType, sc: Scenario, fil: Path, mappe: Path,
|
|
resultater: list[Resultat]) -> str:
|
|
"""Skriv datafilene, sammendraget og resultat.json. Returnerer sammendraget."""
|
|
alle = [nokkeltall(r, k) for r, k in zip(resultater, sc.kjoringer)]
|
|
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_
|
|
|
|
|
|
# --------------------------------------------------------------------------- inngang
|
|
|
|
|
|
PYTHON = "python" if os.name == "nt" else "python3" # tolken heter python på Windows
|
|
BRUK = (f"Bruk: {PYTHON} 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]
|
|
for r in resultater:
|
|
kolonner_for(sc, r)
|
|
try:
|
|
tekst_ = skriv_alt(sim, sc, fil, mappe, resultater)
|
|
except OSError as e:
|
|
grunn = e.strerror or str(e)
|
|
melding = f"Kan ikke skrive i {mappe}: {grunn}. Velg en annen mappe med --ut."
|
|
raise ScenarioFeil(melding) from None
|
|
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:]))
|