From c28bb64b0a8430d968e7813a1e74745bdaec8f83 Mon Sep 17 00:00:00 2001 From: wty-yy <993660140@qq.com> Date: Mon, 29 Dec 2025 21:29:03 +0800 Subject: [PATCH] v1.0.0 prev; add serve, client, search_seeds --- UPDATE.md | 6 + robogauge/scripts/client.py | 122 ++++++++++++++ robogauge/scripts/server.py | 149 ++++++++++++++++++ robogauge/tasks/pipeline/level_pipeline.py | 9 +- robogauge/tasks/pipeline/stress_pipeline.py | 10 +- robogauge/utils/helpers.py | 7 +- robogauge/utils/progress_monitor.py | 2 +- .../utils/visualize/plot_radar_and_bar.py | 2 +- .../utils/visualize/plot_terrain_levels.py | 2 +- run.bash | 30 ++-- setup.py | 7 +- 11 files changed, 319 insertions(+), 27 deletions(-) create mode 100644 robogauge/scripts/client.py create mode 100644 robogauge/scripts/server.py diff --git a/UPDATE.md b/UPDATE.md index 48b3ad5..6801f32 100644 --- a/UPDATE.md +++ b/UPDATE.md @@ -1,4 +1,10 @@ # UPDATE +## 20251229 +### v1.0.0 +1. 完成stress pipeline的客户端, 服务端代码, 支持异步推理 +2. 加入MultiPipeline和LevelPipeline分开的种子数量`seeds, search_seeds`, Level 5个, Multi 3个, 这样可以显著提高速度, 并且保证评估精度不掉 (两种评估总分前三位都是相同的), 速度提升`3m 32s -> 2m 47s` +3. 删除stairs_down任务, 将stairs_up重命名为stairs +4. 对于slope, stairs再分化为两个任务, slope_fd, slope_bd, stairs_fd, stairs_bd, 分别表示正向/背向朝更高处的初始化, target_pos_velocity中指令同时根据是否backward反转 ## 20251228 ### v0.1.18 1. 将final_score改为merge_metrics, 并在multi_pipeline中也进行该评估 diff --git a/robogauge/scripts/client.py b/robogauge/scripts/client.py new file mode 100644 index 0000000..2a5c85d --- /dev/null +++ b/robogauge/scripts/client.py @@ -0,0 +1,122 @@ +import requests +import time +import json +import sys +from typing import Dict, Any, Optional +from robogauge.scripts.server import ResponseStatus + +class RoboGaugeClient: + def __init__(self, base_url: str = "http://127.0.0.1:9973"): + self.base_url = base_url + self.processing_ids = [] + self.task_id2info = {} + self.response_data = {} + + def submit_task(self, + model_path: str, + step: int, + task_name: str, + experiment_name: str, + wait_for_server: bool = True, + retry_interval: int = 2 + ) -> Optional[str]: + """ Submit stress pipeline evaluation + Args: + model_path (str): Torch script model path. + step (int): Model training step. + task_name (str): Registered task name. + experiment_name (str): Experiment name for logging. + wait_for_server (bool): If True, will keep retrying until the server is available. + retry_interval (int): Seconds to wait before retrying connection. + Returns: + Optional[str]: Task ID if submission is successful, else None. + """ + payload = { + "model_path": model_path, + "step": step, + "task_name": task_name, + "experiment_name": experiment_name + } + + print(f"[RoboGaugeClient]📤 Preparing to submit task: {task_name}") + + while True: + try: + response = requests.post(f"{self.base_url}/submit_eval", json=payload) + + if response.status_code == 200: + data = response.json() + task_id = data["task_id"] + print(f"[RoboGaugeClient]✅ Submission successful! Task ID: {task_id}") + self.processing_ids.append(task_id) + self.task_id2info[task_id] = f"{payload['task_name']}_step{payload['step']}_{payload['experiment_name']}_{task_id}" + return task_id + else: + print(f"[RoboGaugeClient]❌ Server returned error: {response.text}") + return None + + except requests.exceptions.ConnectionError: + if not wait_for_server: + print("[RoboGaugeClient]❌ Unable to connect to server (Connection Refused).") + return None + + print(f"[RoboGaugeClient]⏳ Server not responding, retrying in {retry_interval} seconds... (Make sure server.py is running)") + time.sleep(retry_interval) + except Exception as e: + print(f"[RoboGaugeClient]❌ Unknown error: {e}") + return None + + def monitor_tasks(self): + print("[RoboGaugeClient]⏱️ Monitoring submitted tasks...") + """ Monitor all submitted tasks until completion. """ + for task_id in reversed(self.processing_ids): + respone = requests.get(f"{self.base_url}/get_result/{task_id}") + if respone.status_code != 200: + continue + resp_data = respone.json() + status = resp_data['status'] + + if status in [ResponseStatus.PENDING, ResponseStatus.PROCESSING]: + print(f"[RoboGaugeClient]⏳ Task {self.task_id2info[task_id]} is still {status}.") + continue + + if status == ResponseStatus.FINISHED: + self.response_data[task_id] = resp_data + print(f"[RoboGaugeClient]🎉 Task {self.task_id2info[task_id]} finished successfully!") + elif status == ResponseStatus.ERROR: + print(f"[RoboGaugeClient]❌ Task {self.task_id2info[task_id]} encountered an error: {resp_data.get('error_msg')}") + elif status == ResponseStatus.NOT_FOUND: + print(f"[RoboGaugeClient]❓ Task {self.task_id2info[task_id]} not found on server.") + self.processing_ids.remove(task_id) + +if __name__ == "__main__": + SERVER_URL = "http://127.0.0.1:9973" + client = RoboGaugeClient(base_url=SERVER_URL) + + test_payload = { + "model_path": "{ROBOGAUGE_ROOT_DIR}/resources/models/go2/go2_moe_cts_124k.pt", + "step": 124000, + "task_name": "go2_moe", + "experiment_name": "client_debug_001" + } + + print("="*40) + print(" RoboGauge Client Demo") + print("="*40) + + task_id = client.submit_task( + model_path=test_payload["model_path"], + step=test_payload["step"], + task_name=test_payload["task_name"], + experiment_name=test_payload["experiment_name"], + wait_for_server=True + ) + + while True: + client.monitor_tasks() + for task_id, resp in client.response_data.items(): + scores = resp['results']['scores'] + print("[RoboGaugeClient]📊 Scores:") + print(json.dumps(scores, indent=2, ensure_ascii=False)) + client.response_data.clear() + time.sleep(5) diff --git a/robogauge/scripts/server.py b/robogauge/scripts/server.py new file mode 100644 index 0000000..2685664 --- /dev/null +++ b/robogauge/scripts/server.py @@ -0,0 +1,149 @@ +# -*- coding: utf-8 -*- +''' +@File : server.py +@Time : 2025/12/29 10:58:05 +@Author : wty-yy, Gemini3 Pro +@Version : 1.0 +@Blog : https://wty-yy.github.io/ +@Desc : Asynchronous stress pipeline evaluation server +''' +import os +os.environ['MUJOCO_GL'] = 'glfw' # avoid mujoco.Renderer EGL context error +os.environ["OMP_NUM_THREADS"] = "1" +os.environ["MKL_NUM_THREADS"] = "1" + +import multiprocessing +import uvicorn +import queue +import time +import uuid +from fastapi import FastAPI +from pydantic import BaseModel +from typing import Dict, Optional + +from dataclasses import dataclass +from robogauge.utils.helpers import parse_args, class_to_dict +from robogauge.tasks.pipeline.stress_pipeline import StressPipeline +from pprint import pprint + +default_args_list = [ + '--stress-benchmark', + # '--stress-terrain-names', 'flat', 'wave', 'slope', 'stairs_up', 'stairs_down', 'obstacle', + '--stress-terrain-names', 'flat', 'wave', + '--num-processes', '50', + '--seeds', '0', '1', '2', + '--search-seeds', '0', '1', '2', '3', '4', + '--frictions', '0.5', '0.75', '1.0', '1.25', '1.5', '1.75', '2.0', '2.25', '2.5', + '--compress-logs', + '--headless', +] + +@dataclass +class EvalTaskData: + model_path: str + step: int + task_name: str + experiment_name: str + +class EvalRequest(BaseModel): + model_path: str + step: int + task_name: str + experiment_name: str + +class ResponseStatus: + PENDING = "pending" + PROCESSING = "processing" + FINISHED = "finished" + ERROR = "error" + NOT_FOUND = "not_found" + +def run_api_server(input_queue, result_dict, port=9973): + """ + Running in a separate subprocess. + I/O Process: submit requests -> put into queue -> return ID. + """ + app = FastAPI() + + @app.post("/submit_eval") + def submit_eval(req: EvalRequest): + task_id = str(uuid.uuid4()) + task_data = EvalTaskData( + model_path=req.model_path, + step=req.step, + task_name=req.task_name, + experiment_name=req.experiment_name + ) + input_queue.put((task_id, task_data)) + + result_dict[task_id] = {"status": ResponseStatus.PENDING} + return {"task_id": task_id, "message": "Queued"} + + @app.get("/get_result/{task_id}") + def get_result(task_id: str): + if task_id not in result_dict: + return {"status": ResponseStatus.NOT_FOUND} + return result_dict[task_id] + + print(f"📡 API Server listening on port {port}...") + uvicorn.run(app, host="127.0.0.1", port=port, log_level="error") + +def main(): + print("🤖 RoboGauge Evaluation Server Starting...") + ctx = multiprocessing.get_context('spawn') + manager = ctx.Manager() + task_queue = manager.Queue() + results_store = manager.dict() + + api_p = ctx.Process( + target=run_api_server, + args=(task_queue, results_store), + daemon=True + ) + api_p.start() + + print("🚀 Main Process started. Waiting for tasks...") + print(" (StressPipeline will run directly in this Main Process)") + + try: + while True: + try: + task_data: EvalTaskData + task_id, task_data = task_queue.get(timeout=1.0) + + print(f"\n🔄 [Main] Processing Task {task_id} (Step {task_data.step})...") + results_store[task_id] = {"status": ResponseStatus.PROCESSING} + + args_list = default_args_list.copy() + args_list += ['--model-path', task_data.model_path, '--task-name', task_data.task_name, '--experiment-name', task_data.experiment_name] + args = parse_args(args_list) + + print(f"📋 Running with args:") + pprint(class_to_dict(args)) + + pipeline = StressPipeline(args) + stress_results = pipeline.run() + + results_store[task_id] = { + "status": ResponseStatus.FINISHED, + "step": task_data.step, + "results": stress_results + } + print(f"✅ [Main] Task {task_id} Finished.") + + except queue.Empty: + continue + except Exception as e: + print(f"❌ [Main] Error: {e}") + import traceback + traceback.print_exc() + if 'task_id' in locals(): + results_store[task_id] = {"status": ResponseStatus.ERROR, "error": str(e), "error_msg": traceback.format_exc()} + + except KeyboardInterrupt: + print("\n🛑 Shutting down...") + api_p.terminate() + api_p.join() + +if __name__ == "__main__": + main() diff --git a/robogauge/tasks/pipeline/level_pipeline.py b/robogauge/tasks/pipeline/level_pipeline.py index e3f8867..c0525e4 100644 --- a/robogauge/tasks/pipeline/level_pipeline.py +++ b/robogauge/tasks/pipeline/level_pipeline.py @@ -8,6 +8,7 @@ @Desc : Level Pipeline for Robogauge ''' import yaml +from copy import deepcopy from robogauge.tasks.pipeline.multi_pipeline import MultiPipeline from robogauge.utils.logger import Logger @@ -19,7 +20,7 @@ level_logger = Logger() # LevelPipeline logger class LevelPipeline: def __init__(self, args, console_output=True, progress_data: ProgressData = None): self.args = args - self.seeds = args.seeds + self.seeds = args.search_seeds self.console_output = console_output self.progress_data = progress_data parent_log_dir = getattr(args, 'parent_log_dir', None) @@ -65,8 +66,10 @@ class LevelPipeline: def test_level(self, level: int): level_logger.info(f"🔍 Testing level {level}...") - self.args.level = level - multi_pipeline = MultiPipeline(self.args, console_output=self.console_output) + args = deepcopy(self.args) + args.level = level + args.seeds = self.seeds + multi_pipeline = MultiPipeline(args, console_output=self.console_output) aggregated_results = multi_pipeline.run() success_mean = float(aggregated_results['summary']['success']['mean'].split(' ')[0]) all_success = success_mean >= 0.8 diff --git a/robogauge/tasks/pipeline/stress_pipeline.py b/robogauge/tasks/pipeline/stress_pipeline.py index 58c0e4f..d475a40 100644 --- a/robogauge/tasks/pipeline/stress_pipeline.py +++ b/robogauge/tasks/pipeline/stress_pipeline.py @@ -90,7 +90,6 @@ def run_pipeline(args, progress_queue, data): class StressPipeline: def __init__(self, args): self.args = args - self.seeds = args.seeds self.task_robot_model = args.task_name.split('.')[0] self.num_processes = args.num_processes args.experiment_name = self.task_robot_model + '_stress' + ('' if args.cli_experiment_name is None else '_' + args.cli_experiment_name) @@ -109,7 +108,7 @@ class StressPipeline: def run(self): stress_logger.info(f"🚀 Starting Stress Benchmark for '{self.args.experiment_name}'.") - stress_logger.info(f"🔢 Seeds: {self.seeds}") + stress_logger.info(f"🔢 Seeds: {self.args.seeds}, Level Search Seeds: {self.args.search_seeds}") terrain_names = self.args.stress_terrain_names stress_logger.info(f"🌄 Stress Test Terrain Names: {terrain_names}") @@ -161,7 +160,7 @@ class StressPipeline: stress_logger.info("📊 Aggregating Stress Benchmark Results...") finish_msg = ( f"""\n{'='*20} Stress Benchmark Summary {'='*20}\n""" - f"""{'Seeds':^20}{str(self.seeds)}\n""" + f"""{'Seeds':^20}{str(self.args.seeds):^15}{'Level Search Seeds':^20}{str(self.args.search_seeds):^15}\n""" f"""{'Terrain Name':^20}{'Base Mass':^15}{'Friction':^15}{'Max Level':^15}\n""" ) all_results = sorted(all_results, key=lambda x: (x['data']['terrain_name'], x['data'].get('base_mass', 0), x['data'].get('friction', 0))) @@ -178,7 +177,7 @@ class StressPipeline: stress_logger.error("No results to aggregate.") return - summary = {**self.static_info, 'summary': {}, 'robust_score': {}, 'benchmark_score': 0.0} + summary = {**self.static_info, 'summary': {}, 'robust_score': {}, 'benchmark_score': 0.0, 'scores': {}} metric_collections = defaultdict(lambda: defaultdict(list)) terrain_collections = defaultdict(lambda: defaultdict(list)) zero_terrain_count = defaultdict(lambda: 0) @@ -209,14 +208,17 @@ class StressPipeline: robust_score = defaultdict(dict) robust_scores = [] + scores = summary['scores'] for terrain_name, means in terrain_collections.items(): for mean_name, values in means.items(): values.extend([0.0] * zero_terrain_count[terrain_name]) # include zero terrains robust_score[terrain_name][mean_name] = float(np.mean(values)) + scores[terrain_name] = robust_score[terrain_name]['mean@50'] if terrain_name != 'stairs_down': # skip stairs_down for benchmark score calculation robust_scores.append(robust_score[terrain_name]['mean@50']) summary['robust_score'] = dict(robust_score) summary['benchmark_score'] = float(np.mean(robust_scores)) + scores['benchmark'] = summary['benchmark_score'] save_path = stress_logger.log_dir / "stress_benchmark_results.yaml" with open(save_path, 'w') as file: diff --git a/robogauge/utils/helpers.py b/robogauge/utils/helpers.py index ff856ce..09fd03a 100644 --- a/robogauge/utils/helpers.py +++ b/robogauge/utils/helpers.py @@ -60,7 +60,7 @@ def str2bool(v): return False raise TypeError('Boolean value expected.') -def parse_args(): +def parse_args(args_list=None): parser = ArgumentParser() parameters = [ # Single run parameters @@ -81,12 +81,13 @@ def parse_args(): # Multiprocessing parameters, with different seeds {"name": "--multi", "action": "store_true", "default": False, "help": "Enable multiprocessing."}, - {"name": "--seeds", "type": int, "nargs": "+", "default": [0, 1, 2, 3, 4], "help": "List of random seeds for multiple runs."}, + {"name": "--seeds", "type": int, "nargs": "+", "default": [0, 1, 2], "help": "List of random seeds for multiple runs."}, {"name": "--base-masses", "type": float, "nargs": "+", "default": [0], "help": "List of base masses for the model."}, {"name": "--frictions", "type": float, "nargs": "+", "default": [0.5, 0.75, 1.0, 1.25, 1.5, 1.75, 2.0, 2.25, 2.5], "help": "List of friction coefficients for the model."}, # Level pipeline parameters {"name": "--search-max-level", "action": "store_true", "default": False, "help": "Use level pipeline to search maximum level."}, + {"name": "--search-seeds", "type": int, "nargs": "+", "default": [0, 1, 2, 3, 4], "help": "List of random seeds for level search."}, # Stress pipeline parameters {"name": "--stress-benchmark", "action": "store_true", "default": False, "help": "Use stress pipeline to benchmark model robustness."}, @@ -98,7 +99,7 @@ def parse_args(): ] for param in parameters: parser.add_argument(param['name'], **{k: v for k, v in param.items() if k != 'name'}) - args = parser.parse_args() + args = parser.parse_args(args_list) flatten_task_name = args.task_name.replace('.', '_') args.cli_experiment_name = args.experiment_name if args.experiment_name is not None: diff --git a/robogauge/utils/progress_monitor.py b/robogauge/utils/progress_monitor.py index 5c885e3..8713114 100644 --- a/robogauge/utils/progress_monitor.py +++ b/robogauge/utils/progress_monitor.py @@ -2,7 +2,7 @@ ''' @File : progress_monitor.py @Time : 2025/12/27 00:10:07 -@Author : wty-yy (with Gemini 3) +@Author : wty-yy, Gemini3 Pro @Version : 1.0 @Blog : https://wty-yy.github.io/ @Desc : Centralized progress monitoring with Dynamic Slot Management diff --git a/robogauge/utils/visualize/plot_radar_and_bar.py b/robogauge/utils/visualize/plot_radar_and_bar.py index 820dd88..369720a 100644 --- a/robogauge/utils/visualize/plot_radar_and_bar.py +++ b/robogauge/utils/visualize/plot_radar_and_bar.py @@ -2,7 +2,7 @@ ''' @File : plot_radar_and_bar.py @Time : 2025/12/27 22:52:24 -@Author : wty-yy (with Gemini 3) +@Author : wty-yy, Gemini3 Pro @Version : 1.0 @Blog : https://wty-yy.github.io/ @Desc : 可视化评测结果，生成雷达图和柱状图 diff --git a/robogauge/utils/visualize/plot_terrain_levels.py b/robogauge/utils/visualize/plot_terrain_levels.py index dac1a96..2bb11d4 100644 --- a/robogauge/utils/visualize/plot_terrain_levels.py +++ b/robogauge/utils/visualize/plot_terrain_levels.py @@ -2,7 +2,7 @@ ''' @File : plot_terrain_levels.py @Time : 2025/12/27 23:04:01 -@Author : wty-yy (with Gemini 3) +@Author : wty-yy, Gemini3 Pro @Version : 1.0 @Blog : https://wty-yy.github.io/ @Desc : None diff --git a/run.bash b/run.bash index 17dfc41..0b01872 100755 --- a/run.bash +++ b/run.bash @@ -3,71 +3,77 @@ source /root/Programs/miniforge3/bin/activate robot python robogauge/scripts/run.py \ - --task go2_moe \ + --task-name go2_moe \ --experiment-name debug \ --stress-benchmark \ --stress-terrain-names flat slope stairs_up stairs_down wave obstacle \ --num-processes 62 \ - --seeds 0 1 2 3 4 \ + --seeds 0 1 2 \ + --search-seeds 0 1 2 3 4 \ --frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \ --compress-logs \ --headless python robogauge/scripts/run.py \ - --task go2_moe \ + --task-name go2_moe \ --model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_hard_terrain_141k.pt \ --experiment-name debug \ --stress-benchmark \ --stress-terrain-names flat slope stairs_up stairs_down wave obstacle \ --num-processes 62 \ - --seeds 0 1 2 3 4 \ + --seeds 0 1 2 \ + --search-seeds 0 1 2 3 4 \ --frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \ --compress-logs \ --headless python robogauge/scripts/run.py \ - --task go2 \ + --task-name go2 \ --model-path /root/Coding/RoboGauge/mytest/go2_cts_hard_terrain_141k.pt \ --experiment-name debug \ --stress-benchmark \ --stress-terrain-names flat slope stairs_up stairs_down wave obstacle \ --num-processes 62 \ - --seeds 0 1 2 3 4 \ + --seeds 0 1 2 \ + --search-seeds 0 1 2 3 4 \ --frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \ --compress-logs \ --headless python robogauge/scripts/run.py \ - --task go2 \ + --task-name go2 \ --experiment-name debug \ --stress-benchmark \ --stress-terrain-names flat slope stairs_up stairs_down wave obstacle \ --num-processes 62 \ - --seeds 0 1 2 3 4 \ + --seeds 0 1 2 \ + --search-seeds 0 1 2 3 4 \ --frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \ --compress-logs \ --headless python robogauge/scripts/run.py \ - --task go2_moe \ + --task-name go2_moe \ --model-path /root/Coding/RoboGauge/mytest/go2_moe_cts_hard_terrain_100k.pt \ --experiment-name debug \ --stress-benchmark \ --stress-terrain-names flat slope stairs_up stairs_down wave obstacle \ --num-processes 62 \ - --seeds 0 1 2 3 4 \ + --seeds 0 1 2 \ + --search-seeds 0 1 2 3 4 \ --frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \ --compress-logs \ --headless python robogauge/scripts/run.py \ - --task go2 \ + --task-name go2 \ --model-path /root/Coding/RoboGauge/mytest/go2_cts_hard_terrain_100k.pt \ --experiment-name debug \ --stress-benchmark \ --stress-terrain-names flat slope stairs_up stairs_down wave obstacle \ --num-processes 62 \ - --seeds 0 1 2 3 4 \ + --seeds 0 1 2 \ + --search-seeds 0 1 2 3 4 \ --frictions 0.5 0.75 1.0 1.25 1.5 1.75 2.0 2.25 2.5 \ --compress-logs \ --headless diff --git a/setup.py b/setup.py index e39d8d0..7baa498 100644 --- a/setup.py +++ b/setup.py @@ -2,13 +2,14 @@ from setuptools import setup, find_packages setup( name="robogauge", # 包名 - version="0.1.0", # 版本号 - author="wty-yy", # 你的名字 + version="1.0.0", # 版本号 + author="Wu Tianyang", # 你的名字 author_email="993660140@qq.com", description="A generic robot RL model evaluation library based on MuJoCo", url="https://github.com/wty-yy/robot_gauge", # 如果有仓库地址 packages=find_packages(), install_requires=[ + "torch", # Refer: https://pytorch.org/get-started/locally/ "numpy>=1.20.0", "mujoco>=3.0.0", "dm_control>=1.0.14", @@ -18,6 +19,8 @@ setup( "imageio[ffmpeg]", "tensorboard", "PyYAML", + "fastapi", + "uvicorn", ], python_requires=">=3.8",