122 lines
3.6 KiB
Python
122 lines
3.6 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",
|
|
"lp_to_ir",
|
|
]
|
|
|
|
|
|
def lp_to_ir(lp: object) -> IRObject:
|
|
"""Convert a local problem instance to a vendor-agnostic IR Object.
|
|
|
|
This helper is intentionally permissive to accommodate multiple LocalProblem
|
|
shapes that exist in this repo (e.g., core.LocalProblem and admm_lite.LocalProblem).
|
|
It extracts common attributes when present and falls back to sensible defaults.
|
|
"""
|
|
# Fallback defaults
|
|
default_id = getattr(lp, "id", None) or getattr(lp, "name", None) or "lp"
|
|
# Dimension could be named differently across implementations
|
|
dim = getattr(lp, "dimension", None)
|
|
if dim is None:
|
|
dim = getattr(lp, "n", None)
|
|
try:
|
|
dim_int = int(dim)
|
|
except Exception:
|
|
dim_int = 0
|
|
|
|
# Collect a small amount of metadata if available
|
|
meta: Dict[str, object] = {}
|
|
if hasattr(lp, "data"):
|
|
meta["data"] = getattr(lp, "data")
|
|
if hasattr(lp, "lb"):
|
|
meta["lb"] = getattr(lp, "lb")
|
|
if hasattr(lp, "ub"):
|
|
meta["ub"] = getattr(lp, "ub")
|
|
|
|
return IRObject(id=str(default_id), dimension=dim_int, metadata=meta)
|