224 lines
8.2 KiB
Python
224 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Compile process_spec.json into deterministic HIL logic.
|
|
|
|
Input: process_spec.json (ProcessSpec)
|
|
Output: Python HIL logic file implementing the physics model
|
|
|
|
Usage:
|
|
python3 -m tools.compile_process_spec \
|
|
--spec outputs/process_spec.json \
|
|
--out outputs/hil_logic.py
|
|
|
|
With config (to initialize all HIL outputs, not just physics-related):
|
|
python3 -m tools.compile_process_spec \
|
|
--spec outputs/process_spec.json \
|
|
--out outputs/hil_logic.py \
|
|
--config outputs/configuration.json
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
from typing import Dict, Optional, Set
|
|
|
|
from models.process_spec import ProcessSpec
|
|
|
|
|
|
def get_hil_output_keys(config: dict, hil_name: Optional[str] = None) -> Set[str]:
|
|
"""
|
|
Extract all io:"output" physical_values keys from HIL(s) in config.
|
|
|
|
If hil_name is provided, only return keys for that HIL.
|
|
Otherwise, return keys from all HILs (union).
|
|
"""
|
|
output_keys: Set[str] = set()
|
|
for hil in config.get("hils", []):
|
|
if hil_name and hil.get("name") != hil_name:
|
|
continue
|
|
for pv in hil.get("physical_values", []):
|
|
if pv.get("io") == "output":
|
|
key = pv.get("name")
|
|
if key:
|
|
output_keys.add(key)
|
|
return output_keys
|
|
|
|
|
|
def render_water_tank_v1(spec: ProcessSpec, extra_output_keys: Optional[Set[str]] = None) -> str:
|
|
"""
|
|
Render deterministic HIL logic for water_tank_v1 model.
|
|
|
|
Physics:
|
|
d(level)/dt = (Q_in - Q_out) / area
|
|
Q_in = q_in_max if valve_open >= 0.5 else 0
|
|
Q_out = k_out * sqrt(level)
|
|
|
|
Contract:
|
|
- Initialize all physical_values keys (including extra_output_keys from config)
|
|
- Read io:"input" keys (valve_open_key)
|
|
- Update io:"output" keys (tank_level_key, level_measured_key)
|
|
- Clamp level between min and max
|
|
|
|
Args:
|
|
spec: ProcessSpec with physics parameters
|
|
extra_output_keys: Additional output keys from config that need initialization
|
|
"""
|
|
p = spec.params
|
|
s = spec.signals
|
|
dt = spec.dt
|
|
|
|
# Collect all output keys that need initialization
|
|
physics_output_keys = {s.tank_level_key, s.level_measured_key}
|
|
all_output_keys = physics_output_keys | (extra_output_keys or set())
|
|
|
|
lines = []
|
|
lines.append('"""')
|
|
lines.append("HIL logic for water_tank_v1 process model.")
|
|
lines.append("")
|
|
lines.append("Autogenerated by ics-simlab-config-gen (compile_process_spec).")
|
|
lines.append("DO NOT EDIT - regenerate from process_spec.json instead.")
|
|
lines.append('"""')
|
|
lines.append("")
|
|
lines.append("import math")
|
|
lines.append("")
|
|
lines.append("")
|
|
lines.append("def _clamp(x: float, lo: float, hi: float) -> float:")
|
|
lines.append(" return lo if x < lo else hi if x > hi else x")
|
|
lines.append("")
|
|
lines.append("")
|
|
lines.append("def _as_float(x, default: float = 0.0) -> float:")
|
|
lines.append(" try:")
|
|
lines.append(" return float(x)")
|
|
lines.append(" except Exception:")
|
|
lines.append(" return default")
|
|
lines.append("")
|
|
lines.append("")
|
|
lines.append("def logic(physical_values):")
|
|
lines.append(" # === Process Parameters (from process_spec.json) ===")
|
|
lines.append(f" dt = {float(dt)}")
|
|
lines.append(f" level_min = {float(p.level_min)}")
|
|
lines.append(f" level_max = {float(p.level_max)}")
|
|
lines.append(f" level_init = {float(p.level_init)}")
|
|
lines.append(f" area = {float(p.area)}")
|
|
lines.append(f" q_in_max = {float(p.q_in_max)}")
|
|
lines.append(f" k_out = {float(p.k_out)}")
|
|
lines.append("")
|
|
lines.append(" # === Signal Keys ===")
|
|
lines.append(f" TANK_LEVEL_KEY = '{s.tank_level_key}'")
|
|
lines.append(f" VALVE_OPEN_KEY = '{s.valve_open_key}'")
|
|
lines.append(f" LEVEL_MEASURED_KEY = '{s.level_measured_key}'")
|
|
lines.append("")
|
|
lines.append(" # === Initialize all output physical_values ===")
|
|
lines.append(" # Physics outputs (with meaningful defaults)")
|
|
lines.append(f" physical_values.setdefault('{s.tank_level_key}', level_init)")
|
|
if s.level_measured_key != s.tank_level_key:
|
|
lines.append(f" physical_values.setdefault('{s.level_measured_key}', level_init)")
|
|
# Add initialization for extra output keys (from config)
|
|
extra_keys = sorted(all_output_keys - physics_output_keys)
|
|
if extra_keys:
|
|
lines.append(" # Other outputs from config (with zero defaults)")
|
|
for key in extra_keys:
|
|
lines.append(f" physical_values.setdefault('{key}', 0.0)")
|
|
lines.append("")
|
|
lines.append(" # === Read inputs ===")
|
|
lines.append(" valve_open = _as_float(physical_values.get(VALVE_OPEN_KEY, 0.0), 0.0)")
|
|
lines.append("")
|
|
lines.append(" # === Read current state ===")
|
|
lines.append(" level = _as_float(physical_values.get(TANK_LEVEL_KEY, level_init), level_init)")
|
|
lines.append("")
|
|
lines.append(" # === Physics: water tank dynamics ===")
|
|
lines.append(" # Inflow: Q_in = q_in_max if valve_open >= 0.5 else 0")
|
|
lines.append(" q_in = q_in_max if valve_open >= 0.5 else 0.0")
|
|
lines.append("")
|
|
lines.append(" # Outflow: Q_out = k_out * sqrt(level) (gravity-driven)")
|
|
lines.append(" q_out = k_out * math.sqrt(max(level, 0.0))")
|
|
lines.append("")
|
|
lines.append(" # Level change: d(level)/dt = (Q_in - Q_out) / area")
|
|
lines.append(" d_level = (q_in - q_out) / area * dt")
|
|
lines.append(" level = level + d_level")
|
|
lines.append("")
|
|
lines.append(" # Clamp to physical bounds")
|
|
lines.append(" level = _clamp(level, level_min, level_max)")
|
|
lines.append("")
|
|
lines.append(" # === Write outputs ===")
|
|
lines.append(" physical_values[TANK_LEVEL_KEY] = level")
|
|
lines.append(" physical_values[LEVEL_MEASURED_KEY] = level")
|
|
lines.append("")
|
|
lines.append(" return")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def compile_process_spec(spec: ProcessSpec, extra_output_keys: Optional[Set[str]] = None) -> str:
|
|
"""Compile ProcessSpec to HIL logic Python code."""
|
|
if spec.model == "water_tank_v1":
|
|
return render_water_tank_v1(spec, extra_output_keys)
|
|
else:
|
|
raise ValueError(f"Unsupported process model: {spec.model}")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Compile process_spec.json into HIL logic Python file"
|
|
)
|
|
parser.add_argument(
|
|
"--spec",
|
|
required=True,
|
|
help="Path to process_spec.json",
|
|
)
|
|
parser.add_argument(
|
|
"--out",
|
|
required=True,
|
|
help="Output path for HIL logic .py file",
|
|
)
|
|
parser.add_argument(
|
|
"--config",
|
|
default=None,
|
|
help="Path to configuration.json (to initialize all HIL output keys)",
|
|
)
|
|
parser.add_argument(
|
|
"--overwrite",
|
|
action="store_true",
|
|
help="Overwrite existing output file",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
spec_path = Path(args.spec)
|
|
out_path = Path(args.out)
|
|
config_path = Path(args.config) if args.config else None
|
|
|
|
if not spec_path.exists():
|
|
raise SystemExit(f"Spec file not found: {spec_path}")
|
|
if out_path.exists() and not args.overwrite:
|
|
raise SystemExit(f"Output file exists: {out_path} (use --overwrite)")
|
|
if config_path and not config_path.exists():
|
|
raise SystemExit(f"Config file not found: {config_path}")
|
|
|
|
spec_dict = json.loads(spec_path.read_text(encoding="utf-8"))
|
|
spec = ProcessSpec.model_validate(spec_dict)
|
|
|
|
# Get extra output keys from config if provided
|
|
extra_output_keys: Optional[Set[str]] = None
|
|
if config_path:
|
|
config = json.loads(config_path.read_text(encoding="utf-8"))
|
|
extra_output_keys = get_hil_output_keys(config)
|
|
print(f"Loading HIL output keys from config: {len(extra_output_keys)} keys")
|
|
|
|
print(f"Compiling process spec: {spec_path}")
|
|
print(f" Model: {spec.model}")
|
|
print(f" dt: {spec.dt}s")
|
|
|
|
code = compile_process_spec(spec, extra_output_keys)
|
|
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
out_path.write_text(code, encoding="utf-8")
|
|
print(f"Wrote: {out_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|