diff --git a/scripts/endpoint_tcn.py b/scripts/endpoint_tcn.py new file mode 100644 index 0000000..bd5d331 --- /dev/null +++ b/scripts/endpoint_tcn.py @@ -0,0 +1,78 @@ +""" +ONNX-only inference script for Raspberry Pi deployment. This script is for using + +""" + +import onnxruntime +import numpy as np +import argparse +from pathlib import Path +import serial +from collections import deque + +MODEL_PATH = "outputs/model.onnx" +SERIAL_PORT = "/dev/ttyACM0" +BAUD_RATE = 115200 +FEATURE_COUNT = 28 +WINDOW_SIZE = 100 + +def main(): + + + if not Path(MODEL_PATH).exists(): + print(f"Error: Model file '{MODEL_PATH}' not found.") + return + + print(f"Loading model: {MODEL_PATH}") + + session = onnxruntime.InferenceSession(MODEL_PATH, providers=["CPUExecutionProvider"]) + input_name = session.get_inputs()[0].name + history = deque(maxlen=WINDOW_SIZE) + for _ in range(WINDOW_SIZE): + history.append(np.zeros(FEATURE_COUNT)) + + try: + ser = serial.Serial(SERIAL_PORT, 115200, timeout=1) + ser.flush() + except Exception as e: + print(f"Serial Error: {e}") + return + + # 3. Run Inference + + while True: + if ser.in_waiting > 0: + try: + line = ser.readline().decode('utf-8').strip() + if not line: continue + + # Parse the latest single reading from Arduino + new_sample = [float(x) for x in line.split(',')] + if len(new_sample) != FEATURE_COUNT: continue + + # 2. Add new sample to history (automatically drops the oldest) + history.append(new_sample) + + # 3. Convert history to the shape the TCN expects: (1, 100, 28) + input_data = np.array(history, dtype=np.float32).reshape(1, WINDOW_SIZE, FEATURE_COUNT) + + # 4. Run Inference + outputs = session.run(None, {input_name: input_data}) + predictions = outputs[0] # Shape: (1, 100, 4) + + # 5. Extract the LATEST prediction for the 4 motors + # We take index -1 (the most recent time step) + current_pred = predictions[0, -1, :] + + # 6. Send all 4 motor values back as a comma-separated string + pred_str = ",".join([f"{x:.4f}" for x in current_pred]) + ser.write(f"{pred_str}\n".encode('utf-8')) + + except Exception as e: + print(f"Error: {e}") + +if __name__ == "__main__": + main() + + + \ No newline at end of file diff --git a/scripts/export_onnx.py b/scripts/export_onnx.py new file mode 100644 index 0000000..f4904d8 --- /dev/null +++ b/scripts/export_onnx.py @@ -0,0 +1,115 @@ +""" +Export a trained PyTorch TCN checkpoint to ONNX format for Raspberry Pi deployment. + +Usage: + python scripts/export_onnx.py + python scripts/export_onnx.py --model-path outputs/2026-01-08/18-29-05/best_model.pt + python scripts/export_onnx.py --output outputs/model.onnx +""" + +import argparse +from pathlib import Path + +import numpy as np +import onnx +import onnxruntime +import torch +from omegaconf import OmegaConf + +from exoskeleton_ml.models import create_model + +DEFAULT_MODEL_PATH = "outputs/2026-01-08/18-29-05/best_model.pt" + + +def load_model(model_path: Path) -> tuple[torch.nn.Module, int]: + checkpoint = torch.load(model_path, map_location="cpu") + + config_path = model_path.parent / "config.yaml" + if not config_path.exists(): + raise ValueError(f"Config not found at {config_path}") + + config = OmegaConf.load(config_path) + model = create_model(config.model) + model.load_state_dict(checkpoint["model_state_dict"]) + model.eval() + + window_size: int = config.model.effective_history + return model, window_size + + +def export(model_path: Path, output_path: Path) -> None: + print(f"Loading model: {model_path}") + model, window_size = load_model(model_path) + + num_params = sum(p.numel() for p in model.parameters()) + print(f" Parameters: {num_params:,}") + print(f" Window size: {window_size} timesteps") + print(f" Input shape: (1, {window_size}, 28)") + print(f" Output shape: (1, {window_size}, 4)") + + # Representative input for tracing + dummy_input = torch.randn(1, window_size, 28) + + with torch.no_grad(): + pytorch_output = model(dummy_input) + + print(f"\nExporting to ONNX: {output_path}") + output_path.parent.mkdir(parents=True, exist_ok=True) + + onnx_program = torch.onnx.export(model, dummy_input, dynamo=True) + onnx_program.save(str(output_path)) + + print("Checking ONNX model...") + onnx.checker.check_model(str(output_path)) + print(" ONNX check passed.") + + print("\nVerifying outputs match PyTorch...") + session_options = onnxruntime.SessionOptions() + session_options.graph_optimization_level = onnxruntime.GraphOptimizationLevel.ORT_ENABLE_ALL + ort_session = onnxruntime.InferenceSession( + str(output_path), + sess_options=session_options, + providers=["CPUExecutionProvider"], + ) + + input_name = ort_session.get_inputs()[0].name + onnx_output = ort_session.run(None, {input_name: dummy_input.numpy()})[0] + + try: + torch.testing.assert_close( + pytorch_output, + torch.tensor(onnx_output), + rtol=1e-3, + atol=1e-4, + ) + print(" Outputs match within tolerance. Export verified.") + except AssertionError as e: + print(f" WARNING: outputs differ: {e}") + + size_mb = output_path.stat().st_size / (1024 * 1024) + print(f"\nDone. Saved to: {output_path} ({size_mb:.2f} MB)") + print(f"\nSCP to Pi:") + print(f" scp {output_path} pi@:~/exoskeleton-ai/outputs/model.onnx") + + +def main() -> None: + parser = argparse.ArgumentParser(description="Export TCN checkpoint to ONNX") + parser.add_argument( + "--model-path", + type=str, + default=DEFAULT_MODEL_PATH, + help="Path to best_model.pt", + ) + parser.add_argument( + "--output", + type=str, + default="outputs/model.onnx", + help="Output path for the .onnx file", + ) + args = parser.parse_args() + + export(Path(args.model_path), Path(args.output)) + + +if __name__ == "__main__": + main() diff --git a/scripts/server.py b/scripts/server.py new file mode 100644 index 0000000..f2f75e2 --- /dev/null +++ b/scripts/server.py @@ -0,0 +1,175 @@ +""" +FastAPI inference server for the exoskeleton TCN model. + +Runs on the Raspberry Pi. The C++ control system sends a POST request to +/predict with a full (187, 28) window of sensor data and receives 4 motor +torque predictions in response. + +Usage: + uvicorn scripts.server:app --host 0.0.0.0 --port 8000 + uvicorn scripts.server:app --host 127.0.0.1 --port 8000 # localhost only + +Dependencies (not in pyproject.toml, install on Pi): + pip install fastapi uvicorn[standard] onnxruntime numpy msgpack + +Input shape: (1, WINDOW_SIZE, 28) — WINDOW_SIZE defaults to 187 for standard TCN +Output shape: (1, WINDOW_SIZE, 4) — server returns only the final timestep's 4 values +""" + +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Annotated + +import time + +import msgpack +import numpy as np +import onnxruntime as ort +from fastapi import FastAPI, HTTPException, Request, Response +from pydantic import BaseModel, Field, model_validator + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +MODEL_PATH = "outputs/model.onnx" + +# Must match the effective receptive field of the deployed model variant: +# standard TCN (kernel=7, 5 layers): 187 +# small TCN (kernel=5, 4 layers): 61 +# large TCN (kernel=9, 6 layers): 505 +WINDOW_SIZE = 187 +FEATURE_COUNT = 28 # 24 IMU + 4 joint angles +OUTPUT_COUNT = 4 # hip_l, hip_r, knee_l, knee_r (Nm/kg) + +# --------------------------------------------------------------------------- +# App lifespan: load model once at startup +# --------------------------------------------------------------------------- + +_session: ort.InferenceSession | None = None +_input_name: str | None = None + + +@asynccontextmanager +async def lifespan(app: FastAPI): + global _session, _input_name + + model_path = Path(MODEL_PATH) + if not model_path.exists(): + raise RuntimeError(f"Model file not found: {model_path.resolve()}") + + print(f"Loading ONNX model: {model_path.resolve()}") + _session = ort.InferenceSession(str(model_path), providers=["CPUExecutionProvider"]) + _input_name = _session.get_inputs()[0].name + print(f"Model loaded. Input: '{_input_name}' | Expected shape: (1, {WINDOW_SIZE}, {FEATURE_COUNT})") + + yield + + _session = None + _input_name = None + print("Server shutting down.") + + +app = FastAPI( + title="Exoskeleton TCN Inference Server", + lifespan=lifespan, +) + +# --------------------------------------------------------------------------- +# Request / response schemas +# --------------------------------------------------------------------------- + + +class PredictRequest(BaseModel): + # List of WINDOW_SIZE rows, each with FEATURE_COUNT values. + # C++ sends: {"window": [[f0..f27], [f0..f27], ...]} (187 rows) + window: Annotated[ + list[list[float]], + Field(description=f"Sensor window: {WINDOW_SIZE} timesteps × {FEATURE_COUNT} features"), + ] + + @model_validator(mode="after") + def validate_shape(self) -> "PredictRequest": + rows = len(self.window) + if rows != WINDOW_SIZE: + raise ValueError(f"Expected {WINDOW_SIZE} timesteps, got {rows}") + for i, row in enumerate(self.window): + if len(row) != FEATURE_COUNT: + raise ValueError( + f"Timestep {i}: expected {FEATURE_COUNT} features, got {len(row)}" + ) + return self + + +class PredictResponse(BaseModel): + # Torque predictions for the 4 joints at the latest timestep (Nm/kg) + hip_left: float + hip_right: float + knee_left: float + knee_right: float + inference_ms: float # ONNX session.run() time only, excludes HTTP/JSON overhead + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +def _run_inference(window_bytes: np.ndarray) -> tuple[np.ndarray, float]: + """Shared inference logic. Input must already be shaped (1, WINDOW_SIZE, FEATURE_COUNT).""" + t0 = time.perf_counter() + outputs = _session.run(None, {_input_name: window_bytes}) # type: ignore[index] + inference_ms = (time.perf_counter() - t0) * 1000 + preds = outputs[0][0, -1, :] + return preds, inference_ms + + +@app.get("/health") +def health() -> dict[str, str]: + """Quick liveness check — the C++ side can poll this on startup.""" + if _session is None: + raise HTTPException(status_code=503, detail="Model not loaded") + return {"status": "ok"} + + +@app.post("/predict", response_model=PredictResponse) +def predict(req: PredictRequest) -> PredictResponse: + if _session is None or _input_name is None: + raise HTTPException(status_code=503, detail="Model not loaded") + + input_array = np.array(req.window, dtype=np.float32)[np.newaxis, ...] + preds, inference_ms = _run_inference(input_array) + + return PredictResponse( + hip_left=float(preds[0]), + hip_right=float(preds[1]), + knee_left=float(preds[2]), + knee_right=float(preds[3]), + inference_ms=round(inference_ms, 3), + ) + + +@app.post("/predict_msgpack") +async def predict_msgpack(request: Request) -> Response: + """ + Msgpack endpoint for lower-overhead inference. + + Request body: msgpack-encoded flat list of 187*28=5236 float32 values (row-major) + Response body: msgpack-encoded dict {hip_left, hip_right, knee_left, knee_right, inference_ms} + """ + if _session is None or _input_name is None: + raise HTTPException(status_code=503, detail="Model not loaded") + + raw = await request.body() + flat = msgpack.unpackb(raw, raw=False) + input_array = np.array(flat, dtype=np.float32).reshape(1, WINDOW_SIZE, FEATURE_COUNT) + preds, inference_ms = _run_inference(input_array) + + result = { + "hip_left": float(preds[0]), + "hip_right": float(preds[1]), + "knee_left": float(preds[2]), + "knee_right": float(preds[3]), + "inference_ms": round(inference_ms, 3), + } + return Response(content=msgpack.packb(result), media_type="application/x-msgpack") diff --git a/scripts/test_endpoint.cpp b/scripts/test_endpoint.cpp new file mode 100644 index 0000000..0c03b90 --- /dev/null +++ b/scripts/test_endpoint.cpp @@ -0,0 +1,336 @@ +/** + * test_endpoint.cpp + * + * C++ test client for the exoskeleton TCN inference server. + * Tests both JSON and msgpack endpoints and compares latency. + * + * Dependencies (install on Pi): + * sudo apt install libcurl4-openssl-dev nlohmann-json3-dev libmsgpack-dev + * + * Compile: + * g++ -O2 -std=c++17 test_endpoint.cpp -lcurl -o test_endpoint + * + * Run: + * ./test_endpoint + * ./test_endpoint --runs 50 + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +// --------------------------------------------------------------------------- +// Config — must match server.py +// --------------------------------------------------------------------------- + +static const char* HEALTH_URL = "http://127.0.0.1:8000/health"; +static const char* JSON_URL = "http://127.0.0.1:8000/predict"; +static const char* MSGPACK_URL = "http://127.0.0.1:8000/predict_msgpack"; +static const int WINDOW_SIZE = 187; +static const int FEATURE_COUNT = 28; +static const int DEFAULT_RUNS = 20; +static const double BUDGET_MS = 10.0; // 100 Hz + +// --------------------------------------------------------------------------- +// libcurl helpers +// --------------------------------------------------------------------------- + +struct CurlResponse { + std::string body; + long http_code = 0; +}; + +static size_t write_callback(char* ptr, size_t size, size_t nmemb, void* userdata) { + auto* resp = static_cast(userdata); + resp->body.append(ptr, size * nmemb); + return size * nmemb; +} + +static CurlResponse http_get(CURL* curl, const std::string& url) { + CurlResponse resp; + curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); + curl_easy_setopt(curl, CURLOPT_HTTPGET, 1L); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_callback); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, &resp); + curl_easy_setopt(curl, CURLOPT_TIMEOUT, 5L); + curl_easy_perform(curl); + curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &resp.http_code); + return resp; +} + +static CurlResponse http_post(CURL* curl, const std::string& url, + const char* data, size_t size, const char* content_type) { + CurlResponse resp; + struct curl_slist* headers = nullptr; + headers = curl_slist_append(headers, content_type); + + curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); + curl_easy_setopt(curl, CURLOPT_POST, 1L); + curl_easy_setopt(curl, CURLOPT_POSTFIELDS, data); + curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, (long)size); + curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_callback); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, &resp); + curl_easy_setopt(curl, CURLOPT_TIMEOUT, 5L); + curl_easy_perform(curl); + curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &resp.http_code); + + curl_slist_free_all(headers); + return resp; +} + +// --------------------------------------------------------------------------- +// JSON request/response +// --------------------------------------------------------------------------- + +static std::string build_json_request(const std::vector& flat) { + nlohmann::json window = nlohmann::json::array(); + for (int t = 0; t < WINDOW_SIZE; ++t) { + nlohmann::json row = nlohmann::json::array(); + for (int f = 0; f < FEATURE_COUNT; ++f) { + row.push_back(flat[t * FEATURE_COUNT + f]); + } + window.push_back(row); + } + return nlohmann::json{{"window", window}}.dump(); +} + +struct Prediction { + double hip_left, hip_right, knee_left, knee_right, inference_ms; +}; + +static Prediction parse_json_response(const std::string& body) { + auto j = nlohmann::json::parse(body); + return { + j["hip_left"].get(), + j["hip_right"].get(), + j["knee_left"].get(), + j["knee_right"].get(), + j["inference_ms"].get(), + }; +} + +// --------------------------------------------------------------------------- +// Msgpack request/response +// The server expects: flat array of WINDOW_SIZE * FEATURE_COUNT float32 values +// The server returns: map with hip_left, hip_right, knee_left, knee_right, inference_ms +// --------------------------------------------------------------------------- + +static std::string build_msgpack_request(const std::vector& flat) { + msgpack::sbuffer buf; + msgpack::pack(buf, flat); + return std::string(buf.data(), buf.size()); +} + +static Prediction parse_msgpack_response(const std::string& body) { + msgpack::object_handle oh = msgpack::unpack(body.data(), body.size()); + msgpack::object obj = oh.get(); + + // Response is a map — convert to std::map to look up keys + std::map result; + obj.convert(result); + + return { + result["hip_left"], + result["hip_right"], + result["knee_left"], + result["knee_right"], + result["inference_ms"], + }; +} + +// --------------------------------------------------------------------------- +// Stats helpers +// --------------------------------------------------------------------------- + +static double mean(const std::vector& v) { + return std::accumulate(v.begin(), v.end(), 0.0) / v.size(); +} + +static double percentile(std::vector v, double p) { + std::sort(v.begin(), v.end()); + size_t idx = static_cast(std::ceil(p / 100.0 * v.size())) - 1; + return v[std::min(idx, v.size() - 1)]; +} + +static double vmin(const std::vector& v) { return *std::min_element(v.begin(), v.end()); } +static double vmax(const std::vector& v) { return *std::max_element(v.begin(), v.end()); } + +static std::string fmt(double v) { + std::ostringstream s; + s.precision(1); + s << std::fixed << v; + return s.str(); +} + +static void print_table_row(const char* label, + double json_rt, double mp_rt, + double inference, + double json_oh, double mp_oh) { + printf(" %-8s %10s %10s %10s %10s %10s ms\n", + label, + fmt(json_rt).c_str(), fmt(mp_rt).c_str(), + fmt(inference).c_str(), + fmt(json_oh).c_str(), fmt(mp_oh).c_str()); +} + +// --------------------------------------------------------------------------- +// Main +// --------------------------------------------------------------------------- + +int main(int argc, char* argv[]) { + int num_runs = DEFAULT_RUNS; + for (int i = 1; i < argc; ++i) { + if (std::string(argv[i]) == "--runs" && i + 1 < argc) { + num_runs = std::stoi(argv[++i]); + } + } + + curl_global_init(CURL_GLOBAL_DEFAULT); + CURL* curl = curl_easy_init(); + if (!curl) { + std::cerr << "Failed to init curl\n"; + return 1; + } + + std::cout << std::string(60, '=') << "\n"; + std::cout << "Exoskeleton C++ Endpoint Test\n"; + std::cout << std::string(60, '=') << "\n"; + + // ------------------------------------------------------------------ + // [1] Health check + // ------------------------------------------------------------------ + std::cout << "\n[1] Health check...\n"; + auto health_resp = http_get(curl, HEALTH_URL); + if (health_resp.http_code != 200) { + std::cerr << " Server not ready (HTTP " << health_resp.http_code << ")\n"; + std::cerr << " Is the server running?\n"; + curl_easy_cleanup(curl); + curl_global_cleanup(); + return 1; + } + std::cout << " OK\n"; + + // ------------------------------------------------------------------ + // [2] Single prediction — JSON, zero input + // ------------------------------------------------------------------ + std::cout << "\n[2] Single prediction — JSON (zero input)...\n"; + { + std::vector flat(WINDOW_SIZE * FEATURE_COUNT, 0.0f); + std::string body = build_json_request(flat); + auto resp = http_post(curl, JSON_URL, body.c_str(), body.size(), + "Content-Type: application/json"); + if (resp.http_code != 200) { + std::cerr << " Failed (HTTP " << resp.http_code << "): " << resp.body << "\n"; + curl_easy_cleanup(curl); + curl_global_cleanup(); + return 1; + } + auto p = parse_json_response(resp.body); + printf(" hip_left: %f Nm/kg\n", p.hip_left); + printf(" hip_right: %f Nm/kg\n", p.hip_right); + printf(" knee_left: %f Nm/kg\n", p.knee_left); + printf(" knee_right: %f Nm/kg\n", p.knee_right); + printf(" inference: %.3f ms\n", p.inference_ms); + } + + // ------------------------------------------------------------------ + // [3] Single prediction — msgpack, zero input + // ------------------------------------------------------------------ + std::cout << "\n[3] Single prediction — msgpack (zero input)...\n"; + { + std::vector flat(WINDOW_SIZE * FEATURE_COUNT, 0.0f); + std::string body = build_msgpack_request(flat); + auto resp = http_post(curl, MSGPACK_URL, body.c_str(), body.size(), + "Content-Type: application/x-msgpack"); + if (resp.http_code != 200) { + std::cerr << " Failed (HTTP " << resp.http_code << "): " << resp.body << "\n"; + curl_easy_cleanup(curl); + curl_global_cleanup(); + return 1; + } + auto p = parse_msgpack_response(resp.body); + printf(" hip_left: %f Nm/kg\n", p.hip_left); + printf(" hip_right: %f Nm/kg\n", p.hip_right); + printf(" knee_left: %f Nm/kg\n", p.knee_left); + printf(" knee_right: %f Nm/kg\n", p.knee_right); + printf(" inference: %.3f ms\n", p.inference_ms); + } + + // ------------------------------------------------------------------ + // [4] Latency benchmark — JSON vs msgpack + // ------------------------------------------------------------------ + std::cout << "\n[4] Latency benchmark — JSON vs msgpack (" << num_runs << " runs each)...\n"; + { + std::mt19937 rng(42); + std::normal_distribution dist(0.0f, 1.0f); + std::vector flat(WINDOW_SIZE * FEATURE_COUNT); + for (auto& v : flat) v = dist(rng); + + std::string json_body = build_json_request(flat); + std::string msgpack_body = build_msgpack_request(flat); + + std::vector json_rt, json_inf; + std::vector mp_rt, mp_inf; + + std::cout << " JSON...\n"; + for (int i = 0; i < num_runs; ++i) { + auto t0 = std::chrono::high_resolution_clock::now(); + auto resp = http_post(curl, JSON_URL, json_body.c_str(), json_body.size(), + "Content-Type: application/json"); + auto t1 = std::chrono::high_resolution_clock::now(); + json_rt.push_back(std::chrono::duration(t1 - t0).count()); + json_inf.push_back(parse_json_response(resp.body).inference_ms); + if ((i + 1) % 5 == 0) std::cout << " " << (i+1) << "/" << num_runs << "\n"; + } + + std::cout << " msgpack...\n"; + for (int i = 0; i < num_runs; ++i) { + auto t0 = std::chrono::high_resolution_clock::now(); + auto resp = http_post(curl, MSGPACK_URL, msgpack_body.c_str(), msgpack_body.size(), + "Content-Type: application/x-msgpack"); + auto t1 = std::chrono::high_resolution_clock::now(); + mp_rt.push_back(std::chrono::duration(t1 - t0).count()); + mp_inf.push_back(parse_msgpack_response(resp.body).inference_ms); + if ((i + 1) % 5 == 0) std::cout << " " << (i+1) << "/" << num_runs << "\n"; + } + + // Overhead = round-trip minus inference + std::vector json_oh, mp_oh; + for (int i = 0; i < num_runs; ++i) { + json_oh.push_back(json_rt[i] - json_inf[i]); + mp_oh.push_back(mp_rt[i] - mp_inf[i]); + } + + std::cout << "\n"; + printf(" %-8s %10s %10s %10s %10s %10s\n", + "", "JSON rt", "MP rt", "inference", "JSON oh", "MP oh"); + printf(" %-8s %10s %10s %10s %10s %10s\n", + "", "(ms)", "(ms)", "(ms)", "(ms)", "(ms)"); + std::cout << " " << std::string(60, '-') << "\n"; + print_table_row("mean", mean(json_rt), mean(mp_rt), mean(mp_inf), mean(json_oh), mean(mp_oh)); + print_table_row("min", vmin(json_rt), vmin(mp_rt), vmin(mp_inf), vmin(json_oh), vmin(mp_oh)); + print_table_row("max", vmax(json_rt), vmax(mp_rt), vmax(mp_inf), vmax(json_oh), vmax(mp_oh)); + print_table_row("p95", percentile(json_rt, 95), percentile(mp_rt, 95), percentile(mp_inf, 95), percentile(json_oh, 95), percentile(mp_oh, 95)); + + std::cout << "\n Control loop budget (100 Hz): " << BUDGET_MS << " ms\n"; + printf(" JSON: %s\n", mean(json_rt) < BUDGET_MS ? "PASS" : "WARN — exceeds budget"); + printf(" msgpack: %s\n", mean(mp_rt) < BUDGET_MS ? "PASS" : "WARN — exceeds budget"); + } + + std::cout << "\n" << std::string(60, '=') << "\n"; + + curl_easy_cleanup(curl); + curl_global_cleanup(); + return 0; +} diff --git a/scripts/test_server.py b/scripts/test_server.py new file mode 100644 index 0000000..847d970 --- /dev/null +++ b/scripts/test_server.py @@ -0,0 +1,168 @@ +""" +Quick server test script for Raspberry Pi. +Tests the inference server with dummy data and reports latency. + +Usage: + python scripts/test_server.py + python scripts/test_server.py --port 8000 --runs 20 +""" + +import argparse +import json +import time +import urllib.request +import urllib.error + +import msgpack +import numpy as np + +DEFAULT_HOST = "http://127.0.0.1" +DEFAULT_PORT = 8000 +WINDOW_SIZE = 187 +FEATURE_COUNT = 28 + + +def check_health(base_url: str) -> bool: + try: + resp = urllib.request.urlopen(f"{base_url}/health", timeout=3) + data = json.loads(resp.read()) + return data.get("status") == "ok" + except Exception as e: + print(f" Health check failed: {e}") + return False + + +def predict(base_url: str, window: list) -> dict: + body = json.dumps({"window": window}).encode() + req = urllib.request.Request( + f"{base_url}/predict", + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + resp = urllib.request.urlopen(req, timeout=5) + return json.loads(resp.read()) + + +def predict_msgpack(base_url: str, flat: list) -> dict: + # Send flat list of 187*28 floats, receive msgpack dict + body = msgpack.packb(flat) + req = urllib.request.Request( + f"{base_url}/predict_msgpack", + data=body, + headers={"Content-Type": "application/x-msgpack"}, + method="POST", + ) + resp = urllib.request.urlopen(req, timeout=5) + return msgpack.unpackb(resp.read(), raw=False) + + +def main(): + parser = argparse.ArgumentParser(description="Test exoskeleton inference server") + parser.add_argument("--host", type=str, default=DEFAULT_HOST) + parser.add_argument("--port", type=int, default=DEFAULT_PORT) + parser.add_argument("--runs", type=int, default=20, help="Number of latency runs") + args = parser.parse_args() + + base_url = f"{args.host}:{args.port}" + + print("=" * 50) + print("Exoskeleton Server Test") + print("=" * 50) + print(f"Server: {base_url}") + + # Health check + print("\n[1] Health check...") + if not check_health(base_url): + print("Server not ready. Is it running?") + print(f" uvicorn scripts.server:app --host 127.0.0.1 --port {args.port}") + return + print(" OK") + + # Single prediction with zeros + print("\n[2] Single prediction (zero input)...") + window = np.zeros((WINDOW_SIZE, FEATURE_COUNT)).tolist() + result = predict(base_url, window) + print(f" hip_left: {result['hip_left']:.6f} Nm/kg") + print(f" hip_right: {result['hip_right']:.6f} Nm/kg") + print(f" knee_left: {result['knee_left']:.6f} Nm/kg") + print(f" knee_right: {result['knee_right']:.6f} Nm/kg") + + # Single prediction with random data + print("\n[3] Single prediction (random input)...") + window = np.random.randn(WINDOW_SIZE, FEATURE_COUNT).tolist() + result = predict(base_url, window) + print(f" hip_left: {result['hip_left']:.6f} Nm/kg") + print(f" hip_right: {result['hip_right']:.6f} Nm/kg") + print(f" knee_left: {result['knee_left']:.6f} Nm/kg") + print(f" knee_right: {result['knee_right']:.6f} Nm/kg") + + # Latency benchmark — JSON + print(f"\n[4] Latency benchmark — JSON ({args.runs} runs)...") + window = np.random.randn(WINDOW_SIZE, FEATURE_COUNT).tolist() + roundtrip_times = [] + inference_times = [] + for i in range(args.runs): + t0 = time.perf_counter() + result = predict(base_url, window) + roundtrip_times.append((time.perf_counter() - t0) * 1000) + inference_times.append(result["inference_ms"]) + if (i + 1) % 5 == 0: + print(f" {i + 1}/{args.runs}") + + roundtrip_times = np.array(roundtrip_times) + inference_times = np.array(inference_times) + overhead = roundtrip_times - inference_times + + print(f"\n {'':30s} {'round-trip':>10s} {'inference':>10s} {'overhead':>10s}") + print(f" {'':30s} {'(HTTP+JSON)':>10s} {'(ONNX only)':>10s} {'(diff)':>10s}") + print(f" {'-'*64}") + print(f" {'mean':30s} {roundtrip_times.mean():>10.1f} {inference_times.mean():>10.1f} {overhead.mean():>10.1f} ms") + print(f" {'median':30s} {np.median(roundtrip_times):>10.1f} {np.median(inference_times):>10.1f} {np.median(overhead):>10.1f} ms") + print(f" {'min':30s} {roundtrip_times.min():>10.1f} {inference_times.min():>10.1f} {overhead.min():>10.1f} ms") + print(f" {'max':30s} {roundtrip_times.max():>10.1f} {inference_times.max():>10.1f} {overhead.max():>10.1f} ms") + print(f" {'p95':30s} {np.percentile(roundtrip_times, 95):>10.1f} {np.percentile(inference_times, 95):>10.1f} {np.percentile(overhead, 95):>10.1f} ms") + + # Latency benchmark — msgpack + print(f"\n[5] Latency benchmark — msgpack ({args.runs} runs)...") + flat = np.random.randn(WINDOW_SIZE, FEATURE_COUNT).flatten().tolist() + mp_roundtrip_times = [] + mp_inference_times = [] + for i in range(args.runs): + t0 = time.perf_counter() + result = predict_msgpack(base_url, flat) + mp_roundtrip_times.append((time.perf_counter() - t0) * 1000) + mp_inference_times.append(result["inference_ms"]) + if (i + 1) % 5 == 0: + print(f" {i + 1}/{args.runs}") + + mp_roundtrip_times = np.array(mp_roundtrip_times) + mp_inference_times = np.array(mp_inference_times) + mp_overhead = mp_roundtrip_times - mp_inference_times + + budget_ms = 10.0 # 100 Hz + + print(f"\n {'':20s} {'JSON rt':>8s} {'MP rt':>8s} {'inference':>10s} {'JSON oh':>8s} {'MP oh':>8s}") + print(f" {'-'*70}") + + def row(label, json_rt, mp_rt, inf, json_oh, mp_oh): + print(f" {label:20s} {json_rt:>8.1f} {mp_rt:>8.1f} {inf:>10.1f} {json_oh:>8.1f} {mp_oh:>8.1f} ms") + + row("mean", roundtrip_times.mean(), mp_roundtrip_times.mean(), mp_inference_times.mean(), overhead.mean(), mp_overhead.mean()) + row("median", np.median(roundtrip_times), np.median(mp_roundtrip_times), np.median(mp_inference_times), np.median(overhead), np.median(mp_overhead)) + row("min", roundtrip_times.min(), mp_roundtrip_times.min(), mp_inference_times.min(), overhead.min(), mp_overhead.min()) + row("max", roundtrip_times.max(), mp_roundtrip_times.max(), mp_inference_times.max(), overhead.max(), mp_overhead.max()) + row("p95", np.percentile(roundtrip_times,95), np.percentile(mp_roundtrip_times,95), np.percentile(mp_inference_times,95), np.percentile(overhead,95), np.percentile(mp_overhead,95)) + + print(f"\n Control loop budget (100 Hz): {budget_ms:.0f} ms") + for label, times in [("JSON", roundtrip_times), ("msgpack", mp_roundtrip_times)]: + if times.mean() < budget_ms: + print(f" {label}: PASS") + else: + print(f" {label}: WARN — exceeds budget") + + print("\n" + "=" * 50) + + +if __name__ == "__main__": + main()