250 lines
9.2 KiB
Python
250 lines
9.2 KiB
Python
"""Pravidla pro GET /sites/{id}/notifications (ceny, EV, predikce záporných cen)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from datetime import date, datetime, time, timedelta, timezone
|
||
from decimal import Decimal
|
||
from typing import Any, Literal
|
||
from zoneinfo import ZoneInfo
|
||
|
||
PRAGUE = ZoneInfo("Europe/Prague")
|
||
|
||
NotificationLevel = Literal["success", "info", "warning", "error"]
|
||
NotificationAction = Literal["connect_ev", "replan", "import_prices", "switch_auto"]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PriceSlot:
|
||
interval_start: datetime
|
||
buy: float
|
||
sell: float
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class EvSessionRow:
|
||
id: int
|
||
charger_id: int
|
||
energy_delivered_wh: float
|
||
target_soc_pct: float | None
|
||
session_start: datetime
|
||
battery_capacity_kwh: float | None
|
||
make: str | None
|
||
model: str | None
|
||
default_target_soc_pct: float | None
|
||
charger_code: str
|
||
soc_at_connect_pct: float | None
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class NegWindowRow:
|
||
predicted_date: date
|
||
window_start_hour: int
|
||
window_end_hour: int
|
||
probability_pct: int
|
||
|
||
|
||
def _num(v: Any) -> float | None:
|
||
if v is None:
|
||
return None
|
||
if isinstance(v, Decimal):
|
||
return float(v)
|
||
return float(v)
|
||
|
||
|
||
def _ev_connect_hint(ev_sessions: list[EvSessionRow], current_price: float, avg_buy: float | None) -> str:
|
||
if ev_sessions:
|
||
return ""
|
||
kwh = 30.0
|
||
if current_price < 0:
|
||
return f"Připojíš-li Teslu, dostaneš zaplaceno za ~{kwh * abs(current_price):.0f} Kč za 30 kWh."
|
||
if avg_buy is None:
|
||
return ""
|
||
savings = avg_buy - current_price
|
||
return f"Připojíš-li Teslu, ušetříš ~{kwh * max(0, savings):.0f} Kč."
|
||
|
||
|
||
def _estimate_ev_soc(s: EvSessionRow) -> float:
|
||
cap = s.battery_capacity_kwh
|
||
delivered_kwh = (s.energy_delivered_wh or 0) / 1000.0
|
||
if s.soc_at_connect_pct is not None and cap and cap > 0:
|
||
return min(95.0, float(s.soc_at_connect_pct) + (delivered_kwh / cap) * 100.0)
|
||
if cap and cap > 0 and s.energy_delivered_wh:
|
||
return min(95.0, (delivered_kwh / cap) * 100.0 + 20.0)
|
||
return 50.0
|
||
|
||
|
||
def _estimate_needed_kwh(soc_pct: float, battery_kwh: float, ev_sessions: list[EvSessionRow]) -> float:
|
||
bat_needed = max(0.0, (80.0 - soc_pct) / 100.0 * battery_kwh)
|
||
ev_needed = 0.0
|
||
for s in ev_sessions:
|
||
cap = s.battery_capacity_kwh or 60.0
|
||
tgt = (s.target_soc_pct if s.target_soc_pct is not None else None) or (
|
||
s.default_target_soc_pct if s.default_target_soc_pct is not None else 80.0
|
||
)
|
||
delivered = (s.energy_delivered_wh or 0) / 1000.0
|
||
want = max(0.0, (tgt / 100.0) * cap - delivered)
|
||
ev_needed += want
|
||
return bat_needed + ev_needed
|
||
|
||
|
||
def _estimate_ev_free_kwh(s: EvSessionRow) -> float:
|
||
cap = s.battery_capacity_kwh or 60.0
|
||
delivered = (s.energy_delivered_wh or 0) / 1000.0
|
||
return max(0.0, cap * 0.95 - delivered)
|
||
|
||
|
||
def _tesla_potential_kwh(ev_sessions: list[EvSessionRow]) -> float:
|
||
"""Odhad „kolik lze ještě dobít“ pro typickou Teslu, pokud není připojena."""
|
||
if ev_sessions:
|
||
return 0.0
|
||
return 55.0
|
||
|
||
|
||
def _date_label(d: date) -> str:
|
||
today = datetime.now(PRAGUE).date()
|
||
if d == today:
|
||
return "dnes"
|
||
if d == today + timedelta(days=1):
|
||
return "zítra"
|
||
return f"{d.day}. {d.month}."
|
||
|
||
|
||
def _hours_until(predicted_date: date, start_hour: int) -> float:
|
||
start_local = datetime.combine(predicted_date, time(hour=start_hour, minute=0), tzinfo=PRAGUE)
|
||
now = datetime.now(PRAGUE)
|
||
delta = start_local - now
|
||
return max(0.0, delta.total_seconds() / 3600.0)
|
||
|
||
|
||
def build_smart_notifications(
|
||
*,
|
||
prices: list[PriceSlot],
|
||
avg_buy: float | None,
|
||
soc_pct: float | None,
|
||
battery_kwh: float | None,
|
||
ev_sessions: list[EvSessionRow],
|
||
neg_windows: list[NegWindowRow],
|
||
mode: str,
|
||
sell_price_now: float | None,
|
||
) -> list[dict[str, Any]]:
|
||
notifications: list[dict[str, Any]] = []
|
||
|
||
soc = float(soc_pct) if soc_pct is not None else 50.0
|
||
bat_kwh = float(battery_kwh) if battery_kwh is not None and battery_kwh > 0 else 10.0
|
||
|
||
current_buy = prices[0].buy if prices else None
|
||
current_sell = prices[0].sell if prices else None
|
||
|
||
# 1. Záporná cena právě teď
|
||
if current_buy is not None and current_buy < 0:
|
||
bat_free_kwh = (100.0 - soc) / 100.0 * bat_kwh
|
||
ev_hint = _ev_connect_hint(ev_sessions, current_buy, avg_buy)
|
||
notifications.append(
|
||
{
|
||
"id": "neg_price_now",
|
||
"level": "success",
|
||
"title": f"Záporné ceny právě teď ({current_buy:.3f} Kč/kWh)",
|
||
"body": (
|
||
f"Dostaneš zaplaceno za každý odebraný kWh. "
|
||
f"Baterie může pojmout ještě {bat_free_kwh:.1f} kWh. "
|
||
+ ev_hint
|
||
),
|
||
"eta_minutes": 0,
|
||
"action": "connect_ev" if not ev_sessions else None,
|
||
}
|
||
)
|
||
|
||
# 2. Levná cena v příštích 6 h
|
||
avg_ok = avg_buy is not None and avg_buy > 0
|
||
if prices and avg_ok and not (current_buy is not None and current_buy < 0):
|
||
horizon = prices[:24]
|
||
cheap_slots = [p for p in horizon if p.buy < avg_buy * 0.60]
|
||
if cheap_slots:
|
||
cheapest = min(cheap_slots, key=lambda p: p.buy)
|
||
now_utc = datetime.now(timezone.utc)
|
||
istart = cheapest.interval_start
|
||
if istart.tzinfo is None:
|
||
istart = istart.replace(tzinfo=timezone.utc)
|
||
eta_min = int((istart - now_utc).total_seconds() / 60)
|
||
eta_min = max(0, eta_min)
|
||
savings_per_kwh = avg_buy - cheapest.buy
|
||
|
||
ev_plug_useful = (not ev_sessions) or any(_estimate_ev_soc(s) < 60.0 for s in ev_sessions)
|
||
bat_low = soc < 70.0
|
||
|
||
if ev_plug_useful or bat_low:
|
||
needed_kwh = _estimate_needed_kwh(soc, bat_kwh, ev_sessions)
|
||
savings_czk = needed_kwh * max(0.0, savings_per_kwh)
|
||
extra_body = ""
|
||
if ev_plug_useful and not ev_sessions:
|
||
extra_body = " Připoj auto před tímto oknem."
|
||
notifications.append(
|
||
{
|
||
"id": "cheap_price_soon",
|
||
"level": "info",
|
||
"title": f"Levná elektřina za {eta_min} min ({cheapest.buy:.3f} Kč/kWh)",
|
||
"body": (
|
||
f"Cena bude o {savings_per_kwh:.2f} Kč/kWh nižší než průměr. "
|
||
f"Potenciální úspora: ~{savings_czk:.0f} Kč."
|
||
+ extra_body
|
||
),
|
||
"eta_minutes": eta_min,
|
||
"action": "connect_ev" if ev_plug_useful and not ev_sessions else None,
|
||
}
|
||
)
|
||
|
||
# 3. Predikované záporné ceny
|
||
for window in neg_windows[:2]:
|
||
if any(n["id"] == "neg_price_now" for n in notifications):
|
||
continue
|
||
date_label = _date_label(window.predicted_date)
|
||
window_str = f"{window.window_start_hour:02d}:00–{window.window_end_hour:02d}:00"
|
||
hours_until = _hours_until(window.predicted_date, window.window_start_hour)
|
||
bat_free = (100.0 - soc) / 100.0 * bat_kwh
|
||
ev_free = sum(_estimate_ev_free_kwh(s) for s in ev_sessions)
|
||
total_free = bat_free + ev_free
|
||
tesla_kwh = _tesla_potential_kwh(ev_sessions)
|
||
ev_hint = (
|
||
f" Připojíš-li Teslu, lze dobít až {tesla_kwh:.0f} kWh navíc." if not ev_sessions else ""
|
||
)
|
||
lvl: NotificationLevel = "success" if window.probability_pct >= 70 else "info"
|
||
notifications.append(
|
||
{
|
||
"id": f"neg_pred_{window.predicted_date}_{window.window_start_hour}",
|
||
"level": lvl,
|
||
"title": (
|
||
f"Záporné ceny {date_label} {window_str} ({window.probability_pct}% jistota)"
|
||
),
|
||
"body": (
|
||
f"Solver naplánuje max. odběr ze sítě. Lze dobít ~{total_free:.0f} kWh zdarma."
|
||
+ ev_hint
|
||
),
|
||
"eta_minutes": int(hours_until * 60),
|
||
"action": "connect_ev" if not ev_sessions else None,
|
||
}
|
||
)
|
||
|
||
# 4. Manuální režim + drahá cena + plná baterie
|
||
mode_u = (mode or "").strip().upper()
|
||
sell = sell_price_now if sell_price_now is not None else (current_sell if current_sell is not None else None)
|
||
if mode_u != "AUTO" and avg_ok and sell is not None and sell > avg_buy * 1.30 and soc > 70.0:
|
||
pct_above = ((sell / avg_buy) - 1.0) * 100.0 if avg_buy else 0.0
|
||
notifications.append(
|
||
{
|
||
"id": "manual_expensive",
|
||
"level": "warning",
|
||
"title": "Drahá elektřina – systém není v AUTO",
|
||
"body": (
|
||
f"Cena prodeje {sell:.3f} Kč/kWh (+{pct_above:.0f}% nad průměr). "
|
||
f"Baterie {soc:.0f}%. Přepnutím na AUTO solver využije cenové okno."
|
||
),
|
||
"eta_minutes": None,
|
||
"action": "switch_auto",
|
||
}
|
||
)
|
||
|
||
priority = {"error": 0, "success": 1, "warning": 2, "info": 3}
|
||
notifications.sort(key=lambda n: priority.get(n["level"], 9))
|
||
return notifications
|