102 lines
2.5 KiB
Python
102 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List
|
|
|
|
|
|
@dataclass
|
|
class Object:
|
|
"""Canonical object describing a device or asset."""
|
|
|
|
id: str
|
|
type: str
|
|
properties: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.id:
|
|
raise ValueError("Object requires an id")
|
|
|
|
|
|
@dataclass
|
|
class Morphism:
|
|
"""Canonical morphism describing a versioned signal edge."""
|
|
|
|
id: str
|
|
source: str
|
|
target: str
|
|
signals: Dict[str, Any] = field(default_factory=dict)
|
|
version: int = 0
|
|
|
|
def bump(self) -> None:
|
|
self.version += 1
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
"""Incremental islanding or load-shedding update."""
|
|
|
|
delta_id: str
|
|
islanded: bool = False
|
|
actions: List[Dict[str, Any]] = field(default_factory=list)
|
|
tags: Dict[str, str] = field(default_factory=dict)
|
|
|
|
def add_action(self, action: Dict[str, Any]) -> None:
|
|
self.actions.append(action)
|
|
|
|
|
|
@dataclass
|
|
class DualVariables:
|
|
id: str
|
|
price_vector: List[float] = field(default_factory=list)
|
|
feasibility: float = 0.0
|
|
version: int = 0
|
|
|
|
|
|
@dataclass
|
|
class AuditLog:
|
|
entry: str
|
|
signer: str
|
|
timestamp: str
|
|
contract_id: str
|
|
version: int = 0
|
|
|
|
|
|
@dataclass
|
|
class PrivacyBudget:
|
|
signal: str
|
|
budget: float
|
|
remaining: float
|
|
expiry: str
|
|
|
|
|
|
@dataclass
|
|
class RegistryEntry:
|
|
adapter_id: str
|
|
contract_version: str
|
|
data_contract: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class DeltaStore:
|
|
def __init__(self) -> None:
|
|
self.objects: Dict[str, Object] = {}
|
|
self.morphisms: Dict[str, Morphism] = {}
|
|
self.deltas: List[PlanDelta] = []
|
|
|
|
def add_object(self, obj: Object) -> None:
|
|
self.objects[obj.id] = obj
|
|
|
|
def add_morphism(self, morph: Morphism) -> None:
|
|
self.morphisms[morph.id] = morph
|
|
|
|
def add_delta(self, delta: PlanDelta) -> None:
|
|
self.deltas.append(delta)
|
|
|
|
|
|
def build_sample_world() -> DeltaStore:
|
|
ds = DeltaStore()
|
|
ds.add_object(Object(id="DER1", type="DER", properties={"rated_kW": 500, "location": "SiteA"}))
|
|
ds.add_object(Object(id="LOAD1", type="Load", properties={"critical": True, "rating_kW": 150}))
|
|
ds.add_morphism(Morphism(id="SIG1", source="DER1", target="LOAD1", signals={"voltage": 1.0}, version=1))
|
|
ds.add_delta(PlanDelta(delta_id="DELTA1", islanded=True, actions=[{"type": "island", "target": "LOAD1"}], tags={"sig": "v1"}))
|
|
return ds
|