diff --git a/CMD.md b/CMD.md index 8e30749..d5b0732 100644 --- a/CMD.md +++ b/CMD.md @@ -32,7 +32,16 @@ python robogauge/scripts/run.py \ # Stress Pipeline ```bash - +python robogauge/scripts/run.py \ + --task go2_moe \ + --experiment-name debug \ + --stress-benchmark \ + --stress-terrain-names flat slope stairs_up stairs_down wave \ + --stress-num-processes 2 \ + --num-processes 3 \ + --seeds 0 1 2 \ + --frictions 0.5 1.0 1.5 2.0 2.5 \ + --headless ``` # Radar/Bar Plot diff --git a/UPDATE.md b/UPDATE.md index e1873ad..82c8507 100644 --- a/UPDATE.md +++ b/UPDATE.md @@ -1,8 +1,10 @@ # UPDATE ## 20251226 ### v0.1.16 -Bugs: level_results会存储到MultiPipeline下, 因为MultiPipeline不是子进程启动的, 会覆盖LevelPipeline的Logger冲突 -1. 基本完成StressPipeline, 继续调试完毕 +1. 基本完成StressPipeline, 但是绘制进度信息还有问题 +2. wave地形的穿模判定非常容易触发, 加入最多穿模判定重启次数为1次, 超过该次数后穿模就不再自动重启了 + +FIX Bugs: (level_results会存储到MultiPipeline下, 因为MultiPipeline不是子进程启动的, 会覆盖LevelPipeline的Logger冲突) 通过创建新的Logger实现 ## 20251225 ### v0.1.15 1. 统一terrain的大小为10x10m, 某一边的中点在(0,0,0), 对全部的带等级的terrain都加上边界墙高度10m diff --git a/robogauge/scripts/run.py b/robogauge/scripts/run.py index d992b6d..ed41e22 100644 --- a/robogauge/scripts/run.py +++ b/robogauge/scripts/run.py @@ -13,8 +13,7 @@ os.environ["OMP_NUM_THREADS"] = "1" os.environ["MKL_NUM_THREADS"] = "1" from robogauge.tasks import * -from robogauge.tasks.pipeline.multi_pipeline import MultiPipeline -from robogauge.tasks.pipeline.level_pipeline import LevelPipeline +from robogauge.tasks.pipeline import * from robogauge.utils.task_register import task_register from robogauge.utils.helpers import parse_args @@ -23,7 +22,10 @@ from robogauge.utils.logger import logger if __name__ == '__main__': args = parse_args() - if args.multi: + if args.stress_benchmark: + stress_pipeline = StressPipeline(args) + stress_pipeline.run() + elif args.multi: multi_pipeline = MultiPipeline(args) multi_pipeline.run() elif args.search_max_level: diff --git a/robogauge/tasks/gauge/gauge_configs/terrain_levels_config.py b/robogauge/tasks/gauge/gauge_configs/terrain_levels_config.py index ff5fa1a..bc4219f 100644 --- a/robogauge/tasks/gauge/gauge_configs/terrain_levels_config.py +++ b/robogauge/tasks/gauge/gauge_configs/terrain_levels_config.py @@ -28,18 +28,6 @@ class TerrainSearchLevelsConfig(Config): [4.8, 0, 2.092 + 0.1], [4.8, 0, 2.280 + 0.1], ] - spawns = [ - [0.5, 0, 0], - [0.5, 0, 0], - [0.5, 0, 0], - [0.5, 0, 0], - [0.5, 0, 0], - [0.5, 0, 0], - [0.5, 0, 0], - [0.5, 0, 0], - [0.5, 0, 0], - [0.5, 0, 0], - ] class wave: levels = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] diff --git a/robogauge/tasks/gauge/gauge_configs/wave_gauge_config.py b/robogauge/tasks/gauge/gauge_configs/wave_gauge_config.py index 50e04fc..98854eb 100644 --- a/robogauge/tasks/gauge/gauge_configs/wave_gauge_config.py +++ b/robogauge/tasks/gauge/gauge_configs/wave_gauge_config.py @@ -19,7 +19,7 @@ class WaveGaugeConfig(BaseGaugeConfig): '{ROBOGAUGE_ROOT_DIR}/resources/terrains/wave/wave_10.xml', '{ROBOGAUGE_ROOT_DIR}/resources/terrains/wall/10x10_wall.xml', ] - terrain_spawn_pos = [1.5, 1.25, 1.0] # x y z [m], robot freejoint spawn position on the terrain + terrain_spawn_pos = [1.5, 1.25, 0.0] # x y z [m], robot freejoint spawn position on the terrain class goals: class target_pos_velocity: # goal to reach a target position by velocity command diff --git a/robogauge/tasks/pipeline/level_pipeline.py b/robogauge/tasks/pipeline/level_pipeline.py index 73f80a1..a2c6c59 100644 --- a/robogauge/tasks/pipeline/level_pipeline.py +++ b/robogauge/tasks/pipeline/level_pipeline.py @@ -15,10 +15,11 @@ from robogauge.utils.logger import Logger level_logger = Logger() # LevelPipeline logger class LevelPipeline: - def __init__(self, args): + def __init__(self, args, console_output: bool = True): self.args = args self.seeds = args.seeds - level_logger.create(args.experiment_name+'_level', args.run_name) + self.console_output = console_output + level_logger.create(args.experiment_name+'_level', args.run_name, console_output=console_output) def run(self): level_logger.info(f"🚀 Starting Level Searcher for '{self.args.experiment_name}'.") @@ -52,9 +53,9 @@ class LevelPipeline: def test_level(self, level: int): level_logger.info(f"🔍 Testing level {level}...") self.args.level = level - multi_pipeline = MultiPipeline(self.args) + multi_pipeline = MultiPipeline(self.args, console_output=self.console_output) aggregated_results = multi_pipeline.run() - success_mean = float(aggregated_results['success']['mean'].split(' ')[0]) + success_mean = float(aggregated_results['summary']['success']['mean'].split(' ')[0]) all_success = success_mean >= 0.8 if all_success: level_logger.info(f"✅ Level {level} passed all tests.") diff --git a/robogauge/tasks/pipeline/multi_pipeline.py b/robogauge/tasks/pipeline/multi_pipeline.py index cdad959..49868bc 100644 --- a/robogauge/tasks/pipeline/multi_pipeline.py +++ b/robogauge/tasks/pipeline/multi_pipeline.py @@ -22,6 +22,7 @@ from robogauge.tasks.pipeline.base_pipeline import BasePipeline from robogauge.utils.task_register import task_register from robogauge.utils.logger import Logger +from robogauge.utils.process_utils import NoDaemonPool multi_logger = Logger() # MultiPipeline logger @@ -62,14 +63,14 @@ def run_single_process(args, data): return ret class MultiPipeline: - def __init__(self, args): + def __init__(self, args, console_output: bool = True): self.args = args self.seeds = args.seeds self.frictions = args.frictions self.base_masses = args.base_masses self.num_processes = args.num_processes self.static_info = {} - multi_logger.create(args.experiment_name+'_multi', args.run_name+'_multi') + multi_logger.create(args.experiment_name+'_multi', args.run_name+'_multi', console_output=console_output) def add_static_info(self, key: str, value): if key not in self.static_info: @@ -85,7 +86,7 @@ class MultiPipeline: ctx = multiprocessing.get_context('spawn') worker_func = functools.partial(run_single_process, self.args) results_list = [] - with ctx.Pool(processes=self.num_processes) as pool: + with NoDaemonPool(processes=self.num_processes, context=ctx) as pool: iterator = pool.imap_unordered(worker_func, workers_data) for results in tqdm(iterator, total=len(workers_data), desc="Evaluation"): results_list.append(results) @@ -104,7 +105,7 @@ class MultiPipeline: """ Process results from all processes and aggregate them. """ multi_logger.info("📊 Aggregating Results from all runs...") - summary = {'success': {}, **self.static_info} + summary = {'success': {}, **self.static_info, 'summary': {}} finish_msg = ( f"""\n{'='*20} Run Finish Summary {'='*20}\n""" f"""{'Seed':^10}{'Base Mass':^15}{'Friction':^15}{'Status':^10}\n""" @@ -133,9 +134,9 @@ class MultiPipeline: value_collections[metric][mean_name].append(float(mean_value.split(' ')[0])) for metric, means in value_collections.items(): - summary[metric] = {} + summary['summary'][metric] = {} for mean_name, values in means.items(): - summary[metric][mean_name] = f"{float(np.mean(values)):.4f} ± {float(np.std(values)):.4f}" + summary['summary'][metric][mean_name] = f"{float(np.mean(values)):.4f} ± {float(np.std(values)):.4f}" save_path = multi_logger.log_dir / "aggregated_results.yaml" with open(save_path, 'w') as file: @@ -143,9 +144,9 @@ class MultiPipeline: multi_logger.info("✅ Aggregated execution finished.") multi_logger.info(f"📁 Aggregated results saved to: {save_path}") - multi_logger.info( - f"""\n{'='*20} Multi-Run Summary {'='*20}\n""" - f"""{yaml.dump(summary, allow_unicode=True)}""" - f"""{'='*60}""" - ) + # multi_logger.info( + # f"""\n{'='*20} Multi-Run Summary {'='*20}\n""" + # f"""{yaml.dump(summary, allow_unicode=True)}""" + # f"""{'='*60}""" + # ) return summary diff --git a/robogauge/tasks/pipeline/stress_pipeline.py b/robogauge/tasks/pipeline/stress_pipeline.py index 8660c3c..fefcd53 100644 --- a/robogauge/tasks/pipeline/stress_pipeline.py +++ b/robogauge/tasks/pipeline/stress_pipeline.py @@ -9,12 +9,15 @@ ''' import yaml import functools +import numpy as np from tqdm import tqdm import multiprocessing from copy import deepcopy from itertools import product +from collections import defaultdict from robogauge.utils.logger import Logger +from robogauge.utils.process_utils import NoDaemonPool from robogauge.tasks.pipeline import MultiPipeline, LevelPipeline from robogauge.tasks.gauge.gauge_configs.terrain_levels_config import TerrainSearchLevelsConfig @@ -36,20 +39,22 @@ def run_pipeline(args, data): level = None # flat terrain if search: - level, results = LevelPipeline(args).run() + level, results = LevelPipeline(args, console_output=False).run() if level == 0: # no valid level found results = { 'success': False, 'results': results, 'data': data, + 'level': 0, } else: args.level = level results = { 'success': True, - 'results': MultiPipeline(args).run(), + 'results': MultiPipeline(args, console_output=False).run(), 'data': data, + 'level': level, } return results @@ -95,19 +100,22 @@ class StressPipeline: } if search_max_level: for friction, base_mass in product(self.args.frictions, self.args.base_masses): - data.update({ + now_data = deepcopy(data) + now_data.update({ 'friction': friction, 'base_mass': base_mass, }) - workers_data.append(data) + workers_data.append(now_data) + else: + workers_data.append(data) ### Run and collect results ### results_list = [] - with ctx.Pool(processes=self.num_processes) as pool: + with NoDaemonPool(processes=self.num_processes, context=ctx) as pool: iterator = pool.imap_unordered(worker_func, workers_data) for results in tqdm(iterator, total=len(workers_data), desc="Stress Benchmark"): results_list.append(results) - self.add_static_info('model_path', results['results']['model_path']) + self.add_static_info('model_path', results['results'].pop('model_path', None)) stress_logger.info("✅ Stress Benchmark Completed.") stress_results = self.aggregate_results(results_list) @@ -115,17 +123,48 @@ class StressPipeline: def aggregate_results(self, all_results): stress_logger.info("📊 Aggregating Stress Benchmark Results...") - summary = {} finish_msg = ( f"""\n{'='*20} Stress Benchmark Summary {'='*20}\n""" - f"""{'Terrain Name':^20}{'Base Mass':^15}{'Friction':^15}{'Status':^10}\n""" + f"""{'Seeds':^20}{str(self.seeds)}\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))) for result in all_results: terrain_name = result['data']['terrain_name'] base_mass = result['data'].get('base_mass', self.args.base_masses) friction = result['data'].get('friction', self.args.frictions) - status = "✅" if result['success'] else "❌" - finish_msg += f"{terrain_name:^20}{str(base_mass):^15}{str(friction):^15}{status:^10}\n" + status = f"{result['level']}" if result['success'] else "❌" + finish_msg += f"{terrain_name:^20}{str(base_mass):^15}{str(friction):^15}{status:^15}\n" finish_msg += f"""{'='*66}""" stress_logger.info(finish_msg) + + if not all_results: + stress_logger.error("No results to aggregate.") + return + + summary = {**self.static_info, 'summary': {}} + value_collections = defaultdict(lambda: defaultdict(list)) + for result in all_results: + terrain_name = result['data']['terrain_name'] + terrain_level = result['level'] # None, 0, 1, ..., 10 + key = terrain_name + if terrain_level is not None: + key += f'_{terrain_level}' + key += f'_baseMass{result["data"]["base_mass"]}_friction{result["data"]["friction"]}' + summary[key] = result['results'] if terrain_level != 0 else None + + for metric, means in result['results']['summary'].items(): + for mean_name, mean_value in means.items(): + value_collections[metric][mean_name].append(float(mean_value.split(' ')[0])) + + for metric, means in value_collections.items(): + summary['summary'][metric] = {} + for mean_name, values in means.items(): + summary['summary'][metric][mean_name] = f"{float(np.mean(values)):.4f} ± {float(np.std(values)):.4f}" + + save_path = stress_logger.log_dir / "stress_benchmark_results.yaml" + with open(save_path, 'w') as file: + yaml.dump(summary, file, allow_unicode=True, sort_keys=False) + stress_logger.info(f"✅ Stress benchmark aggregated execution finished.") + stress_logger.info(f"📁 Stress benchmark results saved to: {save_path}") + return summary diff --git a/robogauge/tasks/simulator/mujoco_config.py b/robogauge/tasks/simulator/mujoco_config.py index 170521e..9750e05 100644 --- a/robogauge/tasks/simulator/mujoco_config.py +++ b/robogauge/tasks/simulator/mujoco_config.py @@ -52,3 +52,4 @@ class MujocoConfig(Config): penetration_threshold = -0.035 # [m], if any contact penetration depth is below this threshold, truncate episode skip_penetration_geoms = ['wall', 'floor'] # Geometries to skip penetration check skip_self_penetration = True # Whether to check self-penetration + penetration_max_reset_num = 1 # Max number of resets due to penetration per run diff --git a/robogauge/tasks/simulator/mujoco_simulator.py b/robogauge/tasks/simulator/mujoco_simulator.py index d0eee6e..666b7c3 100644 --- a/robogauge/tasks/simulator/mujoco_simulator.py +++ b/robogauge/tasks/simulator/mujoco_simulator.py @@ -44,6 +44,7 @@ class MujocoSimulator: self.n_step = 0 self.sim_time = 0.0 self.target_pos = None + self.penetration_reset_count = 0 def load( self, @@ -158,6 +159,7 @@ class MujocoSimulator: self._pause = False self.n_step = 0 self.sim_time = 0.0 + self.penetration_reset_count = 0 self.load_dof_limits() self.preload_sensors() @@ -269,6 +271,8 @@ class MujocoSimulator: return sim_data def check_penetration(self, threshold: float = -0.02): + if self.penetration_reset_count >= self.cfg.truncation.penetration_max_reset_num: + return False, None, None, None for i in range(self.mj_data.ncon): contact = self.mj_data.contact[i] if contact.dist < threshold: @@ -285,16 +289,17 @@ class MujocoSimulator: is_penetrated, geom1, geom2, dist = self.check_penetration(self.cfg.truncation.penetration_threshold) if is_penetrated: - flat = True + is_err = True if self.cfg.truncation.skip_penetration_geoms is not None and ( any(skip_geom in geom1.lower() for skip_geom in self.cfg.truncation.skip_penetration_geoms) or any(skip_geom in geom2.lower() for skip_geom in self.cfg.truncation.skip_penetration_geoms) ): - flat = False + is_err = False if self.cfg.truncation.skip_self_penetration: if geom1.split('/')[0] == geom2.split('/')[0]: - flat = False - if flat: + is_err = False + if is_err: + self.penetration_reset_count += 1 raise RuntimeError(f"[Penetration Error] Episode truncated: Penetration ({geom1} <-> {geom2}), distance: {dist}") def reset(self): diff --git a/robogauge/utils/process_utils.py b/robogauge/utils/process_utils.py new file mode 100644 index 0000000..55d3311 --- /dev/null +++ b/robogauge/utils/process_utils.py @@ -0,0 +1,29 @@ +# -*- coding: utf-8 -*- +''' +@File : process_utils.py +@Time : 2025/12/26 21:59:53 +@Author : wty-yy +@Version : 1.0 +@Blog : https://wty-yy.github.io/ +@Desc : No Daemon Pool for Multiprocessing +''' +import multiprocessing +import multiprocessing.pool + +class NoDaemonProcess(multiprocessing.Process): + @property + def daemon(self): + return False + @daemon.setter + def daemon(self, value): + pass + +class NoDaemonPool(multiprocessing.pool.Pool): + def __init__(self, *args, **kwargs): + """ context=ctx in kwargs is required """ + super(NoDaemonPool, self).__init__(*args, **kwargs) + + def Process(self, *args, **kwds): + proc = super(NoDaemonPool, self).Process(*args, **kwds) + proc.__class__ = NoDaemonProcess + return proc