From ca473c62bb5f1236a535430d6186c0c8fea123b6 Mon Sep 17 00:00:00 2001 From: Per Magnus Petersen Date: Sat, 14 Feb 2026 16:25:59 +0100 Subject: [PATCH] Add Eloverblik data pipeline: fetch metering points, readings, and charges to PostgreSQL - Add eloverblik controller with API client for metering points, time series, and charges - Add migrations for metering_points, meter_readings, and charges tables - Add HugSQL queries (upsert-metering-point!, insert-meter-reading!, insert-charge!) - Add POST /eloverblik/sync route to trigger sync from browser Co-Authored-By: Claude Opus 4.6 --- ...0214140000-create-metering-points.down.sql | 1 + ...260214140000-create-metering-points.up.sql | 15 ++ ...60214150000-create-meter-readings.down.sql | 1 + ...0260214150000-create-meter-readings.up.sql | 9 + .../20260214160000-create-charges.down.sql | 1 + .../20260214160000-create-charges.up.sql | 16 ++ resources/queries.sql | 36 +++ .../elprice/web/controllers/eloverblik.clj | 255 ++++++++++++++++++ src/clj/pmagnus/elprice/web/routes/ui.clj | 13 +- 9 files changed, 346 insertions(+), 1 deletion(-) create mode 100644 resources/migrations/20260214140000-create-metering-points.down.sql create mode 100644 resources/migrations/20260214140000-create-metering-points.up.sql create mode 100644 resources/migrations/20260214150000-create-meter-readings.down.sql create mode 100644 resources/migrations/20260214150000-create-meter-readings.up.sql create mode 100644 resources/migrations/20260214160000-create-charges.down.sql create mode 100644 resources/migrations/20260214160000-create-charges.up.sql create mode 100644 src/clj/pmagnus/elprice/web/controllers/eloverblik.clj diff --git a/resources/migrations/20260214140000-create-metering-points.down.sql b/resources/migrations/20260214140000-create-metering-points.down.sql new file mode 100644 index 0000000..05e84e9 --- /dev/null +++ b/resources/migrations/20260214140000-create-metering-points.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS metering_points; diff --git a/resources/migrations/20260214140000-create-metering-points.up.sql b/resources/migrations/20260214140000-create-metering-points.up.sql new file mode 100644 index 0000000..90f998f --- /dev/null +++ b/resources/migrations/20260214140000-create-metering-points.up.sql @@ -0,0 +1,15 @@ +CREATE TABLE metering_points ( + metering_point_id VARCHAR(18) PRIMARY KEY, + type_of_mp VARCHAR(50), + balance_supplier VARCHAR(100), + street_name VARCHAR(200), + building_number VARCHAR(20), + postcode VARCHAR(10), + city_name VARCHAR(100), + has_relation BOOLEAN, + settlement_method VARCHAR(50), + reading_occurrence VARCHAR(50), + consumer_start_date DATE, + first_consumer_name VARCHAR(200), + fetched_at TIMESTAMPTZ NOT NULL DEFAULT now() +); diff --git a/resources/migrations/20260214150000-create-meter-readings.down.sql b/resources/migrations/20260214150000-create-meter-readings.down.sql new file mode 100644 index 0000000..d7c3786 --- /dev/null +++ b/resources/migrations/20260214150000-create-meter-readings.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS meter_readings; diff --git a/resources/migrations/20260214150000-create-meter-readings.up.sql b/resources/migrations/20260214150000-create-meter-readings.up.sql new file mode 100644 index 0000000..68834c2 --- /dev/null +++ b/resources/migrations/20260214150000-create-meter-readings.up.sql @@ -0,0 +1,9 @@ +CREATE TABLE meter_readings ( + metering_point_id VARCHAR(18) NOT NULL REFERENCES metering_points(metering_point_id), + time_dk TIMESTAMP NOT NULL, + quantity_kwh NUMERIC(10,3), + quality VARCHAR(50), + PRIMARY KEY (metering_point_id, time_dk) +); +--;; +CREATE INDEX idx_meter_readings_time_dk ON meter_readings(time_dk); diff --git a/resources/migrations/20260214160000-create-charges.down.sql b/resources/migrations/20260214160000-create-charges.down.sql new file mode 100644 index 0000000..092e820 --- /dev/null +++ b/resources/migrations/20260214160000-create-charges.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS charges; diff --git a/resources/migrations/20260214160000-create-charges.up.sql b/resources/migrations/20260214160000-create-charges.up.sql new file mode 100644 index 0000000..a90c0aa --- /dev/null +++ b/resources/migrations/20260214160000-create-charges.up.sql @@ -0,0 +1,16 @@ +CREATE TABLE charges ( + id SERIAL, + metering_point_id VARCHAR(18) NOT NULL REFERENCES metering_points(metering_point_id), + charge_type VARCHAR(20) NOT NULL, + name VARCHAR(200) NOT NULL, + description VARCHAR(500), + owner VARCHAR(100) NOT NULL, + valid_from_date DATE NOT NULL, + valid_to_date DATE, + period_type VARCHAR(50), + price NUMERIC(12,6), + quantity INTEGER, + position INTEGER NOT NULL DEFAULT 0, + fetched_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (metering_point_id, charge_type, name, owner, valid_from_date, position) +); diff --git a/resources/queries.sql b/resources/queries.sql index e2ccf51..7cc30a4 100644 --- a/resources/queries.sql +++ b/resources/queries.sql @@ -11,3 +11,39 @@ FROM day_ahead_prices WHERE time_dk::date = :date AND price_area = :price-area ORDER BY time_dk; + +-- :name upsert-metering-point! :! :n +-- :doc Upsert a metering point, refreshing data on re-sync +INSERT INTO metering_points (metering_point_id, type_of_mp, balance_supplier, street_name, + building_number, postcode, city_name, has_relation, settlement_method, + reading_occurrence, consumer_start_date, first_consumer_name) +VALUES (:metering-point-id, :type-of-mp, :balance-supplier, :street-name, + :building-number, :postcode, :city-name, :has-relation, :settlement-method, + :reading-occurrence, :consumer-start-date, :first-consumer-name) +ON CONFLICT (metering_point_id) DO UPDATE SET + type_of_mp = EXCLUDED.type_of_mp, + balance_supplier = EXCLUDED.balance_supplier, + street_name = EXCLUDED.street_name, + building_number = EXCLUDED.building_number, + postcode = EXCLUDED.postcode, + city_name = EXCLUDED.city_name, + has_relation = EXCLUDED.has_relation, + settlement_method = EXCLUDED.settlement_method, + reading_occurrence = EXCLUDED.reading_occurrence, + consumer_start_date = EXCLUDED.consumer_start_date, + first_consumer_name = EXCLUDED.first_consumer_name, + fetched_at = now(); + +-- :name insert-meter-reading! :! :n +-- :doc Insert a meter reading, skip if already exists +INSERT INTO meter_readings (metering_point_id, time_dk, quantity_kwh, quality) +VALUES (:metering-point-id, :time-dk, :quantity-kwh, :quality) +ON CONFLICT (metering_point_id, time_dk) DO NOTHING; + +-- :name insert-charge! :! :n +-- :doc Insert a charge record, skip if already exists +INSERT INTO charges (metering_point_id, charge_type, name, description, owner, + valid_from_date, valid_to_date, period_type, price, quantity, position) +VALUES (:metering-point-id, :charge-type, :name, :description, :owner, + :valid-from-date, :valid-to-date, :period-type, :price, :quantity, :position) +ON CONFLICT (metering_point_id, charge_type, name, owner, valid_from_date, position) DO NOTHING; diff --git a/src/clj/pmagnus/elprice/web/controllers/eloverblik.clj b/src/clj/pmagnus/elprice/web/controllers/eloverblik.clj new file mode 100644 index 0000000..7315743 --- /dev/null +++ b/src/clj/pmagnus/elprice/web/controllers/eloverblik.clj @@ -0,0 +1,255 @@ +(ns pmagnus.elprice.web.controllers.eloverblik + (:require + [clojure.data.json :as json] + [clojure.string :as str] + [clojure.tools.logging :as log]) + (:import + [java.net URI] + [java.net.http HttpClient HttpRequest HttpRequest$BodyPublishers HttpResponse$BodyHandlers] + [java.time LocalDate LocalDateTime ZoneId] + [java.time.format DateTimeFormatter] + [java.sql Timestamp Date])) + +(def ^:private base-url "https://api.eloverblik.dk/customerapi") +(def ^:private dk-zone (ZoneId/of "Europe/Copenhagen")) +(def ^:private date-fmt (DateTimeFormatter/ofPattern "yyyy-MM-dd")) + +;; --- Auth --- + +(defn- read-refresh-token [] + (str/trim (slurp ".token"))) + +(defn- get-access-token + "Exchange refresh token for a short-lived access token." + [refresh-token] + (let [client (HttpClient/newHttpClient) + request (-> (HttpRequest/newBuilder) + (.uri (URI/create (str base-url "/api/token"))) + (.header "Authorization" (str "Bearer " refresh-token)) + (.header "Accept" "application/json") + (.GET) + (.build)) + response (.send client request (HttpResponse$BodyHandlers/ofString))] + (when (= 200 (.statusCode response)) + (let [body (json/read-str (.body response) :key-fn keyword)] + (:result body))))) + +;; --- HTTP helpers --- + +(defn- api-get + "GET with Bearer access token, returns parsed JSON." + [access-token path] + (let [client (HttpClient/newHttpClient) + request (-> (HttpRequest/newBuilder) + (.uri (URI/create (str base-url path))) + (.header "Authorization" (str "Bearer " access-token)) + (.header "Accept" "application/json") + (.GET) + (.build)) + response (.send client request (HttpResponse$BodyHandlers/ofString))] + (log/info "GET" path "->" (.statusCode response)) + (when (= 200 (.statusCode response)) + (json/read-str (.body response) :key-fn keyword)))) + +(defn- api-post + "POST with Bearer access token + JSON body, returns parsed JSON." + [access-token path body] + (let [client (HttpClient/newHttpClient) + json-str (json/write-str body) + request (-> (HttpRequest/newBuilder) + (.uri (URI/create (str base-url path))) + (.header "Authorization" (str "Bearer " access-token)) + (.header "Content-Type" "application/json") + (.header "Accept" "application/json") + (.POST (HttpRequest$BodyPublishers/ofString json-str)) + (.build)) + response (.send client request (HttpResponse$BodyHandlers/ofString))] + (log/info "POST" path "->" (.statusCode response)) + (when (= 200 (.statusCode response)) + (json/read-str (.body response) :key-fn keyword)))) + +;; --- Metering Points --- + +(defn- fetch-metering-points [access-token] + (let [data (api-get access-token "/api/meteringpoints/meteringpoints?includeAll=true")] + (:result data))) + +(defn- parse-date [s] + (when (and s (not (str/blank? s))) + (Date/valueOf (LocalDate/parse (subs s 0 10))))) + +(defn- upsert-mp! [query-fn mp] + (query-fn :upsert-metering-point! + {:metering-point-id (:meteringPointId mp) + :type-of-mp (:typeOfMP mp) + :balance-supplier (:balanceSupplierName mp) + :street-name (:streetName mp) + :building-number (:buildingNumber mp) + :postcode (:postcode mp) + :city-name (:cityName mp) + :has-relation (:hasRelation mp) + :settlement-method (:settlementMethod mp) + :reading-occurrence (:meterReadingOccurrence mp) + :consumer-start-date (parse-date (:consumerStartDate mp)) + :first-consumer-name (:firstConsumerPartyName mp)})) + +(defn- save-metering-points! [query-fn metering-points] + (let [all-mps (mapcat (fn [mp] + (cons mp (:childMeteringPoints mp))) + metering-points)] + (doseq [mp all-mps] + (upsert-mp! query-fn mp)) + (log/info "Saved" (count all-mps) "metering points"))) + +;; --- Time Series --- + +(defn- fetch-time-series [access-token metering-point-ids from to] + (api-post access-token + (str "/api/meterdata/gettimeseries/" + (.format from date-fmt) "/" + (.format to date-fmt) "/Hour") + {:meteringPoints {:meteringPoint metering-point-ids}})) + +(defn- parse-time-series + "Parse the nested time series response into flat reading maps." + [response metering-point-ids] + (let [results (:result response)] + (mapcat + (fn [result mp-id] + (let [doc (:MyEnergyData_MarketDocument result) + series (or (:TimeSeries doc) [])] + (mapcat + (fn [ts] + (mapcat + (fn [period] + (let [start-str (get-in period [:timeInterval :start]) + start (LocalDateTime/parse + start-str + (DateTimeFormatter/ofPattern "yyyy-MM-dd'T'HH:mm:ss'Z'")) + start-dk (-> start + (.atZone (ZoneId/of "UTC")) + (.withZoneSameInstant dk-zone) + (.toLocalDateTime))] + (map (fn [point] + (let [pos (dec (Long/parseLong (str (:position point)))) + time-dk (.plusHours start-dk pos) + raw-qty (get point (keyword "out_Quantity.quantity")) + quantity (when raw-qty (Double/parseDouble (str raw-qty)))] + {:metering-point-id mp-id + :time-dk (Timestamp/valueOf time-dk) + :quantity-kwh quantity + :quality (get point (keyword "out_Quantity.quality"))})) + (:Point period)))) + (:Period ts))) + series))) + results + metering-point-ids))) + +(defn- save-time-series! [query-fn readings] + (doseq [r readings] + (query-fn :insert-meter-reading! r)) + (log/info "Saved" (count readings) "meter readings")) + +;; --- Charges --- + +(defn- fetch-charges [access-token metering-point-ids] + (api-post access-token + "/api/meteringpoints/meteringpoint/getcharges" + {:meteringPoints {:meteringPoint metering-point-ids}})) + +(defn- save-charges! [query-fn charges-response] + (let [results (:result charges-response)] + (doseq [result results] + (let [mp-id (:id result) + inner (:result result)] + ;; Subscriptions + (doseq [sub (:subscriptions inner)] + (query-fn :insert-charge! + {:metering-point-id mp-id + :charge-type "subscription" + :name (:name sub) + :description (:description sub) + :owner (:owner sub) + :valid-from-date (parse-date (:validFromDate sub)) + :valid-to-date (parse-date (:validToDate sub)) + :period-type (:periodType sub) + :price (:price sub) + :quantity (:quantity sub) + :position 0})) + ;; Tariffs (one row per position/price) + (doseq [tar (:tariffs inner)] + (let [prices (or (:prices tar) [])] + (if (seq prices) + (doseq [p prices] + (query-fn :insert-charge! + {:metering-point-id mp-id + :charge-type "tariff" + :name (:name tar) + :description (:description tar) + :owner (:owner tar) + :valid-from-date (parse-date (:validFromDate tar)) + :valid-to-date (parse-date (:validToDate tar)) + :period-type (:periodType tar) + :price (:price p) + :quantity (:quantity p) + :position (some-> (:position p) str Long/parseLong)})) + (query-fn :insert-charge! + {:metering-point-id mp-id + :charge-type "tariff" + :name (:name tar) + :description (:description tar) + :owner (:owner tar) + :valid-from-date (parse-date (:validFromDate tar)) + :valid-to-date (parse-date (:validToDate tar)) + :period-type (:periodType tar) + :price nil + :quantity nil + :position 0})))) + ;; Fees + (doseq [fee (:fees inner)] + (query-fn :insert-charge! + {:metering-point-id mp-id + :charge-type "fee" + :name (:name fee) + :description (:description fee) + :owner (:owner fee) + :valid-from-date (parse-date (:validFromDate fee)) + :valid-to-date (parse-date (:validToDate fee)) + :period-type (:periodType fee) + :price (:price fee) + :quantity (:quantity fee) + :position 0})))))) + +;; --- Orchestrator --- + +(defn fetch-and-save-all! + "Fetch all Eloverblik data and save to database." + [query-fn] + (log/info "Starting Eloverblik fetch") + (let [refresh-token (read-refresh-token) + access-token (get-access-token refresh-token)] + (if-not access-token + (do (log/error "Failed to get Eloverblik access token") + {:error "Failed to get access token"}) + (let [mps (fetch-metering-points access-token) + all-ids (->> mps + (mapcat (fn [mp] + (cons (:meteringPointId mp) + (map :meteringPointId + (:childMeteringPoints mp))))) + (distinct) + (vec)) + from (LocalDate/of 2026 2 1) + to (LocalDate/of 2026 2 2)] + (log/info "Found" (count all-ids) "metering points:" all-ids) + (save-metering-points! query-fn mps) + ;; Time series + (let [ts-resp (fetch-time-series access-token all-ids from to) + readings (parse-time-series ts-resp all-ids)] + (save-time-series! query-fn readings)) + ;; Charges + (let [ch-resp (fetch-charges access-token all-ids)] + (save-charges! query-fn ch-resp)) + {:metering-points (count all-ids) + :from (.format from date-fmt) + :to (.format to date-fmt)})))) diff --git a/src/clj/pmagnus/elprice/web/routes/ui.clj b/src/clj/pmagnus/elprice/web/routes/ui.clj index 52cd1ab..6ed4a4a 100644 --- a/src/clj/pmagnus/elprice/web/routes/ui.clj +++ b/src/clj/pmagnus/elprice/web/routes/ui.clj @@ -6,6 +6,7 @@ [pmagnus.elprice.web.htmx :refer [page fragment]] [pmagnus.elprice.web.controllers.prices :as prices] [pmagnus.elprice.web.controllers.tariffs :as tariffs] + [pmagnus.elprice.web.controllers.eloverblik :as eloverblik] [integrant.core :as ig] [reitit.ring.middleware.muuntaja :as muuntaja] [reitit.ring.middleware.parameters :as parameters]) @@ -174,12 +175,22 @@ (fragment (price-content :today (prices/get-today-prices query-fn) (has-tomorrow? query-fn)))) +(defn sync-eloverblik [query-fn request] + (let [result (eloverblik/fetch-and-save-all! query-fn)] + (fragment + (if (:error result) + [:p {:class "text-red-600 font-medium"} (:error result)] + [:p {:class "text-green-700 font-medium"} + (str "Synced " (:metering-points result) + " metering points (" (:from result) " to " (:to result) ")")])))) + ;; Routes (defn ui-routes [{:keys [query-fn]}] [["/" {:get (partial home query-fn)}] ["/prices/today" {:get (partial prices-today query-fn)}] ["/prices/tomorrow" {:get (partial prices-tomorrow query-fn)}] - ["/prices/fetch" {:post (partial fetch-prices query-fn)}]]) + ["/prices/fetch" {:post (partial fetch-prices query-fn)}] + ["/eloverblik/sync" {:post (partial sync-eloverblik query-fn)}]]) (def route-data {:muuntaja formats/instance