Skip to content

Feature negative dc charging #3795

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

Draft
AlexanderHa98 wants to merge 8 commits into
base: master
Choose a base branch
from
Draft

Feature negative dc charging #3795

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
6b25cc2
Add Bidi to instant-charging
AlexanderHa98 Aug 6, 2026
02820e3
Save Zwischenstand
AlexanderHa98 Aug 10, 2026
bddd3c7
Fix counts in bidi
AlexanderHa98 Aug 11, 2026
04c0a5f
Fix Counter
AlexanderHa98 Aug 11, 2026
5d5463d
Refactor and Improve Bidi charging logic
AlexanderHa98 Aug 12, 2026
c77c482
Remove local debugging setup
AlexanderHa98 Aug 12, 2026
6e52bad
Add tests and fix bidi_charging
AlexanderHa98 Aug 12, 2026
5fe579b
Add Copilot suggestions
AlexanderHa98 Aug 17, 2026
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
135 changes: 113 additions & 22 deletions packages/control/algorithm/bidi_charging.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,18 @@
import logging
from control import data
from control.chargemode import Chargemode
from control.algorithm.chargemodes import CONSIDERED_CHARGE_MODES_BIDI_DISCHARGE
from control.algorithm.filter_chargepoints import get_chargepoints_with_required_current_by_chargemode
from helpermodules.phase_handling import voltages_mean

from control.limiting_value import LoadmanagementLimit
from control.loadmanagement import Loadmanagement

from control.chargepoint.chargepoint import Chargepoint
import control.algorithm.common as common
from typing import List


log = logging.getLogger(__name__)


Expand All @@ -14,33 +23,115 @@ def __init__(self):
def set_bidi(self):
grid_counter = data.data.counter_all_data.get_evu_counter()
log.debug(f"Nullpunktanpassung {grid_counter.data.set.surplus_power_left}W")
zero_point_adjustment = grid_counter
for mode_tuple in CONSIDERED_CHARGE_MODES_BIDI_DISCHARGE:
preferenced_cps = get_chargepoints_with_required_current_by_chargemode(mode_tuple)
if preferenced_cps:
log.info(
f"Mode-Tuple {mode_tuple[0]} - {mode_tuple[1]} - {mode_tuple[2]}, Zähler {grid_counter.num}")

while len(preferenced_cps):
cp = preferenced_cps[0]
zero_point_adjustment = grid_counter.data.set.surplus_power_left / len(preferenced_cps)
log.debug(f"Nullpunktanpassung für LP{cp.num}: verbleibende Leistung {zero_point_adjustment}W")
missing_currents = [zero_point_adjustment / cp.data.get.phases_in_use /
230 for i in range(0, cp.data.get.phases_in_use)]
missing_currents += [0] * (3 - len(missing_currents))
if zero_point_adjustment > 0:
if cp.data.set.charging_ev_data.charge_template.bidi_charging_allowed(
cp.data.control_parameter.current_plan, cp.data.set.charging_ev_data.data.get.soc):
for index in range(0, 3):
missing_currents[index] = min(cp.data.control_parameter.required_current,
missing_currents[index])
else:
log.info(f"LP{cp.num}: Nur bidirektional entladen erlaubt, da SoC-Limit erreicht.")
missing_currents = [0, 0, 0]
else:
for index in range(0, 3):
missing_currents[index] = cp.check_min_max_current(missing_currents[index],
cp.data.get.phases_in_use)
grid_counter.update_surplus_values_left(missing_currents, voltages_mean(cp.data.get.voltages))
cp.data.set.current = missing_currents[0]
log.info(f"LP{cp.num}: Stromstärke {missing_currents}A")
counts = self.get_counts(cp)

cp.data.set.target_current = 0

missing_currents = self.get_missing_currents(preferenced_cps, grid_counter)
log.debug(f"Bidi-LP{cp.num}: missing currents {missing_currents}A")

counters = data.data.counter_all_data.get_counters_to_check(cp.num)
for counter in counters:
Comment on lines +41 to +42

@LKuemmel LKuemmel Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Gibt es einen Grund, dass Du nicht die Methode mode_and_counter_generator verwendest, Die auch min_current,.. verwenden, um über die Zähler zu iterieren?

@AlexanderHa98 AlexanderHa98 Aug 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wenn ich mode_and_counter_generator verwende, schlägt der Test test_cp3_cp4_bidi_discharge fehl.
In dem Test wird geprüft, ob bei mehreren Bidi-CPs (im Zielladen) alle gleichmäßig entladen werden und nicht nur einer und die anderen nicht.

-> Deswegen habe ich das so gelöst. Ich brauche eine Liste mit allen CPs, die sich in diesem Mode befinden, um das Entladen entsprechend aufteilen zu können.

available_currents, limit = Loadmanagement().get_available_currents_bidi(
missing_currents, voltages_mean(cp.data.get.voltages), data.data.counter_data[counter])

if limit.limiting_value is not None:
cp.data.control_parameter.limit = limit

available_for_cp = common.available_current_for_cp(
cp, counts, available_currents, missing_currents, bidi_mode=True)

# Der neue Strom darf nicht höher als der bisher gesetzte Strom sein
current = common.get_current_to_set(
cp.data.set.current, available_for_cp, cp.data.set.target_current)

cp.data.set.current = current
log.info(f"LP{cp.num}: Stromstärke {current}A")

# Ausgabe LIMIT-MSG
self._set_loadmangement_message(current, limit, cp)

common.set_current_counterdiff(cp.data.set.target_current, current, cp, surplus=True)

preferenced_cps.pop(0)

def _set_loadmangement_message(self,
current: float,
limit: LoadmanagementLimit,
chargepoint: Chargepoint) -> None:
# Strom muss an diesem Zähler geändert werden
log.debug(
f"current {current} target {chargepoint.data.set.target_current} set current {chargepoint.data.set.current}"
f" required currents {chargepoint.data.control_parameter.required_currents}")
required_currents = chargepoint.data.control_parameter.required_currents
required_current = min(required_currents) if current < 0 else max(required_currents)
if (limit.message and
# Strom erreicht nicht die vorgegebene Stromstärke
round(current, 2) != round(required_current, 2)):
if current == 0:
chargepoint.set_state_and_log(f"Es kann nicht mit der vorgegebenen Stromstärke geladen/entladen werden"
f"{limit.message}")
elif current < 0:
chargepoint.set_state_and_log(f"Es kann nicht mit der vorgegebenen Stromstärke entladen werden"
f"{limit.message}")
else:
chargepoint.set_state_and_log(f"Es kann nicht mit der vorgegebenen Stromstärke geladen werden"
f"{limit.message}")

def get_counts(self, chargepoint: Chargepoint) -> List[int]:

counts = [0]*3
required_currents = chargepoint.data.control_parameter.required_currents
for i in range(3):
if required_currents[i] != 0:
counts[i] += 1
return counts

def get_missing_currents(self, preferenced_cps: List[Chargepoint], grid_counter) -> List[float]:
cp = preferenced_cps[0]
missing_currents = [0, 0, 0]
if cp.data.control_parameter.chargemode == Chargemode.INSTANT_CHARGING:
# Entladen im Instant-Charging-Modus
if cp.data.set.charging_ev_data.data.get.soc > 0:
# Auto-bat ist nicht leer
Comment on lines +103 to +106

@LKuemmel LKuemmel Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if cp.data.set.charging_ev_data.data.get.soc is None:
raise ValueError(f"LP{cp.num}: Auto-Bat SoC unbekannt, daher keine Entladung möglich.")
if cp.data.set.charging_ev_data.data.get.soc > 0:
# Auto-bat ist nicht leer

Ob das Auto leer ist oder aus anderen Gründen die Entladung verweigert (fast leer, Temperatur, etc.) wird ja dadurch abgefangen, dass nur die tatsächlich entladenen Leistung zum Laden von anderen Fahrzeugen verwendet wird. Dann kann diese Restriktion entfallen.


dc_current = cp.data.set.charging_ev_data.charge_template.data.chargemode.instant_charging.dc_current
if dc_current < 0:
# Phasen in use berücksichtigen
missing_currents = [dc_current for i in range(0, cp.data.get.phases_in_use)]
missing_currents += [0] * (3 - len(missing_currents))

for index in range(0, 3):
missing_currents[index] = cp.check_min_max_current(
missing_currents[index], cp.data.get.phases_in_use)
Comment on lines +108 to +116

@LKuemmel LKuemmel Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Müsste das nicht control_parameter.required_currents entsprechen?

@AlexanderHa98 AlexanderHa98 Aug 21, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ja das ist genau der Punkt, denn wir schonmal besprochen haben:
Durch das self.surplus_controlled.set_required_current_to_max()in calc_current() steht im control_parameter.required_currents immer der Maximalwert.
-> Wir brauchen hier aber den im UI eingestellten Wert.


else:
# Default Nullpunktanpassung
zero_point_adjustment = grid_counter.data.set.surplus_power_left / len(preferenced_cps)
log.debug(f"Nullpunktanpassung für LP{cp.num}: verbleibende Leistung {zero_point_adjustment}W")
missing_currents = [zero_point_adjustment / cp.data.get.phases_in_use /
230 for i in range(0, cp.data.get.phases_in_use)]
missing_currents += [0] * (3 - len(missing_currents))
if zero_point_adjustment > 0:
if cp.data.set.charging_ev_data.charge_template.bidi_charging_allowed(
cp.data.control_parameter.current_plan, cp.data.set.charging_ev_data.data.get.soc):
for index in range(0, 3):
missing_currents[index] = min(cp.data.control_parameter.required_current,
missing_currents[index])
else:
log.info(f"LP{cp.num}: Nur bidirektional entladen erlaubt, da SoC-Limit erreicht.")
missing_currents = [0, 0, 0]
else:
for index in range(0, 3):
missing_currents[index] = cp.check_min_max_current(missing_currents[index],
cp.data.get.phases_in_use)

return missing_currents
48 changes: 25 additions & 23 deletions packages/control/algorithm/chargemodes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,32 +2,34 @@

# Lademodi in absteigender Priorität
# Tupel-Inhalt:(eingestellter Modus, tatsächlich genutzter Modus, Priorität)
CHARGEMODES = ((Chargemode.SCHEDULED_CHARGING, Chargemode.INSTANT_CHARGING, True),
(Chargemode.SCHEDULED_CHARGING, Chargemode.INSTANT_CHARGING, False),
(None, Chargemode.TIME_CHARGING, True),
(None, Chargemode.TIME_CHARGING, False),
(Chargemode.INSTANT_CHARGING, Chargemode.INSTANT_CHARGING, True),
(Chargemode.INSTANT_CHARGING, Chargemode.INSTANT_CHARGING, False),
(Chargemode.ECO_CHARGING, Chargemode.INSTANT_CHARGING, True),
(Chargemode.ECO_CHARGING, Chargemode.INSTANT_CHARGING, False),
(Chargemode.PV_CHARGING, Chargemode.INSTANT_CHARGING, True),
(Chargemode.PV_CHARGING, Chargemode.INSTANT_CHARGING, False),
(Chargemode.SCHEDULED_CHARGING, Chargemode.PV_CHARGING, True),
(Chargemode.SCHEDULED_CHARGING, Chargemode.PV_CHARGING, False),
(Chargemode.ECO_CHARGING, Chargemode.PV_CHARGING, True),
(Chargemode.ECO_CHARGING, Chargemode.PV_CHARGING, False),
(Chargemode.PV_CHARGING, Chargemode.PV_CHARGING, True),
(Chargemode.PV_CHARGING, Chargemode.PV_CHARGING, False),
CHARGEMODES = ((Chargemode.SCHEDULED_CHARGING, Chargemode.INSTANT_CHARGING, True), # 0
(Chargemode.SCHEDULED_CHARGING, Chargemode.INSTANT_CHARGING, False), # 1
(None, Chargemode.TIME_CHARGING, True), # 2
(None, Chargemode.TIME_CHARGING, False), # 3
(Chargemode.INSTANT_CHARGING, Chargemode.INSTANT_CHARGING, True), # 4
(Chargemode.INSTANT_CHARGING, Chargemode.INSTANT_CHARGING, False), # 5
(Chargemode.ECO_CHARGING, Chargemode.INSTANT_CHARGING, True), # 6
(Chargemode.ECO_CHARGING, Chargemode.INSTANT_CHARGING, False), # 7
(Chargemode.PV_CHARGING, Chargemode.INSTANT_CHARGING, True), # 8
(Chargemode.PV_CHARGING, Chargemode.INSTANT_CHARGING, False), # 9
(Chargemode.SCHEDULED_CHARGING, Chargemode.PV_CHARGING, True), # 10
(Chargemode.SCHEDULED_CHARGING, Chargemode.PV_CHARGING, False), # 11
(Chargemode.ECO_CHARGING, Chargemode.PV_CHARGING, True), # 12
(Chargemode.ECO_CHARGING, Chargemode.PV_CHARGING, False), # 13
(Chargemode.PV_CHARGING, Chargemode.PV_CHARGING, True), # 14
(Chargemode.PV_CHARGING, Chargemode.PV_CHARGING, False), # 15
# niedrigere Priorität soll nachrangig geladen, aber zuerst entladen werden
(Chargemode.SCHEDULED_CHARGING, Chargemode.BIDI_CHARGING, False),
(Chargemode.SCHEDULED_CHARGING, Chargemode.BIDI_CHARGING, True),
(None, Chargemode.STOP, True),
(None, Chargemode.STOP, False))
(Chargemode.INSTANT_CHARGING, Chargemode.BIDI_CHARGING, False), # 16
(Chargemode.INSTANT_CHARGING, Chargemode.BIDI_CHARGING, True), # 17
(Chargemode.SCHEDULED_CHARGING, Chargemode.BIDI_CHARGING, False), # 18
(Chargemode.SCHEDULED_CHARGING, Chargemode.BIDI_CHARGING, True), # 19
(None, Chargemode.STOP, True), # 20
(None, Chargemode.STOP, False)) # 21

CONSIDERED_CHARGE_MODES_SURPLUS = CHARGEMODES[0:2] + CHARGEMODES[6:16]
CONSIDERED_CHARGE_MODES_PV_ONLY = CHARGEMODES[10:16]
CONSIDERED_CHARGE_MODES_ADDITIONAL_CURRENT = CHARGEMODES[0:10]
CONSIDERED_CHARGE_MODES_MIN_CURRENT = CHARGEMODES[0:-4]
CONSIDERED_CHARGE_MODES_NO_CURRENT = CHARGEMODES[18:20]
CONSIDERED_CHARGE_MODES_BIDI_DISCHARGE = CHARGEMODES[16:18]
CONSIDERED_CHARGE_MODES_MIN_CURRENT = CHARGEMODES[0:-6]
CONSIDERED_CHARGE_MODES_NO_CURRENT = CHARGEMODES[20:22]
CONSIDERED_CHARGE_MODES_BIDI_DISCHARGE = CHARGEMODES[16:20]
CONSIDERED_CHARGE_MODES_CHARGING = CHARGEMODES[0:16]
42 changes: 33 additions & 9 deletions packages/control/algorithm/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,10 +67,9 @@ def set_current_counterdiff(diff_current: float,
required_currents = chargepoint.data.control_parameter.required_currents
considered_current = consider_less_charging_chargepoint_in_loadmanagement(
chargepoint, current)
# gar nicht ladende Autos?
diff = max(considered_current - diff_current, 0)
diff = considered_current - diff_current
diffs = [diff if required_currents[i] != 0 else 0 for i in range(3)]
if max(diffs) > 0:
if any(diff_value != 0 for diff_value in diffs):
counters = data.data.counter_all_data.get_counters_to_check(chargepoint.num)
for counter in counters:
if surplus:
Expand All @@ -95,10 +94,16 @@ def get_current_to_set(set_current: float, diff: float, prev_current: float) ->
wird."""
new_current = prev_current + diff
if set_current is not None:
if new_current > set_current:
log.debug("Neuer Soll-Strom darf nicht höher als bisher gesetzter sein: "
f"bisher {set_current}A, neuer {new_current}")
return set_current
if diff < 0:
if new_current < set_current:
log.debug("Neuer Soll-Strom darf beim Entladen nicht niedriger als bisher gesetzter sein: "
f"bisher {set_current}A, neuer {new_current}")
return set_current
else:
if new_current > set_current:
log.debug("Neuer Soll-Strom darf nicht höher als bisher gesetzter sein: "
f"bisher {set_current}A, neuer {new_current}")
return set_current
return new_current

# tested
Expand All @@ -107,10 +112,29 @@ def get_current_to_set(set_current: float, diff: float, prev_current: float) ->
def available_current_for_cp(chargepoint: Chargepoint,
counts: List[int],
available_currents: List[float],
missing_currents: List[float]) -> float:
missing_currents: List[float],
bidi_mode: bool = False) -> float:
control_parameter = chargepoint.data.control_parameter
available_current = float("inf")
missing_current_cp = control_parameter.required_current - chargepoint.data.set.target_current

if bidi_mode:
is_discharge = missing_current_cp < 0
available_current = float("-inf") if is_discharge else float("inf")
for i in range(0, 3):
if control_parameter.required_currents[i] == 0 or counts[i] == 0:
continue
phase_available_current = available_currents[i] / counts[i]
if is_discharge:
available_current = max(
max(missing_current_cp, phase_available_current), available_current)
else:
available_current = min(
min(missing_current_cp, phase_available_current), available_current)
if available_current in [float("inf"), float("-inf")]:
available_current = missing_current_cp
return available_current

available_current = float("inf")
for i in range(0, 3):
if (control_parameter.required_currents[i] != 0 and
missing_currents[i] != available_currents[i]):
Expand Down
Loading