239 lines
8.7 KiB
TypeScript
239 lines
8.7 KiB
TypeScript
// B11 EP-Online (RVO energy labels): the monthly total file ("mutatiebestand", CSV in a zip,
|
||
// ~1.6 GB unpacked, every currently registered label) aggregated to the number of homes per
|
||
// energy label for the Netherlands, Gouda (postcodes 2800–2809) and each Gouda PC4. Only these
|
||
// aggregates are stored; no address-level data leaves the run. Needs EP_ONLINE_API_KEY (loaded
|
||
// by the runner from ~/.config/stats/ep-online.env; never logged). One snapshot per publication
|
||
// date, kept over time, so the shares become a monthly series.
|
||
import { mkdtemp, rm } from 'node:fs/promises';
|
||
import { tmpdir } from 'node:os';
|
||
import { join } from 'node:path';
|
||
import {
|
||
http,
|
||
insertChunks,
|
||
log,
|
||
run,
|
||
sql,
|
||
writeSeries,
|
||
type ObsRow,
|
||
type SeriesRow
|
||
} from './common';
|
||
|
||
const API = 'https://public.ep-online.nl/api/v5';
|
||
/** Best to worst (NTA 8800 scale since 2021). */
|
||
export const LABELS = ['A++++', 'A+++', 'A++', 'A+', 'A', 'B', 'C', 'D', 'E', 'F', 'G'] as const;
|
||
const GOUDA_PC4 = Array.from({ length: 10 }, (_, i) => String(2800 + i));
|
||
const AREA_NAME: Record<string, string> = {
|
||
NL: 'Netherlands',
|
||
GOUDA: 'Gouda (postcodes 2800–2809)',
|
||
...Object.fromEntries(GOUDA_PC4.map((p) => [p, `Gouda ${p}`]))
|
||
};
|
||
const AREA_REGION: Record<string, string> = { NL: 'NLD', GOUDA: 'GM0513' };
|
||
|
||
/** CSV split on ';' with optional "…" quoting ("" escapes). */
|
||
function splitLine(line: string): string[] {
|
||
if (!line.includes('"')) return line.split(';');
|
||
const out: string[] = [];
|
||
let cur = '';
|
||
let q = false;
|
||
for (let i = 0; i < line.length; i++) {
|
||
const c = line[i];
|
||
if (q && c === '"' && line[i + 1] === '"') {
|
||
cur += '"';
|
||
i++;
|
||
} else if (c === '"') {
|
||
q = !q;
|
||
} else if (!q && c === ';') {
|
||
out.push(cur);
|
||
cur = '';
|
||
} else {
|
||
cur += c;
|
||
}
|
||
}
|
||
out.push(cur);
|
||
return out;
|
||
}
|
||
|
||
async function* lines(stream: ReadableStream<Uint8Array>) {
|
||
const decoder = new TextDecoder();
|
||
let buf = '';
|
||
for await (const chunk of stream) {
|
||
buf += decoder.decode(chunk, { stream: true });
|
||
let nl: number;
|
||
let start = 0;
|
||
while ((nl = buf.indexOf('\n', start)) >= 0) {
|
||
yield buf.slice(start, nl).replace(/\r$/, '');
|
||
start = nl + 1;
|
||
}
|
||
buf = buf.slice(start);
|
||
}
|
||
if (buf) yield buf;
|
||
}
|
||
|
||
/** Streams the CSV out of the zip and counts homes per label and area (one label per home). */
|
||
export async function aggregate(zipPath: string) {
|
||
const proc = Bun.spawn(['unzip', '-p', zipPath], { stdout: 'pipe', stderr: 'ignore' });
|
||
const it = lines(proc.stdout as ReadableStream<Uint8Array>);
|
||
let published: string | null = null;
|
||
let ix: Record<string, number> | null = null;
|
||
// Newest registration per home: BAG verblijfsobject id, or the address when there is none.
|
||
// Value packs label index (4 bits) + Gouda PC4 slot (4 bits, 15 = elsewhere) + yyyymmdd.
|
||
const byVbo = new Map<number, number>();
|
||
const byAddr = new Map<string, number>();
|
||
let rows = 0;
|
||
let skipped = 0;
|
||
for await (const line of it) {
|
||
if (!ix) {
|
||
const m = line.match(/^PublicatieDatum;(\d\d)-(\d\d)-(\d{4})/);
|
||
if (m) published = `${m[3]}-${m[2]}-${m[1]}`;
|
||
if (line.startsWith('Registratiedatum;'))
|
||
ix = Object.fromEntries(line.split(';').map((h, i) => [h, i]));
|
||
continue;
|
||
}
|
||
if (!line) continue;
|
||
rows++;
|
||
const f = splitLine(line);
|
||
if (f[ix.Gebouwklasse] !== 'W') continue;
|
||
// Labels issued on a building-permit design are not a home's label yet.
|
||
if (f[ix.Status]?.startsWith('Vergunningsaanvraag')) {
|
||
skipped++;
|
||
continue;
|
||
}
|
||
const label = LABELS.indexOf(f[ix.Energieklasse] as (typeof LABELS)[number]);
|
||
if (label < 0) {
|
||
skipped++;
|
||
continue;
|
||
}
|
||
const pc = f[ix.Postcode] ?? '';
|
||
const slot = pc >= '2800' && pc < '2810' ? Number(pc[3]) : 15;
|
||
const date = Number(f[ix.Registratiedatum]) || 0;
|
||
const packed = date * 256 + slot * 16 + label;
|
||
const vbo = f[ix.BAGVerblijfsobjectID];
|
||
if (vbo && /^\d{16}$/.test(vbo)) {
|
||
const key = Number(vbo);
|
||
const prev = byVbo.get(key);
|
||
if (prev === undefined || prev < packed) byVbo.set(key, packed);
|
||
} else {
|
||
const key = [pc, f[ix.Huisnummer], f[ix.Huisletter], f[ix.Huisnummertoevoeging]].join('|');
|
||
const prev = byAddr.get(key);
|
||
if (prev === undefined || prev < packed) byAddr.set(key, packed);
|
||
}
|
||
}
|
||
const code = await proc.exited;
|
||
if (code !== 0) throw new Error(`unzip exited ${code}`);
|
||
if (!ix || !published) throw new Error('unexpected EP-Online file header');
|
||
const counts = new Map<string, number[]>();
|
||
const bump = (area: string, label: number) => {
|
||
let c = counts.get(area);
|
||
if (!c) counts.set(area, (c = LABELS.map(() => 0)));
|
||
c[label]++;
|
||
};
|
||
for (const v of [...byVbo.values(), ...byAddr.values()]) {
|
||
const label = v % 16;
|
||
const slot = Math.floor(v / 16) % 16;
|
||
bump('NL', label);
|
||
if (slot !== 15) {
|
||
bump('GOUDA', label);
|
||
bump(String(2800 + slot), label);
|
||
}
|
||
}
|
||
return { published, counts, rows, skipped, homes: byVbo.size + byAddr.size };
|
||
}
|
||
|
||
export async function eponline(opts: { force: boolean; job: string }) {
|
||
await run('eponline', opts.job, async (note) => {
|
||
const key = process.env.EP_ONLINE_API_KEY;
|
||
if (!key) throw new Error('EP_ONLINE_API_KEY is not set (see ~/.config/stats/ep-online.env)');
|
||
const info = (await (
|
||
await http(`${API}/Mutatiebestand/DownloadInfo?fileType=csv`, {
|
||
headers: { authorization: key, accept: 'application/json' }
|
||
})
|
||
).json()) as { bestandsnaam: string; downloadUrl: string };
|
||
const m = info.bestandsnaam?.match(/v(\d{4})(\d\d)(\d\d)/);
|
||
if (!m || !info.downloadUrl) throw new Error('DownloadInfo returned no file');
|
||
const snapshot = `${m[1]}-${m[2]}-${m[3]}`;
|
||
const [{ have }] = await sql`
|
||
select count(*)::int as have from stats.ep_label_agg where snapshot = ${snapshot}`;
|
||
if (have && !opts.force) {
|
||
log(`eponline: ${info.bestandsnaam} already loaded`);
|
||
await rebuildSeries();
|
||
return 0;
|
||
}
|
||
const dir = await mkdtemp(join(tmpdir(), 'eponline-'));
|
||
try {
|
||
const zip = join(dir, 'labels.zip');
|
||
// The download URL carries a token: never put it in logs or fetch_run errors.
|
||
const res = await fetch(info.downloadUrl, {
|
||
headers: { 'user-agent': 'stanrunge.dev-stats/1.0' },
|
||
signal: AbortSignal.timeout(30 * 60_000)
|
||
}).catch((e) => {
|
||
throw new Error(`download failed (${e instanceof Error ? e.name : 'error'})`);
|
||
});
|
||
if (!res.ok || !res.body) throw new Error(`download failed: HTTP ${res.status}`);
|
||
await Bun.write(zip, res);
|
||
log(
|
||
`eponline: downloaded ${info.bestandsnaam} (${(Bun.file(zip).size / 1e6).toFixed(0)} MB)`
|
||
);
|
||
const agg = await aggregate(zip);
|
||
if (agg.published !== snapshot)
|
||
note(`file says published ${agg.published}, name says ${snapshot}`);
|
||
const nl = agg.counts.get('NL');
|
||
if (!nl || nl.reduce((a, b) => a + b, 0) < 1_000_000)
|
||
throw new Error(`implausible total: ${agg.homes} homes`);
|
||
const out: Record<string, unknown>[] = [];
|
||
for (const [area, c] of agg.counts)
|
||
c.forEach((homes, i) => out.push({ snapshot, area, label: LABELS[i], homes }));
|
||
await sql.begin(async (tx) => {
|
||
await tx`delete from stats.ep_label_agg where snapshot = ${snapshot}`;
|
||
await insertChunks(tx, 'ep_label_agg', out);
|
||
});
|
||
log(
|
||
`eponline: ${agg.rows} rows, ${agg.homes} homes with a label (${agg.skipped} permit/blank skipped)`
|
||
);
|
||
} finally {
|
||
await rm(dir, { recursive: true, force: true });
|
||
}
|
||
return rebuildSeries();
|
||
});
|
||
}
|
||
|
||
/** stats.obs: share of homes per label (%), per area, one point per monthly snapshot. */
|
||
async function rebuildSeries(): Promise<number> {
|
||
const rows = await sql<{ snapshot: string; area: string; label: string; homes: number }[]>`
|
||
select snapshot::text, area, label, homes from stats.ep_label_agg`;
|
||
const totals = new Map<string, number>();
|
||
for (const r of rows)
|
||
totals.set(`${r.snapshot}|${r.area}`, (totals.get(`${r.snapshot}|${r.area}`) ?? 0) + r.homes);
|
||
const series: SeriesRow[] = [];
|
||
const obs: ObsRow[] = [];
|
||
const areas = [...new Set(rows.map((r) => r.area))];
|
||
for (const area of areas)
|
||
for (const label of [...LABELS, 'TOTAL']) {
|
||
const id = `eponline:${label === 'TOTAL' ? 'homes' : `share:${label}`}:${area}`;
|
||
series.push({
|
||
series_id: id,
|
||
source_id: 'eponline',
|
||
source_code: `${area}:${label}`,
|
||
title:
|
||
label === 'TOTAL'
|
||
? `Homes with a registered energy label: ${AREA_NAME[area] ?? area}`
|
||
: `Share of labelled homes with label ${label}: ${AREA_NAME[area] ?? area}`,
|
||
unit: label === 'TOTAL' ? 'homes' : '%',
|
||
freq: 'M',
|
||
region_code: AREA_REGION[area] ?? null,
|
||
theme: 'energy',
|
||
meta: { area, label }
|
||
});
|
||
for (const r of rows)
|
||
if (r.area === area && (label === 'TOTAL' ? r.label === LABELS[0] : r.label === label)) {
|
||
const total = totals.get(`${r.snapshot}|${area}`) ?? 0;
|
||
obs.push({
|
||
series_id: id,
|
||
period: `${r.snapshot.slice(0, 7)}-01`,
|
||
value: label === 'TOTAL' ? total : total ? (100 * r.homes) / total : null,
|
||
status: null
|
||
});
|
||
}
|
||
}
|
||
return writeSeries(series, obs);
|
||
}
|