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 <noreply@anthropic.com>

resources/migrations/20260214140000-create-metering-points.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS metering_points;

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()
+);

resources/migrations/20260214150000-create-meter-readings.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS meter_readings;

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);

resources/migrations/20260214160000-create-charges.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS charges;

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)
+);

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;

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)}))))

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