84 lines
2.3 KiB
Python
84 lines
2.3 KiB
Python
from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Dict, List, Optional
|
|
import time
|
|
|
|
|
|
@dataclass
|
|
class LocalProblem:
|
|
name: str
|
|
region: Optional[str] = None
|
|
metadata: Dict[str, Any] = field(default_factory=dict)
|
|
created_at: float = field(default_factory=lambda: time.time())
|
|
|
|
def __repr__(self) -> str:
|
|
return f"LocalProblem(name={self.name!r}, region={self.region!r})"
|
|
|
|
|
|
@dataclass
|
|
class SharedVariables:
|
|
description: str = ""
|
|
data: Dict[str, Any] = field(default_factory=dict)
|
|
aggregate_type: str = "sum" # placeholder for privacy-preserving aggregation
|
|
created_at: float = field(default_factory=lambda: time.time())
|
|
|
|
def __repr__(self) -> str:
|
|
return f"SharedVariables(desc={self.description!r}, keys={list(self.data.keys())})"
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
delta_id: str
|
|
actions: List[Dict[str, Any]] = field(default_factory=list)
|
|
timestamp: float = field(default_factory=lambda: time.time())
|
|
annotations: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"PlanDelta(delta_id={self.delta_id!r}, actions={len(self.actions)} actions)"
|
|
|
|
|
|
@dataclass
|
|
class Policy:
|
|
policy_id: str
|
|
rules: Dict[str, Any] = field(default_factory=dict)
|
|
effective_at: float = field(default_factory=lambda: time.time())
|
|
|
|
def __repr__(self) -> str:
|
|
return f"Policy(id={self.policy_id!r}, rules={len(self.rules)})"
|
|
|
|
|
|
@dataclass
|
|
class AttestationHint:
|
|
hint_id: str
|
|
scheme: str = "dummy"
|
|
payload: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
def __repr__(self) -> str:
|
|
return f"AttestationHint(id={self.hint_id!r}, scheme={self.scheme!r})"
|
|
|
|
|
|
@dataclass
|
|
class AuditLog:
|
|
entry_id: str
|
|
message: str
|
|
timestamp: float = field(default_factory=lambda: time.time())
|
|
signature: Optional[str] = None
|
|
|
|
def __repr__(self) -> str:
|
|
return f"AuditLog(id={self.entry_id!r}, ts={self.timestamp})"
|
|
|
|
|
|
@dataclass
|
|
class PrivacyBudget:
|
|
budget_id: str
|
|
limit: float
|
|
used: float = 0.0
|
|
def consume(self, amount: float) -> None:
|
|
self.used = min(self.used + amount, self.limit)
|
|
|
|
def remaining(self) -> float:
|
|
return self.limit - self.used
|
|
|
|
def __repr__(self) -> str:
|
|
return f"PrivacyBudget(id={self.budget_id!r}, remaining={self.remaining()})"
|