75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
"""NeuPlan DSL: minimal LocalProblem / PlanDelta / SharedVariables model.
|
|
|
|
This is a lightweight, easily testable sketch translating planning problems
|
|
into a toy neuromorphic intermediate representation (N-IR).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, List, Any
|
|
import time
|
|
|
|
|
|
@dataclass
|
|
class LocalProblem:
|
|
asset: str
|
|
constraints: Dict[str, Any] = field(default_factory=dict)
|
|
objective: Dict[str, float] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
delta_id: str
|
|
changes: Dict[str, Any] = field(default_factory=dict)
|
|
timestamp: float = field(default_factory=lambda: time.time())
|
|
|
|
|
|
@dataclass
|
|
class SharedVariables:
|
|
variables: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
def to_nir(local_problems: List[LocalProblem], deltas: List[PlanDelta], shared: SharedVariables) -> Dict[str, Any]:
|
|
"""Translate a set of problems/deltas into a toy neuromorphic IR.
|
|
|
|
The real project would generate a graph of spiking neurons with temporal
|
|
dynamics encoding constraints; here we emit a deterministic, testable
|
|
toy representation for MVP validation and integration testing.
|
|
"""
|
|
nodes: List[Dict[str, Any]] = []
|
|
edges: List[Dict[str, Any]] = []
|
|
|
|
# Create nodes for LocalProblems
|
|
for idx, lp in enumerate(local_problems):
|
|
n = {
|
|
"id": f"LP:{idx}:{lp.asset}",
|
|
"type": "LocalProblem",
|
|
"asset": lp.asset,
|
|
"constraints": lp.constraints,
|
|
"objective": lp.objective,
|
|
}
|
|
nodes.append(n)
|
|
|
|
# Create nodes for each PlanDelta
|
|
for d in deltas:
|
|
n = {
|
|
"id": f"DELTA:{d.delta_id}",
|
|
"type": "PlanDelta",
|
|
"delta_id": d.delta_id,
|
|
"changes": d.changes,
|
|
"timestamp": d.timestamp,
|
|
}
|
|
nodes.append(n)
|
|
|
|
# Shared variables as a single hub node if present
|
|
if shared and shared.variables:
|
|
nodes.append({"id": "SharedVariables:root", "type": "SharedVariables", "payload": shared.variables})
|
|
|
|
# Simplified edges: connect LocalProblems to Delta changes if named in constraints
|
|
for lp in local_problems:
|
|
for d in deltas:
|
|
if lp.asset in (d.changes.get("requires", []) or []):
|
|
edges.append({"src": f"LP:{local_problems.index(lp)}:{lp.asset}", "dst": f"DELTA:{d.delta_id}"})
|
|
|
|
return {"nodes": nodes, "edges": edges}
|