energietarieven.py

#!/usr/bin/python3.11
from datetime import datetime, timedelta
import sqlite3
import requests
import json
from time import sleep
from logger_setup import get_logger
from shared_store import set_key
from config import (
    ENERGY_DB, TARIEVEN_JSON, READER_DATA_JSON, WEERLIVE_TOKEN, WEER_LOCATIE,
    COP_A, COP_B, COP_C,
    DEGREE_DAY_GAS_FACTOR, DEGREE_DAY_YEAR_FACTOR, DEGREE_DAY_DIVISOR,
    WP_CAPACITY_KW,
    INKOOPOPSLAG, VASTE_LEVERINGSKOSTEN, ENERGIEBELASTING_FACTOR, VERKOOPOPSLAG,
)

logger = get_logger("tarieven")


#  1. Ophalen van verwachte min en max temperatuur
def getTemp():
   if not WEERLIVE_TOKEN:
      logger.error("Weerlive_token ontbreekt in omgeving (.env)")
      return None

   API_URL = f"http://weerlive.nl/api/json-data-10min.php?key={WEERLIVE_TOKEN}&locatie={WEER_LOCATIE}"
   for attempt in range(4):
      try:
         r = requests.get(API_URL, timeout=5)
         r.raise_for_status()
         data = r.json()
         if "liveweer" in data and data["liveweer"]:
            tmax = float(data["liveweer"][0]["d1tmax"])
            tmin = float(data["liveweer"][0]["d1tmin"])
            # Tmin (minimumtemperatuur-voorspelling) is een apart gegeven
            # t.o.v. WP_reader's live WPoutsideT - hier apart wegschrijven
            # zodat WPmanager()'s nachttak deze kan gebruiken zonder de
            # live buitentemperatuur te overschrijven.
            set_key("KNMI", {"Tmin": tmin}, path=READER_DATA_JSON)
            return ((tmax + tmin)/2)
      except Exception as e:
         logger.info(f"KNMIdata fout poging {attempt+1}: {e}")
         sleep(5)
   return None

#  2. Berekenen van draaiuren WP op basis van graaddagen
def WPbedrijf():
   tavg = getTemp()
   if tavg is None:
      logger.error("WPbedrijf: geen temperatuurdata ontvangen, WP_uren op 0")
      return 0

   COP = 1.5 + (tavg+20) * COP_A + (((tavg+20) * COP_B) ** 2) / COP_C

   graaddagen = max(0, 15 - tavg)
   # 1.3 Nm3 gas per graaddag: 1600 graaddagen per jaar -> 2100 Nm3/jaar
   warmte_cal = round(int(graaddagen * DEGREE_DAY_GAS_FACTOR * DEGREE_DAY_YEAR_FACTOR / DEGREE_DAY_DIVISOR), 2)
   WP_uren = warmte_cal / COP / WP_CAPACITY_KW
   if warmte_cal == 0:
      WP_uren = 0
   else:
      WP_uren = min(24, int(WP_uren) + 1)
   WP_uren = max(0, WP_uren)   # guard tegen negatieve waarde bij extreme/afwijkende COP
   logger.info(f"WP_uren: {WP_uren}")
   return WP_uren

#  3. Get prices from database
def prices():
   try:
      WP_uren = WPbedrijf()

      with sqlite3.connect(ENERGY_DB) as db:
         db.row_factory = sqlite3.Row
         cur = db.cursor()
         datum = (datetime.now()-timedelta(days=0)).strftime("%Y-%m-%d")
         cur.execute("SELECT * FROM kaleprijs_1 WHERE datum = ?",(datum,))
         kaleprijs = cur.fetchone()
         if kaleprijs is None:
            logger.error("ophalen kaleprijs mislukt")
            return

      kaleprijs = dict(kaleprijs)
      datum = kaleprijs.pop("datum")

      datum_obj = datetime.strptime(datum, "%Y-%m-%d").date()
      is_today = (datum_obj == datetime.now().date())
      if not is_today:
         logger.error(f"geen prijzen voor vandaag geladen {datum}")

      #  Levering:      kaleprijs + inkoopopslag + belastingen (standaard dynamisch model).
      #  Teruglevering: kaleprijs - verkoopopslag.
      prijs       = {key: round((kaleprijs[key] + INKOOPOPSLAG + VASTE_LEVERINGSKOSTEN) * ENERGIEBELASTING_FACTOR, 3) for key in kaleprijs.keys()}
      prijs_terug = {key: round( kaleprijs[key] - VERKOOPOPSLAG, 3) for key in kaleprijs.keys()}

#  4. Rank dict based on prices
      i = 1
      prijs_rank = {}
      prijs_sorted = dict(sorted(prijs.items(), key=lambda x: float(x[1]))) # sorteer op oplopende prijzen
      for k in prijs_sorted:
         prijs_rank[k] = {"prijs": prijs_sorted[k], "prijs_terug": prijs_terug[k], "rank":i}
         i = i + 1

#  5. Determine when WP can run
      sorted_WPuren = sorted(
         list(prijs_sorted.items())[:WP_uren], key=lambda x: x[0]     # WP uren op basis van laagste prijzen
      )

#  6. Rank dict back on hours
      prijs_sorted = dict(sorted(prijs_rank.items(), key=lambda x: float(x[0]))) # sorteer op oplopende uren (incl. ranking)

#  7. Determine hours for power consumption
      numeric_values = [float(prijs[k]) for k in prijs]
      max_value = max(numeric_values)
      min_value = min(numeric_values)
      avg_value = sum(numeric_values) / len(numeric_values)

      i = block = 0
      for k in reversed(prijs_sorted):
         if prijs_sorted[k]["prijs"] > avg_value and prijs_sorted[k]["rank"] > WP_uren:
            prijs_rank[k]["block"] = 1
            prijs_rank[k]["fc"] = i
            i += 1
            block += 1
         else:
            prijs_rank[k]["block"] = 0
            prijs_rank[k]["fc"] = i
            i = max(0, i - 1)

#  8. Determine negative pricing hours to block solar production
      for k in prijs_terug:
         if prijs_terug[k] < 0:
            logger.info(f"neg. prijs {k} {prijs_terug[k]}")
            prijs_sorted[k]["block_solar"] = 1
         else:
            prijs_sorted[k]["block_solar"] = 0

      logger.info(f"check voldoende WP_uren: beschikbaar: {24 - block} nodig: {WP_uren} {24 - block >= WP_uren}")
      logger.info(f"{datum} avg: {avg_value:.3f} max: {max_value:.3f} min: {min_value:.3f}")

#  9. Schrijven van dict met prijzen, ranking en blokken naar json file
      tmp_file = TARIEVEN_JSON.with_suffix(".json.tmp")
      TARIEVEN_JSON.parent.mkdir(parents=True, exist_ok=True)

      with open(tmp_file, "w") as f:
         json.dump(prijs_sorted, f, indent=2)

      tmp_file.rename(TARIEVEN_JSON)
   except Exception as error:
      logger.error(f"Fout met ophalen energietarieven {error}")


if __name__ == "__main__":
   prices()