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 10 commits into
base: master
Choose a base branch
from
Draft

Feature negative dc charging #3795

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 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
6e43fff
Add requested changes
AlexanderHa98 Aug 21, 2026
5695be1
Remove unused import
AlexanderHa98 Aug 21, 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
145 changes: 123 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,17 @@
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 @@ -12,35 +20,128 @@ def __init__(self):
pass

def set_bidi(self):
"""Setzt den verfügbaren Strom für bidirektionales Laden und Entladen.
Dafür gibt es jeweils eine Grenze für das Laden und Entladen
(raw_currents_left und raw_exported_currents_left).
Entladen erhöht die verfügbare Grenze für das Laden, während Laden
die verfügbare Grenze für das Entladen erhöht.
Das Lastmanagement stellt sicher, dass die Summe der Ströme die am
Zähler verfügbare Leistung nicht überschreitet.
"""
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
# CPs nach Modus filtern, dann verfügbare Ströme gegen alle relevanten Counter-Limits begrenzen.
# -> Um bei mehreren Bidi-CPs das Entladen bei der Nullpunktanpassung gleichmäßig zu verteilen

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 +51 to +52

@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

# cp.control_parameter.required_currents enthält duch self.surplus_controlled.set_required_current_to_max()
# in calc_current nur noch die maximalen Ströme
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)

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