From d2a16cc075f1d01ad7ad4cba7eff9e38476c9df3 Mon Sep 17 00:00:00 2001 From: wty-yy <993660140@qq.com> Date: Wed, 25 Mar 2026 15:32:39 +0800 Subject: [PATCH] v1.0.2-rc2; fix robogauge eval return None bug --- UPDATE.md | 3 ++ rsl_rl/rsl_rl/runners/on_policy_runner.py | 52 +++++++++++++------ rsl_rl/rsl_rl/runners/on_policy_runner_cts.py | 52 +++++++++++++------ tools/logs_merge.py | 50 +++++++++++++++--- 4 files changed, 117 insertions(+), 40 deletions(-) diff --git a/UPDATE.md b/UPDATE.md index bc85292..192fa6e 100644 --- a/UPDATE.md +++ b/UPDATE.md @@ -1,3 +1,6 @@ +# 20260325 +## v1.0.2-rc2 +1. 修复robogauge评估中返回None导致的训练中断问题 # 20260126 ## v1.0.2-rc1 1. 修改高速移动的训练文件到最终版，删除配置中无用注释 diff --git a/rsl_rl/rsl_rl/runners/on_policy_runner.py b/rsl_rl/rsl_rl/runners/on_policy_runner.py index 1ef82e2..03d266e 100644 --- a/rsl_rl/rsl_rl/runners/on_policy_runner.py +++ b/rsl_rl/rsl_rl/runners/on_policy_runner.py @@ -253,38 +253,56 @@ class OnPolicyRunner: if self.robogauge_client is None: return - if it % 500 == 0 or last_model: - # export jit model - jit_dir = os.path.join(self.log_dir, 'jit_models') - jit_path = os.path.join(jit_dir, f'policy_jit_{it}.pt') - export_policy_as_jit(self.alg.actor_critic, jit_dir, filename=f'policy_jit_{it}.pt') - # upload to robogauge - task_name = 'go2' - self.robogauge_client.submit_task( - model_path=jit_path, - step=it, - task_name=task_name, - experiment_name=self.cfg["experiment_name"] - ) + try: + if it % 500 == 0 or last_model: + # export jit model + jit_dir = os.path.join(self.log_dir, 'jit_models') + jit_path = os.path.join(jit_dir, f'policy_jit_{it}.pt') + export_policy_as_jit(self.alg.actor_critic, jit_dir, filename=f'policy_jit_{it}.pt') + # upload to robogauge + task_name = 'go2' + self.robogauge_client.submit_task( + model_path=jit_path, + step=it, + task_name=task_name, + experiment_name=self.cfg["experiment_name"] + ) + except Exception as e: + print(f"[WARN] RoboGauge submit failed at step {it}: {e}") + return check_times = 1 if last_model: check_times = int(1e9) # keep checking until the last model is evaluated while check_times > 0: check_times -= 1 - self.robogauge_client.monitor_tasks() + try: + self.robogauge_client.monitor_tasks() + except Exception as e: + print(f"[WARN] RoboGauge monitor failed at step {it}: {e}") + break results_dir = os.path.join(self.log_dir, 'robogauge_results') os.makedirs(results_dir, exist_ok=True) result_received = False for task_id, resp in self.robogauge_client.response_data.items(): - scores = resp['results']['scores'] - step = resp['step'] + if not isinstance(resp, dict): + print(f"[WARN] RoboGauge returned an invalid response for task {task_id}: {resp}") + continue + results = resp.get('results') + step = resp.get('step', it) + if results is None: + print(f"[WARN] RoboGauge returned empty results for task {task_id} at step {step}.") + continue + scores = results.get('scores') + if scores is None: + print(f"[WARN] RoboGauge results for task {task_id} at step {step} do not contain 'scores'.") + continue if step == it: result_received = True for key, val in scores.items(): self.writer.add_scalar(f'RoboGauge/{key}', val, step) results_path = os.path.join(results_dir, f'results_{step}.yaml') with open(results_path, 'w', encoding='utf-8') as f: - yaml.dump(resp['results'], f, allow_unicode=True, sort_keys=False) + yaml.dump(results, f, allow_unicode=True, sort_keys=False) if last_model and result_received: print(f"RoboGauge result for step {it} received. Exiting wait loop.") diff --git a/rsl_rl/rsl_rl/runners/on_policy_runner_cts.py b/rsl_rl/rsl_rl/runners/on_policy_runner_cts.py index 05eee9e..8b34115 100644 --- a/rsl_rl/rsl_rl/runners/on_policy_runner_cts.py +++ b/rsl_rl/rsl_rl/runners/on_policy_runner_cts.py @@ -298,38 +298,56 @@ class OnPolicyRunnerCTS: if self.robogauge_client is None: return - if it % 500 == 0 or last_model: - # export jit model - jit_dir = os.path.join(self.log_dir, 'jit_models') - jit_path = os.path.join(jit_dir, f'policy_jit_{it}.pt') - export_policy_as_jit(self.alg.model, jit_dir, filename=f'policy_jit_{it}.pt') - # upload to robogauge - task_name = 'go2_moe' # Both cts, moe-cts actor return a tuple `action, (latent, ...)` - self.robogauge_client.submit_task( - model_path=jit_path, - step=it, - task_name=task_name, - experiment_name=self.cfg["experiment_name"] - ) + try: + if it % 500 == 0 or last_model: + # export jit model + jit_dir = os.path.join(self.log_dir, 'jit_models') + jit_path = os.path.join(jit_dir, f'policy_jit_{it}.pt') + export_policy_as_jit(self.alg.model, jit_dir, filename=f'policy_jit_{it}.pt') + # upload to robogauge + task_name = 'go2_moe' # Both cts, moe-cts actor return a tuple `action, (latent, ...)` + self.robogauge_client.submit_task( + model_path=jit_path, + step=it, + task_name=task_name, + experiment_name=self.cfg["experiment_name"] + ) + except Exception as e: + print(f"[WARN] RoboGauge submit failed at step {it}: {e}") + return check_times = 1 if last_model: check_times = int(1e9) # keep checking until manually stopped while check_times > 0: check_times -= 1 - self.robogauge_client.monitor_tasks() + try: + self.robogauge_client.monitor_tasks() + except Exception as e: + print(f"[WARN] RoboGauge monitor failed at step {it}: {e}") + break results_dir = os.path.join(self.log_dir, 'robogauge_results') os.makedirs(results_dir, exist_ok=True) result_received = False for task_id, resp in self.robogauge_client.response_data.items(): - scores = resp['results']['scores'] - step = resp['step'] + if not isinstance(resp, dict): + print(f"[WARN] RoboGauge returned an invalid response for task {task_id}: {resp}") + continue + results = resp.get('results') + step = resp.get('step', it) + if results is None: + print(f"[WARN] RoboGauge returned empty results for task {task_id} at step {step}.") + continue + scores = results.get('scores') + if scores is None: + print(f"[WARN] RoboGauge results for task {task_id} at step {step} do not contain 'scores'.") + continue if step == it: result_received = True for key, val in scores.items(): self.writer.add_scalar(f'RoboGauge/{key}', val, step) results_path = os.path.join(results_dir, f'results_{step}.yaml') with open(results_path, 'w', encoding='utf-8') as f: - yaml.dump(resp['results'], f, allow_unicode=True, sort_keys=False) + yaml.dump(results, f, allow_unicode=True, sort_keys=False) if last_model and result_received: print(f"RoboGauge result for step {it} received. Exiting wait loop.") diff --git a/tools/logs_merge.py b/tools/logs_merge.py index 8a72ddc..92fc62f 100644 --- a/tools/logs_merge.py +++ b/tools/logs_merge.py @@ -36,7 +36,39 @@ def fast_read(event_file_path, tag_names): if value.tag in tag_names: tag_data[event.step][value.tag] = value.simple_value - return pd.DataFrame(tag_data).T + df = pd.DataFrame(tag_data).T + df.index.name = 'step' + return df + +def normalize_tb_df(tb_df): + tb_df = tb_df.copy() + + if 'step' not in tb_df.columns: + first_col = tb_df.columns[0] if len(tb_df.columns) > 0 else None + if first_col is not None and str(first_col).startswith('Unnamed:'): + tb_df = tb_df.rename(columns={first_col: 'step'}) + elif tb_df.index.name == 'step': + tb_df = tb_df.reset_index() + else: + tb_df = tb_df.reset_index().rename(columns={'index': 'step'}) + + tb_df['step'] = pd.to_numeric(tb_df['step'], errors='coerce') + tb_df = tb_df.dropna(subset=['step']) + tb_df['step'] = tb_df['step'].astype(int) + return tb_df + +def get_tb_value(tb_df, step, candidate_tags): + row = tb_df[tb_df['step'] == step] + if row.empty: + raise KeyError(f"No tensorboard entry found for step={step}.") + + for tag in candidate_tags: + if tag not in row.columns: + continue + values = row[tag].dropna().values + if len(values) > 0: + return float(values[0]) + raise KeyError(f"No tensorboard value found for step={step} in tags: {candidate_tags}") class Collector: def __init__(self, log_dirs): @@ -58,14 +90,14 @@ class Collector: self.output_tb = self.output_dir / "tb.csv" if self.output_tb.exists(): print(f"Loading existing tensorboard data from {self.output_tb}") - self.tb_df = pd.read_csv(self.output_tb) + self.tb_df = normalize_tb_df(pd.read_csv(self.output_tb)) else: start_time = time.time() print(f"Start reading tensorboard events at {time.ctime(start_time)}") - self.tb_df = fast_read(str(self.log_dirs.glob("events.out.tfevents.*").__next__()), [ + self.tb_df = normalize_tb_df(fast_read(str(self.log_dirs.glob("events.out.tfevents.*").__next__()), [ 'Terrain/terrain_level_all', 'Episode/terrain_level_all', 'RoboGauge/benchmark' - ]) + ])) print(f"Finished reading tensorboard events in {time.time() - start_time:.2f} seconds.") self.tb_df.to_csv(self.output_tb, index=False) print(f"Saved tensorboard data to {self.output_tb}") @@ -109,7 +141,11 @@ class Collector: self.datas[f'{terrain_name}_mean@25'].append(float(data['robust_score'][terrain_name]['mean@25'])) self.datas[f'{terrain_name}_mean@50'].append(float(data['robust_score'][terrain_name]['mean@50'])) - self.datas['terrain_level'].append(float(self.tb_df[self.tb_df['step'] == it]['value'].values[0])) + self.datas['terrain_level'].append(get_tb_value( + self.tb_df, + it, + ['Terrain/terrain_level_all', 'Episode/terrain_level_all'] + )) df = pd.DataFrame(self.datas) df.to_csv(self.output_csv, index=False) print(f"Saved merged results to {self.output_csv}") @@ -117,6 +153,8 @@ class Collector: if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument("--log-dirs") + parser.add_argument("--read-robogauge", default=True, type=lambda x: (str(x).lower() in ['true', '1']), help="Whether to read robogauge_results") args = parser.parse_args() collector = Collector(args.log_dirs) - # collector.collect() + if args.read_robogauge: + collector.collect()