91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
"""CatOpt-Grid Bridge: Canonical Interoperability Layer (minimal).
|
|
|
|
This module provides a small, production-friendly scaffold to map the
|
|
CatOpt-Grid primitives to a vendor-agnostic intermediate representation
|
|
(IR). It is intentionally lightweight and intended as a starting point for the
|
|
MVP wiring described in the community plan.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, List, Optional
|
|
import time
|
|
|
|
@dataclass
|
|
class PrivacyBudget:
|
|
budget: float
|
|
unit: str = "epsilon"
|
|
|
|
|
|
@dataclass
|
|
class AuditLogEntry:
|
|
message: str
|
|
timestamp: float = field(default_factory=lambda: time.time())
|
|
signature: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
delta: List[float]
|
|
version: int
|
|
timestamp: float = field(default_factory=lambda: time.time())
|
|
|
|
|
|
@dataclass
|
|
class IRObject:
|
|
"""Canonical Object: a local optimization problem (per-asset task).
|
|
|
|
This is intentionally aligned with LocalProblem from catopt_grid.core/admm_lite
|
|
to ease adoption. Users can extend the IR with extra metadata as needed.
|
|
"""
|
|
id: str
|
|
dimension: int
|
|
# Optional per-object metadata
|
|
metadata: Dict[str, object] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class IRMorphism:
|
|
"""Canonical Morphism: a channel carrying shared data among agents."""
|
|
name: str
|
|
value: List[float]
|
|
version: int = 0
|
|
|
|
|
|
class GraphOfContracts:
|
|
"""Lightweight registry mapping adapters to their contract metadata.
|
|
|
|
This is a very small stand-in for a more feature-rich contract registry
|
|
(GoC) that would live in a real deployment. It records versions and simple
|
|
schemas for cross-domain interoperability.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._contracts: Dict[str, Dict[str, object]] = {}
|
|
|
|
def register(self, adapter_name: str, contract: Dict[str, object]) -> None:
|
|
self._contracts[adapter_name] = {
|
|
"contract": contract,
|
|
"version": contract.get("version", 1),
|
|
"timestamp": time.time(),
|
|
}
|
|
|
|
def get_contract(self, adapter_name: str) -> Optional[Dict[str, object]]:
|
|
return self._contracts.get(adapter_name, None)
|
|
|
|
def list_contracts(self) -> List[Dict[str, object]]:
|
|
items = []
|
|
for name, meta in self._contracts.items():
|
|
items.append({"adapter": name, **meta})
|
|
return sorted(items, key=lambda x: x.get("timestamp", 0))
|
|
|
|
|
|
__all__ = [
|
|
"PrivacyBudget",
|
|
"AuditLogEntry",
|
|
"PlanDelta",
|
|
"IRObject",
|
|
"IRMorphism",
|
|
"GraphOfContracts",
|
|
]
|