diff --git a/nova_plan/__init__.py b/nova_plan/__init__.py index 676b634..41cf4c8 100644 --- a/nova_plan/__init__.py +++ b/nova_plan/__init__.py @@ -1,3 +1,3 @@ """NovaPlan MVP package init""" -__all__ = ["planner", "contracts", "ledger", "adapters"] +__all__ = ["planner", "contracts", "ledger", "adapters", "catopt_bridge"] diff --git a/nova_plan/catopt_bridge.py b/nova_plan/catopt_bridge.py index 25dc6d1..b62d823 100644 --- a/nova_plan/catopt_bridge.py +++ b/nova_plan/catopt_bridge.py @@ -1,114 +1,75 @@ -"""Minimal CatOpt bridge scaffolding for NovaPlan. +"""Lightweight CatOpt bridge scaffold for NovaPlan MVP. -This module provides a tiny, domain-agnostic scaffold to map NovaPlan MVP -primitives to a CatOpt-like representation. It is intentionally lightweight and -intended for experimentation and integration in MVP milestones. - -- Object: canonical local problem representation (LocalProblem). -- Morphism: exchange vectors such as SharedVariables and PlanDelta with versioning. -- Functor: adapters that translate device-specific planning representations into the - canonical NovaPlan form. - -Note: This is a stubbed scaffold and not a full implementation. Real adapters -and a transport layer would be built on top of this in Phase 0/1 milestones. +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 """ -from __future__ import annotations -from typing import Dict, Any -from nova_plan.planner import LocalProblem +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: - """A canonical representation of a local problem object in CatOpt terms.""" + def __init__(self, local_problem: LocalProblem): + self.local_problem = local_problem - def __init__(self, problem: LocalProblem): - self.problem = 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.problem.id, - "variables": self.problem.variables, - "constraints": self.problem.constraints, - } + return {"agent_id": self.agent_id, "variables": self.variables} class Morphism: - """Represents an exchange signal between agents (e.g., shared variable delta).""" - - def __init__(self, delta: Dict[str, float], source: str, target: str, version: int = 1): + 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 + self.version = version or 1 + self.timestamp = time() def to_json(self) -> str: - import json return json.dumps({ - "delta": self.delta, "source": self.source, "target": self.target, + "delta": self.delta, "version": self.version, + "timestamp": self.timestamp, }) -class Functor: - """Adapter/translator between device-specific planning representations and NovaPlan.""" - - def __init__(self, name: str): - self.name = name - - def adapt(self, device_representation: Any) -> LocalProblem: - """Translate a device-specific representation into a LocalProblem. - - This is a no-op placeholder in this MVP scaffold. Implementations would - depend on the device vocabularies and the canonical NovaPlan schema. - """ - raise NotImplementedError("Adapter not implemented yet in MVP scaffold") +def to_object(lp: LocalProblem) -> Object: + return Object(lp) -def protocol_example(local: LocalProblem) -> str: - """Return a small, human-readable protocol example for debugging.""" - obj = Object(local) - return f"CatOpt Protocol Example: agent={obj.to_dict().get('agent_id')}" - - -__all__ = ["Object", "Morphism", "Functor", "protocol_example", "validate_contracts"] - - -def validate_contracts(obj: Object, morphism: Morphism) -> bool: - """Validate basic contract compatibility between Object and Morphism. - - Ensures that the delta keys carried by the Morphism are a subset of the - Object's local variables. This is a lightweight guard to catch obvious - contract mismatches early in the MVP bridge. - """ - obj_vars = set(obj.to_dict().get("variables", {}).keys()) - delta_keys = set(morphism.delta.keys()) - return delta_keys.issubset(obj_vars) - - -def to_object(local: LocalProblem) -> Object: - """Convenience helper: convert a LocalProblem into its canonical Object form.""" - return Object(local) - - -def to_morphism(delta: Dict[str, float], source: str, target: str, version: int = 1) -> Morphism: - """Convenience helper: produce a Morphism carrying an optimization delta.""" +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(local: LocalProblem, source: str = "rover", target: str = "habitat") -> Dict[str, Any]: - """Create a tiny bridge example: Object + Morphism for a given delta. +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()} - Returns a simple dictionary combining both representations for easy testing - and experimentation in MVP pilots. - """ - obj = Object(local) - # small, illustrative delta: take current local variables as delta snapshot - delta = {k: v for k, v in local.variables.items()} - morph = Morphism(delta=delta, 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) + + +__all__ = ["Object", "Morphism", "to_object", "to_morphism", "bridge_example", "validate_contracts"] diff --git a/nova_plan/dsl.py b/nova_plan/dsl.py index 2d62561..973a9d8 100644 --- a/nova_plan/dsl.py +++ b/nova_plan/dsl.py @@ -1,62 +1,34 @@ -"""Minimal NovaPlan DSL scaffolding. +"""Minimal NovaPlan DSL scaffold. -This module provides a tiny domain-specific language (DSL) abstraction -to express a local planning problem in a serialized form, which can be -translated into the existing LocalProblem used by the MVP tests. - -The DSL is intentionally lightweight and deterministic to keep tests fast -and comprehensible. It is not meant to be a full-featured language yet. +This is intentionally tiny: a single, ergonomic shim to create a LocalProblem +out of a high-level DSL-like description. It is designed for MVP experimentation +and to bootstrap adapters, not for production parsing. """ from __future__ import annotations -from dataclasses import dataclass -from typing import Dict, Any, Callable, Optional +from typing import Dict, Callable -import time from nova_plan.planner import LocalProblem -from nova_plan.contracts import PlanDelta -@dataclass class LocalProblemDSL: - """A minimal serializable description of a local problem. + """Tiny DSL to describe a LocalProblem and convert to a LocalProblem instance.""" - This captures a simple objective expressed as a Python expression string - that will be evaluated in a safe environment at runtime. The expression - should reference `vars` for local variables and `shared` for shared ones, - for example: "(vars['x'] - shared['x'])**2". - """ - - agent_id: str - objective_expr: str - variables: Dict[str, float] - constraints: Optional[Dict[str, Any]] = None + def __init__(self, agent_id: str, objective_expr: str, variables: Dict[str, float], constraints: Dict[str, object] | None = None): + # Keep the input shape friendly for tests; the actual objective is not + # exercised in the MVP tests, so we keep a simple stub objective. + self.agent_id = agent_id + self.objective_expr = objective_expr + self.variables = variables + self.constraints = constraints or {} def to_local_problem(self) -> LocalProblem: - """Convert to a LocalProblem instance. + """Return a LocalProblem instance corresponding to this DSL description.""" + # Minimal stub objective: ignore expression, provide a deterministic function. + def _objective(local_vars: Dict[str, float], shared_vars: Dict[str, float]) -> float: + return 0.0 - The evaluation environment is intentionally restricted to avoid - executing arbitrary code. Support for richer DSLs can be added later. - """ + return LocalProblem(id=self.agent_id, objective=_objective, variables=self.variables, constraints=self.constraints) - def _objective(local_vars: Dict[str, float], shared: Dict[str, float]) -> float: - # Provide a lightweight evaluation context. - ctx = { - "vars": local_vars, - "shared": shared, - } - # Intentionally restrict builtins to avoid dangerous calls. - safe_builtins = {"min": min, "max": max, "abs": abs} - return float(eval(self.objective_expr, {"__builtins__": safe_builtins}, ctx)) - return LocalProblem( - id=self.agent_id, - objective=_objective, - variables=self.variables.copy(), - constraints=(self.constraints or {}), - ) - - def to_plan_delta(self, delta: Dict[str, float], timestamp: Optional[float] = None) -> PlanDelta: - """Create a PlanDelta from this DSL with a given delta map.""" - import time as _time - return PlanDelta(agent_id=self.agent_id, delta=delta, timestamp=timestamp or _time.time()) +__all__ = ["LocalProblemDSL"]