"""Flask app: JSON API + static wizard, everything same-origin.
The linearize and design endpoints deliberately go *through the YAML files*:
they write model.yaml / design.yaml to the session workdir and run the same
loaders the CLIs use, so exported artifacts are byte-identical to what
produced the numbers on screen.
"""
import math
import os
import warnings
import numpy as np
import yaml
from flask import Flask, Response, jsonify, request, send_file
from urdf_state_space.config import build_from_yaml
from . import benchmark as bench_mod
from .equilibrium import find_equilibrium
from .export import write_bundle
from .introspect import parse_robot
from .resources import load_urdf, safe_share_path
from .schemas import list_controllers
from .session import WizardSession
from .validation import validate_urdf
from .yamlgen import design_yaml_dict, dump_yaml, model_yaml_dict
MESH_TYPES = {
'.stl': 'application/octet-stream',
'.dae': 'model/vnd.collada+xml',
'.obj': 'text/plain',
'.png': 'image/png',
'.jpg': 'image/jpeg',
}
[docs]
def to_jsonable(obj):
"""numpy/complex/NaN-safe JSON conversion (NaN/Inf -> None)."""
if isinstance(obj, dict):
return {k: to_jsonable(v) for k, v in obj.items()}
if isinstance(obj, (list, tuple)):
return [to_jsonable(v) for v in obj]
if isinstance(obj, np.ndarray):
return to_jsonable(obj.tolist())
if isinstance(obj, (np.integer,)):
return int(obj)
if isinstance(obj, (np.floating,)):
obj = float(obj)
if isinstance(obj, (complex, np.complexfloating)):
return {'re': to_jsonable(float(obj.real)),
'im': to_jsonable(float(obj.imag))}
if isinstance(obj, float):
return obj if math.isfinite(obj) else None
if isinstance(obj, (np.bool_,)):
return bool(obj)
return obj
def _poles_json(poles) -> list:
return [{'re': to_jsonable(float(p.real)), 'im': to_jsonable(float(p.imag))}
for p in np.sort_complex(np.asarray(poles, dtype=complex))]
def _urdf_source_for_yaml(session: WizardSession) -> str:
"""What to write into model.yaml's model.urdf key: keep package:// URIs
and absolute paths as the user gave them; anything else must be the
resolved absolute path so the YAML works from any directory."""
src = session.urdf_source or ''
if src.startswith('package://') or os.path.isabs(src):
return src
return session.urdf_path
def _model_json(model, dt=None) -> dict:
state_names = [f'{j}.q' for j in model.joint_names] + \
[f'{j}.qd' for j in model.joint_names]
out = {
'A': to_jsonable(model.A), 'B': to_jsonable(model.B),
'C': to_jsonable(model.C), 'D': to_jsonable(model.D),
'poles': _poles_json(model.poles()),
'controllable': bool(model.is_controllable()),
'observable': bool(model.is_observable()),
'q_eq': to_jsonable(model.q_eq),
'u_eq': to_jsonable(model.u_eq),
'joint_names': list(model.joint_names),
'actuated_joint_names': list(model.actuated_joint_names),
'output_names': list(model.output_names),
'state_names': state_names,
'n_states': model.n_states,
'n_inputs': model.n_inputs,
'n_outputs': model.n_outputs,
'summary': model.summary(),
}
if dt:
disc = model.c2d(float(dt))
out['discrete'] = {'dt': float(dt), 'A': to_jsonable(disc.A),
'B': to_jsonable(disc.B),
'poles': _poles_json(np.linalg.eigvals(disc.A))}
return out
[docs]
def create_app(session: WizardSession = None, prefill_urdf: str = None) -> Flask:
app = Flask(__name__)
if session is None:
session = WizardSession()
app.config['SESSION'] = session
app.config['PREFILL_URDF'] = prefill_urdf
@app.errorhandler(ValueError)
@app.errorhandler(KeyError)
def bad_request(exc):
msg = str(exc) if not isinstance(exc, KeyError) \
else f'missing required field {exc.args[0]!r}'
return jsonify({'error': msg}), 400
@app.errorhandler(FileNotFoundError)
@app.errorhandler(LookupError)
def not_found(exc):
return jsonify({'error': str(exc)}), 404
@app.get('/')
def index():
from flask import render_template
return render_template('index.html')
@app.get('/api/state')
def state():
return jsonify(to_jsonable({
'prefill_urdf': app.config.get('PREFILL_URDF'),
'output_dir': session.output_dir,
'load': session.responses.get('load'),
'linearize': session.responses.get('linearize'),
'design': session.responses.get('design'),
'benchmark': session.responses.get('benchmark'),
'export': session.responses.get('export'),
}))
@app.post('/api/robot/load')
def robot_load():
body = request.get_json(force=True) or {}
source = body['source']
session.reset_robot()
path, urdf_xml = load_urdf(source)
robot = parse_robot(urdf_xml)
findings = validate_urdf(urdf_xml)
session.urdf_source = source
session.urdf_path = path
session.urdf_xml = urdf_xml
session.robot_name = robot.name
session.joints = robot.joints
session.validation = findings
payload = {
'robot_name': robot.name,
'links': robot.links,
'joints': [j.to_dict() for j in robot.joints],
'validation': findings,
'blocked': any(f['level'] == 'error' for f in findings),
}
session.responses['load'] = payload
return jsonify(to_jsonable(payload))
@app.get('/api/robot/urdf')
def robot_urdf():
if session.urdf_xml is None:
raise FileNotFoundError('no robot loaded')
return Response(session.urdf_xml, mimetype='text/xml')
@app.get('/pkg/<package>/<path:relative>')
def package_resource(package, relative):
path = safe_share_path(package, relative) # LookupError -> 404
if path is None:
raise FileNotFoundError(
f'{relative!r} not found in package {package!r}')
ext = os.path.splitext(path)[1].lower()
return send_file(
path, mimetype=MESH_TYPES.get(ext, 'application/octet-stream'))
@app.post('/api/model/equilibrium')
def model_equilibrium():
if session.urdf_xml is None:
raise ValueError('load a robot first')
body = request.get_json(force=True) or {}
out = find_equilibrium(
session.urdf_xml,
actuated_joints=body['actuated'],
q_seed=body.get('q_seed') or {},
floating_base=bool(body.get('floating_base', False)),
)
return jsonify(to_jsonable(out))
@app.post('/api/model/linearize')
def model_linearize():
if session.urdf_xml is None:
raise ValueError('load a robot first')
body = request.get_json(force=True) or {}
outputs = body.get('outputs') or {}
cfg = model_yaml_dict(
urdf_source=_urdf_source_for_yaml(session),
joints=body.get('joints') or {},
velocities=bool(outputs.get('velocities', False)),
output_joints=outputs.get('joints'),
floating_base=bool(body.get('floating_base', False)),
)
cfg_path = os.path.join(session.workdir, 'model.yaml')
dump_yaml(cfg, cfg_path)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter('always')
model, _ = build_from_yaml(cfg_path)
session.model = model
session.model_cfg = cfg
session.design_cfg = None
session.result = None
session.benchmark = None
for key in ('design', 'benchmark', 'export'):
session.responses.pop(key, None)
payload = _model_json(model, dt=body.get('dt'))
payload['warnings'] = [str(w.message) for w in caught]
session.responses['linearize'] = payload
return jsonify(to_jsonable(payload))
@app.get('/api/controllers')
def controllers():
return jsonify(to_jsonable({'controllers': list_controllers()}))
@app.post('/api/controller/design')
def controller_design():
if session.model is None:
raise ValueError('linearize a model first')
body = request.get_json(force=True) or {}
cfg = design_yaml_dict(body['controller'], body.get('params') or {})
cfg_path = os.path.join(session.workdir, 'design.yaml')
dump_yaml(cfg, cfg_path)
# Read the file back and design exactly the way ss_design does, so
# the exported design.yaml provably reproduces this result.
from state_space_control.base import Plant
with open(cfg_path) as f:
spec = yaml.safe_load(f) or {}
plant = Plant.from_model(session.model)
try:
result = bench_mod.design_controller(
plant, spec['controller'], spec.get('params') or {},
timeout=app.config.get('SYNTH_TIMEOUT',
bench_mod.SYNTH_TIMEOUT))
except ValueError:
raise
except Exception as exc: # synthesis failures (e.g. slycot) -> 400
msg = f"{spec['controller']} synthesis failed: {exc}".strip()
if 'rank' in str(exc):
msg += ('\nHint: a joint with no restoring torque (e.g. a '
'free azimuth axis) gives the plant a pure-integrator '
'pole at s=0, and standard mixed-sensitivity H∞ '
'assumptions exclude imaginary-axis poles. Common '
'workarounds: stabilize with an inner loop first, or '
'apply a small pole-shift/bilinear transform before '
'synthesis.')
raise ValueError(msg) from exc
session.design_cfg = cfg
session.result = result
session.responses.pop('export', None)
metrics = bench_mod.simulate_metrics(result)
from state_space_control.analysis import damping_report
payload = {
'controller': spec['controller'],
'K': to_jsonable(result.K) if result.K is not None else None,
'dynamic': None if result.controller is None else {
'A': to_jsonable(result.controller.A),
'B': to_jsonable(result.controller.B),
'C': to_jsonable(result.controller.C),
'D': to_jsonable(result.controller.D),
'n_states': result.controller.n_states,
},
'open_loop_poles': _poles_json(session.model.poles()),
'closed_loop_poles': _poles_json(result.closed_loop_poles()),
'stable': bool(result.is_stable()),
'info': to_jsonable({k: v for k, v in result.info.items()
if np.isscalar(v)}),
'summary': result.summary(),
'damping_report': damping_report(result.closed_loop()),
'step': {
't': to_jsonable(metrics['t']),
'y': to_jsonable(metrics['y']),
'u': to_jsonable(metrics['u']),
'output_names': list(session.model.output_names),
'settling_time': to_jsonable(metrics['settling_time']),
'overshoot_pct': to_jsonable(metrics['overshoot_pct']),
'control_effort': to_jsonable(metrics['control_effort']),
},
'state_names': session.responses['linearize']['state_names']
if 'linearize' in session.responses else [],
}
session.responses['design'] = payload
return jsonify(to_jsonable(payload))
@app.post('/api/benchmark')
def run_benchmark():
if session.model is None:
raise ValueError('linearize a model first')
body = request.get_json(force=True) or {}
entries = body.get('entries') or []
if not entries:
raise ValueError('benchmark needs at least one controller entry')
from state_space_control.base import Plant
plant = Plant.from_model(session.model)
out = bench_mod.run_benchmark(
plant, entries,
timeout=app.config.get('SYNTH_TIMEOUT', bench_mod.SYNTH_TIMEOUT))
session.benchmark = out
payload = to_jsonable(out)
session.responses['benchmark'] = payload
return jsonify(payload)
@app.post('/api/export')
def export():
body = request.get_json(force=True) or {}
output_dir = body.get('output_dir') or session.output_dir
if not output_dir:
raise ValueError('output_dir is required')
out = write_bundle(
session, output_dir,
include_mat=bool(body.get('include_mat', False)),
dt=body.get('dt'))
payload = to_jsonable(out)
session.responses['export'] = payload
return jsonify(payload)
return app