70 lines
2.7 KiB
Python
70 lines
2.7 KiB
Python
"""Minimal DSL seeds for NovaPlan interoperability.
|
|
|
|
This module provides a tiny Python-based DSL surface to seed LocalProblem
|
|
instances and produce basic PlanDelta-like deltas, enabling adapters and tests
|
|
to bootstrap interoperability workflows without requiring a full production DSL.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import List, Dict, Any
|
|
|
|
from .planner import LocalProblem
|
|
from .contracts import PlanDelta
|
|
|
|
|
|
@dataclass
|
|
class LocalProblemDSL:
|
|
"""A compact DSL representation of a LocalProblem seed.
|
|
|
|
This is deliberately simple and focuses on interoperability wiring:
|
|
- id, domain, and assets describe the problem scope
|
|
- objective is a simple string key understood by the translator
|
|
- constraints is a list of human-readable constraints (kept as strings)
|
|
"""
|
|
|
|
id: str
|
|
domain: str
|
|
assets: List[str]
|
|
objective: str
|
|
constraints: List[str]
|
|
|
|
def to_local_problem(self) -> LocalProblem:
|
|
"""Translate this DSL seed into a minimal LocalProblem instance.
|
|
|
|
The objective is translated into a lightweight, deterministic callable
|
|
that users can override or extend in adapters during integration tests.
|
|
"""
|
|
# Simple objective placeholder: sum of all local variables (initialized to 0)
|
|
def objective(variables: Dict[str, float], shared_vars: Dict[str, float]) -> float:
|
|
# Basic heuristic: minimize the sum of local variables (toy objective)
|
|
return sum(float(v) for v in variables.values())
|
|
|
|
# Initialize a tiny set of local decision variables for each asset
|
|
variables: Dict[str, float] = {f"{a}_var": 0.0 for a in self.assets}
|
|
# Constraints are retained for provenance; not evaluated in this MVP seed
|
|
constraints = {"domain": self.domain, "constraints": self.constraints}
|
|
|
|
return LocalProblem(id=self.id, objective=objective, variables=variables, constraints=constraints)
|
|
|
|
|
|
def seed_delta_for_local_problem(
|
|
lp: LocalProblem, shared_vars: Dict[str, float] | None = None, agent_id: str | None = None
|
|
) -> PlanDelta:
|
|
"""Create a minimal PlanDelta reflecting the current local problem delta.
|
|
|
|
This uses the tiny delta computation used in the MVP to bootstrap federation
|
|
and tests. It returns a PlanDelta that adapters can sign/propagate.
|
|
"""
|
|
if shared_vars is None:
|
|
shared_vars = {}
|
|
delta = {}
|
|
for k, v in lp.variables.items():
|
|
delta[k] = v - shared_vars.get(k, 0.0)
|
|
ts = __import__("time").time()
|
|
return PlanDelta(agent_id=agent_id or getattr(lp, "id", "unknown"), delta=delta, timestamp=ts)
|
|
|
|
|
|
__all__ = ["LocalProblemDSL", "seed_delta_for_local_problem"]
|