214 lines
6.8 KiB
Python
214 lines
6.8 KiB
Python
"""Deye reg 340 (max solar power) z plánu a capu z DB."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from services.control.exporter_monolith import (
|
|
OperatingModeInfo,
|
|
_DictRecord,
|
|
_build_setpoints,
|
|
compute_pv_a_reg340_max_solar_w,
|
|
deye_reg_triggers_self_sustain_after_verify_exhaust,
|
|
)
|
|
from services.control.setpoints import plan_skips_deye_reg340_write
|
|
|
|
|
|
def _auto_mode() -> OperatingModeInfo:
|
|
return OperatingModeInfo(
|
|
mode_code="AUTO",
|
|
battery_mode="auto",
|
|
grid_mode="auto",
|
|
ev_enabled=True,
|
|
heat_pump_enabled_def=True,
|
|
loxone_mode_value=0,
|
|
)
|
|
|
|
|
|
def _pi_base(**kwargs: object) -> _DictRecord:
|
|
d: dict[str, object] = {
|
|
"grid_setpoint_w": 0,
|
|
"battery_setpoint_w": 0,
|
|
"battery_soc_target_pct": None,
|
|
"heat_pump_enabled": False,
|
|
"effective_sell_price": 1.0,
|
|
"pv_a_forecast_solver_w": 8000,
|
|
"pv_a_curtailed_w": 0,
|
|
}
|
|
d.update(kwargs)
|
|
return _DictRecord(d)
|
|
|
|
|
|
class ComputePvAReg340Tests(unittest.TestCase):
|
|
def test_full_cap_when_no_curtail(self) -> None:
|
|
self.assertEqual(compute_pv_a_reg340_max_solar_w(10_000, 8000, 0), 10_000)
|
|
|
|
def test_curtailed_value(self) -> None:
|
|
self.assertEqual(compute_pv_a_reg340_max_solar_w(10_000, 8000, 2000), 6000)
|
|
|
|
def test_clamped_to_cap_when_forecast_high(self) -> None:
|
|
self.assertEqual(compute_pv_a_reg340_max_solar_w(10_000, 12_000, 0), 10_000)
|
|
|
|
def test_curtail_floor_zero(self) -> None:
|
|
self.assertEqual(compute_pv_a_reg340_max_solar_w(10_000, 1000, 5000), 0)
|
|
|
|
def test_min_clamp_when_positive(self) -> None:
|
|
self.assertEqual(
|
|
compute_pv_a_reg340_max_solar_w(32_000, 5000, 4600, min_w=400),
|
|
400,
|
|
)
|
|
|
|
def test_min_not_applied_when_curtail_to_zero(self) -> None:
|
|
self.assertEqual(compute_pv_a_reg340_max_solar_w(10_000, 1000, 5000, min_w=400), 0)
|
|
|
|
|
|
class BuildSetpointsReg340Tests(unittest.TestCase):
|
|
def test_with_cap_sets_pv_a_allowed(self) -> None:
|
|
sp = _build_setpoints(
|
|
_auto_mode(),
|
|
_pi_base(pv_a_forecast_solver_w=8000, pv_a_curtailed_w=2000),
|
|
pv_a_cap_w=10_000,
|
|
reg340_pv_a_control_enabled=True,
|
|
)
|
|
assert sp is not None
|
|
self.assertEqual(sp.pv_a_allowed_w, 6000)
|
|
|
|
def test_skipped_when_cap_zero(self) -> None:
|
|
sp = _build_setpoints(
|
|
_auto_mode(),
|
|
_pi_base(),
|
|
pv_a_cap_w=0,
|
|
reg340_pv_a_control_enabled=True,
|
|
)
|
|
assert sp is not None
|
|
self.assertIsNone(sp.pv_a_allowed_w)
|
|
|
|
def test_self_sustain_no_pv_a_allowed(self) -> None:
|
|
mode = OperatingModeInfo(
|
|
mode_code="SELF_SUSTAIN",
|
|
battery_mode="x",
|
|
grid_mode="x",
|
|
ev_enabled=False,
|
|
heat_pump_enabled_def=False,
|
|
loxone_mode_value=0,
|
|
)
|
|
sp = _build_setpoints(mode, None, pv_a_cap_w=10_000)
|
|
assert sp is not None
|
|
self.assertIsNone(sp.pv_a_allowed_w)
|
|
|
|
def test_neg_buy_and_sell_with_pv_b_forces_pv_a_off(self) -> None:
|
|
sp = _build_setpoints(
|
|
_auto_mode(),
|
|
_pi_base(
|
|
effective_buy_price=-3.0,
|
|
effective_sell_price=-2.0,
|
|
pv_b_forecast_solver_w=5000,
|
|
pv_a_forecast_solver_w=0,
|
|
pv_a_curtailed_w=0,
|
|
),
|
|
pv_a_cap_w=3333,
|
|
reg340_pv_a_control_enabled=True,
|
|
)
|
|
assert sp is not None
|
|
self.assertEqual(sp.pv_a_allowed_w, 0)
|
|
|
|
def test_skipped_low_pv_forecast_with_mi_no_curtail(self) -> None:
|
|
"""BA81 úsvit: slabý forecast, bez curtail — EMS neposílá reg 340."""
|
|
sp = _build_setpoints(
|
|
_auto_mode(),
|
|
_pi_base(
|
|
pv_a_forecast_solver_w=405,
|
|
pv_b_forecast_solver_w=49,
|
|
pv_a_curtailed_w=0,
|
|
grid_setpoint_w=-100,
|
|
battery_setpoint_w=0,
|
|
export_mode="PV_SURPLUS",
|
|
export_limit_w=100,
|
|
),
|
|
pv_a_cap_w=32_000,
|
|
reg340_pv_a_control_enabled=True,
|
|
)
|
|
assert sp is not None
|
|
self.assertIsNone(sp.pv_a_allowed_w)
|
|
|
|
def test_skipped_when_no_export_no_charge_no_curtail(self) -> None:
|
|
sp = _build_setpoints(
|
|
_auto_mode(),
|
|
_pi_base(
|
|
grid_setpoint_w=0,
|
|
battery_setpoint_w=0,
|
|
export_mode="NONE",
|
|
export_limit_w=0,
|
|
pv_a_curtailed_w=0,
|
|
),
|
|
pv_a_cap_w=10_000,
|
|
reg340_pv_a_control_enabled=True,
|
|
)
|
|
assert sp is not None
|
|
self.assertIsNone(sp.pv_a_allowed_w)
|
|
|
|
def test_writes_reg340_when_curtail_planned(self) -> None:
|
|
sp = _build_setpoints(
|
|
_auto_mode(),
|
|
_pi_base(
|
|
grid_setpoint_w=0,
|
|
battery_setpoint_w=0,
|
|
export_mode="NONE",
|
|
pv_a_curtailed_w=3000,
|
|
),
|
|
pv_a_cap_w=10_000,
|
|
reg340_pv_a_control_enabled=True,
|
|
)
|
|
assert sp is not None
|
|
self.assertEqual(sp.pv_a_allowed_w, 5000)
|
|
|
|
def test_writes_reg340_when_battery_charging_without_export(self) -> None:
|
|
sp = _build_setpoints(
|
|
_auto_mode(),
|
|
_pi_base(
|
|
grid_setpoint_w=0,
|
|
battery_setpoint_w=5000,
|
|
export_mode="NONE",
|
|
pv_a_curtailed_w=0,
|
|
),
|
|
pv_a_cap_w=10_000,
|
|
reg340_pv_a_control_enabled=True,
|
|
)
|
|
assert sp is not None
|
|
self.assertEqual(sp.pv_a_allowed_w, 10_000)
|
|
|
|
def test_plan_skips_helper(self) -> None:
|
|
self.assertTrue(
|
|
plan_skips_deye_reg340_write(
|
|
battery_setpoint_w=0,
|
|
grid_setpoint_w=0,
|
|
export_mode="NONE",
|
|
export_limit_w=0,
|
|
pv_a_curtailed_w=0,
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
plan_skips_deye_reg340_write(
|
|
battery_setpoint_w=0,
|
|
grid_setpoint_w=-2000,
|
|
export_mode="PV_SURPLUS",
|
|
export_limit_w=2000,
|
|
pv_a_curtailed_w=0,
|
|
)
|
|
)
|
|
|
|
def test_skipped_when_reg340_control_disabled(self) -> None:
|
|
sp = _build_setpoints(
|
|
_auto_mode(),
|
|
_pi_base(pv_a_forecast_solver_w=8000, pv_a_curtailed_w=2000),
|
|
pv_a_cap_w=10_000,
|
|
reg340_pv_a_control_enabled=False,
|
|
)
|
|
assert sp is not None
|
|
self.assertIsNone(sp.pv_a_allowed_w)
|
|
|
|
|
|
class Reg340VerifyPolicyTests(unittest.TestCase):
|
|
def test_reg340_not_critical_for_self_sustain(self) -> None:
|
|
self.assertFalse(deye_reg_triggers_self_sustain_after_verify_exhaust(340))
|