90 lines
2.5 KiB
Python
90 lines
2.5 KiB
Python
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
@dataclass
|
|
class LocalProblem:
|
|
id: str
|
|
domain: str
|
|
assets: List[str]
|
|
constraints: Dict[str, Any] = field(default_factory=dict)
|
|
objective: Dict[str, Any] = field(default_factory=dict)
|
|
solver_hint: Optional[str] = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.id:
|
|
raise ValueError("LocalProblem.id is required")
|
|
if not self.domain:
|
|
raise ValueError("LocalProblem.domain is required")
|
|
if not self.assets:
|
|
raise ValueError("LocalProblem.assets must not be empty")
|
|
|
|
|
|
@dataclass
|
|
class SharedVariables:
|
|
version: int
|
|
signals: Dict[str, Any] = field(default_factory=dict)
|
|
privacy_tag: Optional[str] = None
|
|
aggregation_method: str = "mean"
|
|
|
|
def __post_init__(self) -> None:
|
|
if self.version < 1:
|
|
raise ValueError("SharedVariables.version must be >= 1")
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
delta: Dict[str, Any]
|
|
timestamp: datetime
|
|
author: str
|
|
contract_id: str
|
|
signature: Optional[str] = None
|
|
safety_tags: Optional[List[str]] = None
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.author:
|
|
raise ValueError("PlanDelta.author is required")
|
|
if not self.contract_id:
|
|
raise ValueError("PlanDelta.contract_id is required")
|
|
|
|
|
|
@dataclass
|
|
class DualVariables:
|
|
lagrange_multipliers: Dict[str, float] = field(default_factory=dict)
|
|
convergence_status: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class AuditLog:
|
|
entries: List[Dict[str, Any]] = field(default_factory=list)
|
|
last_updated: Optional[datetime] = None
|
|
|
|
def record(self, event: str, **details: Any) -> None:
|
|
self.entries.append({"event": event, **details})
|
|
self.last_updated = datetime.utcnow()
|
|
|
|
|
|
@dataclass
|
|
class PrivacyBudget:
|
|
budget: float
|
|
spent: float = 0.0
|
|
policy: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def consume(self, amount: float) -> None:
|
|
if amount < 0:
|
|
raise ValueError("Privacy budget consumption must be positive")
|
|
if self.spent + amount > self.budget:
|
|
raise ValueError("Privacy budget exhausted")
|
|
self.spent += amount
|
|
|
|
|
|
@dataclass
|
|
class RegistryMetadata:
|
|
contract_version: str
|
|
adapter_id: str
|
|
contract_id: Optional[str] = None
|
|
schema_name: Optional[str] = None
|
|
created_at: Optional[datetime] = None
|
|
replay_hint: Optional[str] = None
|