Strømpriser i Norge by @mmy

#!/usr/bin/env python3
"""Norwegian day-ahead power prices -> the `strompriser` novem space.

Step one of the `strompriser` job. It fetches the Nord Pool day-ahead prices
for all five bidding zones (NO1-NO5) from hvakosterstrommen.no, keeps them in
the space as one CSV per day, and refreshes the `strompriser` plot (NO1,
yesterday/today/tomorrow, incl. VAT) from that dataset. Later chain steps
(and other jobs) read the same space to render their own plots.

The space is mounted at /mnt/prices (the job's config/spaces):

    README.md                      the dataset's schema and provenance
    day-ahead/<YYYY>/<YYYY-MM-DD>.csv
        time_start,time_end,zone,nok_per_kwh,eur_per_kwh,exr
        one row per zone per published slot (24/day hourly; 96 when the
        source publishes quarter-hours), prices excl. VAT and grid tariffs

A day file is only written once all five zones are published, so a present
file is always complete and never rewritten. Each run fetches the days after
the newest file through tomorrow (tomorrow appears ~13:00 CET).

    python3 strompriser.py [--prices DIR] [--since YYYY-MM-DD] [--plot NAME]

Fetching and the dataset are stdlib only; refreshing the plot uses the novem
Python SDK (requirements.txt), which reads the run's NOVEM_TOKEN from the
environment.
"""

import argparse
import csv
import datetime as dt
import inspect
import io
import json
import sys
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from zoneinfo import ZoneInfo

ZONES = ["NO1", "NO2", "NO3", "NO4", "NO5"]
FIELDS = ["time_start", "time_end", "zone", "nok_per_kwh", "eur_per_kwh", "exr"]
SOURCE = "https://www.hvakosterstrommen.no/api/v1/prices/{d:%Y}/{d:%m}-{d:%d}_{z}.json"
MVA = 0.25  # Norwegian VAT, applied for the strompriser plot only
OSLO = ZoneInfo("Europe/Oslo")

README = """\
# Strømpriser — Norwegian day-ahead power prices

Nord Pool day-ahead prices for Norway's five bidding zones (NO1 Oslo, NO2
Kristiansand, NO3 Trondheim, NO4 Tromsø, NO5 Bergen), as published by
hvakosterstrommen.no. Maintained by the `strompriser` novem job
(repo `strompriser-aux`), which adds each new day as soon as all five zones
are published (tomorrow's prices appear around 13:00 CET).

## Layout

    day-ahead/<YYYY>/<YYYY-MM-DD>.csv

One file per delivery day (Europe/Oslo), complete and never rewritten:

| column        | meaning                                        |
| ------------- | ---------------------------------------------- |
| `time_start`  | slot start, ISO 8601 with Oslo offset          |
| `time_end`    | slot end                                       |
| `zone`        | `NO1` … `NO5`                                  |
| `nok_per_kwh` | price in NOK/kWh, excl. VAT and grid tariffs   |
| `eur_per_kwh` | price in EUR/kWh                               |
| `exr`         | EUR→NOK exchange rate used by the source       |

Slots are hourly (24 per day; 23/25 on DST days), or quarter-hourly where
the source publishes 15-minute prices. Group by `time_start[:13]` for hours.
"""


def fetch(day: dt.date, zone: str) -> list | None:
    """One zone-day from the source, or None if it isn't published (yet)."""
    req = urllib.request.Request(SOURCE.format(d=day, z=zone),
                                 headers={"User-Agent": "novem-strompriser/2.0"})
    for attempt in range(4):
        try:
            with urllib.request.urlopen(req, timeout=30) as r:
                return json.loads(r.read())
        except urllib.error.HTTPError as e:
            if e.code == 404:
                return None
        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError):
            pass
        time.sleep(2 ** attempt)
    raise RuntimeError(f"giving up on {zone} {day}")


def day_path(root: Path, day: dt.date) -> Path:
    return root / "day-ahead" / f"{day:%Y}" / f"{day:%Y-%m-%d}.csv"


def day_csv(by_zone: dict[str, list]) -> str:
    """A day's file content from the source's per-zone JSON."""
    buf = io.StringIO()
    w = csv.writer(buf, lineterminator="\n")
    w.writerow(FIELDS)
    for z in ZONES:
        for e in by_zone[z]:
            w.writerow([e["time_start"], e["time_end"], z,
                        e["NOK_per_kWh"], e["EUR_per_kWh"], e["EXR"]])
    return buf.getvalue()


def read_day(root: Path, day: dt.date) -> list[dict] | None:
    f = day_path(root, day)
    return list(csv.DictReader(f.open())) if f.exists() else None


def newest_day(root: Path) -> dt.date | None:
    files = sorted((root / "day-ahead").glob("*/*.csv"))
    return dt.date.fromisoformat(files[-1].stem) if files else None


def ingest(root: Path, since: dt.date, until: dt.date) -> list[dt.date]:
    """Fetch and store every complete day in [since, until] not yet stored."""
    days = [since + dt.timedelta(n) for n in range((until - since).days + 1)]
    days = [d for d in days if not day_path(root, d).exists()]
    jobs = [(d, z) for d in days for z in ZONES]
    got: dict[dt.date, dict[str, list]] = {}
    with ThreadPoolExecutor(max_workers=5) as pool:
        for (d, z), raw in zip(jobs, pool.map(lambda j: fetch(*j), jobs)):
            if raw:
                got.setdefault(d, {})[z] = raw
    written = []
    for d in days:
        if len(got.get(d, {})) == len(ZONES):
            out = day_path(root, d)
            out.parent.mkdir(parents=True, exist_ok=True)
            out.write_text(day_csv(got[d]))
            written.append(d)
        else:
            print(f"  {d}: not published yet ({len(got.get(d, {}))}/5 zones)")
    return written


def update_plot(root: Path, name: str, today: dt.date) -> None:
    """NO1 hourly prices incl. VAT for yesterday/today/tomorrow, in the
    plot's `hour,<date>,<date>,<date>` shape, plus this script as its
    description."""
    cols: dict[str, dict[int, list[float]]] = {}
    for d in (today - dt.timedelta(1), today, today + dt.timedelta(1)):
        rows = read_day(root, d)
        if not rows:
            continue
        hours: dict[int, list[float]] = {}
        for r in rows:
            if r["zone"] == "NO1":
                hours.setdefault(int(r["time_start"][11:13]), []).append(float(r["nok_per_kwh"]))
        cols[d.isoformat()] = hours
    out = io.StringIO()
    w = csv.writer(out, lineterminator="\n")
    w.writerow(["hour", *cols])
    for h in range(24):
        w.writerow([h, *("" if h not in hs else f"{sum(hs[h]) / len(hs[h]) * (1 + MVA):.4f}"
                         for hs in cols.values())])

    from novem import Plot  # the SDK is only needed for the plot

    plot = Plot(name)
    plot.data = out.getvalue()
    plot.description = f"```python\n{inspect.getsource(sys.modules[__name__])}```\n"
    print(f"updated plot {name}: {', '.join(cols)}")


def main(argv: list[str]) -> None:
    p = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    p.add_argument("--prices", type=Path, default=Path("/mnt/prices"),
                   help="dataset root (the mounted space); default /mnt/prices")
    p.add_argument("--since", type=dt.date.fromisoformat,
                   help="first day to fetch when the dataset is empty (default: 30 days ago)")
    p.add_argument("--plot", default="strompriser",
                   help="novem plot to refresh from the dataset ('' to skip)")
    args = p.parse_args(argv)

    today = dt.datetime.now(OSLO).date()
    root = args.prices
    if not (root / "README.md").exists():
        root.mkdir(parents=True, exist_ok=True)
        (root / "README.md").write_text(README)

    newest = newest_day(root)
    since = newest + dt.timedelta(1) if newest else (args.since or today - dt.timedelta(30))
    until = today + dt.timedelta(1)
    if since > until:
        print(f"dataset through {newest}; up to date")
    else:
        print(f"dataset through {newest or '(empty)'}; fetching {since} .. {until}")
        for d in ingest(root, since, until):
            print(f"  wrote {day_path(root, d).relative_to(root)}")

    if args.plot:
        update_plot(root, args.plot, today)


if __name__ == "__main__":
    main(sys.argv[1:])
Strømpriser i Norge

Description
