Skip to content

fix(modes): reopen curtailed PV and route surplus into loads in one-sided modes #10

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bvweerd merged 1 commit into from
Jul 4, 2026
Merged

fix(modes): reopen curtailed PV and route surplus into loads in one-sided modes #10

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
171 changes: 90 additions & 81 deletions custom_components/zero_grid_controller/control_engine.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -343,19 +343,29 @@ async def run_cycle(
load_setpoints=dict(self._current_load_setpoints),
)

# 5b. Mode idle check — only activate PID on the "forbidden" side
skip_pid = False
# 5b. Mode check — on the "allowed" side only a safe subset of the
# actuators participates. Opening our own PV on import is always
# free energy and routing surplus into loads on export is always
# useful; what the one-sided modes skip is fighting the allowed
# direction (reducing loads in zero_export, curtailing PV in
# zero_import). Previously the whole PID was skipped, which left
# PV curtailed forever in zero_export once a single export event
# had occurred (nothing ever reopened it while importing).
allow_arrays = True
allow_loads = True
cycle_status: ControllerStatus = STATUS_ACTIVE
if self._mode == ControllerMode.ZERO_IMPORT and filtered < -self._deadband_w:
# Exporting: allowed direction — idle, battery charge layer still runs below
self._pid.freeze_integrator()
skip_pid = True
# Exporting is allowed: never curtail PV, but do route the
# surplus into controllable loads (battery layer runs below).
allow_arrays = False
cycle_status = ControllerStatus.IDLE_EXPORT_OK
elif self._mode == ControllerMode.ZERO_EXPORT and filtered > self._deadband_w:
# Importing: allowed direction — idle, battery discharge layer still runs below
self._pid.freeze_integrator()
skip_pid = True
# Importing is allowed: never reduce loads for it, but do reopen
# curtailed PV (battery layer runs below).
allow_loads = False
cycle_status = ControllerStatus.IDLE_IMPORT_OK
arrays_active = arrays if allow_arrays else []
loads_active = loads if allow_loads else []

# 6. Scheduled batteries follow an external optimizer
# (e.g. battery_controller) with a real-time grid correction on top.
Expand Down Expand Up @@ -432,84 +442,83 @@ async def run_cycle(
residual -= scheduled_pending_w
residual += recovery_w

# 7. PID + actuator distribution (skipped in one-sided idle modes)
pid_output = 0.0
if not skip_pid:
# 7a. PID on residual
# Negate: positive output when importing (open PV / reduce load)
pid_output = self._pid.compute(-residual, dt)

# Back-calculation anti-windup (sign contradiction):
# When the integral has wound up so far that it opposes the actual
# grid direction, immediately reset it and use the P-only output for
# this cycle. Threshold of 3×deadband avoids triggering on normal
# near-zero oscillation.
if abs(pid_output) > 3 * self._deadband_w:
wrong_direction = (pid_output < 0 and filtered > self._deadband_w) or (
pid_output > 0 and filtered < -self._deadband_w
)
if wrong_direction:
p_only = self._pid.kp * residual
_LOGGER.info(
"Integrator wind-up corrected: output %.0f W → %.0f W"
" (grid %.1f W)",
pid_output,
p_only,
filtered,
)
self._pid.set_integral(0.0)
pid_output = p_only

# 7b. Numeric correction
_load_sp_before = {
ld.name: self._current_load_setpoints.get(ld.name, ld.setpoint_min)
for ld in loads
if not ld.is_switch
}
_load_w_per_unit = {
ld.name: ld.w_per_unit for ld in loads if not ld.is_switch
}

if pid_output < 0:
# Exporting surplus → increase loads, then curtail arrays
after_loads = await self._distribute_to_numeric_loads(
pid_output, now, loads
)
final_remaining = await self._distribute_to_numeric_arrays(
after_loads, now, arrays
)
array_absorbed_w = after_loads - final_remaining
else:
# Importing deficit → open arrays, then reduce loads
after_arrays = await self._distribute_to_numeric_arrays(
pid_output, now, arrays
)
array_absorbed_w = pid_output - after_arrays
final_remaining = await self._distribute_to_numeric_loads(
after_arrays, now, loads
# 7. PID + actuator distribution (restricted to the allowed actuator
# subset in one-sided modes)
# 7a. PID on residual
# Negate: positive output when importing (open PV / reduce load)
pid_output = self._pid.compute(-residual, dt)

# Back-calculation anti-windup (sign contradiction):
# When the integral has wound up so far that it opposes the actual
# grid direction, immediately reset it and use the P-only output for
# this cycle. Threshold of 3×deadband avoids triggering on normal
# near-zero oscillation.
if abs(pid_output) > 3 * self._deadband_w:
wrong_direction = (pid_output < 0 and filtered > self._deadband_w) or (
pid_output > 0 and filtered < -self._deadband_w
)
if wrong_direction:
p_only = self._pid.kp * residual
_LOGGER.info(
"Integrator wind-up corrected: output %.0f W → %.0f W"
" (grid %.1f W)",
pid_output,
p_only,
filtered,
)
self._pid.set_integral(0.0)
pid_output = p_only

# 7b. Numeric correction
_load_sp_before = {
ld.name: self._current_load_setpoints.get(ld.name, ld.setpoint_min)
for ld in loads_active
if not ld.is_switch
}
_load_w_per_unit = {
ld.name: ld.w_per_unit for ld in loads_active if not ld.is_switch
}

# Saturation anti-windup: freeze integrator when actuators cannot
# absorb the requested correction (settling timers or at physical
# limits). Prevents further wind-up while the system is saturated.
if abs(pid_output) > 1.0 and abs(final_remaining) >= 0.95 * abs(pid_output):
self._pid.freeze_integrator()

# Actual watts absorbed by numeric loads in this cycle
load_absorbed_w = sum(
(self._current_load_setpoints.get(name, before) - before)
* _load_w_per_unit[name]
for name, before in _load_sp_before.items()
if pid_output < 0:
# Exporting surplus → increase loads, then curtail arrays
after_loads = await self._distribute_to_numeric_loads(
pid_output, now, loads_active
)

# 7c. Switch loads: feedforward of numeric load AND array changes
# so switch decisions do not double-take corrections made this cycle
residual = await self._apply_load_switch_hysteresis(
residual + load_absorbed_w - array_absorbed_w, now, loads
final_remaining = await self._distribute_to_numeric_arrays(
after_loads, now, arrays_active
)
array_absorbed_w = after_loads - final_remaining
else:
# Importing deficit → open arrays, then reduce loads
after_arrays = await self._distribute_to_numeric_arrays(
pid_output, now, arrays_active
)
array_absorbed_w = pid_output - after_arrays
final_remaining = await self._distribute_to_numeric_loads(
after_arrays, now, loads_active
)

# Saturation anti-windup: freeze integrator when actuators cannot
# absorb the requested correction (settling timers or at physical
# limits). Prevents further wind-up while the system is saturated.
if abs(pid_output) > 1.0 and abs(final_remaining) >= 0.95 * abs(pid_output):
self._pid.freeze_integrator()

# Actual watts absorbed by numeric loads in this cycle
load_absorbed_w = sum(
(self._current_load_setpoints.get(name, before) - before)
* _load_w_per_unit[name]
for name, before in _load_sp_before.items()
)

# 7c. Switch loads: feedforward of numeric load AND array changes
# so switch decisions do not double-take corrections made this cycle
residual = await self._apply_load_switch_hysteresis(
residual + load_absorbed_w - array_absorbed_w, now, loads_active
)

# 7d. Switch arrays: hysteresis on updated residual
residual = await self._apply_switch_hysteresis(residual, now, arrays)
# 7d. Switch arrays: hysteresis on updated residual
residual = await self._apply_switch_hysteresis(residual, now, arrays_active)

return ControlCycleResult(
grid_raw_w=grid_raw,
Expand Down
134 changes: 126 additions & 8 deletions tests/test_control_modes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,14 +139,18 @@ async def test_zero_import_importing_is_active(hass):


async def test_zero_import_exporting_is_idle(hass):
"""When exporting (grid < 0), zero_import mode should be idle."""
"""When exporting (grid < 0), zero_import reports the allowed-side status.

The PID still runs so surplus can be routed into controllable loads,
but with no loads configured nothing is actuated.
"""
entry = _make_entry(control_mode="zero_import")
entry.add_to_hass(hass)
coordinator = ZeroGridCoordinator(hass, entry)
_set_grid(hass, 0, 200)
result = await coordinator._async_update_data()
assert result.status == ControllerStatus.IDLE_EXPORT_OK
assert result.pid_output_w == 0.0
assert result.setpoints == {}


async def test_zero_import_within_deadband_is_deadband(hass):
Expand Down Expand Up @@ -221,14 +225,18 @@ async def test_zero_export_exporting_is_active(hass):


async def test_zero_export_importing_is_idle(hass):
"""When importing (grid > 0), zero_export mode should be idle."""
"""When importing (grid > 0), zero_export reports the allowed-side status.

The PID still runs so curtailed PV can be reopened, but with no arrays
configured nothing is actuated.
"""
entry = _make_entry(control_mode="zero_export")
entry.add_to_hass(hass)
coordinator = ZeroGridCoordinator(hass, entry)
_set_grid(hass, 200, 0)
result = await coordinator._async_update_data()
assert result.status == ControllerStatus.IDLE_IMPORT_OK
assert result.pid_output_w == 0.0
assert result.load_setpoints == {}


async def test_zero_export_within_deadband_is_deadband(hass):
Expand Down Expand Up @@ -641,8 +649,12 @@ async def _mock_write_setpoint(array_cfg, value: float) -> None:
assert written == [], f"No array writes expected in idle, got: {written}"


async def test_zero_export_idle_does_not_write_array_setpoints(hass):
"""No array setpoint writes should happen when idle in zero_export mode."""
async def test_zero_export_importing_reopens_curtailed_arrays(hass):
"""Importing in zero_export mode reopens curtailed PV (free energy).

Previously the whole PID was skipped on the allowed side, so a single
export event left the arrays curtailed forever while importing.
"""
entry = _make_entry(
control_mode="zero_export",
subentries_data=(
Expand All @@ -655,7 +667,7 @@ async def test_zero_export_idle_does_not_write_array_setpoints(hass):
),
)
entry.add_to_hass(hass)
_set_grid(hass, 200, 0) # importing → idle
_set_grid(hass, 200, 0) # importing → allowed side
_set_state(hass, "number.inverter_limit", 80)

written: list[tuple[str, float]] = []
Expand All @@ -668,7 +680,113 @@ async def _mock_write_setpoint(array_cfg, value: float) -> None:

result = await coordinator._async_update_data()
assert result.status == ControllerStatus.IDLE_IMPORT_OK
assert written == [], f"No array writes expected in idle, got: {written}"
assert written, "Curtailed PV must be reopened while importing"
assert written[0][1] > 80


async def test_zero_export_importing_does_not_reduce_loads(hass):
"""Importing in zero_export mode must not reduce controllable loads."""
entry = _make_entry(
control_mode="zero_export",
subentries_data=(
{
"subentry_id": "load1",
"subentry_type": LOAD_SUBENTRY_TYPE,
"title": "EV",
"data": {
"load_name": "EV",
"load_type": "numeric",
"setpoint_entity": "number.ev",
"setpoint_min": 0.0,
"setpoint_max": 16.0,
"w_per_unit": 230.0,
"settling_time_s": 0,
"load_priority": 10,
},
},
),
)
entry.add_to_hass(hass)
coordinator = ZeroGridCoordinator(hass, entry)
coordinator._engine._current_load_setpoints["EV"] = 16.0 # charging full
_set_grid(hass, 2000, 0) # importing → allowed in zero_export

with patch.object(
coordinator._actuators, "write_numeric_entity", new=AsyncMock()
) as mock_write:
result = await coordinator._async_update_data()

assert result.status == ControllerStatus.IDLE_IMPORT_OK
mock_write.assert_not_awaited()
assert coordinator._engine._current_load_setpoints["EV"] == 16.0


async def test_zero_import_exporting_routes_surplus_into_loads(hass):
"""Exporting in zero_import mode increases loads instead of doing nothing."""
entry = _make_entry(
control_mode="zero_import",
subentries_data=(
{
"subentry_id": "load1",
"subentry_type": LOAD_SUBENTRY_TYPE,
"title": "EV",
"data": {
"load_name": "EV",
"load_type": "numeric",
"setpoint_entity": "number.ev",
"setpoint_min": 0.0,
"setpoint_max": 16.0,
"w_per_unit": 230.0,
"settling_time_s": 0,
"load_priority": 10,
},
},
),
)
entry.add_to_hass(hass)
coordinator = ZeroGridCoordinator(hass, entry)
coordinator._engine._current_load_setpoints["EV"] = 0.0
_set_state(hass, "number.ev", 0)
_set_grid(hass, 0, 2300) # exporting → allowed in zero_import

with patch.object(
coordinator._actuators, "write_numeric_entity", new=AsyncMock()
) as mock_write:
result = await coordinator._async_update_data()

assert result.status == ControllerStatus.IDLE_EXPORT_OK
mock_write.assert_awaited()
assert coordinator._engine._current_load_setpoints["EV"] > 0.0


async def test_zero_import_exporting_still_never_curtails_arrays(hass):
"""Even with loads absorbing, arrays are never curtailed in zero_import."""
entry = _make_entry(
control_mode="zero_import",
subentries_data=(
{
"subentry_id": "arr1",
"subentry_type": ARRAY_SUBENTRY_TYPE,
"title": "Roof",
"data": _ARRAY_DATA,
},
),
)
entry.add_to_hass(hass)
_set_grid(hass, 0, 3000) # big export, no loads to absorb it
_set_state(hass, "number.inverter_limit", 100)

written: list[tuple[str, float]] = []

async def _mock_write_setpoint(array_cfg, value: float) -> None:
written.append((array_cfg.setpoint_entity, value))

coordinator = ZeroGridCoordinator(hass, entry)
coordinator._actuators.write_setpoint = _mock_write_setpoint

result = await coordinator._async_update_data()
assert result.status == ControllerStatus.IDLE_EXPORT_OK
assert written == [], f"Arrays must never be curtailed in zero_import: {written}"


# ---------------------------------------------------------------------------
Expand Down
Loading