refactor export limit semantics

backend/services/control/deye_helpers.py

@@ -19,9 +19,6 @@ DEYE_CLOCK_RESYNC_INTERVAL_HOURS = 24
 # Deye LV baterie: převod výkon -> proud pro registry 108/109.
 BATT_VOLTAGE_V = 51.2
 
-# Reg 143 ve SELL: min(|grid_setpoint_w|, ...) nesmí klesnout pod tuto podlahu (W).
-REG143_SELL_CAP_MIN_W = 200
-
 # Reg 178 - bitové pole: bity 4-5 (peak shaving switch) a bity 0-1 (MI export cutoff).
 REG178_SELL = 0b00100000
 REG178_PASSIVE = 0b00110000

backend/services/control/exporter_monolith.py

@@ -13,7 +13,6 @@ from services.control.deye_helpers import (
     DEYE_TOU_INACTIVE_HHMM,
     DEYE_TOU_POWER_REGS,
     PRAGUE_TZ,
-    REG143_SELL_CAP_MIN_W,
     REG178_MI_EXPORT_DISABLE,
     REG178_MI_EXPORT_ENABLE,
     REG178_MI_EXPORT_MASK,

backend/services/control/inverter.py

@@ -15,7 +15,6 @@ from services.control.deye_helpers import (
     DEYE_CLOCK_RESYNC_INTERVAL_HOURS,
     DEYE_TOU_INACTIVE_HHMM,
     PRAGUE_TZ,
-    REG143_SELL_CAP_MIN_W,
     REG178_MI_EXPORT_DISABLE,
     REG178_MI_EXPORT_ENABLE,
     REG178_MI_EXPORT_MASK,
@@ -103,8 +102,6 @@ async def write_inverter_setpoints(
         selling_mode = 0 if deye_mode == "SELL" else zero_exp_mode
         solar_sell = 0 if (setpoints_now.export_ban and deye_mode != "SELL") else 1
         export_limit = export_lim
-        if deye_mode == "SELL" and grid_w < 0:
-            export_limit = min(export_lim, max(REG143_SELL_CAP_MIN_W, abs(grid_w)))
         reg178_val = REG178_SELL if deye_mode == "SELL" else REG178_PASSIVE
 
         logger.info(

backend/services/control/models.py

@@ -39,6 +39,7 @@ class InverterConfig:
 @dataclass
 class ControlSetpoints:
     battery_w: int | None
+    #: Tvrdý limit exportu do sítě v daném slotu (W), ne forecastová cílová hodnota.
     grid_export_limit: int
     ev1_current_a: int
     ev2_current_a: int

backend/services/control/setpoints.py

@@ -81,6 +81,8 @@ def _build_setpoints(
         if pi is None:
             return None
         grid_sp = int(pi["grid_setpoint_w"] or 0)
+        export_limit_raw = pi.get("export_limit_w")
+        export_limit = int(export_limit_raw) if export_limit_raw is not None else abs(min(grid_sp, 0))
         ev1_w = int(pi["ev1_setpoint_w"] or 0) if "ev1_setpoint_w" in pi else 0
         ev2_w = int(pi["ev2_setpoint_w"] or 0) if "ev2_setpoint_w" in pi else 0
         hp_en = bool(pi["heat_pump_enabled"])
@@ -90,6 +92,10 @@ def _build_setpoints(
         pm: str | None = str(pm_raw).strip().upper() if pm_raw is not None else None
         sell_raw = pi.get("effective_sell_price")
         sell_f: float | None = float(sell_raw) if sell_raw is not None else None
+        export_mode_raw = pi.get("export_mode")
+        export_mode = str(export_mode_raw).strip().upper() if export_mode_raw is not None else None
+        if export_mode == "NONE":
+            export_limit = 0
         # Záporný výkup sám o sobě neblokuje export, pokud plán export explicitně žádá.
         export_ban = sell_f is not None and float(sell_f) < 0 and grid_sp >= 0
         gen_cutoff_raw = pi.get("deye_gen_cutoff_enabled")
@@ -112,7 +118,7 @@ def _build_setpoints(
                 pv_a_allowed = 0
         return ControlSetpoints(
             battery_w=int(pi["battery_setpoint_w"] or 0),
-            grid_export_limit=abs(min(grid_sp, 0)),
+            grid_export_limit=max(0, export_limit),
             ev1_current_a=watts_to_amps(ev1_w, phases=3),
             ev2_current_a=watts_to_amps(ev2_w, phases=1),
             heat_pump_enable=hp_en,

backend/services/planning_engine.py

@@ -319,6 +319,8 @@ class DispatchResult:
     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
@@ -851,6 +853,10 @@ def solve_dispatch(
         batt_w     = round(pulp.value(bc[t]) - pulp.value(bd[t]))
         grid_w     = round(pulp.value(gi[t]) - pulp.value(ge[t]))
         soc_pct    = round(pulp.value(soc[t]) / battery.usable_capacity_wh * 100, 1)
+        export_limit_w = int(grid.max_export_power_w) if grid_w < 0 else 0
+        export_mode = "NONE"
+        if grid_w < 0:
+            export_mode = "BATTERY_SELL" if batt_w < 0 else "PV_SURPLUS"
 
         # Primární klasifikace fyzického režimu pro Deye: explicitně do plánu (Variant A).
         # Default PASSIVE; SELL při export+vybíjení; CHARGE při import+nabíjení.
@@ -874,6 +880,8 @@ def solve_dispatch(
             battery_setpoint_w   = batt_w,
             battery_soc_target   = soc_pct,
             grid_setpoint_w      = grid_w,
+            export_limit_w       = export_limit_w,
+            export_mode          = export_mode,
             deye_physical_mode   = deye_mode,
             deye_gen_cutoff_enabled = deye_gen_cutoff,
             ev1_setpoint_w       = round(pulp.value(ev_direct[0][t]) + pulp.value(ev_via_bat[0][t]))
@@ -1319,6 +1327,8 @@ async def _save_planning_run(
             "battery_setpoint_w": r.battery_setpoint_w,
             "battery_soc_target_pct": r.battery_soc_target,
             "grid_setpoint_w": r.grid_setpoint_w,
+            "export_limit_w": r.export_limit_w,
+            "export_mode": r.export_mode,
             "deye_physical_mode": r.deye_physical_mode,
             "deye_gen_cutoff_enabled": r.deye_gen_cutoff_enabled,
             "ev1_setpoint_w": r.ev1_setpoint_w,

backend/services/signal_service.py

@@ -178,6 +178,10 @@ async def compute_export_ban_active(site_id: int, conn: asyncpg.Connection) -> b
     if bool(pi.get("is_predicted_price")):
         return True
 
+    export_mode = str(pi.get("export_mode") or "").upper()
+    if export_mode in ("PV_SURPLUS", "BATTERY_SELL"):
+        return False
+
     sell_raw = pi.get("effective_sell_price")
     grid_sp = int(pi.get("grid_setpoint_w") or 0)
     if sell_raw is None:

backend/tests/test_control_exporter_tou.py

@@ -15,6 +15,8 @@ from services.control.exporter_monolith import (
     deye_reg_triggers_self_sustain_after_verify_exhaust,
     get_deye_mode,
 )
+from services.control.models import OperatingModeInfo
+from services.control.setpoints import _build_setpoints
 
 
 def _inv(*, min_soc: int | None = 12, reserve_soc: int | None = 20) -> InverterConfig:
@@ -110,6 +112,30 @@ class DeyeTouParamsTests(unittest.TestCase):
         )
         self.assertEqual(get_deye_mode(sp), "PASSIVE")
 
+    def test_build_setpoints_uses_explicit_export_limit(self) -> None:
+        mode = OperatingModeInfo(
+            mode_code="AUTO",
+            battery_mode="AUTO",
+            grid_mode="AUTO",
+            ev_enabled=False,
+            heat_pump_enabled_def=False,
+            loxone_mode_value=1,
+        )
+        pi = {
+            "battery_setpoint_w": 0,
+            "grid_setpoint_w": -3000,
+            "export_limit_w": 13_500,
+            "export_mode": "PV_SURPLUS",
+            "ev1_setpoint_w": 0,
+            "ev2_setpoint_w": 0,
+            "heat_pump_enabled": False,
+            "battery_soc_target_pct": 50,
+            "effective_sell_price": 1.0,
+        }
+        sp = _build_setpoints(mode, pi)
+        self.assertIsNotNone(sp)
+        self.assertEqual(sp.grid_export_limit, 13_500)
+
     def test_pv_led_export_with_small_battery_is_sell(self) -> None:
         """Obě záporné → SELL (bez porovnání |bat| vs |grid|)."""
         sp = ControlSetpoints(

backend/tests/test_planning_dispatch_milp.py

@@ -254,6 +254,47 @@ class PlanningDispatchMilpTests(unittest.TestCase):
         self.assertEqual(int(results[0].pv_a_curtailed_w), 5000)
         self.assertGreaterEqual(int(results[0].grid_setpoint_w), 0)
 
+    def test_pv_surplus_export_uses_hard_export_cap(self) -> None:
+        slots = [
+            _slot(load=0, buy=3.0, sell=2.5, pv_a=20_000, pv_b=0),
+        ]
+        battery = _battery()
+        hp = SimpleNamespace(
+            rated_heating_power_w=0,
+            tuv_min_temp_c=45.0,
+            tuv_target_temp_c=55.0,
+        )
+        grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=13_500)
+        vehicles = [
+            SimpleNamespace(
+                max_charge_power_w=0,
+                battery_capacity_kwh=1.0,
+                default_target_soc_pct=80.0,
+            ),
+            SimpleNamespace(
+                max_charge_power_w=0,
+                battery_capacity_kwh=1.0,
+                default_target_soc_pct=80.0,
+            ),
+        ]
+        soc0 = battery.soc_max_wh
+        results, _ms = solve_dispatch(
+            slots,
+            battery,
+            hp,
+            grid,
+            [None, None],
+            vehicles,
+            soc0,
+            50.0,
+            tuv_delta_stats=None,
+            operating_mode="AUTO",
+        )
+        self.assertEqual(len(results), 1)
+        self.assertEqual(results[0].export_mode, "PV_SURPLUS")
+        self.assertEqual(results[0].export_limit_w, 13_500)
+        self.assertGreater(results[0].pv_a_curtailed_w, 0)
+
     def test_two_tier_soc_solves_optimal(self) -> None:
         slots = [_slot()]
         battery = _battery()