35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
"""Toy DER Aggregator Adapter with deterministic replay support."""
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
@dataclass
|
|
class DERAggregatorAdapter:
|
|
aggregator_id: str
|
|
samples: Optional[List[Dict[str, Any]]] = None
|
|
_cursor: int = field(init=False, default=0)
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.samples is None:
|
|
self.samples = []
|
|
|
|
def aggregate(self) -> Dict[str, Any]:
|
|
if self.samples:
|
|
sample_list = self.samples
|
|
if self._cursor >= len(sample_list):
|
|
raise IndexError("No more DER aggregation samples available")
|
|
reading = dict(sample_list[self._cursor])
|
|
self._cursor += 1
|
|
reading.setdefault("aggregator_id", self.aggregator_id)
|
|
reading.setdefault("timestamp", self._cursor)
|
|
reading.setdefault("status", "nominal")
|
|
return reading
|
|
|
|
return {
|
|
"aggregator_id": self.aggregator_id,
|
|
"timestamp": self._cursor,
|
|
"total_power_kw": 500.0, # fixed placeholder for deterministic behavior
|
|
"status": "nominal",
|
|
}
|