152 lines
6.1 KiB
Python
152 lines
6.1 KiB
Python
import datetime
|
|
import calendar
|
|
from typing import List, Optional
|
|
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import RedirectResponse, HTMLResponse
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
from calculator.miner import PriceNotFound, get_energy_prices, get_eur_czk_ratio
|
|
from calculator.schema import CheapestHours, DayPrice, MostExpensiveHours, 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.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
@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>
|
|
**num_most_expensive_hours** - number of the most expensive 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>
|
|
```
|
|
rest:
|
|
- resource: https://pricepower2.rostiapp.cz/price/day
|
|
scan_interval: 60
|
|
sensor:
|
|
- name: "energy_price"
|
|
value_template: "{{ value_json.total.now|round(2) }}"
|
|
unit_of_measurement: "CZK/kWh"
|
|
- name: "energy_market_price"
|
|
value_template: "{{ value_json.spot.now|round(2) }}"
|
|
unit_of_measurement: "CZK/kWh"
|
|
- name: "energy_market_price_sell"
|
|
value_template: "{{ value_json.sell.now|round(2) }}"
|
|
unit_of_measurement: "CZK/kWh"
|
|
binary_sensor:
|
|
- 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: Optional[datetime.date]=None,
|
|
hour: Optional[int]=None,
|
|
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,
|
|
num_most_expensive_hours:int = 8,
|
|
) -> DayPrice:
|
|
|
|
if not date:
|
|
date = datetime.date.today()
|
|
if not hour:
|
|
hour = datetime.datetime.now().hour
|
|
|
|
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 = {}
|
|
try:
|
|
spot_data = get_energy_prices(date, no_cache=no_cache)
|
|
except PriceNotFound:
|
|
raise HTTPException(status_code=404, detail="prices not found")
|
|
for key, value in spot_data.items():
|
|
kwh_fees = kwh_fees_low if int(key) in low_tariff_hours_parsed else kwh_fees_high
|
|
|
|
spot_hours[key] = value * currency_ratio / 1000
|
|
spot_hours_total[key] = (value * currency_ratio / 1000 + kwh_fees) * VAT
|
|
spot_hours_for_sell[key] = value * currency_ratio / 1000 - 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]]
|
|
most_expensive_hours = [int(k) for k, v in list(reversed(spot_hours_total_sorted.items()))[0:num_most_expensive_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,
|
|
hour=hour,
|
|
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),
|
|
most_expensive_hours=MostExpensiveHours(hours=most_expensive_hours, is_the_most_expensive=hour in most_expensive_hours if is_today else None),
|
|
)
|
|
return data
|
|
|
|
@app.get("/widget", response_class=HTMLResponse)
|
|
def get_widget():
|
|
with open("calculator/index.html", "r") as file:
|
|
html_content = file.read()
|
|
return HTMLResponse(content=html_content)
|