77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
"""Minimal CatOpt bridge scaffolding for NovaPlan MVP.
|
|
|
|
This module provides tiny, well-scoped helpers to map NovaPlan primitives
|
|
to a canonical CatOpt-like representation suitable for interoperability
|
|
in MVP experiments.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Optional
|
|
|
|
from nova_plan.contracts import PlanDelta
|
|
from nova_plan.planner import LocalProblem
|
|
|
|
|
|
@dataclass
|
|
class ObjectI:
|
|
"""Canonical object representation (per-agent LocalProblem).
|
|
|
|
This is a lightweight wrapper intended for interop exploration.
|
|
"""
|
|
|
|
id: str
|
|
payload: Dict[str, Any]
|
|
|
|
|
|
@dataclass
|
|
class Morphism:
|
|
"""Canonical morphism carrying summarized signals or delta information."""
|
|
|
|
source: str
|
|
target: str
|
|
data: Dict[str, float] # summarized signals or delta payload
|
|
version: int = 1
|
|
contract_id: str = "default"
|
|
|
|
|
|
def to_object(local: LocalProblem) -> ObjectI:
|
|
"""Map a LocalProblem to a canonical ObjectI representation."""
|
|
payload = {
|
|
"variables": local.variables,
|
|
"constraints": local.constraints,
|
|
"id": local.id,
|
|
}
|
|
return ObjectI(id=local.id, payload=payload)
|
|
|
|
|
|
def delta_to_morphism(delta: PlanDelta, source: str = "local", target: str = "global", contract_id: Optional[str] = None) -> Morphism:
|
|
"""Convert a PlanDelta into a canonical Morphism."""
|
|
return Morphism(
|
|
source=delta.agent_id or source,
|
|
target=target,
|
|
data=delta.delta,
|
|
version=1,
|
|
contract_id=contract_id or delta.__dict__.get("contract_id", "default"),
|
|
)
|
|
|
|
|
|
def bridge_example():
|
|
"""Small helper to illustrate a mapping between NovaPlan and CatOpt forms.
|
|
|
|
This is intentionally lightweight and for demonstration in tests/examples.
|
|
Returns a tuple of (ObjectI, Morphism).
|
|
"""
|
|
# Minimal synthetic LocalProblem
|
|
lp = LocalProblem(id="demo-agent", objective=lambda v, s: sum(v.values()) + sum(s.values()), variables={"a": 1.0}, constraints={})
|
|
obj = to_object(lp)
|
|
# Fake delta
|
|
delta = {"a": -0.1}
|
|
import time
|
|
d = PlanDelta(agent_id=lp.id, delta=delta, timestamp=time.time())
|
|
morph = delta_to_morphism(d, source=lp.id)
|
|
return obj, morph
|
|
|
|
|
|
__all__ = ["ObjectI", "Morphism", "to_object", "delta_to_morphism", "bridge_example"]
|