"""Build ~/data/ice-saints/daily.parquet from DWD's open daily KL files: station, date, tmk, tnk.
Source: opendata.dwd.de/climate_environment/CDC/observations_germany/climate/daily/kl/historical/
Run after the seal; design.py is the only thing that reads rows before it."""
import io, re, sys, zipfile, urllib.request
from pathlib import Path
import pandas as pd
sys.path.insert(0, str(Path(__file__).parent))
import rule
U = "https://opendata.dwd.de/climate_environment/CDC/observations_germany/climate/daily/kl/historical/"
OUT = Path.home() / "data/ice-saints/daily.parquet"
idx = urllib.request.urlopen(U, timeout=60).read().decode()
parts = []
for sid in rule.FRAME:
    name = re.search(rf"tageswerte_KL_{sid}_\d+_\d+_hist\.zip", idx).group(0)
    z = zipfile.ZipFile(io.BytesIO(urllib.request.urlopen(U + name, timeout=180).read()))
    f = [n for n in z.namelist() if n.startswith("produkt_klima_tag")][0]
    df = pd.read_csv(z.open(f), sep=";", skipinitialspace=True, dtype=str, encoding="latin1")
    df.columns = [c.strip() for c in df.columns]
    out = pd.DataFrame({"station": sid,
        "date": pd.to_datetime(df["MESS_DATUM"].str.strip(), format="%Y%m%d"),
        "tmk": pd.to_numeric(df["TMK"], errors="coerce"),
        "tnk": pd.to_numeric(df["TNK"], errors="coerce")})
    out.loc[out.tmk <= -900, "tmk"] = None
    out.loc[out.tnk <= -900, "tnk"] = None
    parts.append(out)
    print(sid, len(out), flush=True)
pd.concat(parts).to_parquet(OUT)
print("wrote", OUT)