""" # @Time : 2021/7/2 5:22 下午 # @Author : hezhiqiang # @Email : tinyzqh@163.com # @File : env_discrete.py """ import gym from gym import spaces import numpy as np from .env_core import EnvCore class DiscreteActionEnv(object): """ 对于离散动作环境的封装 Wrapper for discrete action environment. """ def __init__(self, config): self.env = EnvCore(config) self.num_agent = self.env.agent_num self.signal_obs_dim = self.env.obs_dim self.signal_action_dim = self.env.action_dim self.cap_dim = self.env.cap_dim self.task_num = self.env.task_num self.per_task_feat = self.env.per_task_feat self.ego_dim = self.env.ego_feat_dim self.other_agent_feat_dim = self.env.other_agent_feat_dim self.global_agent_feat_dim = self.env.global_agent_feat_dim self.global_task_feat_dim = self.env.global_task_feat_dim self.global_feat_dim = self.env.global_feat_dim # if true, action is a number 0...N, otherwise action is a one-hot N-dimensional vector # Edited self.discrete_action_input = True self.movable = True # configure spaces self.action_space = [] self.observation_space = [] self.share_observation_space = [] total_action_space = [] for agent_idx in range(self.num_agent): # physical action space u_action_space = spaces.Discrete(self.signal_action_dim) # signal_action_dim 个离散的动作 total_action_space.append(u_action_space) self.action_space.append(total_action_space[agent_idx]) # observation space: policy keys plus structured graph features. observation_spaces = { "ego": spaces.Box( low=-np.inf, high=+np.inf, shape=(self.ego_dim,), dtype=np.float32, ), "tasks": spaces.Box( low=-np.inf, high=+np.inf, shape=(self.task_num, self.per_task_feat), dtype=np.float32, ), "others": spaces.Box( low=-np.inf, high=+np.inf, shape=(self.num_agent, self.other_agent_feat_dim), dtype=np.float32, ), } if self.env.reconfiguration_context_features: observation_spaces["reconfiguration"] = spaces.Box( low=-np.inf, high=+np.inf, shape=(self.env.reconfiguration_context_dim,), dtype=np.float32, ) self.observation_space.append(spaces.Dict(observation_spaces)) state_space = spaces.Dict({ "agents": spaces.Box( low=-np.inf, high=+np.inf, shape=(self.num_agent, self.global_agent_feat_dim), dtype=np.float32, ), "tasks": spaces.Box( low=-np.inf, high=+np.inf, shape=(self.task_num, self.global_task_feat_dim), dtype=np.float32, ), "global": spaces.Box( low=-np.inf, high=+np.inf, shape=(self.global_feat_dim,), dtype=np.float32, ), "focal": spaces.Box( low=0.0, high=1.0, shape=(self.num_agent, 1), dtype=np.float32, ), }) self.share_observation_space = [state_space for _ in range(self.num_agent)] def step(self, actions): """ 输入actions维度假设： # actions shape = (5, 2, 5) # 5个线程的环境，里面有2个智能体，每个智能体的动作是一个one_hot的5维编码 Input actions dimension assumption: # actions shape = (5, 2, 5) # 5 threads of the environment, with 2 intelligent agents inside, and each intelligent agent's action is a 5-dimensional one_hot encoding """ obs, reward, dones, info, active_mask, bad_mask, available_actions = self.env.step(actions) obs = self._format_obs_dict(obs) return obs, reward, dones, info, active_mask, bad_mask, available_actions def reset(self, episode): obs, active_mask, available_actions = self.env.reset(episode) obs = self._format_obs_dict(obs) return obs, active_mask, available_actions def _format_obs_dict(self, obs): return {k: v.astype(np.float32) for k, v in obs.items()} def get_global_state(self): return self.env.get_global_state() def close(self): pass def render(self, mode="rgb_array"): pass def set_reset_mode(self, mode): self.env.set_reset_mode(mode) def seed(self, seed): return self.env.seed(seed) class MultiDiscrete: """ - The multi-discrete action space consists of a series of discrete action spaces with different parameters - It can be adapted to both a Discrete action space or a continuous (Box) action space - It is useful to represent game controllers or keyboards where each key can be represented as a discrete action space - It is parametrized by passing an array of arrays containing [min, max] for each discrete action space where the discrete action space can take any integers from `min` to `max` (both inclusive) Note: A value of 0 always need to represent the NOOP action. e.g. Nintendo Game Controller - Can be conceptualized as 3 discrete action spaces: 1) Arrow Keys: Discrete 5 - NOOP[0], UP[1], RIGHT[2], DOWN[3], LEFT[4] - params: min: 0, max: 4 2) Button A: Discrete 2 - NOOP[0], Pressed[1] - params: min: 0, max: 1 3) Button B: Discrete 2 - NOOP[0], Pressed[1] - params: min: 0, max: 1 - Can be initialized as MultiDiscrete([ [0,4], [0,1], [0,1] ]) """ def __init__(self, array_of_param_array): super().__init__() self.low = np.array([x[0] for x in array_of_param_array]) self.high = np.array([x[1] for x in array_of_param_array]) self.num_discrete_space = self.low.shape[0] self.n = np.sum(self.high) + 2 def sample(self): """Returns a array with one sample from each discrete action space""" # For each row: round(random .* (max - min) + min, 0) random_array = np.random.rand(self.num_discrete_space) return [int(x) for x in np.floor(np.multiply((self.high - self.low + 1.0), random_array) + self.low)] def contains(self, x): return ( len(x) == self.num_discrete_space and (np.array(x) >= self.low).all() and (np.array(x) <= self.high).all() ) @property def shape(self): return self.num_discrete_space def __repr__(self): return "MultiDiscrete" + str(self.num_discrete_space) def __eq__(self, other): return np.array_equal(self.low, other.low) and np.array_equal(self.high, other.high) if __name__ == "__main__": DiscreteActionEnv().step(actions=None)