87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
"""Lightweight CatOpt bridge scaffold for NovaPlan MVP.
|
|
|
|
Public API used by tests:
|
|
- Object: minimal wrapper around a LocalProblem-like object
|
|
- Morphism: wrapper for delta signals between source and destination
|
|
- to_object(lp): convert a LocalProblem-like instance to Object
|
|
- to_morphism(delta, source, target, version=None): create a Morphism from delta
|
|
- bridge_example(lp, source, target): package as a dict with object and serialized morphism
|
|
- validate_contracts(obj, morph): basic compatibility check
|
|
"""
|
|
|
|
import json
|
|
from typing import Any, Dict
|
|
from time import time
|
|
|
|
from nova_plan.contracts import PlanDelta
|
|
from nova_plan.planner import LocalProblem
|
|
|
|
|
|
class Object:
|
|
def __init__(self, local_problem: LocalProblem):
|
|
self.local_problem = local_problem
|
|
|
|
@property
|
|
def agent_id(self) -> str:
|
|
return getattr(self.local_problem, "id", "unknown")
|
|
|
|
@property
|
|
def variables(self) -> Dict[str, float]:
|
|
return getattr(self.local_problem, "variables", {})
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {"agent_id": self.agent_id, "variables": self.variables}
|
|
|
|
|
|
class Morphism:
|
|
def __init__(self, delta: Dict[str, float], source: str, target: str, version: int | None = None):
|
|
self.delta = delta
|
|
self.source = source
|
|
self.target = target
|
|
self.version = version or 1
|
|
self.timestamp = time()
|
|
|
|
def to_json(self) -> str:
|
|
return json.dumps({
|
|
"source": self.source,
|
|
"target": self.target,
|
|
"delta": self.delta,
|
|
"version": self.version,
|
|
"timestamp": self.timestamp,
|
|
})
|
|
|
|
|
|
def to_object(lp: LocalProblem) -> Object:
|
|
return Object(lp)
|
|
|
|
|
|
def to_morphism(delta: Dict[str, float], source: str, target: str, version: int | None = None) -> Morphism:
|
|
return Morphism(delta=delta, source=source, target=target, version=version)
|
|
|
|
|
|
def bridge_example(lp: LocalProblem, source: str, target: str) -> Dict[str, Any]:
|
|
obj = to_object(lp)
|
|
morph = to_morphism(delta={k: v for k, v in lp.variables.items()}, source=source, target=target, version=1)
|
|
return {"object": obj.to_dict(), "morphism": morph.to_json()}
|
|
|
|
|
|
def validate_contracts(obj: Object, morph: Morphism) -> bool:
|
|
# Basic validation: every delta key must exist in the object's variables
|
|
obj_keys = set(obj.variables.keys())
|
|
delta_keys = set(morph.delta.keys())
|
|
return delta_keys.issubset(obj_keys)
|
|
|
|
|
|
def bridge_pair(lp: LocalProblem, source: str, target: str, version: int | None = 1) -> Dict[str, Any]:
|
|
"""Convenience helper returning a CatOpt-style pair for a given LocalProblem.
|
|
|
|
This is a small wrapper mirrors bridge_example but allows explicit version
|
|
control and is intended for use by external adapters that want a stable
|
|
tuple without constructing the JSON themselves.
|
|
"""
|
|
obj = to_object(lp)
|
|
morph = to_morphism(delta={k: v for k, v in lp.variables.items()}, source=source, target=target, version=version)
|
|
return {"object": obj.to_dict(), "morphism": morph.to_json()}
|
|
|
|
__all__ = ["Object", "Morphism", "to_object", "to_morphism", "bridge_example", "validate_contracts"]
|