Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion core/bess/growatt_min_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -187,7 +187,9 @@ def _group_periods_by_mode(self, start_period: int = 0) -> list[dict]:

for period in range(start_period, num_periods):
intent = self.strategic_intents[period]
mode = self.INTENT_TO_MODE.get(intent, "load_first")
mode = self._effective_mode_for_intent(
intent, self.INTENT_TO_MODE.get(intent, "load_first")
)

if mode != current_mode:
# Save previous group if exists
Expand Down
17 changes: 15 additions & 2 deletions core/bess/growatt_sph_controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,19 @@ def active_tou_intervals(self) -> list[dict]:

# ── SPH period grouping ───────────────────────────────────────────────────

def _effective_charge_intents(self) -> frozenset[str]:
"""Intents that produce an AC charge period, honoring external_solar_mode.

SOLAR_STORAGE is normally excluded — a DC-coupled SPH charges from
its own MPPT without an explicit period. On AC-coupled PV setups
(external_solar_mode) the SPH has no DC solar input, so SOLAR_STORAGE
must become an AC charge period or the battery never charges during
the planned solar window.
"""
if self.battery_settings.external_solar_mode:
return self.CHARGE_INTENTS | {"SOLAR_STORAGE"}
return self.CHARGE_INTENTS

def _group_sph_periods(self) -> tuple[list[dict], list[dict]]:
"""Group consecutive strategic intent periods into charge and discharge blocks.

Expand All @@ -94,7 +107,7 @@ def _group_sph_periods(self) -> tuple[list[dict], list[dict]]:
discharge_blocks: list[dict] = []

for _category, target_list, intent_set in [
("charge", charge_blocks, self.CHARGE_INTENTS),
("charge", charge_blocks, self._effective_charge_intents()),
("discharge", discharge_blocks, self.DISCHARGE_INTENTS),
]:
current_block: dict | None = None
Expand Down Expand Up @@ -636,7 +649,7 @@ def log_detailed_schedule(self, header: str = "") -> None:
is_current = run_start <= current_period <= run_end
marker = "*" if is_current else " "

if intent in self.CHARGE_INTENTS:
if intent in self._effective_charge_intents():
action = "charge"
elif intent in self.DISCHARGE_INTENTS:
action = "discharge"
Expand Down
78 changes: 78 additions & 0 deletions core/bess/tests/unit/test_external_solar_mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@

import pytest

from core.bess.growatt_min_controller import GrowattMinController
from core.bess.growatt_sph_controller import GrowattSphController
from core.bess.settings import BatterySettings
from core.bess.solax_controller import SolaxController

Expand Down Expand Up @@ -126,3 +128,79 @@ def test_detailed_period_groups_apply_mode_override(self) -> None:
for group in groups:
assert group["mode"] == "battery_first"
assert group["grid_charge"] is True


class TestExternalSolarModeGrowattMinTouPath:
"""The MIN controller's TOU grouping (_group_periods_by_mode) builds the
segments actually written to hardware. It must apply the
external_solar_mode override, otherwise SOLAR_STORAGE stays load_first,
no TOU segment is created (load_first groups are skipped), and an
AC-coupled battery never charges during the solar window.
"""

def _controller(self, *, external_solar_mode: bool) -> GrowattMinController:
ctrl = GrowattMinController(
battery_settings=_settings(external_solar_mode=external_solar_mode)
)
ctrl.strategic_intents = ["IDLE"] * 40 + ["SOLAR_STORAGE"] * 16 + ["IDLE"] * 40
return ctrl

def test_solar_storage_grouped_as_battery_first_when_enabled(self) -> None:
ctrl = self._controller(external_solar_mode=True)
groups = ctrl._group_periods_by_mode()
modes = [g["mode"] for g in groups]
assert "battery_first" in modes
solar_group = next(g for g in groups if g["mode"] == "battery_first")
assert solar_group["start_period"] == 40
assert solar_group["end_period"] == 55
assert set(solar_group["intents"]) == {"SOLAR_STORAGE"}

def test_solar_storage_grouped_as_load_first_when_disabled(self) -> None:
ctrl = self._controller(external_solar_mode=False)
groups = ctrl._group_periods_by_mode()
assert all(g["mode"] == "load_first" for g in groups)

def test_solar_storage_produces_tou_interval_when_enabled(self) -> None:
ctrl = self._controller(external_solar_mode=True)
groups = ctrl._group_periods_by_mode()
intervals = ctrl._groups_to_tou_intervals(groups)
assert len(intervals) == 1
assert intervals[0]["batt_mode"] == "battery_first"

def test_no_tou_interval_when_disabled(self) -> None:
ctrl = self._controller(external_solar_mode=False)
groups = ctrl._group_periods_by_mode()
assert ctrl._groups_to_tou_intervals(groups) == []


class TestExternalSolarModeSphChargePeriods:
"""SPH normally excludes SOLAR_STORAGE from charge periods (a DC-coupled
SPH charges from its own MPPT). With external_solar_mode the SPH has no
DC solar input, so SOLAR_STORAGE must produce an AC charge period.
"""

def _controller(self, *, external_solar_mode: bool) -> GrowattSphController:
ctrl = GrowattSphController(
battery_settings=_settings(external_solar_mode=external_solar_mode)
)
ctrl.strategic_intents = ["IDLE"] * 40 + ["SOLAR_STORAGE"] * 16 + ["IDLE"] * 40
return ctrl

def test_solar_storage_becomes_charge_block_when_enabled(self) -> None:
ctrl = self._controller(external_solar_mode=True)
charge_blocks, discharge_blocks = ctrl._group_sph_periods()
assert len(charge_blocks) == 1
assert charge_blocks[0]["start_period"] == 40
assert charge_blocks[0]["end_period"] == 55
assert discharge_blocks == []

def test_solar_storage_not_a_charge_block_when_disabled(self) -> None:
ctrl = self._controller(external_solar_mode=False)
charge_blocks, _ = ctrl._group_sph_periods()
assert charge_blocks == []

def test_grid_charging_still_a_charge_block_when_disabled(self) -> None:
ctrl = self._controller(external_solar_mode=False)
ctrl.strategic_intents = ["GRID_CHARGING"] * 8 + ["IDLE"] * 88
charge_blocks, _ = ctrl._group_sph_periods()
assert len(charge_blocks) == 1