""" # @Time : 2021/7/1 7:15 下午 # @Author : hezhiqiang01 # @Email : hezhiqiang01@baidu.com # @File : env_runner.py """ # 原 import import numpy as np import torch from .base_runner import Runner # 数据处理 import os import csv import json import math import pickle import subprocess import sys from datetime import datetime from pathlib import Path import time import yaml # from copy import deepcopy def _t2n(x): return x.detach().cpu().numpy() def _concat_obs_dict(obs): if not isinstance(obs, dict): return np.concatenate(obs) return {k: np.concatenate(v) for k, v in obs.items()} class MACFEnvRunner(Runner): """Runner class to perform training, evaluation. and data collection for the MPEs. See parent class for details.""" def __init__(self, config): super(MACFEnvRunner, self).__init__(config) self.episode = 0 self.env_episode = 0 self.info_buffer = [] self.writer = self.writter print(getattr(self.writer, "logdir", self.tb_dir)) self.best_eval_score = None self.best_eval_summary = None self.last_save_env_episode = -1 self.last_eval_env_episode = -1 @staticmethod def _episode_interval_due(current_episode, last_episode, interval): """Check whether an episode-based trigger is due. Supports a single interval or a list of intervals. A trigger fires at episode 0 and every ``interval`` episodes afterwards (i.e. at ``interval - 1, 2 * interval - 1, ...``) for each configured interval. This matches 0-indexed episode counting. """ if current_episode < 0: return False intervals = interval if isinstance(interval, (list, tuple)) else [interval] intervals = [int(i) for i in intervals if int(i) > 0] if not intervals: return False def _next_node_after(ep, n): # Nodes are 0, n-1, 2n-1, 3n-1, ... if ep < 0: return 0 k = (ep + 1) // n candidate = k * n - 1 if candidate > ep: return max(0, candidate) return (k + 1) * n - 1 next_nodes = [_next_node_after(last_episode, n) for n in intervals] return current_episode >= min(next_nodes) def _info_log_interval(self): if self.env_episode < 10: return 1 if self.env_episode < 100: return 10 return int(self.log_info_interval) def _publish_eval_progress(self, completed, state="running"): progress_file = getattr(self.config, "eval_progress_file", None) if not progress_file: return path = Path(progress_file) path.parent.mkdir(parents=True, exist_ok=True) payload = { "state": str(state), "completed_episodes": int(completed), "total_episodes": int(self.eval_episodes), "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"), } temporary = path.with_name(f"{path.name}.{os.getpid()}.tmp") with temporary.open("w", encoding="utf-8") as file: json.dump(payload, file, ensure_ascii=False, indent=2) temporary.replace(path) @staticmethod def _print_eval_summary(summary): print( "eval: episodes={}, success={:.1%}, reward={:.2f}±{:.2f}, " "length={:.1f}±{:.1f}, successful_length={}".format( summary["num_episodes"], summary["success"], summary["episode_reward"], summary["episode_reward_std"], summary["episode_length"], summary["episode_length_std"], "NA" if summary["successful_episode_length"] is None else f"{summary['successful_episode_length']:.1f}", ), flush=True, ) def _log_eval_scalars(self, summary, total_num_steps): for key, value in summary.items(): if value is not None and isinstance(value, (int, float)): self.writer.add_scalar(f"eval/{key}", value, total_num_steps) def _external_periodic_eval(self, total_num_steps, checkpoint_path, eval_cuda): """Evaluate a frozen checkpoint on another CUDA device. The training process waits at the checkpoint boundary, preserving the existing best/latest selection order. Other jobs on the training GPU continue, while this job's evaluation model and environments live in a short-lived subprocess on ``eval_cuda``. """ config_file = getattr(self.config, "config_file", None) if not config_file or not Path(config_file).is_file(): raise RuntimeError( "training_eval_cuda requires training to use a readable config_file" ) root = Path(self.run_dir) / "training_eval_external" step_dir = root / f"step_{int(total_num_steps):012d}" step_dir.mkdir(parents=True, exist_ok=True) with Path(config_file).open("r", encoding="utf-8") as file: eval_config = yaml.safe_load(file) or {} eval_config.update( cuda=int(eval_cuda), use_eval=True, model_dir=None, eval_progress_file=None, ) eval_config_path = step_dir / "config.yaml" with eval_config_path.open("w", encoding="utf-8") as file: yaml.safe_dump(eval_config, file, sort_keys=False, allow_unicode=True) manifest = { "name": f"training-eval-step-{int(total_num_steps):012d}", "output_root": str(step_dir / "results"), "models": [ { "name": "checkpoint", "checkpoint": str(Path(checkpoint_path).resolve()), "overrides": {}, } ], } manifest_path = step_dir / "manifest.yaml" with manifest_path.open("w", encoding="utf-8") as file: yaml.safe_dump(manifest, file, sort_keys=False, allow_unicode=True) progress_path = step_dir / "progress.json" command = [ sys.executable, str(Path(__file__).resolve().parents[2] / "train" / "eval.py"), "--config_file", str(eval_config_path), "--models_file", str(manifest_path), "--progress_file", str(progress_path), ] guard = eval_config.get("gpu_thermal_guard", {}) or {} if guard.get("enabled", False): command = [ sys.executable, str( Path(__file__).resolve().parents[2] / "scripts" / "gpu_thermal_guard.py" ), "--gpu", str(int(eval_cuda)), "--pause-at", str(int(guard.get("pause_at_c", 93))), "--resume-at", str(int(guard.get("resume_at_c", 85))), "--poll-seconds", str(float(guard.get("poll_seconds", 2))), "--safe-readings", str(int(guard.get("safe_readings", 3))), "--query-failures", str(int(guard.get("query_failures", 2))), "--status-file", str(step_dir / "thermal_status.json"), "--", *command, ] environment = os.environ.copy() environment["PYTHONUNBUFFERED"] = "1" process = subprocess.Popen( command, cwd=Path(__file__).resolve().parents[2], env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) output_lines = [] suite_dir = None assert process.stdout is not None for line in process.stdout: output_lines.append(line) stripped = line.strip() if stripped.startswith("eval progress:"): print(stripped, flush=True) if "Evaluation suite written to" in stripped: suite_dir = stripped.split("Evaluation suite written to", 1)[1].strip() return_code = process.wait() if return_code != 0 or not suite_dir: tail = "".join(output_lines[-30:]) raise RuntimeError( f"external periodic eval failed on cuda:{eval_cuda} " f"with return code {return_code}\n{tail}" ) summary_path = Path(suite_dir) / "summary.json" with summary_path.open("r", encoding="utf-8") as file: payload = json.load(file) records = payload.get("models") or [] if len(records) != 1: raise RuntimeError(f"Unexpected external eval summary: {summary_path}") summary = { key: value for key, value in records[0].items() if key not in ("name", "checkpoint") } self._log_eval_scalars(summary, total_num_steps) self._print_eval_summary(summary) return summary def periodic_eval(self, total_num_steps, checkpoint_path): eval_cuda = getattr(self.config, "training_eval_cuda", None) if eval_cuda is None or int(eval_cuda) == int(self.config.cuda): return self.eval(total_num_steps) return self._external_periodic_eval( total_num_steps, checkpoint_path, int(eval_cuda) ) def run(self): # 初始化 self.warmup() start = datetime.now() episodes = int(self.num_env_steps) // self.episode_length // self.n_rollout_threads max_train_episodes = getattr(self.config, "max_train_episodes", None) if max_train_episodes is not None: max_train_episodes = int(max_train_episodes) if max_train_episodes <= 0: raise ValueError("max_train_episodes must be positive or None.") episodes = min(episodes, max_train_episodes) # 在文件夹不存在时创建文件夹 os.makedirs(os.path.join(self.log_dir, 'infos'), exist_ok=True) # 时间步循环 episode_reward = 0 sent = 0 sent_goal = 0 last_episode_sent_percentage = 0.0 for self.episode in range(episodes): episode = self.episode # self.warmup() rollout_reward = 0 success_flags = 0 if self.use_linear_lr_decay: self.trainer.policy.lr_decay(episode, episodes) for step in range(self.episode_length): # Sample actions 动作采样 ( values, actions, action_log_probs, gate_actions, rnn_states, rnn_states_critic, ) = self.collect(step) # Obser reward and next obs 环境交互 obs, rewards, dones, infos, active_masks, bad_masks, available_actions = self.envs.step(actions) state = self.envs.get_global_state() sent += infos[0]['sent'] sent_goal = infos[0]['sent_goal'] episode_reward += rewards.mean() rollout_reward += rewards.mean() data = ( obs, state, rewards, dones, infos, values, actions, action_log_probs, gate_actions, rnn_states, rnn_states_critic, active_masks, bad_masks, available_actions, ) # insert data into buffer 插入经验回放缓冲区 self.insert(data) # buffer 写入后再记录 scalar if infos[0]['end']: tc = sum( 1 for t in infos[0]['tasks'] if int(t.get("type", -1)) == 1 and t["completed"] ) sp = sent / sent_goal if sent_goal > 0 else 0 last_episode_sent_percentage = sp self.writer.add_scalar("reward/episode", episode_reward, self.env_episode) self.writer.add_scalar("reward/sent_percentage", sp, self.env_episode) self.writer.add_scalar("reward/task_completed", tc, self.env_episode) self.writer.add_scalar("reward/episode_length", infos[0]['end'], self.env_episode) reconfiguration = infos[0].get("reconfiguration", {}) reconfiguration_decisions = int( reconfiguration.get("total_decisions", 0) ) reconfiguration_changes = int( reconfiguration.get("total_coalition_changes", 0) ) self.writer.add_scalar( "reconfiguration/decisions", reconfiguration_decisions, self.env_episode, ) self.writer.add_scalar( "reconfiguration/coalition_changes", reconfiguration_changes, self.env_episode, ) self.writer.add_scalar( "reconfiguration/change_per_decision", reconfiguration_changes / max(1, reconfiguration_decisions), self.env_episode, ) alive_agent_steps = int( reconfiguration.get("total_alive_agent_steps", 0) ) self.writer.add_scalar( "reconfiguration/decisions_per_100_alive_steps", 100.0 * reconfiguration_decisions / max(1, alive_agent_steps), self.env_episode, ) self.writer.add_scalar( "reconfiguration/changes_per_100_alive_steps", 100.0 * reconfiguration_changes / max(1, alive_agent_steps), self.env_episode, ) self.writer.add_scalar( "reconfiguration/mean_response_time", float(reconfiguration.get("mean_response_time", 0.0)), self.env_episode, ) self.writer.add_scalar( "reconfiguration/urgency_pressure", float(reconfiguration.get("urgency_pressure", 0.0)), self.env_episode, ) # ===== 记录零初始化门控 ===== if ( self.use_gat and getattr(self.trainer.policy, "graph_model", None) == "legacy_gat" and hasattr( self.trainer.policy, "get_gat_gate_values" ) ): actor_alpha, critic_alpha = ( self.trainer.policy.get_gat_gate_values() ) self.writer.add_scalar( "gat/actor_alpha", actor_alpha, self.env_episode ) self.writer.add_scalar( "gat/critic_alpha", critic_alpha, self.env_episode ) if self.use_gat and hasattr( self.trainer.policy, "get_graph_diagnostics" ): for tag, value in self.trainer.policy.get_graph_diagnostics().items(): self.writer.add_scalar(tag, value, self.env_episode) # ========================== # 保存 info_buffer log_info_interval = self._info_log_interval() if self.log_info_interval > 0: if self.env_episode % log_info_interval == 0: with open(os.path.join(self.log_dir, 'infos', f'{self.env_episode:04d}.pkl'), 'wb') as f: pickle.dump(self.info_buffer[1:], f) print(f'{self.env_episode:04d}.pkl generated') self.info_buffer = [] self.env_episode += 1 episode_reward = 0 sent = 0 if ( self.config.curriculum_switch_ep > 0 and self.env_episode >= self.config.curriculum_switch_ep and getattr(self, "_curriculum_switched", False) is False ): for env in self.envs.envs: env.set_reset_mode("full_random") self._curriculum_switched = True print( f"curriculum: switched to full_random at env_episode {self.env_episode}" ) if ( max_train_episodes is not None and self.env_episode >= max_train_episodes and step + 1 < self.episode_length ): # A partial rollout cannot be used by the fixed-size PPO buffer. # Preserve the latest trained weights and stop at the exact env-episode cap. self.save() return # if np.all(dones): # break t0 = t = datetime.now().timestamp() ta = [] # info_buffer 可能已在 env episode 结束时清空，用最后一条 info last_info = self.info_buffer[-1][0] if self.info_buffer else infos[0] sent_goal = last_info['sent_goal'] task_completed = sum( 1 for t in last_info['tasks'] if int(t.get("type", -1)) == 1 and t["completed"] ) t = datetime.now().timestamp() ta.append(t - t0) t0 = t # compute return and update network self.compute() t = datetime.now().timestamp() ta.append(t - t0) t0 = t train_infos = self.train() t = datetime.now().timestamp() ta.append(t - t0) t0 = t # post process total_num_steps = (episode + 1) * self.episode_length * self.n_rollout_threads # save model save_due = self._episode_interval_due( self.env_episode, self.last_save_env_episode, self.save_interval, ) if save_due or episode == episodes - 1: self.save() if getattr(self.config, "save_episode_checkpoints", False): self.save(tag=f"ep{self.env_episode:06d}") self.last_save_env_episode = self.env_episode # log information if episode % self.log_interval == 0: end = datetime.now() print( "\n rollout {}/{}, env_episode {}, {}% sent, {} completed, ep_reward: {:>7.2f}, timesteps {}/{}, FPS {}, time cost {}, expected time {}.\n".format( episode, episodes, self.env_episode, last_episode_sent_percentage * 100, task_completed, rollout_reward, total_num_steps, self.num_env_steps, int(total_num_steps / (end - start).total_seconds()), (end - start).total_seconds(), float(episodes)/(1 if episode == 0 else episode)*((end - start).total_seconds()) ) ) train_infos["average_episode_rewards"] = np.mean(self.buffer.rewards) * self.episode_length print("average episode rewards is {}".format(train_infos["average_episode_rewards"])) self.log_train(train_infos, total_num_steps) # self.log_env(env_infos, total_num_steps) # eval eval_due = self._episode_interval_due( self.env_episode, self.last_eval_env_episode, self.eval_interval, ) if eval_due and self.use_eval: tag = f"ep{self.env_episode:06d}" self.save(tag=tag) summary = self.periodic_eval( total_num_steps, Path(self.save_dir) / f"actor_{tag}.pt", ) self._save_eval_checkpoints(summary, total_num_steps) self.last_eval_env_episode = self.env_episode if ( max_train_episodes is not None and self.env_episode >= max_train_episodes ): break t = datetime.now().timestamp() ta.append(t - t0) t0 = t def warmup(self): # reset env — obs is dict obs, active_masks, available_actions = self.envs.reset(self.episode) # obs: dict[str, ndarray] state = self.envs.get_global_state() self.info_buffer = [] share_obs = self._build_share_obs(state) for k in self.buffer.share_obs: self.buffer.share_obs[k][0] = share_obs[k].copy() for k in self.buffer.obs: self.buffer.obs[k][0] = obs[k].copy() self.buffer.active_masks[0] = active_masks.copy() self.buffer.available_actions[0] = available_actions.copy() def _build_share_obs(self, state): if not self.use_centralized_V: raise NotImplementedError("Structured local state for use_centralized_V=False is not implemented.") repeated = { k: np.repeat(v[:, None, ...], self.num_agents, axis=1) for k, v in state.items() } focal = np.eye(self.num_agents, dtype=np.float32)[None, :, :, None] repeated["focal"] = np.repeat(focal, self.n_rollout_threads, axis=0) return repeated def _get_actions(self, step, available_actions): obs_batch = _concat_obs_dict({k: v[step] for k, v in self.buffer.obs.items()}) if self.use_gat: return self.trainer.policy.get_actions( {k: np.concatenate(v[step]) for k, v in self.buffer.share_obs.items()}, obs_batch, np.concatenate(self.buffer.rnn_states[step]), np.concatenate(self.buffer.rnn_states_critic[step]), np.concatenate(self.buffer.masks[step]), available_actions=np.concatenate(available_actions) ) else: return self.trainer.policy.get_actions( {k: np.concatenate(v[step]) for k, v in self.buffer.share_obs.items()}, obs_batch, np.concatenate(self.buffer.rnn_states[step]), np.concatenate(self.buffer.rnn_states_critic[step]), np.concatenate(self.buffer.masks[step]), available_actions=np.concatenate(available_actions) ) @torch.no_grad() def collect(self, step): available_actions = self.buffer.available_actions[step] active_masks = self.buffer.active_masks[step] self.trainer.prep_rollout() action_outputs = self._get_actions(step, available_actions) if len(action_outputs) == 6: ( value, action, action_log_prob, gate_action, rnn_states, rnn_states_critic, ) = action_outputs else: # Standard MAPPO has no explicit hierarchical gate action. Keep a # zero placeholder in the runner tuple so the environment loop can # remain shared, but do not pass it into SharedReplayBuffer below. ( value, action, action_log_prob, rnn_states, rnn_states_critic, ) = action_outputs gate_action = torch.zeros_like(action, dtype=torch.float32) # [self.envs, agents, dim] values = np.array(np.split(_t2n(value), self.n_rollout_threads)) # [env_num, agent_num, 1] actions = np.array(np.split(_t2n(action), self.n_rollout_threads)) # [env_num, agent_num, action_dim] action_log_probs = np.array( np.split(_t2n(action_log_prob), self.n_rollout_threads) ) # [env_num, agent_num, 1] gate_actions = np.array( np.split(_t2n(gate_action), self.n_rollout_threads) ) rnn_states = np.array( np.split(_t2n(rnn_states), self.n_rollout_threads) ) # [env_num, agent_num, 1, hidden_size] rnn_states_critic = np.array( np.split(_t2n(rnn_states_critic), self.n_rollout_threads) ) # [env_num, agent_num, 1, hidden_size] if step == 0: last_actions = self.buffer.actions[-1] last_action_log_probs = self.buffer.action_log_probs[-1] else: last_actions = self.buffer.actions[step-1] last_action_log_probs = self.buffer.action_log_probs[step-1] # 添加 active_masks 掩码 actions = active_masks * actions + (1 - active_masks) * last_actions action_log_probs = active_masks * action_log_probs + (1 - active_masks) * last_action_log_probs # inactive agent 沿用上一动作，不产生新的 gate 决策；否则 PPO 会把 # 环境未执行的 redecision 误当作 rollout 样本。 gate_actions = active_masks * gate_actions return ( values, actions, action_log_probs, gate_actions, rnn_states, rnn_states_critic, # active agents ) def insert(self, data): ( obs, state, rewards, dones, infos, values, actions, action_log_probs, gate_actions, rnn_states, rnn_states_critic, active_masks, bad_masks, available_actions, ) = data rnn_states[dones == True] = np.zeros( ((dones == True).sum(), self.recurrent_N, self.hidden_size), dtype=np.float32, ) rnn_states_critic[dones == True] = np.zeros( ((dones == True).sum(), *self.buffer.rnn_states_critic.shape[3:]), dtype=np.float32, ) masks = np.ones((self.n_rollout_threads, self.num_agents, 1), dtype=np.float32) masks[dones == True] = np.zeros(((dones == True).sum(), 1), dtype=np.float32) share_obs = self._build_share_obs(state) if bool(getattr(self.config, "mappo_team_mean_reward", False)): rewards = np.repeat( rewards.mean(axis=1, keepdims=True), self.num_agents, axis=1 ) # rewards = rewards[:, None] # (T,) → (T,1) # masks = masks[:, None] # 同理 common_args = ( share_obs, obs, rnn_states, rnn_states_critic, actions, action_log_probs, ) trailing_args = (values, rewards, masks) if self.use_gat: self.buffer.insert( *common_args, gate_actions, *trailing_args, active_masks=active_masks, bad_masks=bad_masks, available_actions=available_actions, ) else: self.buffer.insert( *common_args, *trailing_args, active_masks=active_masks, bad_masks=bad_masks, available_actions=available_actions, ) self.info_buffer.append(infos) @torch.no_grad() def _run_eval_episode(self, env, scenario_id): scenario_seed = self.eval_seed + scenario_id numpy_state = np.random.get_state() np.random.seed(scenario_seed) try: env.seed(scenario_seed) obs, active_masks, available_actions = env.reset(scenario_id) rnn_states = np.zeros( (self.num_agents, self.recurrent_N, self.hidden_size), dtype=np.float32, ) masks = np.ones((self.num_agents, 1), dtype=np.float32) last_actions = np.zeros((self.num_agents, 1), dtype=np.float32) previous_coalitions = np.array( [agent.coalition for agent in env.env.agents], dtype=np.int64 ) episode_reward = 0.0 sent = 0.0 coalition_changes = 0 premature_no_explorer_steps = 0 reward_components = {} decision_type_counts = {task_type: 0 for task_type in range(4)} executed_type_counts = {task_type: 0 for task_type in range(4)} decision_count = 0 executed_count = 0 gate_decision_count = 0 gate_open_count = 0 gate_mandatory_count = 0 gate_mandatory_open_count = 0 gate_event_trigger_count = 0 gate_event_trigger_open_count = 0 gate_learned_count = 0 gate_learned_open_count = 0 num_explore_actions = int( getattr(env.env, "num_explore_actions", 1) ) sector_decision_counts = np.zeros( (self.num_agents, num_explore_actions), dtype=np.int64 ) explore_legs_planned = 0 explore_planned_distance = 0.0 short_explore_legs = 0 explore_distance_moved = 0.0 unavailable_sector_ratio_sum = 0.0 alive_agent_steps = 0 harmful_idle_agent_steps = 0 useful_idle_agent_steps = 0 committed_payload_shortage_ratio_sum = 0.0 committed_payload_surplus_ratio_sum = 0.0 max_idle_duration = 0.0 final_info = None self.trainer.prep_rollout() eval_step_limit = int( getattr(self.config, "episode_time_limit", None) or self.config.max_steps ) for step in range(eval_step_limit): previous_rnn_states = rnn_states decision_active_masks = np.asarray(active_masks).copy() actions, next_rnn_states = self.trainer.policy.act( obs, rnn_states, masks, available_actions=available_actions, deterministic=getattr(self.config, "eval_deterministic", True), ) gate_actions = getattr( getattr(self.trainer.policy, "actor", None), "last_gate_actions", None, ) actions = _t2n(actions) next_rnn_states = _t2n(next_rnn_states) proposed_actions = actions.astype(np.int64).reshape(-1) decision_mask = np.asarray(active_masks).reshape(-1) > 0.5 if gate_actions is not None: # hierarchical gate 的 z=1 表示本次 active 机会真正交给任务 # decoder；z=0 表示保持 current。这里按 eval episode 口径统计， # 区分环境强制开门和 learned gate 自主开门。 gate_values = _t2n(gate_actions).reshape(-1) > 0.5 active_gate_values = gate_values[decision_mask] gate_decision_count += int(active_gate_values.size) gate_open_count += int(np.count_nonzero(active_gate_values)) reconfig_obs = ( obs.get("reconfiguration") if isinstance(obs, dict) else None ) if reconfig_obs is not None: reconfig_values = np.asarray(reconfig_obs) context_dim = reconfig_values.shape[-1] # Match HATTV20Actor's event override exactly. The # previous metric treated only context[10] as forced, # so event-only evaluation incorrectly reported its # context[0/1/3] openings as "learned" openings. event_trigger = np.zeros( reconfig_values.shape[:-1], dtype=bool ) if context_dim >= 4: event_trigger = ( (reconfig_values[..., 0] > 0.5) | (reconfig_values[..., 1] > 0.5) | (reconfig_values[..., 3] > 0.5) ) mandatory_all = np.zeros_like(event_trigger) if context_dim >= 11: mandatory_all = reconfig_values[..., 10] > 0.5 event_trigger = event_trigger | mandatory_all event_trigger = event_trigger.reshape(-1)[decision_mask] mandatory = mandatory_all.reshape(-1)[decision_mask] gate_mandatory_count += int(np.count_nonzero(mandatory)) gate_mandatory_open_count += int( np.count_nonzero(active_gate_values & mandatory) ) gate_event_trigger_count += int( np.count_nonzero(event_trigger) ) gate_event_trigger_open_count += int( np.count_nonzero(active_gate_values & event_trigger) ) # Keep the historical learned metric for downstream # CSV compatibility, but give it the correct meaning: # openings outside every forced event trigger. learned = ~event_trigger gate_learned_count += int(np.count_nonzero(learned)) gate_learned_open_count += int( np.count_nonzero(active_gate_values & learned) ) for agent_idx in np.flatnonzero(decision_mask): action_id = proposed_actions[agent_idx] task_type = int(env.env.tasks[int(action_id)].type) decision_type_counts[task_type] += 1 decision_count += 1 if ( task_type == 0 and getattr(env.env, "explore_action_mode", None) == "sector" ): sector = int(action_id) - int(env.env.explore_task_start) if 0 <= sector < num_explore_actions: sector_decision_counts[agent_idx, sector] += 1 actions = active_masks * actions + (1.0 - active_masks) * last_actions for action_id in actions.astype(np.int64).reshape(-1): task_type = int(env.env.tasks[int(action_id)].type) executed_type_counts[task_type] += 1 executed_count += 1 ( obs, rewards, dones, info, active_masks, _, available_actions, ) = env.step(actions) final_info = info episode_reward += float(np.mean(rewards)) sent += float(info["sent"]) allocation = info.get("allocation_diagnostics", {}) exploration = info.get("exploration_diagnostics", {}) alive_agent_steps += int(sum(agent["living"] for agent in info["agents"])) harmful_idle_agent_steps += int(allocation.get("harmful_idle_agents", 0)) useful_idle_agent_steps += int(allocation.get("useful_idle_agents", 0)) committed_payload_shortage_ratio_sum += float( allocation.get("committed_payload_shortage_ratio", 0.0) ) committed_payload_surplus_ratio_sum += float( allocation.get("committed_payload_surplus_ratio", 0.0) ) explore_legs_planned += int(exploration.get("legs_planned", 0)) explore_planned_distance += float( exploration.get("planned_distance", 0.0) ) short_explore_legs += int(exploration.get("short_legs", 0)) explore_distance_moved += float( exploration.get("distance_moved", 0.0) ) unavailable_sector_ratio_sum += float( exploration.get("unavailable_sector_ratio", 0.0) ) max_idle_duration = max( max_idle_duration, max( ( float(agent.get("idle_wait_steps", 0.0)) for agent in info["agents"] if int(info["tasks"][int(agent["coalition"])]["type"]) == 2 ), default=0.0, ), ) component_values = np.asarray(info["rewards"], dtype=np.float64).mean(axis=1) for key, value in zip(info["reward_keys"], component_values): reward_components[key] = reward_components.get(key, 0.0) + float(value) coalitions = np.array( [agent["coalition"] for agent in info["agents"]], dtype=np.int64 ) coalition_changes += int(np.count_nonzero(coalitions != previous_coalitions)) previous_coalitions = coalitions delivery_tasks = [task for task in info["tasks"] if int(task["type"]) == 1] discovered_ratio = ( sum(task["discovered"] for task in delivery_tasks) / max(1, len(delivery_tasks)) ) any_explorer = any( agent["living"] and int(info["tasks"][int(agent["coalition"])]["type"]) == 0 for agent in info["agents"] ) if not any_explorer and discovered_ratio <= 0.9: premature_no_explorer_steps += 1 done_mask = np.asarray(dones, dtype=bool) if ( str(getattr(self.config, "qmix_transition_mode", "")) == "event_smdp" ): if not np.any(decision_active_masks > 0.5): next_rnn_states = previous_rnn_states next_rnn_states[done_mask] = 0.0 masks = np.ones((self.num_agents, 1), dtype=np.float32) masks[done_mask] = 0.0 rnn_states = next_rnn_states last_actions = actions if bool(np.all(done_mask)) or info.get("end", 0): break finally: np.random.set_state(numpy_state) if final_info is None: raise RuntimeError("Evaluation episode ended without environment info.") delivery_tasks = [ task for task in final_info["tasks"] if int(task["type"]) == 1 ] tasks_completed = sum(task["completed"] for task in delivery_tasks) tasks_discovered = sum(task["discovered"] for task in delivery_tasks) episode_length = step + 1 result = { "scenario_id": int(scenario_id), "scenario_seed": int(scenario_seed), "success": int(tasks_completed == len(delivery_tasks)), "episode_reward": float(episode_reward), "episode_length": episode_length, "tasks_completed": int(tasks_completed), "task_completion_ratio": float( tasks_completed / max(1, len(delivery_tasks)) ), "sent_percentage": float(sent / max(1.0, final_info["sent_goal"])), "discovered_ratio": float(tasks_discovered / max(1, len(delivery_tasks))), "dead_agents": int(sum(not agent["living"] for agent in final_info["agents"])), "coalition_changes": int(coalition_changes), "reconfiguration_decisions": int( final_info.get("reconfiguration", {}).get( "total_decisions", decision_count ) ), "coalition_changes_per_decision": float( coalition_changes / max( 1, final_info.get("reconfiguration", {}).get( "total_decisions", decision_count ), ) ), "reconfiguration_decisions_per_100_alive_steps": float( 100.0 * final_info.get("reconfiguration", {}).get( "total_decisions", decision_count ) / max( 1, final_info.get("reconfiguration", {}).get( "total_alive_agent_steps", alive_agent_steps ), ) ), "coalition_changes_per_100_alive_steps": float( 100.0 * coalition_changes / max( 1, final_info.get("reconfiguration", {}).get( "total_alive_agent_steps", alive_agent_steps ), ) ), "mean_response_time": float( final_info.get("reconfiguration", {}).get( "mean_response_time", 0.0 ) ), "premature_no_explorer_ratio": float( premature_no_explorer_steps / max(1, episode_length) ), "harmful_idle_agent_step_ratio": float( harmful_idle_agent_steps / max(1, alive_agent_steps) ), "useful_idle_agent_step_ratio": float( useful_idle_agent_steps / max(1, alive_agent_steps) ), "max_idle_duration": float(max_idle_duration), "committed_payload_shortage_ratio": float( committed_payload_shortage_ratio_sum / max(1, episode_length) ), "committed_payload_surplus_ratio": float( committed_payload_surplus_ratio_sum / max(1, episode_length) ), "mean_explore_leg_distance": float( explore_planned_distance / max(1, explore_legs_planned) ), "short_explore_leg_ratio": float( short_explore_legs / max(1, explore_legs_planned) ), "explore_distance_per_alive_step": float( explore_distance_moved / max(1, alive_agent_steps) ), "unavailable_sector_ratio": float( unavailable_sector_ratio_sum / max(1, episode_length) ), "explored_area_ratio": float( final_info.get("exploration_diagnostics", {}).get( "explored_area_ratio", 0.0 ) ), } if getattr(env.env, "explore_action_mode", None) == "sector": agent_entropies = [] agent_dominant_ratios = [] agent_coverage_ratios = [] for counts in sector_decision_counts: total = int(np.sum(counts)) if total <= 0: continue probabilities = counts[counts > 0].astype(np.float64) / total entropy = -float(np.sum(probabilities * np.log(probabilities))) if num_explore_actions > 1: entropy /= float(np.log(num_explore_actions)) agent_entropies.append(entropy) agent_dominant_ratios.append(float(np.max(counts) / total)) agent_coverage_ratios.append( float(np.count_nonzero(counts) / num_explore_actions) ) result["mean_agent_sector_entropy"] = float( np.mean(agent_entropies) if agent_entropies else 0.0 ) result["mean_agent_dominant_sector_ratio"] = float( np.mean(agent_dominant_ratios) if agent_dominant_ratios else 0.0 ) result["mean_agent_sector_coverage_ratio"] = float( np.mean(agent_coverage_ratios) if agent_coverage_ratios else 0.0 ) task_type_names = ("explore", "deliver", "idle", "resupply") for task_type, task_type_name in enumerate(task_type_names): result[f"decision_{task_type_name}_ratio"] = float( decision_type_counts[task_type] / max(1, decision_count) ) result[f"executed_{task_type_name}_ratio"] = float( executed_type_counts[task_type] / max(1, executed_count) ) result["decision_count"] = int(decision_count) result["gate_decision_count"] = int(gate_decision_count) result["gate_open_count"] = int(gate_open_count) result["gate_open_rate"] = float( gate_open_count / max(1, gate_decision_count) ) result["gate_mandatory_open_rate"] = float( gate_mandatory_open_count / max(1, gate_mandatory_count) ) result["gate_event_trigger_open_rate"] = float( gate_event_trigger_open_count / max(1, gate_event_trigger_count) ) result["gate_off_trigger_open_rate"] = float( gate_learned_open_count / max(1, gate_learned_count) ) result["gate_learned_open_rate"] = float( gate_learned_open_count / max(1, gate_learned_count) ) result["coalition_changes_per_gate_open"] = float( coalition_changes / max(1, gate_open_count) ) result.update( {f"reward_{key}": float(value) for key, value in reward_components.items()} ) return result @staticmethod def _summarize_eval(episodes): numeric_keys = [ key for key, value in episodes[0].items() if key not in {"scenario_id", "scenario_seed"} and isinstance(value, (int, float, np.integer, np.floating)) ] summary = { key: float(np.mean([episode[key] for episode in episodes])) for key in numeric_keys } summary["episode_reward_std"] = float( np.std([episode["episode_reward"] for episode in episodes]) ) summary["episode_length_std"] = float( np.std([episode["episode_length"] for episode in episodes]) ) successful_lengths = [ episode["episode_length"] for episode in episodes if episode["success"] ] summary["successful_episode_length"] = ( float(np.mean(successful_lengths)) if successful_lengths else None ) summary["successful_episode_length_median"] = ( float(np.median(successful_lengths)) if successful_lengths else None ) summary["num_episodes"] = len(episodes) return summary def _save_eval_results(self, episodes, summary, total_num_steps): eval_dir = os.path.join(self.log_dir, "eval") os.makedirs(eval_dir, exist_ok=True) stem = f"step_{int(total_num_steps):012d}" csv_path = os.path.join(eval_dir, stem + ".csv") json_path = os.path.join(eval_dir, stem + ".json") with open(csv_path, "w", newline="", encoding="utf-8") as file: writer = csv.DictWriter(file, fieldnames=list(episodes[0].keys())) writer.writeheader() writer.writerows(episodes) with open(json_path, "w", encoding="utf-8") as file: json.dump( {"summary": summary, "episodes": episodes}, file, ensure_ascii=False, indent=2, ) @staticmethod def _eval_score(summary): """Lexicographic score for selecting the best evaluation checkpoint.""" successful_length = summary.get("successful_episode_length") if successful_length is None: successful_length_score = float("-inf") else: successful_length_score = -float(successful_length) return ( float(summary.get("success", 0.0)), float(summary.get("task_completion_ratio", 0.0)), float(summary.get("episode_reward", float("-inf"))), successful_length_score, ) def _save_eval_checkpoints(self, summary, total_num_steps): self.save(tag="eval_latest") score = self._eval_score(summary) is_best = self.best_eval_score is None or score > self.best_eval_score json_score = [value if math.isfinite(value) else None for value in score] metadata = { "total_num_steps": int(total_num_steps), "env_episode": int(self.env_episode), "score": json_score, "summary": summary, } latest_path = os.path.join(self.save_dir, "eval_latest.json") with open(latest_path, "w", encoding="utf-8") as file: json.dump(metadata, file, ensure_ascii=False, indent=2) if is_best: self.best_eval_score = score self.best_eval_summary = dict(summary) self.save(tag="eval_best") best_path = os.path.join(self.save_dir, "eval_best.json") with open(best_path, "w", encoding="utf-8") as file: json.dump(metadata, file, ensure_ascii=False, indent=2) @torch.no_grad() def eval(self, total_num_steps): if self.eval_envs is None: raise RuntimeError("use_eval=True requires eval_envs.") episodes = [] self._publish_eval_progress(0) envs = self.eval_envs.envs for scenario_id in range(self.eval_episodes): env = envs[scenario_id % len(envs)] episodes.append(self._run_eval_episode(env, scenario_id)) completed = scenario_id + 1 self._publish_eval_progress( completed, state="completed" if completed == self.eval_episodes else "running", ) print( f"eval progress: episodes={completed}/{self.eval_episodes}", flush=True, ) summary = self._summarize_eval(episodes) self._save_eval_results(episodes, summary, total_num_steps) self._log_eval_scalars(summary, total_num_steps) self._print_eval_summary(summary) return summary @torch.no_grad() def render(self): """Visualize the env.""" envs = self.envs all_frames = [] for episode in range(self.config.render_episodes): obs = envs.reset(episode) if self.config.save_gifs: # Maybe to fix image = None #envs.render("rgb_array")[0][0] all_frames.append(image) else: envs.render("human") rnn_states = np.zeros( ( self.n_rollout_threads, self.num_agents, self.recurrent_N, self.hidden_size, ), dtype=np.float32, ) masks = np.ones((self.n_rollout_threads, self.num_agents, 1), dtype=np.float32) episode_rewards = [] for step in range(self.episode_length): calc_start = datetime.now() self.trainer.prep_rollout() action, rnn_states = self.trainer.policy.act( _concat_obs_dict(obs), np.concatenate(rnn_states), np.concatenate(masks), deterministic=getattr(self.config, "eval_deterministic", True), ) actions = np.array(np.split(_t2n(action), self.n_rollout_threads)) rnn_states = np.array(np.split(_t2n(rnn_states), self.n_rollout_threads)) # Obser reward and next obs obs, rewards, dones, infos, active_masks, bad_masks, available_agents = envs.step(actions) episode_rewards.append(rewards) rnn_states[dones == True] = np.zeros( ((dones == True).sum(), self.recurrent_N, self.hidden_size), dtype=np.float32, ) masks = np.ones((self.n_rollout_threads, self.num_agents, 1), dtype=np.float32) masks[dones == True] = np.zeros(((dones == True).sum(), 1), dtype=np.float32) if self.config.save_gifs: # Maybe to fix image = None #envs.render("rgb_array")[0][0] all_frames.append(image) calc_end = datetime.now() elapsed = calc_end - calc_start if elapsed < self.config.ifi: time.sleep(self.config.ifi - elapsed) else: envs.render("human") print("average episode rewards is: " + str(np.mean(np.sum(np.array(episode_rewards), axis=0)))) # if self.all_args.save_gifs: # imageio.mimsave(str(self.gif_dir) + '/render.gif', all_frames, duration=self.all_args.ifi)