diff --git a/README.md b/README.md index a3c2d09..4f8e8ae 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,9 @@ New scaffold: CatOpt Bridge (scaffold) A minimal, open-source MVP for decentralized, privacy-preserving multi-agent mission planning in deep-space robotic constellations. +- Includes a lightweight CatOpt bridge scaffold (nova_plan.catopt_bridge) to map NovaPlan primitives to a canonical interoperability representation. + + - Offline-first, privacy-aware coordination across heterogeneous fleets (rovers, drones, habitat bots). - Local problem solving with a tiny ADMM-like core and federation of agents. - A lightweight mission ledger that can anchor decisions when ground links are available. diff --git a/nova_plan/catopt_bridge.py b/nova_plan/catopt_bridge.py index d0fe5a2..779ce33 100644 --- a/nova_plan/catopt_bridge.py +++ b/nova_plan/catopt_bridge.py @@ -1,76 +1,86 @@ -"""Minimal CatOpt bridge scaffolding for NovaPlan MVP. +"""Minimal CatOpt-style bridge for NovaPlan MVP interoperability. -This module provides tiny, well-scoped helpers to map NovaPlan primitives -to a canonical CatOpt-like representation suitable for interoperability -in MVP experiments. +This module provides tiny, drop-in helpers to map NovaPlan core primitives +to a canonical representation that can be consumed by interoperability bridges +or other runtimes in the ecosystem. The goal is to be lightweight and test-friendly +while not impacting the core MVP tests. """ + from __future__ import annotations +from typing import Any, Dict from dataclasses import dataclass -from typing import Any, Dict, Optional -from nova_plan.contracts import PlanDelta -from nova_plan.planner import LocalProblem +from .planner import LocalProblem +from .contracts import PlanDelta @dataclass class ObjectI: - """Canonical object representation (per-agent LocalProblem). - - This is a lightweight wrapper intended for interop exploration. - """ - + """Canonical Object interface used by CatOpt bridges.""" id: str payload: Dict[str, Any] @dataclass class Morphism: - """Canonical morphism carrying summarized signals or delta information.""" - + """Canonical Morphism interface used by CatOpt bridges.""" source: str target: str - data: Dict[str, float] # summarized signals or delta payload - version: int = 1 - contract_id: str = "default" + data: Dict[str, Any] + contract_id: str -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 to_object(local_problem: LocalProblem) -> ObjectI: + """Convert a LocalProblem into a canonical Object representation. + + The object is intentionally minimal: it contains the id, the local variables, + and basic metadata that downstream adapters can enrich. + """ + payload = {"variables": dict(local_problem.variables)} + return ObjectI(id=local_problem.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.""" +def to_morphism(plan_delta: PlanDelta) -> Morphism: + """Convert a PlanDelta into a canonical Morphism representation. + + This helper is a thin wrapper that exposes a Morphism data class. + Used primarily for internal interoperability and tests. + """ 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"), + source=plan_delta.agent_id, + target="global", + data=plan_delta.delta, + contract_id=getattr(plan_delta, "contract_id", "default"), ) -def bridge_example(): - """Small helper to illustrate a mapping between NovaPlan and CatOpt forms. +def delta_to_morphism( + plan_delta: PlanDelta, + contract_id: str = "default", + source: str | None = None, + target: str = "global", +) -> Morphism: + """Convert a PlanDelta into a Morphism, with optional overrides. - This is intentionally lightweight and for demonstration in tests/examples. - Returns a tuple of (ObjectI, Morphism). + - If source is None, uses plan_delta.agent_id + - contract_id defaults to "default" unless explicitly provided + - target defaults to "global" unless overridden + - data is plan_delta.delta """ - # 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 + return Morphism( + source=(source if source is not None else plan_delta.agent_id), + target=target, + data=plan_delta.delta, + contract_id=contract_id, + ) -__all__ = ["ObjectI", "Morphism", "to_object", "delta_to_morphism", "bridge_example"] +def bridge_demo(local_problem: LocalProblem, plan_delta: PlanDelta) -> Dict[str, Any]: + """Tiny end-to-end demo helper that returns both object and morphism representations. + + This function is used by examples or quick tests to verify the bridge wiring. + """ + obj = to_object(local_problem) + morph = delta_to_morphism(plan_delta) + return {"object": obj, "morphism": morph}