Initial commit

This commit is contained in:
Adam Štrauch 2024-08-08 01:31:21 +02:00
commit 0f1cac531e
Signed by: cx
GPG Key ID: 7262DAFE292BCE20
11 changed files with 446 additions and 0 deletions

7
.gitignore vendored Normal file
View File

@ -0,0 +1,7 @@
cache/
.history/
venv/
.rosti.state
cache*.json
Rostifile
__pycache__/

4
.vscode/settings.json vendored Normal file
View File

@ -0,0 +1,4 @@
{
"makefile.extensionOutputFolder": "./.vscode",
"python.analysis.autoImportCompletions": true
}

12
Taskfile.yml Normal file
View File

@ -0,0 +1,12 @@
# https://taskfile.dev
version: '3'
tasks:
deploy:
cmds:
- ssh -p 11335 app@ssh.rosti.cz mkdir -p /srv/app/cache
- rsync -av -e "ssh -p 11335" --delete ./calculator app@ssh.rosti.cz:/srv/app/
- rsync -av -e "ssh -p 11335" ./requirements.txt app@ssh.rosti.cz:/srv/app/
- ssh -p 11335 app@ssh.rosti.cz /srv/venv/bin/pip install -r /srv/app/requirements.txt
- ssh -p 11335 app@ssh.rosti.cz supervisorctl restart app

0
calculator/__init__.py Normal file
View File

1
calculator/consts.py Normal file
View File

@ -0,0 +1 @@
VAT = 1.21

127
calculator/main.py Normal file
View File

@ -0,0 +1,127 @@
import datetime
import calendar
from typing import List
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
from calculator.miner import get_energy_prices, get_eur_czk_ratio
from calculator.schema import CheapestHours, DayPrice, Price
from .consts import VAT
def days_in_month(year: int, month: int) -> int:
return calendar.monthrange(year, month)[1]
app = FastAPI(title="Spot market home calculator", version="0.1", description="Calculate your energy costs based on spot market prices")
@app.get("/", description="Redirect to /docs")
def docs():
return RedirectResponse(url="/docs")
docs = """
Return spot prices for the whole day with all fees included.<br>
<br>
Options:<br>
date - date in format YYYY-MM-DD, default is today<br>
hour - hour of the day, default is current hour, works only when date is today<br>
monthly_fees - monthly fees, default is 509.24 (D57d, BezDodavatele)<br>
daily_fees - daily fees, default is 4.18 (BezDodavatele)<br>
kwh_fees_low - additional fees per kWh in low tariff, usually distribution + other fees, default is 1.62421 (D57d, BezDodavatele)<br>
kwh_fees_high - additional fees per kWh in high tariff, usually distribution + other fees, default is 1.83474 (D57d, BezDodavatele)<br>
sell_fees - selling energy fees, default is 0.45 (BezDodavatele)<br>
num_cheapest_hours - number of cheapest hours to return, default is 8, use this to plan your consumption<br>
low_tariff_hours - list of low tariff hours, default is 0,1,2,3,4,5,6,7,9,10,11,13,14,16,17,18,20,21,22,23 (D57d, ČEZ)<br>
<br>
Output:<br>
spot - current spot prices on our market<br>
total - total price for the day including all fees and VAT<br>
sell - current spot prices minus sell_fees<br>
<br>
The final price on the invoice is calculated as:<br>
monthly_fees + kWh consumption in low tariff * kwh_fees_low + kWh consumption in high tariff * kwh_fees_high<br>
<br>
Except spot and sell prices all prices include VAT.<br>
<br>
<br>
Integration into Home Assistant can be done in configuration.yaml file:<br>
<br>
```
sensor:
- platform: rest
resource: "https://pricepower2.rostiapp.cz/price/day"
name: "energy_price"
value_template: "{{ value_json.total.now|round(2) }}"
unit_of_measurement: "CZK/kWh"
- platform: rest
resource: "https://pricepower2.rostiapp.cz/price/day"
name: "energy_market_price"
value_template: "{{ value_json.spot.now|round(2) }}"
unit_of_measurement: "CZK/kWh"
- platform: rest
resource: "https://pricepower2.rostiapp.cz/price/day"
name: "energy_market_price_sell"
value_template: "{{ value_json.sell.now|round(2) }}"
unit_of_measurement: "CZK/kWh"
- platform: rest
resource: "https://pricepower2.rostiapp.cz/price/day"
name: "energy_cheap_hour"
value_template: "{{ value_json.cheapest_hours.is_cheapest }}"
```
"""
@app.get("/price/day", description=docs)
@app.get("/price/day/{date}", description=docs)
def read_item(
date: datetime.date=datetime.date.today(),
hour:int=datetime.datetime.now().hour,
monthly_fees: float = 509.24,
daily_fees: float=4.18,
kwh_fees_low: float=1.62421,
kwh_fees_high: float=1.83474,
sell_fees: float=0.45,
low_tariff_hours:str="0,1,2,3,4,5,6,7,9,10,11,13,14,16,17,18,20,21,22,23",
no_cache:bool = False,
num_cheapest_hours:int = 8,
) -> DayPrice:
is_today = datetime.date.today() == date
low_tariff_hours_parsed = [int(x.strip()) for x in low_tariff_hours.split(",")]
monthly_fees = (monthly_fees + daily_fees * days_in_month(date.year, date.month))
monthly_fees_hour = monthly_fees / days_in_month(date.year, date.month) / 24
currency_ratio = get_eur_czk_ratio(date, no_cache=no_cache)
spot_hours = {}
spot_hours_for_sell = {}
spot_hours_total = {}
spot_data = get_energy_prices(date, no_cache=no_cache)
for hour, value in spot_data.items():
kwh_fees = kwh_fees_low if int(hour) in low_tariff_hours_parsed else kwh_fees_high
spot_hours[hour] = value / currency_ratio
spot_hours_total[hour] = (value / currency_ratio + kwh_fees) * VAT
spot_hours_for_sell[hour] = value / currency_ratio - sell_fees
spot = Price(hours=spot_hours, now=spot_hours[str(hour)] if is_today else None)
spot_hours_total_sorted = {k: v for k, v in sorted(spot_hours_total.items(), key=lambda item: item[1])}
spot_total = Price(hours=spot_hours_total_sorted, now=spot_hours_total[str(hour)] if is_today else None)
spot_for_sell = Price(hours=spot_hours_for_sell, now=spot_hours_for_sell[str(hour)] if is_today else None)
cheapest_hours = [int(k) for k, v in list(spot_hours_total_sorted.items())[0:num_cheapest_hours]]
data = DayPrice(
monthly_fees=monthly_fees * VAT,
monthly_fees_hour=monthly_fees_hour * VAT,
kwh_fees_low=kwh_fees_low * VAT,
kwh_fees_high=kwh_fees_high * VAT,
sell_fees=sell_fees,
low_tariff_hours=low_tariff_hours_parsed,
vat=VAT,
spot=spot,
total=spot_total,
sell=spot_for_sell,
cheapest_hours=CheapestHours(hours=cheapest_hours, is_cheapest=hour in cheapest_hours if is_today else None)
)
return data

77
calculator/miner.py Normal file
View File

@ -0,0 +1,77 @@
from dataclasses import dataclass
import datetime
import json
import os
from typing import Dict
import requests
class PriceException(Exception): pass
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"
def get_energy_prices(d: datetime.date=datetime.date.today(), no_cache:bool=False) -> Dict[str, float]:
"""
"""
date_str = d.strftime("%Y-%m-%d")
hours = {}
cache_file = os.path.join(CACHE_PATH, "hours-{}.json".format(date_str))
if os.path.isfile(cache_file) and not no_cache:
with open(cache_file, "r") as f:
hours = json.load(f)
if not hours or no_cache:
r = requests.get(url_energy.format(date_str))
for raw in r.json()["data"]["dataLine"][1]["point"]:
hours[str(int(raw["x"])-1)] = raw["y"]
with open(cache_file, "w") as f:
f.write(json.dumps(hours))
return hours
#def get_currency_ratio(currency):
# r = requests.get(url_currency)
# return r.json()["kurzy"][currency]["dev_stred"]
def get_eur_czk_ratio(d: datetime.date=datetime.date.today(), no_cache:bool=False) -> float:
ratio = 0
cache_file = os.path.join(CACHE_PATH, "eur-czk-{}.json".format(d.strftime("%Y-%m-%d")))
if os.path.isfile(cache_file):
with open(cache_file, "r") as f:
ratio = float(f.read())
if not ratio or no_cache:
url = url_currency2.format(day=d.day,month=d.month,year=d.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":
ratio = float(row[4].replace(",", "."))
break
if not ratio:
raise PriceException("EUR not found")
with open(cache_file, "w") as f:
f.write(str(ratio))
return ratio

34
calculator/schema.py Normal file
View File

@ -0,0 +1,34 @@
from dataclasses import dataclass
from typing import Dict, List, Optional
@dataclass
class Price:
hours: Dict[str, float]
now: Optional[float] = None # Price in current hour
@dataclass
class CheapestHours:
hours: List[int]
is_cheapest: Optional[bool] = None
@dataclass
class DayPrice:
monthly_fees: float
monthly_fees_hour: float
kwh_fees_low: float
kwh_fees_high: float
sell_fees: float
low_tariff_hours: List[int]
vat: float
spot: Price
total: Price
sell: Price
cheapest_hours: CheapestHours

21
calculator/tests.py Normal file
View File

@ -0,0 +1,21 @@
import unittest
import datetime
from .miner import get_energy_prices, get_eur_czk_ratio
class TestMiner(unittest.TestCase):
def test_get_energy_prices(self):
data = get_energy_prices(datetime.datetime.now(), no_cache=True)
data_cache = get_energy_prices(datetime.datetime.now())
self.assertGreater(data["0"], 0)
self.assertEqual(data["0"], data_cache["0"])
self.assertEqual(data, data_cache)
def test_currency_ratio(self):
data = get_eur_czk_ratio(datetime.datetime.now(), no_cache=True)
data_cache = get_eur_czk_ratio(datetime.datetime.now())
self.assertGreater(data, 0)
self.assertEqual(data, data_cache)

161
main.py Normal file
View File

@ -0,0 +1,161 @@
#!/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)

2
requirements.txt Normal file
View File

@ -0,0 +1,2 @@
fastapi[standard]
requests