Równoważenie obciążenia podczas ładowania pojazdów elektrycznych: algorytmy czasu rzeczywistego umożliwiające inteligentne ładowanie
Jest wtorek wieczorem, godzina 18:30. Tysiące włoskich kierowców wraca po pracy do domu parkują pojazd elektryczny i podłączają kabel do ładowania. Za kilka minut pytanie moc pobliskiego transformatora strzela w górę. Bez inteligentnego zarządzania ten scenariusz – który powtarza się z każdym dniem coraz intensywniej w miarę rozwoju parku EV – prowadzi do przeciążeń sieci, lokalnych przerw w dostawie prądu i kosztów modernizacji infrastruktury rzędu miliardów euro.
W 2025 roku będą one rozpowszechniane dalej 17 milionów pojazdów elektrycznych w Europie, z czego ok 230 000 we Włoszech i 94 230 nowych rejestracji w samym 2025 r. (+46% w porównaniu z 2024 r.). Prognozy na rok 2030 mówią o ponad 50 milionach pojazdów elektrycznych w Europie: średnio każdy pojazd potrzebuje od 7 do 22 kW mocy podczas ładowania. Pomnożone przez miliony użytkowników, którzy doładowują w tych samych godzinach wieczornych dostajemy bezprecedensowy problem ze stabilnością sieci.
Rozwiązaniem nie jest budowa większej sieci: jest to zbyt kosztowne i powolne. Rozwiązaniem jest inteligentne równoważenie obciążenia: algorytmy czasu rzeczywistego dystrybuujące energię dostępne wśród pojazdów ładujących, z poszanowaniem ograniczeń sieciowych, minimalizując koszty energii i maksymalizując satysfakcję użytkowników. W tym artykule przyjrzymy się całemu stosowi technologiczne: od protokołu OCPP 2.0.1 po algorytmy optymalizacyjne, od uczenia się przez wzmacnianie do integracji V2G, z działającym kodem Pythona i włoskim kontekstem regulacyjnym.
Czego dowiesz się w tym artykule
- Krzywe krzywe i niezarządzany wpływ pojazdów elektrycznych na sieć dystrybucyjną
- Taksonomia algorytmów równoważenia obciążenia: Statyczna, Dynamiczna, Predykcyjna
- Inteligentne ładowanie OCPP 2.0.1: SetChargingProfile, ChargerSchedule, StackLevel
- Implementacja Pythona: Equal Share z dynamicznym przeliczaniem w czasie rzeczywistym
- Algorytm oparty na priorytetach z kolejką sterty (SoC, termin, szybkość)
- Uczenie się przez wzmocnienie (PPO) w celu optymalizacji kosztów energii
- Vehicle-to-Grid (V2G): dwukierunkowy, ISO 15118-20, regulacja częstotliwości
- Integracja fotowoltaiczna: nadwyżka energii słonecznej i kierowanie do pojazdów elektrycznych
- Architektura mikrousług: Kontroler ładowania, Menedżer energii, Forecaster
- Kontekst włoski: taryfy dwugodzinne, CER z EV, przepisy ARERA
Seria EnergyTech - 10 artykułów
| # | Przedmiot | Państwo |
|---|---|---|
| 1 | Inteligentna sieć i IoT: architektura sieci elektroenergetycznej przyszłości | Opublikowany |
| 2 | Architektura DERMS: agregacja milionów rozproszonych zasobów | Opublikowany |
| 3 | System zarządzania akumulatorami: Algorytmy sterowania dla BESS | Opublikowany |
| 4 | Cyfrowy bliźniak sieci elektroenergetycznej z Pythonem i Pandapower | Opublikowany |
| 5 | Prognozowanie energii odnawialnej: ML dla fotowoltaiki i wiatru | Opublikowany |
| 6 | Równoważenie obciążenia ładowania pojazdów elektrycznych: algorytmy czasu rzeczywistego (tutaj jesteś) | Aktualny |
| 7 | MQTT i InfluxDB do telemetrii energii w czasie rzeczywistym | Już wkrótce |
| 8 | IEC 61850: Komunikacja w podstacji elektrycznej | Już wkrótce |
| 9 | Oprogramowanie do rozliczania emisji dwutlenku węgla: pomiar i redukcja emisji | Już wkrótce |
| 10 | Blockchain do handlu energią P2P w jednostkach CER | Już wkrótce |
Szczytowy problem: krzywa kaczki i niezarządzane pojazdy elektryczne
Aby zrozumieć, dlaczego równoważenie obciążenia pojazdów elektrycznych nie jest opcją, ale koniecznością, musimy zacząć z fizyki sieci elektroenergetycznej i koncepcji, której zarządcy sieci obawiają się na co dzień: the kacze krzywe.
Krzywa Kaczki i jej pogorszenie
Krzywa kacza opisuje kształt krzywej zapotrzebowania netto na energię elektryczną w ciągu typowego dnia w systemie o dużej penetracji fotowoltaiki. Tak to się nazywa bo krzywa przypomina profil kaczki: niski brzuch w środku dnia (kiedy energia słoneczna produkuje obficie, a zapotrzebowanie netto i niskie) oraz wysoki garb wieczorem (kiedy zachodzi słońce, spada produkcja fotowoltaiki i gwałtownie rośnie popyt na cele mieszkaniowe).
Bez pojazdów elektrycznych włoska sieć już codziennie radzi sobie z tym problemem. Wraz z rozwojem pojazdów elektrycznych pozostawione bez zarządzania, problem drastycznie się nasila. Badania MDPI (2025) pokazują, że bez optymalizacji szczytowe zapotrzebowanie wzrasta z bazowego poziomu 22 000 MW do 35 000 MW przy 5 milionach pojazdów elektrycznych. Dzięki inteligentnemu ładowaniu ta sama flota dodaje jedynie 2000–3000 MW rozprowadzanych w godzinach nocnych.
| Scenariusz | Czas szczytu | Dodatkowe obciążenie | Ryzyko sieciowe |
|---|---|---|---|
| Wartość bazowa (bez EV) | 19:00-20:00 | 0 MW | Do opanowania |
| 500 000 niezarządzanych pojazdów elektrycznych | 18.30-19.30 | +3500 MW | Lokalne naprężenia na transformatorach |
| 1,5 mln niezarządzanych pojazdów elektrycznych | 18:00-20:00 | +10 500 MW | Powszechne przeciążenie |
| 5 milionów niezarządzanych pojazdów elektrycznych | 18:00-21:00 | +35 000 MW | Systemowe zaciemnienia |
| 1,5 mln EV z inteligentnym ładowaniem | Dostarczono w godzinach 22:00–6:00 | +2100 MW rozcieńczony | Nieistotny |
Oparta na sztucznej inteligencji optymalizacja profilu ładowania może zmniejszyć szczytowe zapotrzebowanie 16% przy 1,5 miliona pojazdów elektrycznych, o 21% przy 3 milionach i o 34% przy 5 milionach (źródło: MDPI Electronics, 2025). Różnica między scenariuszem zarządzanym i niezarządzanym To nie jest kilka punktów procentowych: to różnica pomiędzy stabilną siecią a jej załamaniem.
Wpływ na transformatory rozdzielcze
Problem nie występuje tylko na poziomie systemu przesyłowego: występuje przede wszystkim na poziomie sieci dystrybucja lokalna. Transformator sąsiedzki o mocy 400 kVA zwykle obsługuje 80–120 rodziny. Jeśli 15 z nich będzie miało pojazd elektryczny, którego ładowanie rozpocznie się łącznie o godzinie 18:30 z mocą 11 kW, dodatkowe obciążenie wynosi 165 kW - prawie 41% mocy znamionowej transformatora, co w przypadku konsumpcji krajowej mógłby już wynosić 60–70%. Wynik: przegrzanie, redukcja okresu użytkowania, a w najgorszych przypadkach otwarcie zabezpieczeń.
Koszt bierności
Według Terna, koszt wzmocnienia włoskiej sieci dystrybucji do wsparcia szacuje się, że przejście na pojazdy elektryczne bez inteligentnego ładowania wynosi ok 10-15 miliardów euro do 2030 r. Ze względu na powszechne inteligentne ładowanie samo przejście wymaga inwestycji infrastruktura kosztuje o 60-70% mniej, po prostu przesuwając obciążenie w godzinach, w których sieć ma dostępną przepustowość.
Taksonomia algorytmów równoważenia obciążenia
Nie ma jednego optymalnego algorytmu równoważenia obciążenia: wybór zależy od rozmiaru instalacji, dostępność danych historycznych, wymagania dotyczące opóźnień i złożoności, którą menedżer jest skłonny utrzymać. Klasyfikujemy podejścia na trzy główne rodziny.
Algorytmy statyczne
Algorytmy statyczne definiują a priori zasady rozdziału mocy, bez dynamicznego dostosowywania się do stanu systemu. Są proste w wykonaniu i debugowanie, idealne dla małych, jednorodnych instalacji.
- Okrągły Robin: każde złącze otrzymuje tę samą maksymalną moc na obrót. Proste, ale nieefektywne: prawie naładowany pojazd elektryczny zużywa tę samą moc, co pusty.
- Równy udział: całkowita dostępna moc jest podzielona równo pomiędzy wszystkie aktywne złącza. Skuteczne w przypadku instalacji jednorodnych, ale nie uwzględnia indywidualne potrzeby.
- Kto pierwszy, ten lepszy: pierwsze podłączone pojazdy elektryczne otrzymują maksymalną moc, kolejne dzielą resztę. Karaj tych, którzy spóźniają się.
Algorytmy dynamiczne
Algorytmy dynamiczne dostosowują rozkład w czasie rzeczywistym na podstawie stanu prąd systemu: liczba podłączonych pojazdów elektrycznych, poziom naładowania (SoC), dostępna moc, priorytety użytkownika.
- Proporcjonalny: moc rozkłada się proporcjonalnie do energii wymagane przez każdego EV. Kto ma najniższą baterię, otrzymuje najwięcej mocy.
- Oparte na priorytetach: każdy EV ma wynik priorytetu obliczony jako wielokrotność czynniki (SoC, termin, subskrypcja). Władza przepływa do wyższych priorytetów.
- Rozmyta logika: zasady językowe, takie jak „jeśli SoC i niski termin E i zamknij WTEDY priorytet jest wysoki.” Dobrze radzi sobie z niepewnością danych wejściowych.
Algorytmy predykcyjne (oparte na uczeniu maszynowym)
Algorytmy predykcyjne wykorzystują modele uczenia maszynowego do przewidywania przyszłych zdarzeń (przyjazdy, odloty, wahania cen energii) i optymalizować harmonogramy z wyprzedzeniem, minimalizując koszty całkowite w dłuższej perspektywie.
- Modelowa kontrola predykcyjna (MPC): rozwiązuje problem optymalizacyjny w przyszłym horyzoncie czasowym (np. 4 godziny) w każdym przedziale kontrolnym (np. 15 min), stosując tylko pierwszy krok i przeformułowując w kolejnym kroku.
- Uczenie się głębokiego wzmacniania (DRL): agent uczy się optymalnej polityki współpraca z symulatorem. W produkcji najczęściej stosowane są PPO i SAC.
- Programowanie stochastyczne: wyraźnie uwzględnia niepewność Przyjazdy i odjazdy pojazdów elektrycznych generujące harmonogramy odporne na najgorsze scenariusze.
| Algorytm | Złożoność | Utajenie | Optymalność | Żądano danych | Idealny przypadek użycia |
|---|---|---|---|---|---|
| Równy udział | Niski | <1 ms | Niski | Nikt | Mieszkaniowe, <10 placówek |
| Oparte na priorytetach | Przeciętny | <10 ms | Średnio-wysoki | SoC, ostateczny termin użytkownika | Biura, hotele, floty |
| RPP | Wysoki | 100 ms-1 s | Wysoki | Ceny energii, prognozy | Hub > 50 gniazd, C&I |
| Głęboki RL (PPO) | Bardzo wysoki | <50ms (wnioskowanie) | Bardzo wysoki | Historia ponad 6 miesięcy | Świetne koncentratory, wbudowane V2G |
Inteligentne ładowanie OCPP 2.0.1: Protokół inteligentnego ładowania
OCPP (Open Charge Point Protocol) i uniwersalny język pomiędzy punktami ładowania (CP) oraz system centralny (CSMS – System Zarządzania Stacją Ładowania). W wersji 2.0.1 Wprowadzenie funkcji inteligentnego ładowania w porównaniu do wersji OCPP 1.6 zostało znacznie ulepszone bardziej szczegółowe i niezawodne zarządzanie profilami opłat.
Mechanizm SetChargingProfile
CSMS wysyła wiadomość do punktu ładowania SetChargingProfileRequest zawierający
a ChargingProfile który określa, w jaki sposób stacja musi dostarczać moc w czasie.
A ChargingProfile i składa się z:
-
Profil ładowaniaCel:
ChargePointMaxProfile(ogranicz wszystko stacja),TxDefaultProfile(domyślnie dla nowych transakcji),TxProfile(specyficzne dla trwającej transakcji). - poziom stosu: liczba całkowita określająca priorytet. W przypadku profili nakładają się, wygrywa ten z najwyższym poziomem stosu.
- Harmonogram ładowania: lista okresów (startPeriod w sekundach, limit w A lub W), które określają krzywą ładowania w czasie.
Sekwencja krytyczna: spadek przed wzrostem
W przypadku wysyłania zaktualizowanych profili do wielu stacji istotne jest, aby najpierw je wysłać polecenia redukcji mocy, a następnie polecenia zwiększenia. Jeśli najpierw zwiększysz, ryzykujesz przekroczenie limitu sieci na krótki okres czasu, powodując zabezpieczeń elektrycznych lub generowanie szczytów zapotrzebowania, co wiąże się z karami umownymi. Zasada ta nie podlega negocjacjom w produkcji.
Implementacja Pythona: Generacja profilu OCPP 2.0.1
"""
OCPP 2.0.1 Smart Charging Profile Generator
Genera ChargingProfile compatibili con la specifica OCPP 2.0.1.
"""
from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum
import json
from datetime import datetime, timezone
class ChargingRateUnit(str, Enum):
WATTS = "W"
AMPERES = "A"
class ChargingProfilePurpose(str, Enum):
CHARGE_POINT_MAX = "ChargePointMaxProfile"
TX_DEFAULT = "TxDefaultProfile"
TX = "TxProfile"
class ChargingProfileKind(str, Enum):
ABSOLUTE = "Absolute"
RECURRING = "Recurring"
RELATIVE = "Relative"
@dataclass
class ChargingSchedulePeriod:
"""Singolo intervallo nella schedule di ricarica."""
start_period: int # secondi dall'inizio della schedule
limit: float # limite in W o A
number_phases: Optional[int] = None
@dataclass
class ChargingSchedule:
"""Schedule completa di un profilo di ricarica."""
id: int
charging_rate_unit: ChargingRateUnit
charging_schedule_period: List[ChargingSchedulePeriod]
duration: Optional[int] = None
start_schedule: Optional[str] = None
min_charging_rate: Optional[float] = None
@dataclass
class ChargingProfile:
"""Profilo di ricarica OCPP 2.0.1 completo."""
id: int
stack_level: int
charging_profile_purpose: ChargingProfilePurpose
charging_profile_kind: ChargingProfileKind
charging_schedule: List[ChargingSchedule]
transaction_id: Optional[str] = None
valid_from: Optional[str] = None
valid_to: Optional[str] = None
def build_equal_share_profile(
profile_id: int,
power_limit_watts: float,
duration_seconds: int = 3600,
stack_level: int = 1
) -> ChargingProfile:
"""
Costruisce un profilo OCPP per distribuzione equa (potenza fissa).
Usato dall'algoritmo Equal Share per applicare il limite calcolato.
"""
schedule_period = ChargingSchedulePeriod(
start_period=0,
limit=power_limit_watts,
number_phases=3
)
schedule = ChargingSchedule(
id=profile_id,
charging_rate_unit=ChargingRateUnit.WATTS,
charging_schedule_period=[schedule_period],
duration=duration_seconds,
start_schedule=datetime.now(timezone.utc).isoformat()
)
return ChargingProfile(
id=profile_id,
stack_level=stack_level,
charging_profile_purpose=ChargingProfilePurpose.TX,
charging_profile_kind=ChargingProfileKind.ABSOLUTE,
charging_schedule=[schedule]
)
def build_time_of_use_profile(
profile_id: int,
schedule_periods: List[tuple],
stack_level: int = 2
) -> ChargingProfile:
"""
Costruisce un profilo Time-of-Use con potenza variabile nel tempo.
Esempio: alta potenza di notte (energia economica), bassa di giorno.
Args:
schedule_periods: lista di (start_period_sec, limit_watts)
"""
periods = [
ChargingSchedulePeriod(start_period=start, limit=limit)
for start, limit in schedule_periods
]
schedule = ChargingSchedule(
id=profile_id,
charging_rate_unit=ChargingRateUnit.WATTS,
charging_schedule_period=periods,
start_schedule=datetime.now(timezone.utc).isoformat()
)
return ChargingProfile(
id=profile_id,
stack_level=stack_level,
charging_profile_purpose=ChargingProfilePurpose.TX_DEFAULT,
charging_profile_kind=ChargingProfileKind.ABSOLUTE,
charging_schedule=[schedule]
)
def profile_to_ocpp_dict(profile: ChargingProfile) -> dict:
"""Serializza il profilo nel formato JSON OCPP 2.0.1."""
return {
"id": profile.id,
"stackLevel": profile.stack_level,
"chargingProfilePurpose": profile.charging_profile_purpose.value,
"chargingProfileKind": profile.charging_profile_kind.value,
"chargingSchedule": [
{
"id": sched.id,
"chargingRateUnit": sched.charging_rate_unit.value,
"chargingSchedulePeriod": [
{
"startPeriod": p.start_period,
"limit": p.limit,
}
for p in sched.charging_schedule_period
],
**({"duration": sched.duration} if sched.duration else {}),
**({"startSchedule": sched.start_schedule} if sched.start_schedule else {}),
}
for sched in profile.charging_schedule
],
}
if __name__ == "__main__":
# Profilo Equal Share: 3.3 kW per 2 ore
profile = build_equal_share_profile(
profile_id=101,
power_limit_watts=3300.0,
duration_seconds=7200
)
print(json.dumps(profile_to_ocpp_dict(profile), indent=2))
# Profilo Time-of-Use: 11 kW da adesso, 3.3 kW dopo 4 ore
tou = build_time_of_use_profile(
profile_id=102,
schedule_periods=[
(0, 11000), # da ora: 11 kW (fascia fuori punta)
(4 * 3600, 3300), # dopo 4 ore: 3.3 kW (fascia F1 inizia)
],
stack_level=2
)
print(json.dumps(profile_to_ocpp_dict(tou), indent=2))Algorytm równego udziału z dynamicznym przeliczaniem
Equal Share to najbardziej rozpowszechniony algorytm równoważenia obciążenia w rzeczywistych instalacjach. Jego siła nie leży w matematycznej elegancji, ale w solidności operacyjnej: to proste wyjaśnić użytkownikom, łatwe do debugowania i wystarczająco skuteczne większość instalacji do 30-40 gniazd. Prawdziwa złożoność tkwi w zarządzaj dynamicznymi przeliczeniami: za każdym razem, gdy pojazd elektryczny się łączy, rozłącza lub kiedy dostępne zmiany mocy dla sygnału DSO, system musi przeliczyć i redystrybuować w milisekundach, przestrzegając krytycznej sekwencji redukcji, a następnie wzrostu.
"""
Equal Share Load Balancer
Distribuzione equa con ricalcolo dinamico event-driven.
Thread-safe per deployment in produzione con OCPP WebSocket server.
"""
import asyncio
import logging
from dataclasses import dataclass
from typing import Dict, Optional, Callable, Awaitable
from enum import Enum
import time
logger = logging.getLogger(__name__)
class ConnectorStatus(str, Enum):
AVAILABLE = "Available"
CHARGING = "Charging"
SUSPENDED_EV = "SuspendedEV"
FINISHING = "Finishing"
UNAVAILABLE = "Unavailable"
FAULTED = "Faulted"
@dataclass
class ConnectorState:
"""Stato di un singolo connettore nella stazione."""
connector_id: str
status: ConnectorStatus = ConnectorStatus.AVAILABLE
transaction_id: Optional[str] = None
allocated_power_w: float = 0.0
max_power_w: float = 22000.0 # max fisico del connettore (22 kW AC trifase)
min_power_w: float = 1380.0 # min per mantenere la carica (6A x 230V)
soc_percent: Optional[float] = None
connected_at: Optional[float] = None
@property
def is_charging(self) -> bool:
return self.status == ConnectorStatus.CHARGING
@property
def session_duration_min(self) -> float:
if self.connected_at is None:
return 0.0
return (time.time() - self.connected_at) / 60.0
@dataclass
class StationConfig:
"""Configurazione della stazione di ricarica."""
station_id: str
max_station_power_w: float
dynamic_limit_w: Optional[float] = None
rebalance_interval_s: float = 30.0
@property
def available_power_w(self) -> float:
if self.dynamic_limit_w is not None:
return min(self.max_station_power_w, self.dynamic_limit_w)
return self.max_station_power_w
ProfileSendCallback = Callable[[str, str, float], Awaitable[bool]]
class EqualShareBalancer:
"""
Load balancer Equal Share thread-safe e asincrono.
Gestisce la distribuzione equa della potenza disponibile
tra tutti i connettori attivamente in carica.
"""
def __init__(
self,
config: StationConfig,
profile_sender: ProfileSendCallback
):
self._config = config
self._connectors: Dict[str, ConnectorState] = {}
self._send_profile = profile_sender
self._lock = asyncio.Lock()
self._last_allocation: Dict[str, float] = {}
async def register_connector(
self,
connector_id: str,
max_power_w: float = 22000.0,
min_power_w: float = 1380.0
) -> None:
async with self._lock:
self._connectors[connector_id] = ConnectorState(
connector_id=connector_id,
max_power_w=max_power_w,
min_power_w=min_power_w
)
logger.info("Connettore %s registrato (max=%.0f W)", connector_id, max_power_w)
async def on_ev_connected(self, connector_id: str, transaction_id: str) -> None:
"""Chiamato quando un EV si connette. Triggera ricalcolo immediato."""
async with self._lock:
if connector_id not in self._connectors:
logger.warning("Connettore sconosciuto: %s", connector_id)
return
self._connectors[connector_id].status = ConnectorStatus.CHARGING
self._connectors[connector_id].transaction_id = transaction_id
self._connectors[connector_id].connected_at = time.time()
await self._rebalance()
async def on_ev_disconnected(self, connector_id: str) -> None:
"""Chiamato quando un EV si disconnette. Libera la potenza e redistribuisce."""
async with self._lock:
if connector_id not in self._connectors:
return
connector = self._connectors[connector_id]
connector.status = ConnectorStatus.AVAILABLE
connector.transaction_id = None
connector.allocated_power_w = 0.0
connector.connected_at = None
await self._rebalance()
async def update_dynamic_limit(self, limit_w: float) -> None:
"""Aggiorna il limite DSO in tempo reale. Triggera ricalcolo immediato."""
async with self._lock:
self._config.dynamic_limit_w = limit_w
logger.info("Limite DSO aggiornato: %.0f W (stazione %s)", limit_w, self._config.station_id)
await self._rebalance()
async def _rebalance(self) -> None:
"""
Calcola la nuova allocazione Equal Share e invia i profili OCPP.
REGOLA CRITICA: invia prima le riduzioni, poi gli aumenti.
"""
async with self._lock:
active = [c for c in self._connectors.values() if c.is_charging]
if not active:
return
available = self._config.available_power_w
n = len(active)
raw_share = available / n
new_allocations: Dict[str, float] = {}
# Prima passata: applica limiti min/max per connettore
constrained = []
free = []
for connector in active:
if raw_share <= connector.min_power_w:
new_allocations[connector.connector_id] = connector.min_power_w
constrained.append(connector)
elif raw_share >= connector.max_power_w:
new_allocations[connector.connector_id] = connector.max_power_w
constrained.append(connector)
else:
free.append(connector)
# Seconda passata: redistribuisce il residuo tra i connettori liberi
if free:
allocated_constrained = sum(
new_allocations[c.connector_id] for c in constrained
)
residual = available - allocated_constrained
share_free = residual / len(free)
for connector in free:
new_allocations[connector.connector_id] = max(
connector.min_power_w,
min(connector.max_power_w, share_free)
)
# Ordina: prima le riduzioni, poi gli aumenti (regola OCPP)
reductions = []
increases = []
for connector_id, new_power in new_allocations.items():
old_power = self._last_allocation.get(connector_id, float('inf'))
if new_power < old_power:
reductions.append((connector_id, new_power))
else:
increases.append((connector_id, new_power))
for connector_id, power in reductions:
success = await self._send_profile(self._config.station_id, connector_id, power)
if success:
self._last_allocation[connector_id] = power
logger.info("Riduzione: %s -> %.0f W", connector_id, power)
for connector_id, power in increases:
success = await self._send_profile(self._config.station_id, connector_id, power)
if success:
self._last_allocation[connector_id] = power
logger.info("Aumento: %s -> %.0f W", connector_id, power)
logger.info(
"Ribilanciamento: %d EV attivi, %.0f W disponibili, stazione %s",
len(active), self._config.available_power_w, self._config.station_id
)
async def start_periodic_rebalance(self) -> None:
"""Avvia il ricalcolo periodico come safety net per deriva dello stato."""
async def _loop():
while True:
await asyncio.sleep(self._config.rebalance_interval_s)
await self._rebalance()
asyncio.create_task(_loop())
def get_status(self) -> dict:
return {
"station_id": self._config.station_id,
"available_power_w": self._config.available_power_w,
"connectors": [
{
"connector_id": c.connector_id,
"status": c.status.value,
"allocated_power_w": c.allocated_power_w,
"soc_percent": c.soc_percent,
"session_duration_min": round(c.session_duration_min, 1)
}
for c in self._connectors.values()
]
}
async def demo_equal_share():
async def mock_send(station_id, connector_id, power_w) -> bool:
print(f" OCPP -> {connector_id}: {power_w:.0f} W")
return True
config = StationConfig(station_id="STATION_001", max_station_power_w=44000.0)
balancer = EqualShareBalancer(config, mock_send)
for i in range(1, 5):
await balancer.register_connector(f"C0{i}", max_power_w=22000.0, min_power_w=1380.0)
print("\n[EV1 connesso]")
await balancer.on_ev_connected("C01", "TX001") # C01: 44000 W
print("\n[EV2 connesso]")
await balancer.on_ev_connected("C02", "TX002") # C01,C02: 22000 W ciascuno
print("\n[EV3 e EV4 connessi]")
await balancer.on_ev_connected("C03", "TX003")
await balancer.on_ev_connected("C04", "TX004") # tutti: 11000 W
print("\n[DSO riduce limite a 22 kW]")
await balancer.update_dynamic_limit(22000.0) # tutti: 5500 W
print("\n[EV2 si disconnette]")
await balancer.on_ev_disconnected("C02") # C01,C03,C04: 7333 W
if __name__ == "__main__":
asyncio.run(demo_equal_share())Algorytm oparty na priorytetach z punktacją wieloczynnikową
Równy udział jest sprawiedliwy, ale nie optymalny z punktu widzenia użytkowników. Samochód elektryczny z akumulatorem 5%, których właściciel musi wyjść za 30 minut, ma znacznie pilniejsze potrzeby w porównaniu z pojazdem o mocy 60% EV, który pozostaje zaparkowany przez 8 godzin. Równoważenie oparte na priorytetach odpowiada na tę potrzebę, obliczając wieloczynnikowy wskaźnik pilności, np dystrybucja władzy proporcjonalnie do priorytetów.
Formuła punktacji priorytetów
Wynik łączy cztery wymiary z konfigurowalnymi wagami:
- Pilność SoC (35%): ile i rozładuj akumulator. Funkcja wklęsła do wzmacniania krytycznych przypadków (np. SoC 5% = bardzo wysoka pilność).
- Presja czasu (35%): jak blisko jest odlot. Funkcja wykładnicza, która rośnie szybko w miarę zbliżania się terminu.
- Poziom użytkownika (15%): poziom abonamentu (podstawowy/premium/priorytet). Utwórz model biznesowy ze zróżnicowanymi umowami SLA.
- Efektywność energetyczna (15%): nagradza tych, którzy rzeczywiście potrafią przejąć władzę. Unikaj przydzielania mocy prawie całkowicie naładowanym akumulatorom.
"""
Priority-Based Load Balancer.
Distribuisce potenza proporzionalmente ai punteggi di urgenza,
con algoritmo iterativo per rispettare i vincoli min/max.
"""
import math
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import time
logger = logging.getLogger(__name__)
@dataclass
class ChargingSession:
"""Sessione con tutti i parametri per il calcolo della priorità."""
connector_id: str
transaction_id: str
soc_percent: float
target_soc_percent: float
battery_capacity_kwh: float
departure_time: float # timestamp unix
max_power_w: float
min_power_w: float
user_tier: int = 1 # 1=base, 2=premium, 3=priority
connected_at: float = field(default_factory=time.time)
@property
def energy_needed_kwh(self) -> float:
delta = max(0.0, self.target_soc_percent - self.soc_percent) / 100.0
return self.battery_capacity_kwh * delta
@property
def time_remaining_h(self) -> float:
return max(0.1, (self.departure_time - time.time()) / 3600.0)
class PriorityScoreCalculator:
"""Calcola punteggio di urgenza normalizzato [0, 100]."""
W_SOC = 0.35
W_TIME = 0.35
W_TIER = 0.15
W_EFFICIENCY = 0.15
def calculate(self, session: ChargingSession) -> float:
# 1. Urgenza SoC: concava, amplifica i SoC bassissimi
soc_ratio = session.soc_percent / 100.0
soc_score = (1.0 - soc_ratio) ** 1.5 * 100
# 2. Pressione temporale: esponenziale, cresce rapido vicino alla deadline
time_score = 100.0 * math.exp(-0.3 * session.time_remaining_h)
# 3. Tier: bonus per abbonamenti premium
tier_score = {1: 33.0, 2: 66.0, 3: 100.0}.get(session.user_tier, 33.0)
# 4. Efficienza: premia chi ha batteria da riempire
eff = min(100.0, (session.energy_needed_kwh / max(0.1, session.battery_capacity_kwh)) * 100)
return round(
self.W_SOC * soc_score +
self.W_TIME * time_score +
self.W_TIER * tier_score +
self.W_EFFICIENCY * eff,
2
)
class PriorityBalancer:
"""
Distribuisce potenza proporzionalmente ai punteggi con vincoli min/max.
Algoritmo iterativo: separa i connettori con vincoli attivi e redistribuisce
il residuo tra quelli liberi, ripetendo fino a convergenza.
"""
def __init__(self, station_id: str, max_station_power_w: float, profile_sender):
self._station_id = station_id
self._max_power = max_station_power_w
self._dynamic_limit: Optional[float] = None
self._sessions: Dict[str, ChargingSession] = {}
self._send_profile = profile_sender
self._score_calc = PriorityScoreCalculator()
self._lock = asyncio.Lock()
@property
def available_power_w(self) -> float:
if self._dynamic_limit is not None:
return min(self._max_power, self._dynamic_limit)
return self._max_power
async def add_session(self, session: ChargingSession) -> None:
async with self._lock:
self._sessions[session.connector_id] = session
await self._rebalance()
async def remove_session(self, connector_id: str) -> None:
async with self._lock:
self._sessions.pop(connector_id, None)
await self._rebalance()
async def update_soc(self, connector_id: str, soc_percent: float) -> None:
async with self._lock:
if connector_id in self._sessions:
self._sessions[connector_id].soc_percent = soc_percent
await self._rebalance()
def _compute_allocations(self) -> Dict[str, float]:
"""Algoritmo iterativo per allocazione con vincoli min/max."""
if not self._sessions:
return {}
sessions = list(self._sessions.values())
scores = {s.connector_id: self._score_calc.calculate(s) for s in sessions}
allocations: Dict[str, float] = {}
remaining_power = self.available_power_w
remaining = sessions[:]
for _ in range(10): # max 10 iterazioni per convergenza
if not remaining:
break
total_score = sum(scores[s.connector_id] for s in remaining)
if total_score == 0:
share = remaining_power / len(remaining)
for s in remaining:
allocations[s.connector_id] = max(s.min_power_w, min(s.max_power_w, share))
break
constrained, free_sessions = [], []
for s in remaining:
ratio = scores[s.connector_id] / total_score
proposed = remaining_power * ratio
if proposed <= s.min_power_w:
allocations[s.connector_id] = s.min_power_w
constrained.append(s)
elif proposed >= s.max_power_w:
allocations[s.connector_id] = s.max_power_w
constrained.append(s)
else:
free_sessions.append(s)
if not constrained:
# Nessun vincolo: allocazione finale proporzionale
for s in remaining:
ratio = scores[s.connector_id] / total_score
allocations[s.connector_id] = remaining_power * ratio
break
for s in constrained:
remaining_power -= allocations[s.connector_id]
remaining = free_sessions
return allocations
async def _rebalance(self) -> None:
async with self._lock:
allocs = self._compute_allocations()
# Stampa riepilogo per logging
for cid, power in sorted(allocs.items()):
s = self._sessions.get(cid)
score = self._score_calc.calculate(s) if s else 0
logger.info(
"Priority: %s score=%.1f -> %.0f W (SoC=%.0f%%, %.1fh)",
cid, score, power,
s.soc_percent if s else 0,
s.time_remaining_h if s else 0
)
await self._send_profile(self._station_id, cid, power)
# Demo
async def priority_demo():
async def mock_send(station_id, cid, power_w) -> bool:
print(f" SET {cid}: {power_w:.0f} W")
return True
balancer = PriorityBalancer("HUB_01", 44000.0, mock_send)
now = time.time()
sessions = [
# Urgente: 5% SoC, parte tra 45 min
ChargingSession("C1", "T1", 5.0, 80.0, 77.0, now + 45*60, 22000.0, 1380.0, user_tier=2),
# Normale: 60% SoC, parte tra 8 ore
ChargingSession("C2", "T2", 60.0, 80.0, 40.0, now + 8*3600, 11000.0, 1380.0, user_tier=1),
# Priority tier: 30% SoC, parte tra 2 ore
ChargingSession("C3", "T3", 30.0, 90.0, 100.0, now + 2*3600, 22000.0, 1380.0, user_tier=3),
]
print("\n--- Sessioni aggiunte (44 kW disponibili) ---")
for s in sessions:
await balancer.add_session(s)
print("\n--- SoC C1 sale al 40% (meno urgente) ---")
await balancer.update_soc("C1", 40.0)
if __name__ == "__main__":
asyncio.run(priority_demo())Uczenie się przez wzmacnianie minimalizacji kosztów energii
Poprzednie algorytmy reagują na bieżący stan. Uczenie się przez wzmacnianie wykracza poza: agent uczy się poprzez miliony symulacji optymalnej polityki, którą minimalizuje całkowity koszt energii, biorąc pod uwagę zmienne ceny w czasie, prognozy dostaw i odejście od pojazdów elektrycznych i ograniczenia sieciowe. W produkcji w 2025 roku główni operatorzy koncentratory ładowania wykorzystują warianty Bliższa optymalizacja polityki (PPO) dla obniżyć koszty o 15-25% w porównaniu do algorytmów reaktywnych (źródło: ScienceDirect, 2025).
Formułowanie jako proces decyzyjny Markowa
- Przestrzeń stanu: dla każdego złącza (SoC, pozostały czas, aktualna moc) plus stan sieci (dostępna moc, cena energii, pora dnia, prognoza PV).
- Przestrzeń akcji: wektor ciągły znormalizowanych poziomów mocy [0,1] dla każdego złącza.
- Nagrody: minimalizuje koszty energii, karze pojazdy elektryczne, które nie osiągają docelowy SoC w terminie, karze naruszenia limitów sieci.
"""
Ambiente Gymnasium per EV Charging RL.
Compatibile con Stable-Baselines3 (PPO, SAC).
Ogni step = 15 minuti. Un episodio = 24 ore (96 step).
Installazione: pip install gymnasium stable-baselines3 numpy
"""
import numpy as np
import gymnasium as gym
from gymnasium import spaces
from dataclasses import dataclass
from typing import List, Optional, Tuple
@dataclass
class EVSession:
soc: float # [0, 1] corrente
target_soc: float # [0, 1] obiettivo
battery_kwh: float
time_remaining_h: float
max_power_kw: float
min_power_kw: float
class EVChargingEnv(gym.Env):
"""
Ambiente Gymnasium per scheduling EV charging.
Obiettivo: minimizzare costo energia rispettando deadline SoC.
"""
metadata = {"render_modes": ["human"]}
def __init__(
self,
n_connectors: int = 4,
max_station_power_kw: float = 44.0,
episode_steps: int = 96,
electricity_prices: Optional[np.ndarray] = None
):
super().__init__()
self.n = n_connectors
self.max_power = max_station_power_kw
self.episode_steps = episode_steps
self.prices = electricity_prices if electricity_prices is not None \
else self._italian_biorario_prices()
# Action: potenza normalizzata [0,1] per ogni connettore
self.action_space = spaces.Box(
low=0.0, high=1.0, shape=(n_connectors,), dtype=np.float32
)
# Observation: 5 feature per connettore + 3 globali
obs_dim = n_connectors * 5 + 3
self.observation_space = spaces.Box(
low=-1.0, high=1.0, shape=(obs_dim,), dtype=np.float32
)
self.sessions: List[Optional[EVSession]] = [None] * n_connectors
self.step_count = 0
def _italian_biorario_prices(self) -> np.ndarray:
"""Prezzi biorari italiani simulati: F1 (08-19) = 0.28, F2/F3 = 0.18 EUR/kWh."""
prices = np.zeros(96)
for i in range(96):
hour = (i * 15) // 60
prices[i] = 0.28 if 8 <= hour < 19 else 0.18
# Variabilità mercato spot
return (prices + np.random.normal(0, 0.02, 96)).clip(0.10, 0.40)
def _get_obs(self) -> np.ndarray:
obs = []
for s in self.sessions:
if s is not None:
obs.extend([
s.soc,
s.target_soc,
min(1.0, s.time_remaining_h / 12.0),
1.0,
s.max_power_kw / self.max_power
])
else:
obs.extend([0.0, 0.0, 0.0, 0.0, 0.0])
step_idx = self.step_count % len(self.prices)
hour = (self.step_count * 15 // 60) % 24
obs.extend([
self.prices[step_idx] / 0.40,
np.sin(2 * np.pi * hour / 24),
np.cos(2 * np.pi * hour / 24)
])
return np.array(obs, dtype=np.float32)
def step(self, action: np.ndarray) -> Tuple:
dt_h = 15 / 60
price = self.prices[self.step_count % len(self.prices)]
# Converti azioni in potenze reali con vincoli
actual_powers = np.zeros(self.n)
for i, s in enumerate(self.sessions):
if s is not None:
power = action[i] * s.max_power_kw
actual_powers[i] = np.clip(power, s.min_power_kw, s.max_power_kw)
# Rispetta limite stazione
total = actual_powers.sum()
if total > self.max_power:
actual_powers *= self.max_power / total
# Calcola reward
energy_kwh = actual_powers.sum() * dt_h
reward = -energy_kwh * price # costo negativo (vogliamo minimizzare)
# Aggiorna SoC e penalita deadline
for i, s in enumerate(self.sessions):
if s is None:
continue
delta_soc = (actual_powers[i] * dt_h * 0.92) / s.battery_kwh
s.soc = min(1.0, s.soc + delta_soc)
s.time_remaining_h -= dt_h
if s.time_remaining_h <= 0:
gap = max(0, s.target_soc - s.soc)
reward -= gap * 50.0 # penalita pesante per deadline mancata
# Penalita violazione rete
if total > self.max_power * 1.01:
reward -= (total - self.max_power) * 5.0
# Bonus ricarica off-peak
hour = (self.step_count * 15 // 60) % 24
if not (8 <= hour < 19) and total > 0:
reward += total * 0.015
self.step_count += 1
obs = self._get_obs()
terminated = self.step_count >= self.episode_steps
return obs, float(reward), terminated, False, {"price": price, "total_kw": total}
def reset(self, seed=None, options=None) -> Tuple:
super().reset(seed=seed)
self.step_count = 0
self._spawn_random_sessions()
return self._get_obs(), {}
def _spawn_random_sessions(self) -> None:
"""Genera sessioni EV casuali realistiche per il training."""
for i in range(self.n):
if np.random.random() > 0.3:
self.sessions[i] = EVSession(
soc=float(np.random.uniform(0.05, 0.7)),
target_soc=float(np.random.uniform(0.7, 0.95)),
battery_kwh=float(np.random.choice([40.0, 60.0, 77.0, 100.0])),
time_remaining_h=float(np.random.uniform(1.0, 10.0)),
max_power_kw=float(np.random.choice([7.4, 11.0, 22.0])),
min_power_kw=1.38
)
else:
self.sessions[i] = None
def train_ppo_agent(total_timesteps: int = 500_000):
"""
Addestra un agente PPO sull'ambiente di EV charging.
Richiede: pip install stable-baselines3
"""
try:
from stable_baselines3 import PPO
from stable_baselines3.common.env_checker import check_env
env = EVChargingEnv(n_connectors=4, max_station_power_kw=44.0)
check_env(env)
model = PPO(
"MlpPolicy", env,
learning_rate=3e-4,
n_steps=2048,
batch_size=64,
n_epochs=10,
gamma=0.99,
gae_lambda=0.95,
clip_range=0.2,
verbose=1,
)
model.learn(total_timesteps=total_timesteps, progress_bar=True)
model.save("./models/ppo_ev_charging")
print("Agente PPO salvato in ./models/ppo_ev_charging")
return model
except ImportError:
print("Installa: pip install stable-baselines3")
return NoneVehicle-to-Grid (V2G): Pojazd jako zasób sieciowy
V2G stanowi jakościowy skok w paradygmacie pojazdów elektrycznych: pojazd nie jest już sam konsumentem energii, ale staje się zasób dwukierunkowy. Kiedy cena energii jest wysoka lub sieć jest poddana obciążeniom, akumulatory Pojazdy elektryczne mogą przesyłać energię z powrotem do sieci, zarabiając pieniądze dla właściciela i stabilizację systemu elektroenergetycznego.
Standard V2G w 2025 roku
| Standard | Miotły | Wiadomości V2G | Status |
|---|---|---|---|
| ISO 15118-2 | Komunikacja EV-EVSE AC/DC | Podłącz i ładuj, inteligentne ładowanie | W użyciu |
| ISO 15118-20 | Pełna wersja V2G drugiej generacji | Dwukierunkowa integracja BPT, DER, planowanie dynamiczne | Przyjęcie 2025-2026 |
| OCPP 2.0.1 | Komunikacja CP-CSMS | Integracja z ISO 15118, profile ładowania V2G | W użyciu |
| Rejestracja AFIR UE | Regulacja infrastruktury | ISO 15118 obowiązkowe dla nowych systemów od stycznia 2026 r | Obowiązujący |
W Europie największe komercyjne wdrożenie V2G ma Utrecht (Holandia) – ponad 500 Dwukierunkowe Renault w car-sharingu. Każdy pojazd zarabia pieniądze 600-1500 EUR/rok zapewnienie rezerwy ograniczenia częstotliwości (FCR) holenderskiemu OSP TenneT. V2G we Włoszech i nadal głównie eksperymentalne, ale nowe modele Nissan Leaf e2+, Volkswagen ID.4 Pro i niektóre wersje Tesli Model 3 Highland obsługują już dwukierunkowość.
Redukcja wartości szczytowych i charakterystyka częstotliwościowa w przypadku akumulatorów EV
"""
V2G Controller: peak shaving e frequency response.
Gestisce la bidirezionalita delle sessioni EV compatibili V2G.
"""
from dataclasses import dataclass
from enum import Enum
from typing import Dict, List, Optional
import asyncio
import logging
logger = logging.getLogger(__name__)
class ChargingDirection(str, Enum):
G2V = "G2V" # Grid-to-Vehicle (ricarica normale)
V2G = "V2G" # Vehicle-to-Grid (scarica)
IDLE = "IDLE"
@dataclass
class V2GSession:
"""Sessione V2G con capacità bidirezionale."""
connector_id: str
transaction_id: str
soc_percent: float
battery_capacity_kwh: float
max_charge_kw: float # potenza max di ricarica G2V
max_discharge_kw: float # potenza max di scarica V2G
min_soc_percent: float = 20.0 # SoC minimo garantito (non scarica sotto)
departure_soc_target: float = 80.0
@property
def available_discharge_kwh(self) -> float:
usable = max(0.0, self.soc_percent - self.min_soc_percent) / 100.0
return self.battery_capacity_kwh * usable
class V2GController:
"""Controller V2G per peak shaving e frequency response."""
FREQ_NOMINAL = 50.0
FREQ_DEADBAND = 0.05 # +/- 50 mHz: zona morta
FREQ_FULL_ACTIVATION = 0.5 # +/- 500 mHz: attivazione completa FCR
def __init__(
self,
station_id: str,
max_export_kw: float,
profile_sender
):
self._station_id = station_id
self._max_export = max_export_kw
self._send = profile_sender
self._sessions: Dict[str, V2GSession] = {}
self._lock = asyncio.Lock()
async def add_session(self, session: V2GSession) -> None:
async with self._lock:
self._sessions[session.connector_id] = session
async def peak_shaving(
self,
building_load_kw: float,
peak_threshold_kw: float
) -> dict:
"""
Quando il carico supera la soglia, usa le batterie EV per ridurre l'import.
Restituisce le azioni intraprese per ogni connettore V2G.
"""
if building_load_kw <= peak_threshold_kw:
return {"action": "none", "reason": "under_threshold"}
excess_kw = building_load_kw - peak_threshold_kw
logger.info(
"Peak shaving: %.1f kW > soglia %.1f kW (eccesso %.1f kW)",
building_load_kw, peak_threshold_kw, excess_kw
)
actions = {}
async with self._lock:
sessions = sorted(
self._sessions.values(),
key=lambda s: s.available_discharge_kwh,
reverse=True
)
remaining = excess_kw
for s in sessions:
if remaining <= 0:
break
if s.available_discharge_kwh < 1.0:
continue
discharge = min(s.max_discharge_kw, remaining)
actions[s.connector_id] = {
"direction": ChargingDirection.V2G.value,
"power_kw": discharge
}
# Potenza negativa = V2G discharge in OCPP
await self._send(self._station_id, s.connector_id, -discharge * 1000)
remaining -= discharge
logger.info("V2G: %s scarica %.1f kW (peak shaving)", s.connector_id, discharge)
return {"action": "peak_shaving", "actions": actions, "residual_excess_kw": remaining}
async def frequency_response(self, freq_hz: float) -> dict:
"""
FCR (Frequency Containment Reserve).
Risposta lineare alla deviazione di frequenza, entro 500ms.
Frequenza bassa (< 50 Hz) -> V2G discharge.
Frequenza alta (> 50 Hz) -> aumento ricarica G2V.
"""
deviation = freq_hz - self.FREQ_NOMINAL
abs_dev = abs(deviation)
if abs_dev < self.FREQ_DEADBAND:
return {"action": "none", "frequency_hz": freq_hz}
activation = min(1.0,
(abs_dev - self.FREQ_DEADBAND) / (self.FREQ_FULL_ACTIVATION - self.FREQ_DEADBAND)
)
actions = {}
async with self._lock:
sessions = list(self._sessions.values())
for s in sessions:
if deviation < 0 and s.available_discharge_kwh > 0.5:
# Frequenza bassa: scarica per supportare la rete
power_kw = s.max_discharge_kw * activation
actions[s.connector_id] = {"direction": "V2G", "power_kw": power_kw}
await self._send(self._station_id, s.connector_id, -power_kw * 1000)
elif deviation > 0 and s.soc_percent < 90:
# Frequenza alta: aumenta ricarica per assorbire eccesso
power_kw = s.max_charge_kw * activation
actions[s.connector_id] = {"direction": "G2V", "power_kw": power_kw}
await self._send(self._station_id, s.connector_id, power_kw * 1000)
return {
"frequency_hz": freq_hz,
"deviation_hz": deviation,
"activation_factor": activation,
"actions": actions
}Integracja fotowoltaiczna: nadwyżka energii słonecznej na rzecz pojazdów elektrycznych
Integracja systemu fotowoltaicznego ze stacją ładowania pojazdów elektrycznych jest jednym z przypadków użycia Szybszy zwrot z inwestycji w zarządzanie energią. Kiedy fotowoltaika produkuje więcej niż co zużywa budynek, nadwyżka jest kierowana do pojazdów elektrycznych, a nie zasilana sieciowej (z wynagrodzeniem ~0,07-0,10 EUR/kWh GSE) – pozwalające na wykorzystanie energii w przeciwnym razie pobieranie energii z sieci kosztowałoby 0,28 EUR/kWh. Oszczędności netto tj 0,18-0,21 EUR/kWh za każdą kWh zużytą na własne potrzeby w pojazdach elektrycznych.
| Strategia | Logika | Korzyść | Ograniczenie |
|---|---|---|---|
| Tylko w nadmiarze | Ładuj EV TYLKO nadwyżką PV | Maksymalne zużycie własne, zerowe koszty sieci | Zmienna moc, powolne ładowanie |
| Najpierw Solar | Priorytet dla energii słonecznej, nienaruszony z sieci | Stabilne ładowanie z maksimum słonecznym | Część energii z sieci |
| Zielony maksymalizator | Optymalizuj na horyzoncie 4-godzinnym dzięki prognozie pogody | Maksymalizuj % energii odnawialnej | Wymaga dokładnego prognozowania |
| Optymalizator kosztów | Energia słoneczna + ceny spot + V2G łącznie | Absolutne minimum kosztów | Wysoka złożoność algorytmiczna |
"""
Solar-EV Integration Controller.
Implementa strategie Excess-Only e Solar-First
con isteresi per evitare oscillazioni frequenti.
"""
from dataclasses import dataclass
from typing import Optional
import asyncio
import logging
logger = logging.getLogger(__name__)
@dataclass
class PowerState:
"""Stato istantaneo del sistema energetico."""
pv_production_kw: float
building_load_kw: float # consumo edificio escluso EV
electricity_price_eur: float = 0.25
feed_in_tariff_eur: float = 0.07
@property
def net_surplus_kw(self) -> float:
return max(0.0, self.pv_production_kw - self.building_load_kw)
@property
def savings_per_kwh(self) -> float:
"""Risparmio per kWh autoconsumato invece di cedere+comprare."""
return self.electricity_price_eur - self.feed_in_tariff_eur
class SolarEVController:
"""
Controlla la ricarica EV in funzione del surplus fotovoltaico.
Supporta Excess-Only e Solar-First con isteresi configurabile.
"""
HYSTERESIS_KW = 0.3 # evita comandi continui per piccole oscillazioni
def __init__(
self,
balancer,
strategy: str = "solar_first",
min_ev_power_kw: float = 1.38,
max_grid_supplement_kw: float = 5.0
):
self._balancer = balancer
self._strategy = strategy
self._min_ev_power = min_ev_power_kw
self._max_grid_supp = max_grid_supplement_kw
self._last_limit_kw = 0.0
async def update(self, power_state: PowerState, n_active_evs: int) -> dict:
"""
Ricalcola e applica il limite di potenza EV.
Chiamato ogni 5-15 secondi dal loop di controllo.
"""
if n_active_evs == 0:
return {"action": "no_ev", "allocated_kw": 0}
if self._strategy == "excess_only":
target_kw = self._excess_only(power_state, n_active_evs)
else:
target_kw = self._solar_first(power_state, n_active_evs)
# Applica isteresi
if abs(target_kw - self._last_limit_kw) > self.HYSTERESIS_KW:
await self._balancer.update_dynamic_limit(target_kw * 1000)
self._last_limit_kw = target_kw
logger.info("Solar routing [%s]: %.1f kW -> EV", self._strategy, target_kw)
hourly_savings = target_kw * power_state.savings_per_kwh
return {
"strategy": self._strategy,
"pv_kw": power_state.pv_production_kw,
"surplus_kw": power_state.net_surplus_kw,
"allocated_ev_kw": target_kw,
"savings_eur_h": round(hourly_savings, 3)
}
def _excess_only(self, ps: PowerState, n_evs: int) -> float:
surplus = ps.net_surplus_kw
min_total = self._min_ev_power * n_evs
return surplus if surplus >= min_total else 0.0
def _solar_first(self, ps: PowerState, n_evs: int) -> float:
surplus = ps.net_surplus_kw
min_total = self._min_ev_power * n_evs
if surplus < min_total:
grid_needed = min_total - surplus
return min_total if grid_needed <= self._max_grid_supp else surplus
return min(surplus + self._max_grid_supp, surplus + self._max_grid_supp)
async def solar_routing_demo():
"""Simula una giornata con produzione FV tipica di aprile in Italia."""
hourly_pv = [0, 0, 0, 0, 0, 0.5, 2, 5, 9, 14, 18, 21,
22, 20, 16, 10, 6, 2, 0.5, 0, 0, 0, 0, 0]
print("Ora | PV kW | Bldg kW | Surplus | EV kW | Risparmio €/h")
print("-" * 62)
for h in range(24):
pv = hourly_pv[h]
bldg = 5.0 + (2.0 if 8 <= h < 18 else 0)
surplus = max(0, pv - bldg)
price = 0.28 if 8 <= h < 19 else 0.18
feed_in = 0.07
ev_alloc = min(surplus, 11.0)
savings = ev_alloc * (price - feed_in)
print(
f"{h:02d}:00 | {pv:5.1f} | {bldg:7.1f} | "
f"{surplus:7.1f} | {ev_alloc:5.1f} | {savings:.3f}"
)
if __name__ == "__main__":
asyncio.run(solar_routing_demo())Architektura mikrousług kontrolera ładowania
Produkcyjny system równoważenia obciążenia pojazdów elektrycznych i rozproszona architektura wyspecjalizowane komponenty komunikujące się poprzez zdarzenia. Oto główne usługi:
| Praca | Odpowiedzialność | Półki na książki | stwardnienie zanikowe boczne (ALS). |
|---|---|---|---|
| Brama OCP | Serwer WebSocket, tłumaczenie OCPP 1.6/2.0.1 | Python ocpp, asyncio | 99,9%, <100 ms |
| Kontroler ładowania | Algorytm równoważenia obciążenia, SetChargingProfile | FastAPI, Redis | 99,9%, <500 ms |
| Menedżer ds. Energii | Zmierz PV/BESS/sieć, oblicz dostępność | Modbus TCP, MQTT, Python | 99,5%, odpytywanie 1s |
| Prognozy | Prognoza przyjazdu/wyjazdu, ceny, FV | LightGBM/LSTM, FastAPI | 99%, aktualizacja 15min |
| Brama OSD | Sygnały OpenADR, ograniczenia dynamiczne | Python OpenADR 2.0b | 99%, <2 s |
| Rozliczenia | Pomiary, rozliczenia, sesje | FastAPI, PostgreSQL | 99,5% |
Interfejs API kontrolera ładowania z FastAPI
"""
Charge Controller Service - FastAPI REST API.
Endpoints per load balancing, monitoring e configurazione.
"""
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
from datetime import datetime, timezone
import logging
logger = logging.getLogger(__name__)
app = FastAPI(
title="EV Charge Controller API",
version="2.1.0",
description="Load balancing real-time per stazioni di ricarica EV"
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST", "PUT"],
allow_headers=["*"]
)
# -------------------------------------------------------
# Modelli Pydantic
# -------------------------------------------------------
class ConnectorInfo(BaseModel):
connector_id: str
status: str
allocated_power_w: float
soc_percent: Optional[float]
session_duration_min: float
class StationStatus(BaseModel):
station_id: str
available_power_w: float
algorithm: str
connectors: List[ConnectorInfo]
last_updated: str
class EVConnectedEvent(BaseModel):
connector_id: str
transaction_id: str
initial_soc: Optional[float] = None
battery_capacity_kwh: Optional[float] = None
departure_timestamp: Optional[int] = None
user_tier: int = Field(default=1, ge=1, le=3)
class PowerLimitUpdate(BaseModel):
limit_watts: float = Field(gt=0, le=200_000)
source: str = "manual" # "manual" | "dso" | "solar" | "openadr"
valid_until_ts: Optional[int] = None
class AlgorithmSwitch(BaseModel):
algorithm: str = Field(pattern="^(equal_share|priority|rl_ppo)$")
params: Dict = {}
# In-memory registry (in produzione: Redis)
_station_registry: Dict = {}
_algorithm_map: Dict[str, str] = {}
# -------------------------------------------------------
# Endpoints
# -------------------------------------------------------
@app.get("/health")
async def health():
return {
"status": "healthy",
"ts": datetime.now(timezone.utc).isoformat(),
"version": "2.1.0"
}
@app.get("/stations/{station_id}/status", response_model=StationStatus)
async def get_status(station_id: str):
if station_id not in _station_registry:
raise HTTPException(404, f"Stazione {station_id} non trovata")
balancer = _station_registry[station_id]
raw = balancer.get_status()
return StationStatus(
station_id=station_id,
available_power_w=raw["available_power_w"],
algorithm=_algorithm_map.get(station_id, "equal_share"),
connectors=[ConnectorInfo(**c) for c in raw["connectors"]],
last_updated=datetime.now(timezone.utc).isoformat()
)
@app.post("/stations/{station_id}/events/connected")
async def ev_connected(station_id: str, event: EVConnectedEvent, bg: BackgroundTasks):
if station_id not in _station_registry:
raise HTTPException(404, "Stazione non trovata")
bg.add_task(
_station_registry[station_id].on_ev_connected,
event.connector_id,
event.transaction_id
)
return {"status": "accepted", "rebalancing": True}
@app.delete("/stations/{station_id}/connectors/{connector_id}/session")
async def ev_disconnected(station_id: str, connector_id: str, bg: BackgroundTasks):
if station_id not in _station_registry:
raise HTTPException(404, "Stazione non trovata")
bg.add_task(_station_registry[station_id].on_ev_disconnected, connector_id)
return {"status": "accepted"}
@app.put("/stations/{station_id}/power-limit")
async def set_power_limit(station_id: str, limit: PowerLimitUpdate, bg: BackgroundTasks):
if station_id not in _station_registry:
raise HTTPException(404, "Stazione non trovata")
bg.add_task(_station_registry[station_id].update_dynamic_limit, limit.limit_watts)
logger.info("Limite potenza: %s -> %.0f W (source=%s)", station_id, limit.limit_watts, limit.source)
return {"status": "accepted", "new_limit_w": limit.limit_watts, "source": limit.source}
@app.put("/stations/{station_id}/algorithm")
async def set_algorithm(station_id: str, config: AlgorithmSwitch):
_algorithm_map[station_id] = config.algorithm
return {"status": "ok", "algorithm": config.algorithm}
@app.get("/stations/{station_id}/metrics")
async def get_metrics(station_id: str):
"""KPI della stazione (in produzione: query InfluxDB/TimescaleDB)."""
return {
"station_id": station_id,
"period": "last_24h",
"charging_satisfaction_rate": 0.94,
"peak_reduction_percent": 31.2,
"energy_cost_savings_eur": 47.80,
"solar_share_percent": 42.3,
"grid_stress_index": 0.23,
"total_energy_kwh": 201.6,
"avg_power_kw": 8.4,
"rebalance_latency_ms_p95": 180
}Metryki i wskaźniki KPI do oceny równoważenia obciążenia
Aby ocenić skuteczność systemu równoważenia obciążenia, potrzebne są wymierne wskaźniki KPI które mierzą zarówno zadowolenie użytkowników, jak i optymalizację zużycia energii.
| KPI | Definicja | Cel | Formuła |
|---|---|---|---|
| Wskaźnik satysfakcji z ładowania (CSR) | % sesji osiągających cele SoC w terminie | >90% | ok_sessions / total_sessions |
| Szczytowa redukcja% | Redukcja szczytowa vs brak zarządzania | >25% | (szczyt_niezarządzany - szczyt_zarządzany) / szczyt_niezarządzany |
| Oszczędności kosztów energii% | Oszczędności na kosztach kWh w porównaniu z natychmiastowym ładowaniem | >15% | (koszt_podstawowy - koszt_zarządzany) / koszt_podstawowy |
| % własnego zużycia energii słonecznej | % energii EV z lokalnej fotowoltaiki | >40% | energia_słoneczna_ew / całkowita_energia_ew |
| Indeks naprężenia siatki (GSI) | Średnie ciśnienie na transformatorze [0-1] | <0,3 | średnia(P_rzeczywista / P_nominal_trafo) |
| Opóźnienie przywracania równowagi P95 | Zastosowano czas od zdarzenia do profilu OCPP | <2s | t_ocpp_conf - t_event_received (p95) |
Kontekst włoski: ARERA, CER z EV i zachętami 2025
Wdrożenie systemów równoważenia obciążenia pojazdów elektrycznych we Włoszech stwarza wyzwania regulacyjne i rynku, które bezpośrednio wpływają na wybory architektoniczne.
Taryfy dwugodzinne ARERA i inteligentne ładowanie
Włoski system taryfowy stwarza naturalną zachętę do ładowania poza szczytem:
- Zespół F1 (poniedziałek-piątek 8:00-19:00): około 0,26-0,31 EUR/kWh. Opaska, której należy unikać podczas ładowania pojazdów elektrycznych.
- Pasmo F2 (pn.-pt. 7-8, 19-23; sob. 7-23): ok. 0,20-0,24 EUR/kWh.
- Pasmo F3 (noc 23:00-7:00; niedziele i święta przez cały dzień): około 0,16-0,19 EUR/kWh. Idealny do ładowania w nocy.
Algorytm czasu użytkowania, który przesuwa 60% ładowania z F1 na F3, zmniejsza roczne koszty opłat dla menedżera floty 18-25%.
Wspólnoty energii odnawialnej (CER) wyposażone w pojazdy elektryczne
Dekret legislacyjny nr 199/2021 i dekret ministerialny MASE z dnia 7 grudnia 2023 r. otworzyły konkretne możliwości w przypadku integracji pojazdów elektrycznych we włoskich jednostkach CER:
- Pojazdy elektryczne ładowane nadwyżką energii fotowoltaicznej społeczności generują zachętę do GSE na udziale wspólna energia (zwrot kosztów + premia do 11 centów/kWh).
- Dzięki V2G akumulatory EV działają jak wirtualny magazyn CER, wypierając energia słoneczna od południa do wieczora (przesunięcie szczytowe).
- GSE mierzy energię współdzieloną MACZA, w tym dwukierunkowe przepływy pojazdów elektrycznych V2G w rachunkowości społeczności.
Włoska infrastruktura pojazdów elektrycznych do 2025 r
Włochy mają ponad 58 000 publicznych punktów ładowania (Motus-E, koniec 2024 r.), z czego 22% szybkiego prądu stałego (w porównaniu z 14% w 2023 r.). W 2025 roku zarejestrowano 94 230 pojazdów BEV (+46%), z udziałem w rynku wynoszącym 6,2%. Według Terny 7% populacji bez inteligentnego ładowania jest zjawiskiem powszechnym Podajnik dystrybucyjny zostanie przeciążony do 2027 roku w najgęstszych obszarach woj Północne Włochy. PNRR przeznaczyło na infrastrukturę ponad 740 mln euro doładowania, którego celem jest utworzenie 21 000 nowych punktów publicznych do 2026 r.
Zwróć uwagę na zachęty
Ulgi podatkowe za inteligentne ładowanie w ramach przejścia 5.0 podlegają okresowo sprawdzane. Zawsze sprawdzaj aktualne warunki na stronie internetowej GSE i CSEA przed planowaniem inwestycji opartych na zachętach. Ceny i warunki dostępu mogą się różnić.
Wnioski i plan wdrożenia
Równoważenie obciążenia pojazdów elektrycznych nie jest problemem algorytmicznej elegancji: jest koniecznością infrastrukturę krytyczną dla transformacji energetycznej. Dane są jednoznaczne: przy 5 milionach niezarządzanych pojazdów elektrycznych szczyt wieczorny wzrasta o 35 000 MW; z mądrym samo ładowanie floty pojazdów elektrycznych staje się elastycznym zasobem, który zmniejsza szczyt o 34%.
Dla zespołu programistów wdrażającego te systemy w środowisku produkcyjnym, plan działania optymalne dla kolejnych faz oraz:
- Faza 1 – Równy udział: proste, solidne, wystarczające do instalacji do 30-40 sztuczek. Najwyższy priorytet: prawidłowo wdrożyć przeliczenie sterowane zdarzeniami z sekwencją zmniejszania, a następnie zwiększania. Szacowany czas: 2-3 tygodnie.
-
Faza 2 – Integracja PV: Natychmiastowy zwrot z inwestycji bez ML. Wymaga jedynie
pomiar nadwyżki energii słonecznej poprzez Modbus/MQTT i integracja z balanserem poprzez
update_dynamic_limit(). Szybkie zwycięstwo ze zwrotem w ciągu 6-18 miesięcy. - Faza 3 – Oparta na priorytetach: dodaj, gdy będziesz mieć aplikację mobilną do zebrania SoC i termin rozpoczęcia. Wzrost CSR (współczynnika satysfakcji z ładowania) jest wymierny i wyróżnia usługę na tle konkurencji.
- Faza 4 – Infrastruktura V2G: projektować architekturę dwukierunkową teraz, nawet jeśli nie masz jeszcze pojazdów elektrycznych V2G. OCPP 2.0.1 z ISO 15118-20 staje się obowiązkowe dla nowych zakładów od 2026 r. (EU AFIR). Modernizacja kosztuje 5-10 razy więcej.
- Faza 5 – RL w produkcji: inwestuj w głęboki RL dopiero po ponad 6 miesiącach jakości danych i przy pomocy dedykowanego zespołu MLOps. Wzrost kosztów jest realny (15-25%) ale złożoność operacyjna jest wysoka.
Powiązane artykuły z serii EnergyTech
- Artykuł 4: Cyfrowy bliźniak sieci – symuluj wpływ pojazdów elektrycznych na sieć przed wdrożeniem
- Artykuł 5: Prognozowanie energii odnawialnej – prognoza produkcji fotowoltaiki dla tras fotowoltaicznych
- Artykuł 7: MQTT i InfluxDB – telemetria parametrów ładowania w czasie rzeczywistym
- Artykuł 2: Architektura DERMS – zintegruj pojazdy EV jako DER z rozproszonym systemem zarządzania
Spostrzeżenia między seriami
- Seria MLOps (art. 306-315): wdrożenie modelu PPO z MLflow, monitorowanie dryftu i przekwalifikowanie
- Seria AI Engineering (art. 316-325): RAG na temat dokumentacji OCPP i ISO 15118 z przedsiębiorstwem LLM
- Data & AI Business Series (art. 267-280): ROI, wskaźniki biznesowe i plan działania w zakresie inwestycji w inteligentne ładowanie
- Seria PostgreSQL AI (art. 356-361): serie czasowe przeładowania danych za pomocą TimescaleDB na PostgreSQL