vojacekd/ems
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fáze 1.2: extrakce typů a časových utilit do services/planning/types.py

Browse Source 
PlannerSolverError, PlanningSlot, DispatchResult, SOC_MIN_RELAX_LOOKAHEAD_SLOTS,
_timestamptz_from_db, _slot_float_nullable, _prague_dow_hour, _prague_calendar_date,
_prague_hour, _parse_json_dt, _current_slot_start. Fasáda v planning_engine.py,
beze změny chování (golden 5/5, baseline 4+1 faily beze změny).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dusan Vojacek
2026-06-11 12:34:19 +02:00
parent 4ee5cebf2a
commit d83917da51
3 changed files with 154 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -13,3 +13,4 @@ dist/
*.tsbuildinfo *.tsbuildinfo
frontend/vendor/ frontend/vendor/
frontend/scripts/.native-tmp/ frontend/scripts/.native-tmp/
.claude/settings.local.json

140
backend/services/planning/types.py Normal file
View File

@@ -0,0 +1,140 @@
# backend/services/planning/types.py
#
# EMS plánovač – datové typy a čisté časové utility
# (Fáze 1 dekompozice, čistý přesun z planning_engine.py).
import json
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Optional
from zoneinfo import ZoneInfo
from services.planning.constants import _PRAGUE_TZ
class PlannerSolverError(RuntimeError):
"""Solver selhal po vyčerpání retry řetězce (typicky Infeasible)."""
def __init__(
self,
solver_status: str,
*,
relax_chain: list[str] | None = None,
) -> None:
self.solver_status = solver_status
self.relax_chain = list(relax_chain or [])
super().__init__(f"Solver: {solver_status}")
def _timestamptz_from_db(val: object) -> Optional[datetime]:
if val is None:
return None
if isinstance(val, datetime):
return val if val.tzinfo else val.replace(tzinfo=timezone.utc)
return datetime.fromisoformat(str(val).replace("Z", "+00:00"))
def _slot_float_nullable(d: dict[str, Any], key: str) -> float | None:
v = d.get(key)
if v is None:
return None
return float(v)
def _prague_dow_hour(interval_start: datetime) -> tuple[int, int]:
"""DOW v konvenci PostgreSQL EXTRACT(DOW, Europe/Prague): 0=Ne … 6=So."""
dt = interval_start
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
loc = dt.astimezone(_PRAGUE_TZ)
return (loc.weekday() + 1) % 7, loc.hour
@dataclass
class PlanningSlot:
interval_start: datetime
buy_price: float # Kč/kWh
sell_price: float # Kč/kWh
pv_a_forecast_w: int # W – pole A (řiditelné)
pv_b_forecast_w: int # W – pole B (zelený bonus, pevné)
load_baseline_w: int # W – predikce bazální spotřeby
ev1_connected: bool
ev2_connected: bool
is_predicted_price: bool = False
allow_charge: bool = True
allow_discharge_export: bool = True
#: Měkké LP vstupy z `ems.fn_load_planning_slots_full` (mimo masky allow_*).
night_baseload_target_wh: float | None = None
night_baseload_buffer_wh: float | None = None
safety_soc_target_wh: float | None = None
future_avoided_buy_czk_kwh: float | None = None
future_sell_opportunity_czk_kwh: float | None = None
is_daytime_pv_surplus_slot: bool = False
#: Vážená nákupní / opportunity cena zásoby před prvním exportním oknem (SQL odhad z masek).
charge_acquisition_buy_czk_kwh: float | None = None
charge_acquisition_cutoff_at: datetime | None = None
min_buy_before_cutoff_czk_kwh: float | None = None
pv_charge_wh_ahead: float | None = None
neg_buy_wh_ahead: float | None = None
grid_charge_suppressed_reason: str | None = None
charge_target_wh: float | None = None
pre_window_wh: float | None = None
in_window_wh: float | None = None
charge_slot_wh: float | None = None
charge_cum_wh: float | None = None
charge_layer: str | None = None
charge_slot_reason: str | None = None
#: Pomocny atribut pro green_bonus v planning_interval (Kc/slot); lite default 0.
green_bonus_czk_per_slot: float = 0.0
SOC_MIN_RELAX_LOOKAHEAD_SLOTS = 144
@dataclass
class DispatchResult:
interval_start: datetime
battery_setpoint_w: int # kladné = nabíjení, záporné = vybíjení
battery_soc_target: float # % SoC na konci intervalu
grid_setpoint_w: int # kladné = import, záporné = export
export_limit_w: int # tvrdý limit exportu do sítě; 0 = bez exportu
export_mode: str # NONE / PV_SURPLUS / BATTERY_SELL
#: Explicitní fyzický režim Deye pro control exporter (PASSIVE / SELL / CHARGE).
#: Cíl: odstranit heuristiky z exporteru a nést záměr přímo v plánu.
deye_physical_mode: str
#: True = v daném slotu odpojit GEN port (MI export cutoff) přes reg 178 bits0–1 (0-based; v UI často jako "register 179").
#: None = lokalita tuto funkci nemá / nepoužívá.
deye_gen_cutoff_enabled: bool | None
ev1_setpoint_w: Optional[int]
ev2_setpoint_w: Optional[int]
ev1_via_bat_w: int
ev2_via_bat_w: int
heat_pump_enabled: bool
heat_pump_setpoint_w: int
pv_a_curtailed_w: int
expected_cost_czk: float
effective_buy_price: float
effective_sell_price: float
is_predicted_price: bool # shodné s PlanningSlot (chybí OTE v efektivní ceně → fn_get_predicted_price)
cashflow_czk: float
battery_arbitrage_czk: float
penalty_czk: float
green_bonus_czk: float
def _prague_calendar_date(slot: PlanningSlot):
dt = slot.interval_start
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(ZoneInfo("Europe/Prague")).date()
def _prague_hour(slot: PlanningSlot) -> int:
dt = slot.interval_start
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(ZoneInfo("Europe/Prague")).hour
def _parse_json_dt(val: object) -> Optional[datetime]:
if val is None:
return None
if isinstance(val, datetime):
return val if val.tzinfo else val.replace(tzinfo=timezone.utc)
return datetime.fromisoformat(str(val).replace("Z", "+00:00"))
def _current_slot_start(dt: datetime) -> datetime:
"""Zaokrouhlí čas dolů na začátek aktuálního 15min slotu."""
minute = (dt.minute // 15) * 15
return dt.replace(minute=minute, second=0, microsecond=0)

129
backend/services/planning_engine.py
View File

@@ -24,6 +24,19 @@ from app.config import get_settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
from services.planning.types import (
PlannerSolverError,
_timestamptz_from_db,
_slot_float_nullable,
_prague_dow_hour,
PlanningSlot,
SOC_MIN_RELAX_LOOKAHEAD_SLOTS,
DispatchResult,
_prague_calendar_date,
_prague_hour,
_parse_json_dt,
_current_slot_start,
)
from services.planning.constants import ( from services.planning.constants import (
ACQUISITION_TWO_PASS_EPS_KWH, ACQUISITION_TWO_PASS_EPS_KWH,
SOLVER_RELAX_STEPS, SOLVER_RELAX_STEPS,
@@ -84,18 +97,6 @@ from services.planning.constants import (
_PRAGUE_TZ, _PRAGUE_TZ,
) )
class PlannerSolverError(RuntimeError):
"""Solver selhal po vyčerpání retry řetězce (typicky Infeasible)."""
def __init__(
self,
solver_status: str,
*,
relax_chain: list[str] | None = None,
) -> None:
self.solver_status = solver_status
self.relax_chain = list(relax_chain or [])
super().__init__(f"Solver: {solver_status}")
def _solver_relax_chain( def _solver_relax_chain(
@@ -124,12 +125,6 @@ def _solver_relax_chain(
return chain return chain
def _timestamptz_from_db(val: object) -> Optional[datetime]:
if val is None:
return None
if isinstance(val, datetime):
return val if val.tzinfo else val.replace(tzinfo=timezone.utc)
return datetime.fromisoformat(str(val).replace("Z", "+00:00"))
def _planner_engine_version(explicit: str | None = None) -> str: def _planner_engine_version(explicit: str | None = None) -> str:
@@ -370,66 +365,17 @@ def _soc_security_profile(slots: list["PlanningSlot"], battery) -> tuple[float,
return target_wh, penalty_czk_kwh return target_wh, penalty_czk_kwh
def _slot_float_nullable(d: dict[str, Any], key: str) -> float | None:
v = d.get(key)
if v is None:
return None
return float(v)
def _prague_dow_hour(interval_start: datetime) -> tuple[int, int]:
"""DOW v konvenci PostgreSQL EXTRACT(DOW, Europe/Prague): 0=Ne … 6=So."""
dt = interval_start
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
loc = dt.astimezone(_PRAGUE_TZ)
return (loc.weekday() + 1) % 7, loc.hour
# ============================================================ # ============================================================
# Datové třídy (lze nahradit pydantic modely) # Datové třídy (lze nahradit pydantic modely)
# ============================================================ # ============================================================
@dataclass
class PlanningSlot:
interval_start: datetime
buy_price: float # Kč/kWh
sell_price: float # Kč/kWh
pv_a_forecast_w: int # W – pole A (řiditelné)
pv_b_forecast_w: int # W – pole B (zelený bonus, pevné)
load_baseline_w: int # W – predikce bazální spotřeby
ev1_connected: bool
ev2_connected: bool
is_predicted_price: bool = False
allow_charge: bool = True
allow_discharge_export: bool = True
#: Měkké LP vstupy z `ems.fn_load_planning_slots_full` (mimo masky allow_*).
night_baseload_target_wh: float | None = None
night_baseload_buffer_wh: float | None = None
safety_soc_target_wh: float | None = None
future_avoided_buy_czk_kwh: float | None = None
future_sell_opportunity_czk_kwh: float | None = None
is_daytime_pv_surplus_slot: bool = False
#: Vážená nákupní / opportunity cena zásoby před prvním exportním oknem (SQL odhad z masek).
charge_acquisition_buy_czk_kwh: float | None = None
charge_acquisition_cutoff_at: datetime | None = None
min_buy_before_cutoff_czk_kwh: float | None = None
pv_charge_wh_ahead: float | None = None
neg_buy_wh_ahead: float | None = None
grid_charge_suppressed_reason: str | None = None
charge_target_wh: float | None = None
pre_window_wh: float | None = None
in_window_wh: float | None = None
charge_slot_wh: float | None = None
charge_cum_wh: float | None = None
charge_layer: str | None = None
charge_slot_reason: str | None = None
#: Pomocny atribut pro green_bonus v planning_interval (Kc/slot); lite default 0.
green_bonus_czk_per_slot: float = 0.0
# Lookahead pro relax spodní meze SoC: až 36 h od indexu slotu (pevné OTE ceny v horizontu). # Lookahead pro relax spodní meze SoC: až 36 h od indexu slotu (pevné OTE ceny v horizontu).
SOC_MIN_RELAX_LOOKAHEAD_SLOTS = 144
def _soc_min_wh_series( def _soc_min_wh_series(
@@ -554,35 +500,6 @@ def _soc_panel_min_wh_series(
return out return out
@dataclass
class DispatchResult:
interval_start: datetime
battery_setpoint_w: int # kladné = nabíjení, záporné = vybíjení
battery_soc_target: float # % SoC na konci intervalu
grid_setpoint_w: int # kladné = import, záporné = export
export_limit_w: int # tvrdý limit exportu do sítě; 0 = bez exportu
export_mode: str # NONE / PV_SURPLUS / BATTERY_SELL
#: Explicitní fyzický režim Deye pro control exporter (PASSIVE / SELL / CHARGE).
#: Cíl: odstranit heuristiky z exporteru a nést záměr přímo v plánu.
deye_physical_mode: str
#: True = v daném slotu odpojit GEN port (MI export cutoff) přes reg 178 bits0–1 (0-based; v UI často jako "register 179").
#: None = lokalita tuto funkci nemá / nepoužívá.
deye_gen_cutoff_enabled: bool | None
ev1_setpoint_w: Optional[int]
ev2_setpoint_w: Optional[int]
ev1_via_bat_w: int
ev2_via_bat_w: int
heat_pump_enabled: bool
heat_pump_setpoint_w: int
pv_a_curtailed_w: int
expected_cost_czk: float
effective_buy_price: float
effective_sell_price: float
is_predicted_price: bool # shodné s PlanningSlot (chybí OTE v efektivní ceně → fn_get_predicted_price)
cashflow_czk: float
battery_arbitrage_czk: float
penalty_czk: float
green_bonus_czk: float
# ============================================================ # ============================================================
@@ -846,11 +763,6 @@ def _pre_negative_sell_export_window(
return first_neg, first_neg - 1 return first_neg, first_neg - 1
def _prague_calendar_date(slot: PlanningSlot):
dt = slot.interval_start
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(ZoneInfo("Europe/Prague")).date()
def _neg_sell_phases_enabled(battery: Any) -> bool: def _neg_sell_phases_enabled(battery: Any) -> bool:
@@ -1602,11 +1514,6 @@ def _dispatch_grid_setpoint_w(
return round(float(gi_w) - ge_total), "NONE" return round(float(gi_w) - ge_total), "NONE"
def _prague_hour(slot: PlanningSlot) -> int:
dt = slot.interval_start
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(ZoneInfo("Europe/Prague")).hour
def _morning_pre_neg_zone_peak_sell( def _morning_pre_neg_zone_peak_sell(
@@ -5788,18 +5695,8 @@ async def run_plan_api(
# Pomocné funkce # Pomocné funkce
# ============================================================ # ============================================================
def _current_slot_start(dt: datetime) -> datetime:
"""Zaokrouhlí čas dolů na začátek aktuálního 15min slotu."""
minute = (dt.minute // 15) * 15
return dt.replace(minute=minute, second=0, microsecond=0)
def _parse_json_dt(val: object) -> Optional[datetime]:
if val is None:
return None
if isinstance(val, datetime):
return val if val.tzinfo else val.replace(tzinfo=timezone.utc)
return datetime.fromisoformat(str(val).replace("Z", "+00:00"))
def _ev_session_from_json(obj: object) -> Optional[SimpleNamespace]: def _ev_session_from_json(obj: object) -> Optional[SimpleNamespace]: