EV Şarj Yükü Dengeleme: Akıllı Şarj için Gerçek Zamanlı Algoritmalar
Salı akşamı saat 18.30. Binlerce İtalyan sürücü işten sonra evlerine dönüyor. elektrikli araçlarını park edip şarj kablosunu bağlıyorlar. Birkaç dakika sonra soru Mahalle trafosundaki güç yukarı doğru fırlıyor. Akıllı yönetim olmadan, park büyüdükçe her gün daha yoğun bir şekilde tekrarlanan bu senaryo EV - ağda aşırı yüklenmeye, yerel kesintilere ve altyapı yükseltme maliyetlerine yol açar milyarlarca avro civarında.
2025'te daha da yaygınlaşacaklar Avrupa'da 17 milyon elektrikli araçyaklaşık olarak Yalnızca 2025'te İtalya'da 94.230 yeni kayıtla 230.000 (2024'e göre +%46). 2030 projeksiyonu Avrupa'da 50 milyonun üzerinde elektrikli araçtan bahsediyor: her araç için ortalama Şarj sırasında 7 ile 22 kW arasında güç. Yeniden şarj olan milyonlarca kullanıcıyla çarpıldı Aynı akşam saatlerinde eşi benzeri görülmemiş bir ağ istikrarı sorunuyla karşılaşıyoruz.
Çözüm daha fazla ağ kurmak değil: çok pahalı ve yavaş. Çözüm şu: akıllı yük dengeleme: gücü dağıtan gerçek zamanlı algoritmalar Şarj araçları arasında mevcut, ağ kısıtlamalarına saygı gösteren, enerji maliyetlerini en aza indiren ve kullanıcı memnuniyetini en üst düzeye çıkarmak. Bu yazıda yığının tamamını inceliyoruz teknolojik: OCPP 2.0.1 protokolünden optimizasyon algoritmalarına, takviyeli öğrenmeden Çalışan Python kodu ve İtalyan düzenleyici bağlamıyla V2G entegrasyonuna.
Bu Makalede Neler Öğreneceksiniz?
- Ördek eğrileri ve yönetilmeyen EV'nin dağıtım ağı üzerindeki etkisi
- Yük dengeleme algoritmalarının sınıflandırması: Statik, Dinamik, Tahmine Dayalı
- OCPP 2.0.1 Akıllı Şarj: SetChargingProfile, ChargingSchedule, StackLevel
- Python uygulaması: Gerçek zamanlı dinamik yeniden hesaplamayla Eşit Paylaşım
- Yığın kuyruğuna sahip Öncelik Tabanlı algoritma (SoC, son tarih, oran)
- Enerji maliyeti optimizasyonu için Takviyeli Öğrenme (PPO)
- Araçtan Şebekeye (V2G): çift yönlü, ISO 15118-20, frekans düzenlemesi
- Fotovoltaik entegrasyon: güneş enerjisi fazlası ve EV'ye yönlendirme
- Mikro hizmet mimarisi: Şarj Kontrol Cihazı, Enerji Yöneticisi, Tahminci
- İtalya bağlamı: iki saatlik tarifeler, EV ile CER, ARERA düzenlemeleri
EnergyTech Serisi - 10 Makale
| # | Öğe | Durum |
|---|---|---|
| 1 | Akıllı Şebeke ve Nesnelerin İnterneti: Geleceğin Elektrik Şebekesinin Mimarisi | Yayınlandı |
| 2 | DERMS Mimarisi: Milyonlarca Dağıtılmış Kaynağı Bir araya Getirme | Yayınlandı |
| 3 | Pil Yönetim Sistemi: BESS için Kontrol Algoritmaları | Yayınlandı |
| 4 | Python ve Pandapower ile Elektrik Şebekesinin Dijital İkizi | Yayınlandı |
| 5 | Yenilenebilir Enerji Tahmini: PV ve Rüzgar için ML | Yayınlandı |
| 6 | EV Şarj Yükü Dengeleme: Gerçek Zamanlı Algoritmalar (şu anda buradasınız) | Akım |
| 7 | Gerçek Zamanlı Enerji Telemetrisi için MQTT ve InfluxDB | Yakında gelecek |
| 8 | IEC 61850: Elektrik Trafo Merkezinde İletişim | Yakında gelecek |
| 9 | Karbon Muhasebe Yazılımı: Emisyonların Ölçülmesi ve Azaltılması | Yakında gelecek |
| 10 | CER'lerde P2P Enerji Ticareti için Blockchain | Yakında gelecek |
Zirve Sorunu: Ördek Eğrisi ve Yönetilmeyen EV'ler
EV yük dengelemenin neden bir seçenek değil zorunluluk olduğunu anlamak için elektrik şebekesinin fiziğinden ve şebeke yöneticilerinin her gün korktuğu bir kavramdan: the ördek eğrileri.
Ördek Eğrisi ve kötüleşmesi
Ördek eğrisi net elektrik talebi eğrisinin şeklini tanımlar Yüksek fotovoltaik nüfuza sahip bir sistemde tipik bir gün boyunca. Buna denir çünkü eğri bir ördeğin profilini andırıyor: gün ortasında düşük bir göbek (ne zaman güneş enerjisi bol, net ve düşük talep üretiyor) ve akşamları yüksek bir tümsek (ne zaman güneş batıyor, PV üretimi çöküyor ve konut talebi hızla artıyor).
Elektrikli araçlar olmadan İtalyan şebekesi bu sorunu zaten her gün çözüyor. EV’lerin büyümesiyle Yönetilmeden bırakılırsa sorun büyük ölçüde büyür. MDPI araştırması (2025) şunu gösteriyor: optimizasyonu ile zirve talebi 22.000 MW'tan 5 milyon EV ile 35.000 MW'a çıkıyor. Akıllı şarjla aynı filoya yalnızca gece saatlerinde dağıtılan 2.000-3.000 MW eklenir.
| Senaryo | Yoğun zaman | Ek yük | Ağ riski |
|---|---|---|---|
| Temel (EV yok) | 19.00-20.00 | 0 MW | Yönetilebilir |
| 500.000 yönetilmeyen EV | 18.30-19.30 | +3.500 MW | Transformatörlerde yerel stres |
| 1,5 milyon yönetilmeyen EV | 18:00-20:00 | +10.500 MW | Yaygın aşırı yük |
| 5 milyon yönetilmeyen EV | 18.00-21.00 | +35.000 MW | Sistemik kesintiler |
| Akıllı şarjla 1,5M EV | 22:00-06:00 arası teslim edilir | +2.100 MW seyreltilmiş | İhmal edilebilir |
Şarj profilinin yapay zeka odaklı optimizasyonu, şarj cihazının en yüksek talebini azaltabilir 1,5 milyon EV ile %163 milyon ile %21 ve 5 milyon ile %34 artışla (kaynak: MDPI Electronics, 2025). Yönetilen ve yönetilmeyen senaryo arasındaki fark Bu birkaç yüzde puanı değil: istikrarlı bir ağ ile çöküş arasındaki farktır.
Dağıtım Transformatörlerine Etkisi
Sorun yalnızca iletim sistemi düzeyinde değil: her şeyden önce ağ düzeyinde yerel dağıtım. 400 kVA'lık bir mahalle transformatörü tipik olarak 80-120'ye hizmet eder aileler. Bunlardan 15 tanesinin tamamı 18:30'da 11 kW'ta şarj olmaya başlayan bir EV'ye sahipse, ek yük 165 kW'tır - transformatörün nominal kapasitesinin neredeyse% 41'i; Yurt içi tüketimde zaten yüzde 60-70 civarında olabilir. Sonuç: aşırı ısınma, azalma yararlı ömrün uzatılması ve en kötü durumlarda korumaların açılması.
Hareketsizliğin Maliyeti
Terna'ya göre, İtalyan dağıtım ağını güçlendirmenin maliyeti Akıllı şarj olmadan EV geçişinin şu şekilde olduğu tahmin ediliyor: 10-15 milyar euro 2030 yılına kadar. Akıllı şarjın yaygınlaşmasıyla birlikte geçişin kendisi de yatırım gerektiriyor Sadece yükün çalışma saatleri içerisinde taşınmasıyla altyapı maliyetleri %60-70 daha düşük olur. Ağın kullanılabilir kapasitesi var.
Yük Dengeleme Algoritmalarının Taksonomisi
Tek bir optimum yük dengeleme algoritması yoktur: seçim boyuta bağlıdır Kurulumun özellikleri, geçmiş verilerin kullanılabilirliği, gecikme gereksinimleri ve Yöneticinin sürdürmeye istekli olduğu karmaşıklık. Yaklaşımları üçe ayırıyoruz ana aileler.
Statik Algoritmalar
Statik algoritmalar güç dağıtım kurallarını önceden tanımlar, sistemin durumuna dinamik olarak uyum sağlamadan. Uygulamaları basittir ve hata ayıklama, küçük, homojen kurulumlar için idealdir.
- Yuvarlak Robin: her konektör dönüş başına aynı maksimum gücü alır. Basit ama verimsiz: Neredeyse şarj olmuş bir EV, boş bir EV ile aynı gücü kullanır.
- Eşit Paylaşım: toplam kullanılabilir güç arasında eşit olarak bölünür tüm aktif konektörler. Homojen kurulumlar için etkilidir ancak aşağıdakileri dikkate almaz: bireysel ihtiyaçlar.
- İlk Gelen İlk Hizmet Alır: bağlanan ilk EV'ler maksimum gücü alır, sonrakiler artığı böler. Geç gelenleri cezalandırın.
Dinamik Algoritmalar
Dinamik algoritmalar, duruma göre dağıtımı gerçek zamanlı olarak uyarlar sistem akımı: bağlı EV sayısı, şarj seviyesi (SoC), mevcut güç, kullanıcı öncelikleri.
- Orantılı: güç, enerjiyle orantılı olarak dağıtılır her EV tarafından talep edilir. En düşük bataryaya sahip olan, en fazla gücü alır.
- Önceliğe Dayalı: her EV'nin birden fazla puana göre hesaplanan bir öncelik puanı vardır. faktörler (SoC, son tarih, abonelik). Güç daha yüksek önceliklere akar.
- Bulanık Mantık: "Eğer SoC ve düşük E son tarihi" gibi dil kuralları ve SONRA kapat önceliği yüksektir." Giriş verilerindeki belirsizliği iyi yönetir.
Tahmine Dayalı Algoritmalar (ML tabanlı)
Tahmine dayalı algoritmalar gelecekteki olayları tahmin etmek için makine öğrenimi modellerini kullanır (varışlar, ayrılışlar, enerji fiyatı değişimleri) ve planlamayı optimize edin uzun vadede toplam maliyetleri en aza indirerek ilerleme sağlar.
- Model Tahminli Kontrol (MPC): bir optimizasyon problemini çözer gelecekteki bir zaman diliminde (örneğin 4 saat) her kontrol aralığında (örneğin 15 dakika), yalnızca ilk adımı uygulamak ve bir sonraki adımda yeniden formüle etmek.
- Derin Güçlendirmeli Öğrenme (DRL): bir temsilci en uygun politikayı öğrenir Bir simülatörle etkileşim. PPO ve SAC üretimde en çok kullanılanlardır.
- Stokastik Programlama: belirsizliği açıkça dahil ediyor Elektrikli araçların geliş ve gidişleri, en kötü senaryolara karşı dayanıklı programlar oluşturur.
| Algoritma | Karmaşıklık | Gecikme | Optimallik | Talep edilen veriler | İdeal kullanım durumu |
|---|---|---|---|---|---|
| Eşit Pay | Düşük | <1ms | Düşük | Hiç kimse | Konut, <10 satış noktası |
| Önceliğe Dayalı | Ortalama | <10ms | Orta-Yüksek | SoC, kullanıcı son tarihi | Ofisler, oteller, filolar |
| MPC | Yüksek | 100ms-1s | Yüksek | Enerji fiyatları, tahminler | Hub >50 soket, C&I |
| Derin RL (PPO) | Çok yüksek | <50 ms (çıkarım) | Çok yüksek | Geçmiş 6+ ay | Harika hub'lar, yerleşik V2G |
OCPP 2.0.1 Akıllı Şarj: Akıllı Şarj Protokolü
OCPP (Açık Şarj Noktası Protokolü) ve şarj noktaları (CP) arasındaki evrensel dil ve merkezi sistem (CSMS - Şarj İstasyonu Yönetim Sistemi). 2.0.1 sürümünde, Akıllı Şarj, OCPP 1.6'ya kıyasla büyük ölçüde geliştirildi. Şarj profillerinin daha ayrıntılı ve güvenilir yönetimi.
SetChargingProfile Mekanizması
CSMS şarj noktasına bir mesaj gönderir SetChargingProfileRequest içeren
bir ChargingProfile bu, istasyonun zaman içinde nasıl güç sağlaması gerektiğini tanımlar.
A ChargingProfile ve şunlardan oluşur:
-
şarj ProfiliAmaç:
ChargePointMaxProfile(hepsini sınırla istasyon),TxDefaultProfile(yeni işlemler için varsayılan),TxProfile(devam eden bir işleme özel). - Yığın Düzeyi: önceliği tanımlayan tamsayı. Profiller söz konusu olduğunda örtüşüyorsa, en yüksek stackLevel'e sahip olan kazanır.
- şarjTablosu: dönemlerin listesi (saniye cinsinden startPeriod, Zaman içindeki şarj eğrisini tanımlayan A veya W cinsinden limit.
Kritik Sıra: Artmadan Önce Azalma
Güncellenmiş profilleri birden fazla istasyona gönderirken ilk önce profillerin gönderilmesi önemlidir. güç azaltma komutları ve ardından artırma komutları. Önce sen arttırırsan kısa bir süre için ağ sınırını aşma riskiyle karşı karşıya kalırsınız; elektriksel korumalar veya sözleşmeye bağlı cezalarla sonuçlanan talep zirveleri oluşturma. Bu kural üretimde tartışılamaz.
Python Uygulaması: OCPP Profil Üretimi 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))Dinamik Yeniden Hesaplama ile Eşit Paylaşım algoritması
Eşit Paylaşım, gerçek kurulumlarda en yaygın yük dengeleme algoritmasıdır. Gücü matematiksel zarafetinde değil, operasyonel sağlamlığındadır: basittir kullanıcılara açıklamak, hata ayıklamak kolay ve yeterince etkili Çoğu kurulumda 30-40 sokete kadar. Gerçek karmaşıklık şu dinamik yeniden hesaplamaları yönetin: bir EV her bağlandığında veya bağlantısı kesildiğinde veya Bir DSO sinyali için mevcut güç değişiklikleri, sistemin yeniden hesaplaması ve Kritik dizi azalmaları-sonra-artışlarına saygı göstererek milisaniyeler içinde yeniden dağıtın.
"""
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())Çok Faktörlü Puanlama ile Öncelik Tabanlı Algoritma
Eşit Paylaşım adildir ancak kullanıcıların bakış açısından ideal değildir. Pilli bir EV Sahibi 30 dakika içinde ayrılmak zorunda kalan %5'lik kesimin çok daha acil ihtiyaçları var 8 saat park halinde kalan %60'lık bir EV ile karşılaştırıldığında. Önceliğe Dayalı Dengeleme çok faktörlü bir aciliyet puanı hesaplayarak bu ihtiyaca cevap verir. Gücü önceliklere orantılı olarak dağıtmak.
Öncelik Puanlama Formülü
Puan, dört boyutu yapılandırılabilir ağırlıklarla birleştirir:
- SoC Aciliyeti (%35): pili ne kadar ve deşarj edin. Kritik durumları güçlendirmek için içbükey işlev (örn. SoC %5 = çok yüksek aciliyet).
- Zaman baskısı (%35): ayrılış ne kadar yakın. Son tarih yaklaştıkça hızla büyüyen üstel fonksiyon.
- Kullanıcı katmanı (%15): abonelik düzeyi (temel/premium/öncelik). Farklılaştırılmış SLA'lara sahip bir iş modeli oluşturun.
- Enerji verimliliği (%15): gücü gerçekten özümseyebilenleri ödüllendirir. Gücü neredeyse dolu pillere vermekten kaçının.
"""
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())Enerji Maliyetlerini En Aza İndirmek için Güçlendirme Öğrenimi
Önceki algoritmalar mevcut duruma tepki verir. Takviyeli Öğrenme bunun ötesine geçer: Bir etmen milyonlarca simülasyon aracılığıyla en aza indirdiği optimal politikayı öğrenir Zaman içindeki değişken fiyatlar, varış tahminleri ve EV'lerin ayrılması ve ağ kısıtlamaları. 2025'te üretime girecek başlıca operatörler şarj merkezleri aşağıdakilerin çeşitlerini kullanır: Yakınsal Politika Optimizasyonu (PPO) için reaktif algoritmalara kıyasla maliyetleri %15-25 oranında azaltır (kaynak: ScienceDirect, 2025).
Markov Karar Süreci Olarak Formülasyon
- Durum alanı: her konnektör için (SoC, kalan süre, mevcut güç) artı şebeke durumu (mevcut güç, enerji fiyatı, günün saati, PV tahmini).
- Eylem alanı: normalleştirilmiş güç seviyelerinin sürekli vektörü [0,1] her konektör için.
- Ödüller: enerji maliyetini en aza indirir, hedefe ulaşmayan elektrikli araçları cezalandırır SoC'yi son tarihe kadar hedefleyin, ağ sınırlarının ihlallerini cezalandırır.
"""
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 NoneAraçtan Şebekeye (V2G): Ağ Kaynağı Olarak Araç
V2G, EV paradigmasının niteliksel sıçramasını temsil ediyor: Araç artık yalnız değil enerji tüketicisi ama iki yönlü kaynak. Enerji fiyatı yüksek olduğunda veya şebeke stres altında olduğunda, şebekenin pilleri EV'ler enerjiyi şebekeye geri göndererek sahibine para kazandırabilir ve elektrik sisteminin stabilizasyonu.
2025'te V2G standardı
| Standart | Süpürgeler | V2G haberleri | Durum |
|---|---|---|---|
| ISO 15118-2 | EV-EVSE AC/DC iletişimi | Tak ve Şarj Et, akıllı şarj | Kullanımda |
| ISO 15118-20 | 2. nesil tam V2G | Çift yönlü BPT, DER entegrasyonu, dinamik planlama | Evlat Edinme 2025-2026 |
| OCPP 2.0.1 | CP-CSMS iletişimleri | ISO 15118 entegrasyonu, V2G şarj profilleri | Kullanımda |
| AB AFIR Reg. | Altyapı düzenlemesi | Ocak 2026'dan itibaren yeni sistemler için ISO 15118 zorunlu | yürürlükte |
Avrupa'da Utrecht (Hollanda), 500'den fazla kullanıcıyla en büyük ticari V2G dağıtımına sahiptir. Araba paylaşımında iki yönlü Renault'lar. Her araç para kazandırır 600-1.500 Avro/yıl Hollandalı TSO TenneT'ye Frekans Sınırlama Rezervi (FCR) sağlanması. İtalya'da V2G ve hala çoğunlukla deneysel, ancak yeni modeller Nissan Leaf e2+, Volkswagen ID.4 Pro ve Tesla Model 3 Highland'ın bazı versiyonları zaten çift yönlülüğü destekliyor.
EV Pilleriyle En Yüksek Tıraş ve Frekans Tepkisi
"""
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
}Fotovoltaik Entegrasyon: Elektrikli Araçlara Yönelik Güneş Enerjisi Fazlası
PV sistemi ile EV şarj istasyonu arasındaki entegrasyon, kullanım durumlarından biridir. Enerji yönetiminde daha hızlı yatırım getirisi. Fotovoltaik olduğundan daha fazlasını ürettiğinde binayı tüketiyor, fazlalık enerjiye beslenmek yerine elektrikli araçlara yönlendiriliyor şebekesi (~0,07-0,10 EUR/kWh GSE tutarında ücret ödenir) - enerjinin kullanımına izin verir aksi takdirde şebekeden maliyeti 0,28 EUR/kWh olacaktır. Net tasarruf e 0,18-0,21 Avro/kWh EV aracılığıyla tüketilen her kWh için.
| Strateji | Mantık | Fayda | Sınırlama |
|---|---|---|---|
| Yalnızca Fazlalık | EV'yi YALNIZCA PV fazlası ile şarj edin | Maksimum öz tüketim, sıfır ağ maliyeti | Değişken güç, yavaş şarj |
| Solar-First | Şebekeden etkilenmeden güneş enerjisine öncelik | Solar maksimum ile kararlı şarj | Şebekeden gelen enerjinin bir kısmı |
| Yeşil Maksimize Edici | Hava tahminiyle 4 saatlik ufukta optimizasyon yapın | Yenilenebilir enerji yüzdesini maksimuma çıkarın | Doğru tahmin gerektirir |
| Maliyet Optimize Edici | Güneş enerjisi + spot fiyatlar + V2G birleşimi | Mutlak minimum maliyet | Yüksek algoritmik karmaşıklık |
"""
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())Şarj Kontrol Cihazı Mikro Hizmet Mimarisi
Bir üretim EV yük dengeleme sistemi ve dağıtılmış mimari olaylar aracılığıyla iletişim kuran özel bileşenler. İşte ana hizmetler:
| Hizmet | Sorumluluk | Yığınlar | ALS'li |
|---|---|---|---|
| OCPP Ağ Geçidi | WebSocket sunucusu, OCPP çevirisi 1.6/2.0.1 | Python ocpp, eşzamansız | %99,9, <100ms |
| Şarj Kontrol Cihazı | Yük dengeleme algoritması, SetChargingProfile | FastAPI, Redis | %99,9, <500ms |
| Enerji Müdürü | PV/BESS/ağı ölçün, kullanılabilirliği hesaplayın | Modbus TCP, MQTT, Python | %99,5, oylama 1 saniye |
| Tahminciler | Varış/ayrılış tahmini, fiyatlar, FV | LightGBM/LSTM, FastAPI | %99, güncelleme 15 dk |
| DSO Ağ Geçidi | OpenADR sinyalleri, dinamik sınırlar | Python OpenADR 2.0b | %99, <2sn |
| Faturalandırma | Ölçüm, faturalandırma, oturumlar | FastAPI, PostgreSQL | %99,5 |
FastAPI ile Şarj Kontrol Cihazı API'si
"""
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
}Yük Dengeleme Değerlendirmesine İlişkin Metrikler ve KPI'lar
Yük dengeleme sisteminin etkinliğini değerlendirmek için ölçülebilir KPI'lara ihtiyaç vardır hem kullanıcı memnuniyetini hem de enerji optimizasyonunu ölçer.
| KPI'lar | Tanım | Hedef | Formül |
|---|---|---|---|
| Ücretlendirme Memnuniyet Oranı (CSR) | Son tarihe kadar SoC hedeflerine ulaşan oturumların yüzdesi | >%90 | ok_sessions / total_sessions |
| Tepe Azaltma %'si | Zirve azaltma ve yönetimsizlik karşılaştırması | >%25 | (yönetilmeyen zirve - yönetilen zirve) / yönetilmeyen zirve |
| Enerji Maliyeti Tasarrufu % | Anında şarja kıyasla kWh maliyetinden tasarruf | >%15 | (temel_maliyet - yönetilen_maliyet) / temel_maliyet |
| Güneş Enerjisi Öz Tüketimi % | Yerel PV'den elde edilen EV enerjisinin %'si | >%40 | solar_energy_ev / toplam_enerji_ev |
| Izgara Stres Endeksi (GSI) | Transformatör üzerindeki ortalama basınç [0-1] | <0,3 | ort(P_gerçek / P_nominal_trafo) |
| Gecikme Süresini Yeniden Dengeleme P95 | Etkinlikten OCPP profiline kadar geçen süre uygulandı | <2sn | t_ocpp_conf - t_event_received (p95) |
İtalya bağlamı: EV ve Teşvikler 2025 ile ARERA, CER
İtalya'da EV yük dengeleme sistemlerinin uygulanması düzenleyici zorluklar sunuyor ve mimari tercihleri doğrudan etkileyen bir pazar.
ARERA iki saatlik tarifeler ve Akıllı Şarj
İtalyan tarife sistemi yoğun olmayan saatlerde ücretlendirme için doğal bir teşvik yaratıyor:
- F1 grubu (Pzt-Cuma 8:00-19:00): yaklaşık 0,26-0,31 EUR/kWh. EV şarjı için kaçınılması gereken bant.
- Bant F2 (Pzt-Cum 7-8, 19-23; Cumartesi 7-23): yaklaşık 0,20-0,24 EUR/kWh.
- Bant F3 (gece 23:00-07:00; Pazar ve tatil günleri tüm gün): yaklaşık 0,16-0,19 EUR/kWh. Gece şarjı için idealdir.
Şarjın %60'ını F1'den F3'e kaydıran Kullanım Süresi algoritması, Filo yöneticisinin yıllık ücretlendirme maliyetleri %18-25.
EV ile Yenilenebilir Enerji Toplulukları (CER)
199/2021 sayılı Kanun Hükmünde Kararname ve 7 Aralık 2023 tarihli MASE Bakanlar Kararnamesi somut fırsatlar yarattı İtalyan CER'lerinde EV entegrasyonu için:
- Topluluk PV fazlası ile ücretlendirilen elektrikli araçlar GSE teşviki sağlıyor payına ilişkin paylaşılan enerji (ücret iadesi + teşvik 11 sent/kWh'a kadar).
- V2G ile EV pilleri, sanal CER depolama görevi görüyor ve pillerin yerini alıyor. Öğleden akşama kadar güneş enerjisi (tepe değişimi).
- GSE, paylaşılan enerjiyi ölçer. MACSEdahil Topluluk muhasebesinde V2G EV'lerin çift yönlü akışı.
2025 yılına kadar İtalyan EV altyapısı
İtalya'da 58.000'den fazla halka açık şarj noktası bulunmaktadır (Motus-E, 2024 sonu), bunlardan %22 hızlı DC (2023'teki %14'e kıyasla). 2025 yılında 94.230 BEV kaydedildi (+%46), %6,2 pazar payına sahiptir. Terna'ya göre akıllı şarjın olmadığı nüfusun %7'si yaygın Dağıtım besleyicisi 2027 yılına kadar dünyanın en yoğun bölgelerinde aşırı yüklenecek Kuzey İtalya. PNRR altyapı için 740 milyon Euro'nun üzerinde kaynak ayırdı 2026 yılına kadar 21.000 yeni halka açık nokta hedefiyle tamamlama.
Teşviklere dikkat edin
Geçiş 5.0 kapsamında akıllı şarja yönelik vergi kredileri, periyodik olarak denetlenir. Her zaman GSE web sitesindeki güncellenmiş koşulları kontrol edin ve teşvik bazlı yatırımları planlamadan önce CSEA. Fiyatlar ve erişim koşulları farklılık gösterebilir.
Sonuçlar ve Uygulama Yol Haritası
EV yük dengeleme algoritmik bir zarafet sorunu değil: bir zorunluluktur Enerji geçişi için kritik altyapı. Veriler kesindir: 5 milyon yönetilmeyen EV ile akşam zirvesi 35.000 MW artıyor; akıllı ile EV filosunun şarj edilmesi, zirveyi %34 oranında azaltan esnek bir kaynak haline geliyor.
Bu sistemleri üretimde uygulayan bir geliştirme ekibi için yol haritası sonraki aşamalar için optimaldir ve:
- Aşama 1 - Eşit Paylaşım: basit, sağlam, kurulumlar için yeterli 30-40 numaraya kadar. En büyük öncelik: yeniden hesaplamayı doğru şekilde uygulayın azaltma-sonra-arttırma dizisi ile olaya dayalı. Tahmini süre: 2-3 hafta.
-
Aşama 2 - PV Entegrasyonu: ML olmadan anında yatırım getirisi. Yalnızca şunları gerektirir:
Modbus/MQTT aracılığıyla güneş enerjisi fazlasının ölçümü ve dengeleyici ile entegrasyon
update_dynamic_limit(). 6-18 ayda geri ödemeyle hızlı kazanç. - Aşama 3 - Önceliğe Dayalı: Toplanacak mobil uygulamanız olduğunda ekleyin SoC ve başlangıç tarihi. CSR'deki (Ücretlendirme Memnuniyet Oranı) kazanç ölçülebilir ve hizmeti rakiplerinden farklılaştırır.
- Aşama 4 - V2G Altyapısı: çift yönlü mimari tasarımı henüz V2G EV'leriniz olmasa bile. ISO 15118-20 ile OCPP 2.0.1 zorunlu hale geliyor 2026'dan itibaren yeni tesisler için (AB AFIR). Yenileme maliyeti 5-10 kat daha fazladır.
- Aşama 5 - Üretimde RL: derin RL'ye yalnızca 6+ aydan sonra yatırım yapın kaliteli veriler ve özel bir MLOps ekibiyle. Maliyet kazancı gerçektir (%15-25) ancak operasyonel karmaşıklık yüksektir.
EnergyTech Serisindeki İlgili Makaleler
- Madde 4: Ağ Dijital İkizi - dağıtımdan önce EV'lerin ağ üzerindeki etkisini simüle edin
- Madde 5: Yenilenebilir Enerji Tahmini - Güneş enerjisi yönlendirmesi için PV üretim tahmini
- Madde 7: MQTT ve InfluxDB - şarj parametrelerinin gerçek zamanlı telemetrisi
- Madde 2: DERMS Mimarisi - EV'leri DER olarak dağıtılmış yönetim sistemine entegre edin
Seriler Arası Analizler
- MLOps serisi (mad. 306-315): PPO modelinin MLflow ile dağıtımı, sapma izleme ve yeniden eğitim
- AI Mühendislik Serisi (mad. 316-325): OCPP'de RAG ve LLM kuruluşuyla ISO 15118 belgeleri
- Veri ve Yapay Zeka İş Serisi (mad. 267-280): Akıllı şarj yatırımları için ROI, iş ölçümleri ve yol haritası
- PostgreSQL AI serisi (mad. 356-361): PostgreSQL'de TimescaleDB ile yeniden yükleme verilerinin zaman serisi