diff --git a/examples/cosmosmesh_pilot.py b/examples/cosmosmesh_pilot.py new file mode 100644 index 0000000..d2b357b --- /dev/null +++ b/examples/cosmosmesh_pilot.py @@ -0,0 +1,61 @@ +"""CosmosMesh Interop Pilot + +A small end-to-end demonstration of the CosmosMesh interoperability surface +using the CatOpt-like bridge primitives. It creates a minimal LocalProblem, +SharedVariables, and PlanDelta, converts to the canonical IR via to_catopt, +and back via from_catopt to verify round-tripping semantics. +""" + +from __future__ import annotations + +import json + +from cosmosmesh_privacy_preserving_federated.catopt_bridge import ( + LocalProblem, + SharedVariables, + PlanDelta, + to_catopt, + from_catopt, +) + + +def run_pilot(): + # Build a tiny 2-asset scenario: rover and habitat (represented by IDs) + lp = LocalProblem( + id="lp-demo-1", + domain="space", + assets=["rover-alpha", "habitat-beta"], + objective={"type": "minimize", "payload": 10}, + constraints={"energy": {"limit": 1000}}, + version=1, + ) + + sv = SharedVariables( + forecasts={"energy": 42}, + priors={"energy": 40}, + version=1, + ) + + delta = PlanDelta( + delta={"allocation": {"rover-alpha": 0, "habitat-beta": 1}}, + timestamp="2026-01-01T00:00:00Z", + author="pilot", + contract_id="demo-contract-1", + signature="sig-demo", + ) + + # Canonical IR + catopt = to_catopt(lp, sv, delta) + print("Canonical IR (trimmed):") + print(json.dumps(catopt, indent=2)) + + # Round-trip back from canonical IR + recon = from_catopt(catopt) + print("Recovered payload:") + print(json.dumps(recon, indent=2)) + + return catopt, recon + + +if __name__ == "__main__": + run_pilot() diff --git a/tests/test_pilot.py b/tests/test_pilot.py new file mode 100644 index 0000000..1670367 --- /dev/null +++ b/tests/test_pilot.py @@ -0,0 +1,25 @@ +import json +import sys +import os + +# Ensure the local src layout is importable in test runs +SRC_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "src") +if os.path.isdir(SRC_DIR) and SRC_DIR not in sys.path: + sys.path.insert(0, SRC_DIR) + +from cosmosmesh_privacy_preserving_federated.catopt_bridge import to_catopt, from_catopt, LocalProblem, SharedVariables, PlanDelta + + +def test_pilot_roundtrip_basic(): + lp = LocalProblem( + id="lp-t", domain="space", assets=["rover-1"], objective={"type": "min", "value": 1}, constraints={"time": 60}, version=1 + ) + sv = SharedVariables(forecasts={"energy": 5}, priors={"energy": 4}, version=1) + delta = PlanDelta(delta={"task": "sample"}, timestamp="2026-01-01T00:00:00Z", author="tester", contract_id="c1", signature="sig") + + catopt = to_catopt(lp, sv, delta) + recon = from_catopt(catopt) + + # Basic shape validation of round-trip results + assert isinstance(recon.get("LocalProblem"), dict) + assert isinstance(recon.get("PlanDelta"), dict)