PricePower/main.py

#!/usr/bin/env python3

import datetime
from genericpath import isfile
import json
import os
import requests
import calendar
from flask import Flask
from flask import jsonify

url_energy = "https://www.ote-cr.cz/cs/kratkodobe-trhy/elektrina/denni-trh/@@chart-data?report_date={}"
url_currency = "https://data.kurzy.cz/json/meny/b[1].json"
url_currency2 = "https://www.cnb.cz/cs/financni-trhy/devizovy-trh/kurzy-devizoveho-trhu/kurzy-devizoveho-trhu/denni_kurz.txt?date={day}.{month}.{year}"
CACHE_PATH = "./cache.json"
ENERGY_CACHE_PATH = "./cache-energy.json"
NO_CACHE = False
SELL_FEE = 0.45
BUY_FEE = 0.45
VAT=1.21

class PriceException(Exception):
    pass

def get_distribution_price():
    cez_57d_vt = 0.64862
    cez_57d_nt = 0.43809

    tax = 0.0283
    services = 0.21282
    ote = 0 # this is 4 crown per month
    oze = 0.495

    additional = tax+services+ote+oze

    return {
        "high": cez_57d_vt+additional,
        "low": cez_57d_nt+additional,
    }

def get_breaker_payment():
    # https://www.tzb-info.cz/ceny-paliv-a-energii/14-ceny-elektriny#D57d
    cez_32a = 500

    currentDate = datetime.date.today()
    daysInMonth = calendar.monthrange(currentDate.year, currentDate.month)[1]
    return cez_32a/daysInMonth/24

def get_current_energy_price(date_str, selected_hour):
    selected_hour = str(selected_hour)
    data = {
        "date": "",
        "hours": {},
    }

    try:
        if os.path.isfile(ENERGY_CACHE_PATH) and not NO_CACHE:
            with open(ENERGY_CACHE_PATH, "r") as f:
                data = json.load(f)
    except json.decoder.JSONDecodeError:
        cache = {}

    if data["date"] == date_str:
        return data["hours"][selected_hour]

    r = requests.get(url_energy.format(date_str))

    print(r.text)
    for hour in r.json()["data"]["dataLine"][1]["point"]:
        data["hours"][str(int(hour["x"])-1)] = hour["y"]

    data["date"] = date_str
    with open(ENERGY_CACHE_PATH, "w") as f:
        json.dump(data, f)

    return data["hours"][selected_hour]

#def get_currency_ratio(currency):
#    r = requests.get(url_currency)
#    return r.json()["kurzy"][currency]["dev_stred"]

def get_currency_ratio(currency):
    now = datetime.date.today()
    url = url_currency2.format(day=now.day,month=now.month,year=now.year)
    r = requests.get(url)
    for row in [x.split("|") for x in r.text.split("\n") if x and "|" in x]:
        if row[3] == "EUR":
            return float(row[4].replace(",", "."))

    raise PriceException("EUR not found")

def hour_result():
    current_hour = datetime.datetime.now().hour
    today = datetime.date.today().strftime("%Y-%m-%d")

    cache = {}

    try:
        if os.path.isfile(CACHE_PATH) and not NO_CACHE:
            with open(CACHE_PATH, "r") as f:
                cache = json.load(f)
    except json.decoder.JSONDecodeError:
        cache = {}
    
    if cache.get("date") == today and cache.get("hour") == current_hour:
        cache["cached"] = True
        return cache

    energy_price = get_current_energy_price(today, current_hour)
    eur2czk = get_currency_ratio("EUR")
    breaker = get_breaker_payment()
    distribution = get_distribution_price()

    data = {
        "date": datetime.date.today().strftime("%Y-%m-%d"),
        "hour": datetime.datetime.now().hour,
        "breaker": "32A",
        "tariff": "57d",
        "price": {
            "distribution": distribution,
            "market": energy_price/1000*eur2czk,
            "market_vat": energy_price/1000*eur2czk*VAT,
            "marker_sell": energy_price/1000*eur2czk-SELL_FEE, # Typo
            "market_sell": energy_price/1000*eur2czk-SELL_FEE,
            "breaker": breaker,
            "breaker_vat": breaker*VAT,
            "sell_fee": SELL_FEE,
            "buy_fee": BUY_FEE,
            "total": {
                "high": energy_price/1000*eur2czk+breaker+distribution["high"]+BUY_FEE,
                "low": energy_price/1000*eur2czk+breaker+distribution["low"]+BUY_FEE,
            },
            "total_no_breaker": {
                "high": energy_price/1000*eur2czk+distribution["high"]+BUY_FEE,
                "low": energy_price/1000*eur2czk+distribution["low"]+BUY_FEE,
            },
            "total_vat": {
                "high": (energy_price/1000*eur2czk+breaker+distribution["high"]+BUY_FEE)*VAT,
                "low": (energy_price/1000*eur2czk+breaker+distribution["low"]+BUY_FEE)*VAT,
            },
            "total_no_breaker_vat": {
                "high": (energy_price/1000*eur2czk+distribution["high"]+BUY_FEE)*VAT,
                "low": (energy_price/1000*eur2czk+distribution["low"]+BUY_FEE)*VAT,
            },
        },
        "cached": False,
    }

    with open(CACHE_PATH, "w") as f:
        json.dump(data, f)

    return data

app = Flask(__name__)

@app.route("/")
def result():
    return jsonify(hour_result())

if __name__ == "__main__":
    app.run(debug=True)