198 lines
6.6 KiB
Python
198 lines
6.6 KiB
Python
"""Generation entrypoint for MVP adapters.
|
|
|
|
This module emits two domain adapter skeletons (Energy and Robotics) in Python,
|
|
along with minimal harnesses (solver, conformance, simulator) and a simple
|
|
GoC registry entry. The goal is a minimal, testable scaffold that demonstrates
|
|
the pipeline without requiring a full runtime environment.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Dict, Any
|
|
from dataclasses import is_dataclass, asdict
|
|
|
|
from .dsl import GraphOfContractsSeed, LocalProblem, SharedVariables, PlanDelta, DualVariables, PrivacyBudget, AuditLog, PolicyBlock
|
|
|
|
|
|
def _ensure_dir(path: str) -> None:
|
|
Path(path).mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def _write_file(path: str, content: str) -> None:
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(content)
|
|
|
|
|
|
def _energy_adapter_code() -> str:
|
|
return '''# Energy Adapter Skeleton (TLS-ready)
|
|
import ssl
|
|
from typing import Optional, Dict, Any
|
|
|
|
try:
|
|
# Lightweight local-problem placeholder to avoid external imports
|
|
from dataclasses import dataclass
|
|
@dataclass
|
|
class LocalProblem:
|
|
domain: str
|
|
objective: str
|
|
constraints: dict = None
|
|
except Exception:
|
|
class LocalProblem: # type: ignore
|
|
pass
|
|
|
|
|
|
class EnergyAdapter:
|
|
def __init__(self, config: Optional[Dict[str, Any]] = None):
|
|
self.config = config or {}
|
|
self._tls_context = None
|
|
|
|
def _build_tls_context(self, certfile: str = None, keyfile: str = None) -> ssl.SSLContext:
|
|
ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
|
|
if certfile and keyfile:
|
|
ctx.load_cert_chain(certfile=certfile, keyfile=keyfile)
|
|
self._tls_context = ctx
|
|
return ctx
|
|
|
|
def configure_tls(self, certfile: str, keyfile: str) -> ssl.SSLContext:
|
|
return self._build_tls_context(certfile, keyfile)
|
|
|
|
def start(self, host: str = "127.0.0.1", port: int = 50051):
|
|
# Minimal TLS-ready skeleton; does not bind in MVP tests
|
|
if self._tls_context is None:
|
|
self._build_tls_context()
|
|
return {"status": "initialized", "host": host, "port": port}
|
|
|
|
def solve_local_problem(lp: LocalProblem) -> Dict[str, Any]:
|
|
return {
|
|
"domain": "Energy",
|
|
"lp_objective": getattr(lp, "objective", "minimize_cost"),
|
|
"solution": {k: f"{v}-solved" for k, v in getattr(lp, 'constraints', {}).items()},
|
|
}
|
|
'''
|
|
|
|
|
|
def _robotics_adapter_code() -> str:
|
|
return '''# Robotics Adapter Skeleton (TLS-ready)
|
|
import ssl
|
|
from typing import Optional, Dict, Any
|
|
|
|
try:
|
|
from dataclasses import dataclass
|
|
@dataclass
|
|
class LocalProblem:
|
|
domain: str
|
|
objective: str
|
|
constraints: dict = None
|
|
except Exception:
|
|
class LocalProblem: # type: ignore
|
|
pass
|
|
|
|
|
|
class RoboticsAdapter:
|
|
def __init__(self, config: Optional[Dict[str, Any]] = None):
|
|
self.config = config or {}
|
|
self._tls_context = None
|
|
|
|
def _build_tls_context(self, certfile: str = None, keyfile: str = None) -> ssl.SSLContext:
|
|
ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
|
|
if certfile and keyfile:
|
|
ctx.load_cert_chain(certfile=certfile, keyfile=keyfile)
|
|
self._tls_context = ctx
|
|
return ctx
|
|
|
|
def configure_tls(self, certfile: str, keyfile: str) -> ssl.SSLContext:
|
|
return self._build_tls_context(certfile, keyfile)
|
|
|
|
def start(self, host: str = "127.0.0.1", port: int = 50052):
|
|
if self._tls_context is None:
|
|
self._build_tls_context()
|
|
return {"status": "initialized", "host": host, "port": port}
|
|
|
|
def solve_local_problem(lp: LocalProblem) -> Dict[str, Any]:
|
|
return {
|
|
"domain": "Robotics",
|
|
"lp_objective": getattr(lp, "objective", "maximize_performance"),
|
|
"solution": {k: f"{v}-solved" for k, v in getattr(lp, 'constraints', {}).items()},
|
|
}
|
|
'''
|
|
|
|
|
|
def _conformance_harness_code() -> str:
|
|
return '''# Minimal conformance harness
|
|
def verify_adapter_shape(adapter):
|
|
# Placeholder: in real system, check required methods exist
|
|
required = ["start", "configure_tls"]
|
|
return all(hasattr(adapter, r) for r in required)
|
|
'''
|
|
|
|
|
|
def _serialize(obj: Any) -> Any:
|
|
"""Recursively convert dataclass instances to serializable dicts.
|
|
|
|
This allows storing complex DSL seed structures (which may contain
|
|
dataclass instances) inside JSON without requiring custom encoders.
|
|
"""
|
|
if is_dataclass(obj):
|
|
return asdict(obj)
|
|
if isinstance(obj, dict):
|
|
return {k: _serialize(v) for k, v in obj.items()}
|
|
if isinstance(obj, list):
|
|
return [_serialize(v) for v in obj]
|
|
return obj
|
|
|
|
|
|
def _registry_entry(seed: GraphOfContractsSeed) -> Dict[str, Any]:
|
|
return {
|
|
"name": seed.name,
|
|
"domains": ["Energy", "Robotics"],
|
|
"path": {
|
|
"energy_adapter": "adapters/energy_adapter.py",
|
|
"robotics_adapter": "adapters/robotics_adapter.py",
|
|
},
|
|
"seed_summary": _serialize(seed.seeds) if seed.seeds else {},
|
|
}
|
|
|
|
|
|
def generate_mvp_adapters(seed: GraphOfContractsSeed, out_dir: str = "adapters") -> Dict[str, Any]:
|
|
"""Generate MVP skeleton adapters for two domains.
|
|
|
|
- Creates two Python adapter modules: energy_adapter.py and robotics_adapter.py
|
|
- Includes helper modules: solver.py, conformance.py, sim.py
|
|
- Creates a simple registry entry in registry/registry.json
|
|
Returns a mapping of generated file paths.
|
|
"""
|
|
_ensure_dir(out_dir)
|
|
# Energy adapter
|
|
energy_path = os.path.join(out_dir, "energy_adapter.py")
|
|
robotics_path = os.path.join(out_dir, "robotics_adapter.py")
|
|
_write_file(energy_path, _energy_adapter_code())
|
|
_write_file(robotics_path, _robotics_adapter_code())
|
|
|
|
# Conformance & solver skeletons
|
|
conformance_path = os.path.join(out_dir, "conformance.py")
|
|
solver_path = os.path.join(out_dir, "solver.py")
|
|
sim_path = os.path.join(out_dir, "sim.py")
|
|
_write_file(conformance_path, _conformance_harness_code())
|
|
_write_file(solver_path, """# Placeholder solver module (could be extended)\n""")
|
|
_write_file(sim_path, """# Placeholder simulator module (Two-Domain demo)\n""")
|
|
|
|
# Registry file
|
|
registry_dir = "registry"
|
|
_ensure_dir(registry_dir)
|
|
reg_path = os.path.join(registry_dir, "registry.json")
|
|
reg_entry = _registry_entry(seed)
|
|
with open(reg_path, "w", encoding="utf-8") as f:
|
|
json.dump(reg_entry, f, indent=2)
|
|
|
|
return {
|
|
"energy_adapter": energy_path,
|
|
"robotics_adapter": robotics_path,
|
|
"conformance": conformance_path,
|
|
"solver": solver_path,
|
|
"simulation": sim_path,
|
|
"registry": reg_path,
|
|
"registry_entry": reg_entry,
|
|
}
|