【强化学习系列】Gym库使用——创建自己的强化学习环境2：拆解官方标准模型源码/规范自定义类+打包自定义环境

       

目录

一、 官方标准环境的获取与理解

二、根据官方环境源码修改自定义

        1.初始化__init__()

        2.重置环境 reset()

三、打包环境

        1.注册与创建自定义环境

        2.环境规范化

---

         在本文的早些时候，曾尝试按照自己的想法搭建自定义的基于gym强化学习环境。

        【强化学习系列】Gym库使用——创建自己的强化学习环境1：单一环境创建测试

        但由于学艺不精，在矢量化环境时遇到诸多障碍。仔细检查发现自己编写的环境存在很多不符合gym官方规范的地方，因此，索性从零重新研读官方介绍文档，从头开始解读官方标准环境源码，以帮助规范化自己搭建环境的类参数设置。

        官方文档地址：https://gymnasium.farama.org/

一、 官方标准环境的获取与理解

        通过打印 gym.envs.registry.keys() 获得的字典，就可以获得官方默认的所有可以直接调用的环境。

import gymnasium as gym# 官方支持的所有游戏环境
print(gym.envs.registry.keys())

        通过官方给出的运行和可视化代码即可，先通过“人为”（‘human’）来玩游戏。

import gymnasium as gym# 官方支持的所有游戏环境
print(gym.envs.registry.keys())# 可视化
env = gym.make("MountainCarContinuous-v0", render_mode="human")
observation, info = env.reset()for _ in range(1000):action = env.action_space.sample()  # agent policy that uses the observation and infoobservation, reward, terminated, truncated, info = env.step(action)if terminated or truncated:observation, info = env.reset()
env.close()

        下面展示一些官方游戏的可视化窗口的部分画面。只需根据最开始获得的游戏包装好的字典名，修改上述代码中环境实例化的make中第一个参数名即可，实际运行是动态的。

 

        首先从官方给出的Github地址下载好Gymnaium的源码——Gymnaium官方Github地址

        在官方文档目录的 ENVIRONMENTS 中介绍相关环境的一些基本关键信息，包括其状态动作空间、交互逻辑、奖励逻辑等。

        其中所有环境的源码保存在Github文件的 gymnaium/envs/ 目录下。

        本文选择经典环境下的山地车（Mountain Car）为源码拆解的例子。首先看起官方文档的环境介绍，其动作空间是一个三种可能的离散空间，分别对应小车的向左向右加速或不加速。

        其状态空间是一个连续的二维空间，分别是小车在x轴的位置，取值范围是（-1.2,0.6）；小车速度，取值范围是（-0.07,0.07）。

        目标是让小车登顶，初始化是将小车随机放在山谷的一个位置，初始速度为0，如果小车登顶或超过200步操作，则退出游戏。

二、根据官方环境源码修改自定义

        在理解游戏逻辑的基础上，进入其源码处查看并学习其代码类编写的规范性。

        1.初始化__init__()

        首先是环境的初始化。其在传入参数上设计两个内容：一是选择环境可视化，二是设置汽车的初始速度为0。

        还可以看到其可视化模式的选择是定义了一个metadata的字典，其定义状态空间和动作空间的逻辑和我们之前无异。

        值得注意的规范细节是，对于初始化是numpy数组向量时，都需标准其定义的数据类型dtype

        总结官方环境源码初始化的内容，可以得出初始化的个关键内容——状态环境的取值范围（如所处空间的大小——游戏里的地图，到边缘就不能探索了）、实体状态属性值（如小车的速度、或者游戏里角色的血量蓝条等等）、可视化窗口的一些变量。       

        现在回到自定义环境的逻辑（参考文章——单一自定义环境测试），新的环境中状态空间是图片上的目标检测框，是一个四维的（左上右下四坐标列表）连续空间。动作空间是一个四种可能的离散空间——上扩或缩、下扩或缩。其框坐标的取值范围受到图片高宽的影响。因此初始化必须传入原图地址信息，以获取高宽范围。

        参考总结的内容，可以梳理清晰自定义环境的一些设置，状态空间上肯定超出图片高宽或者小于0的框无意义，因此很容易定义状态空间。更关键的是实体属性，在这里原始的框就是需要移动的实体，像小车会有速度一样，这里我们定义初始框包含几个实际物体对应框（虚拟的框，在状态空间环境中不存在）视为该框的属性值。同时一张图里也可能有多个框存在，如果不希望当前框和环境中其他存在的框存在高度重叠，也可以设置重叠度检查属性。

        2.重置环境 reset()

        查看官方的初始化环境函数，可以看到其对于函数传参有规范化的标准，其初始的状态控制放在options字典数据中传入，并且采取了强制关键字的模式（*后的参数需要以关键字对应方式传入）因此改进我们自定义的reset函数。

         再将真实框可视化进背景，即可运行reset得到可视化结果。

         以上可视化所有完整代码如下。

import pygame
import gym
from gym import spaces
import numpy as np
from typing import Optionalclass My_Env(gym.Env):metadata = {"render_modes": ["human", "rgb_array"], "render_fps": 1}def __init__(self, render_mode: Optional[str] = None, image_path: Optional[str] = None):super(My_Env, self).__init__()# 读取当前图像信息if image_path != None:self.img = pygame.image.load(image_path)self.width, self.height = self.img.get_width(), self.img.get_height()else:self.width, self.height = 100, 100print('env need background')self.real = None  # 真实框数据# 定义框属性值self.box_num = None  # 初始预测框数量self.overlap = None  # 包含真实框个数self.box_iou = None  # 重叠度检查# 创建动作和状态空间范围self.action_space = spaces.Discrete(4)self.observation_space = spaces.Box(low=np.array([0,0,0,0],dtype=np.float32), high=np.array([self.width, self.height,self.width, self.height],dtype=np.float32), shape=(4,), dtype=np.float32)# 定义模式：人类可视化or机器人训练assert render_mode is None or render_mode in self.metadata["render_modes"]self.render_mode = render_modeself.window = None  # 可视化窗口self.clock = None   # 可视化时钟self.window_size = (600,600)  # 窗口大小self.background = None  # 背景图self.scale_x = None     # x横轴缩放比self.scale_y = None     # y竖轴缩放比def reset(self,*,seed: Optional[int] = None,options: Optional[dict] = None) :real = options['real_box']   # 真实框数据box = options['box']   # 初始框数据self.real = np.array(real, dtype=np.float32).reshape(-1, 4)self.state = np.array(box,dtype=np.float32).reshape(-1,4)   # 规范数据格式和形状if self.render_mode == 'human':self.background = pygame.transform.scale(self.img, self.window_size) # 设置背景图# 计算x和y方向的缩放比例self.scale_x = self.window_size[0]/ self.widthself.scale_y = self.window_size[1] / self.heightself.render()return self.statedef step(self, action):  # 动作序列{0：上扩， 1：上缩， 2：下扩， 3：下缩}# 根据action生成移动数组movement_np = np.zeros_like(self.state,dtype=np.float32)for i,act in enumerate(action):if act == 0:movement_np[i,1] = 10elif act == 1:movement_np[i,1] = -10elif act == 2:movement_np[i,3] = 10elif act == 3:movement_np[i,3] = -10# 移动当前状态框self.state += movement_npself.state = np.clip(self.state, self.observation_space.low, self.observation_space.high)return self.statedef render(self):# 初始化窗口和时钟if self.window is None and self.render_mode == 'human':pygame.init()pygame.display.init()self.window = pygame.display.set_mode(self.window_size)if self.clock is None and self.render_mode == 'human':self.clock = pygame.time.Clock()# 重新绘制背景，以清除上一层if self.background is not None:self.window.blit(self.background,(0,0))for real in list(self.real):rect_real = [real[0], real[1], abs(real[2]-real[0]), abs(real[3]-real[1])]# 根据x和y方向的缩放比例计算每个矩形框的新位置stretched_rectangle = [rect_real[0] * self.scale_x,  # x 坐标rect_real[1] * self.scale_y,  # y 坐标rect_real[2] * self.scale_x,  # 宽度rect_real[3] * self.scale_y  # 高度]pygame.draw.rect(self.window, (255, 0, 0), stretched_rectangle, 2)  # 绘制矩形框，线宽为2# 绘制框for box in list(self.state):rect = [box[0], box[1], abs(box[2]-box[0]), abs(box[3]-box[1])]# 根据x和y方向的缩放比例计算每个矩形框的新位置stretched_rectangle = [rect[0] * self.scale_x,  # x 坐标rect[1] * self.scale_y,  # y 坐标rect[2] * self.scale_x,  # 宽度rect[3] * self.scale_y  # 高度]pygame.draw.rect(self.window, (0, 255, 0), stretched_rectangle, 3)  # 绘制矩形框，线宽为3# 更新显示pygame.display.flip()self.clock.tick(self.metadata["render_fps"])def close(self):pygame.quit()if __name__=='__main__':# 加载本地图片image_path = './jpg/000000000025.jpg'  # 替换为你的图片路径# 加载本地框box_path = './box/000000000025.json'with open(box_path, 'r') as f:box_dict_list = json.load(f)box_list = []for box_dict in box_dict_list:box = box_dict['box']box_list.append(box)box2 = [(0,3),(1,0),(1,1),(1,2),(1,3)]box1 = [(0,0),(0,1),(1,0),(1,1),(2,1),(3,1),(4,1),(4,2),(4,3),(5,1),(5,2),(5,3),(5,4),(6,1),(6,2),(6,3),(6,4),(7,1),(7,2),(7,3),(7,4),(7,5),(8,1),(8,2),(8,3),(8,4),(8,5)]real_list = []for i in range(len(box_list)):bbox = box_list[i]wid = bbox[2]-bbox[0]hgt = bbox[3]-bbox[1]gap = 30for w in range(wid//gap):for h in range(hgt//gap):if i ==0:if (h,w) in box1:real = [bbox[0]+w*gap, bbox[1]+h*gap, bbox[0]+(w+1)*gap, bbox[1]+(h+1)*gap]real_list.append(real)elif i ==1:if (h,w) in box2:real = [bbox[0] + w * gap, bbox[1] + h * gap, bbox[0] + (w + 1) * gap, bbox[1] + (h + 1) * gap]real_list.append(real)env = My_Env(render_mode='human', image_path=image_path)state = env.reset(options={'real_box':real_list, 'box':box_list})while True:env.render()

三、打包环境

        1.注册与创建自定义环境

        为了更方便的加载环境以及为后续矢量化做准备，需要将自定义的环境打包并使用官方推荐的gymnasium.make来加载环境，这样的加载方式会帮助检查环境的合规性。

        首先新建项目文件，将所有环境代码放入一个子文件中方便修改，然后再新建一个注册环境的子文件用于测试环境能否被成功注册。

        在测试环境下（有图片和框数据的文件夹），先注册环境在创建环境。

import gymnasium as gym
import json
from gymnasium.envs.registration import register# 注册环境
register(id='detect_env-v0',entry_point='my_gym_env.detect_env:Detect_Env',
)# 加载本地图片
image_path = './jpg/000000000025.jpg'  # 替换为你的图片路径
# 加载本地框
box_path = './box/000000000025.json'
real_path = './real/000000000025.json'
with open(box_path, 'r') as f:box_dict_list = json.load(f)
box_list = []
for box_dict in box_dict_list:box = box_dict['box']box_list.append(box)
with open(real_path, 'r') as f:real_list = json.load(f)env = gym.make('detect_env-v0', render_mode='human', image_path=image_path)state = env.reset(options={'real_box':real_list, 'box':box_list})
while True:env.render()

        2.环境规范化

        运行上面的测试代码，虽然能加载出环境演示，但是make会检测环境创建源码，并对不规范报错，这方便我们进行修改规范源码。

        ① reset 报错与规范

         这里报错的原因是，在定义环境重置 reset 时，没有返回 info 信息。补一个空字典即可。

UserWarning: WARN: The result returned by `env.reset()` was not a tuple of the form `(obs, info)`, where `obs` is a observation and `info` is a dictionary containing additional information. Actual type: `<class 'numpy.ndarray'>`

        ② observation_space 状态空间报错与规范

        改完后报新的错。这是说在重置环境时传入的状态空间与__init__ 中定义的状态空间形状不匹配。

UserWarning: WARN: The obs returned by the `reset()` method is not within the observation space.logger.warn(f"{pre} is not within the observation space.")

        检查环境中定义的空间必须是固定形状的，（，4）的定义会默认为（1,4），其不能自由匹配维度，但是作为图片中的目标框，其第一个维度数量是不固定的。因此一种解决方式是选择所有图片内的最大框数作为初始化状态空间第一维度，不够的地方用零填充。

# 设置最大框数
self.box_max_num = 20
# 创建动作和状态空间范围
self.action_space = spaces.Discrete(4)
self.observation_space = spaces.Box(low=np.zeros((self.box_max_num,4),dtype=np.float32),high=np.tile([self.width, self.height,self.width, self.height],(self.box_max_num,1)),shape=(self.box_max_num,4), dtype=np.float32)

# 初始传入框信息
state = np.array(box,dtype=np.float32).reshape(-1,4)   # 规范数据格式和形状# 填充零使得状态达到指定维度
self.state = np.pad(state,((0,self.box_max_num-state.shape[0]),(0,0)), mode='constant', constant_values=0)

        规范后不再报错，此时打印环境的状态检查。

        ③ step 报错与规范

        继续在加入动作空间的输入，解决step中的规范性问题。首先如果要输入动作，需要知道当前图片中实际需要操作的框的数量（不能根据状态空间判断，因为状态中框数（第一行维度）被强制设为最大框数）， 因此在重置环境时就要返回当前图片的框数，作为每个框生成多少个动作的标准，在官方的规范代码中，可将这些额外信息写入info字典中返回。

info = {'box_num':state.shape[0]}
return self.state, info

        在测试代码中添加采样动作和step操作，并打印状态的变化numpy数组。注意此处环境还没有设计奖励函数，因此定义环境操作reward都是0。

env = gym.make('detect_env-v0', render_mode='human', image_path=image_path)state,info = env.reset(options={'real_box':real_list, 'box':box_list})
print(0, state)
for epid in range(50):action = [env.action_space.sample() for _ in range(info['box_num'])]state, reward, _, _, _ = env.step(action)print(epid+1, state)env.render()

         下图中，红色箭头表示右较大框的状态变化，黄色箭头代表左较小框的状态变化。此处为方便可视化，将最大框数量设为3。以20像素为步长变化检测框。

         加上具体动作打印结果可视化。

        至此，环境规范化告一段落，完整代码放在下面方便复制取走。终于可以开始后续矢量化环境操作了。

        自定义检测环境完整代码。

import pygame
import gymnasium as gym
from gymnasium import spaces
import numpy as np
from typing import Optionalclass Detect_Env(gym.Env):metadata = {"render_modes": ["human", "rgb_array"], "render_fps": 1}def __init__(self, render_mode: Optional[str] = None, image_path: Optional[str] = None):super(Detect_Env, self).__init__()# 读取当前图像信息if image_path != None:self.img = pygame.image.load(image_path)self.width, self.height = self.img.get_width(), self.img.get_height()else:self.width, self.height = 100, 100print('env need background')self.real = None  # 真实框数据# 定义框属性值self.box_num = None  # 初始预测框数量self.overlap = None  # 包含真实框个数self.box_iou = None  # 重叠度检查# 设置最大框数self.box_max_num = 3# 创建动作和状态空间范围self.action_space = spaces.Discrete(4)self.observation_space = spaces.Box(low=np.zeros((self.box_max_num,4),dtype=np.float32),high=np.tile([self.width, self.height,self.width, self.height],(self.box_max_num,1)),shape=(self.box_max_num,4), dtype=np.float32)# 定义模式：人类可视化or机器人训练assert render_mode is None or render_mode in self.metadata["render_modes"]self.render_mode = render_modeself.window = None  # 可视化窗口self.clock = None   # 可视化时钟self.window_size = (600,600)  # 窗口大小self.background = None  # 背景图self.scale_x = None     # x横轴缩放比self.scale_y = None     # y竖轴缩放比def reset(self,*,seed: Optional[int] = None,options: Optional[dict] = None) :real = options['real_box']   # 真实框数据box = options['box']   # 初始框数据self.real = np.array(real, dtype=np.float32).reshape(-1, 4)state = np.array(box,dtype=np.float32).reshape(-1,4)   # 规范数据格式和形状# 填充零使得状态达到指定维度self.state = np.pad(state,((0,self.box_max_num-state.shape[0]),(0,0)), mode='constant', constant_values=0)if self.render_mode == 'human':self.background = pygame.transform.scale(self.img, self.window_size) # 设置背景图# 计算x和y方向的缩放比例self.scale_x = self.window_size[0]/ self.widthself.scale_y = self.window_size[1] / self.heightself.render()info = {'box_num':state.shape[0]}return self.state, infodef step(self, action):  # 动作序列{0：上扩， 1：上缩， 2：下扩， 3：下缩}# 根据action生成移动数组movement_np = np.zeros_like(self.state,dtype=np.float32)for i,act in enumerate(action):if act == 0:movement_np[i,1] = 20elif act == 1:movement_np[i,1] = -20elif act == 2:movement_np[i,3] = 20elif act == 3:movement_np[i,3] = -20# 移动当前状态框self.state += movement_npself.state = np.clip(self.state, self.observation_space.low, self.observation_space.high)return np.array(self.state, dtype=np.float32),0,False,False,{}def render(self):# 初始化窗口和时钟if self.window is None and self.render_mode == 'human':pygame.init()pygame.display.init()self.window = pygame.display.set_mode(self.window_size)if self.clock is None and self.render_mode == 'human':self.clock = pygame.time.Clock()# 重新绘制背景，以清除上一层if self.background is not None:self.window.blit(self.background,(0,0))for real in list(self.real):rect_real = [real[0], real[1], abs(real[2]-real[0]), abs(real[3]-real[1])]# 根据x和y方向的缩放比例计算每个矩形框的新位置stretched_rectangle = [rect_real[0] * self.scale_x,  # x 坐标rect_real[1] * self.scale_y,  # y 坐标rect_real[2] * self.scale_x,  # 宽度rect_real[3] * self.scale_y  # 高度]pygame.draw.rect(self.window, (255, 0, 0), stretched_rectangle, 2)  # 绘制矩形框，线宽为2# 绘制框for box in list(self.state):rect = [box[0], box[1], abs(box[2]-box[0]), abs(box[3]-box[1])]# 根据x和y方向的缩放比例计算每个矩形框的新位置stretched_rectangle = [rect[0] * self.scale_x,  # x 坐标rect[1] * self.scale_y,  # y 坐标rect[2] * self.scale_x,  # 宽度rect[3] * self.scale_y  # 高度]pygame.draw.rect(self.window, (0, 255, 0), stretched_rectangle, 3)  # 绘制矩形框，线宽为3# 更新显示pygame.display.flip()self.clock.tick(self.metadata["render_fps"])def close(self):pygame.quit()

        项目文件层级结构。

        测试可视化完整代码。

import gymnasium as gym
import json
from gymnasium.envs.registration import register# 注册环境
register(id='detect_env-v0',entry_point='my_gym_env.detect_env:Detect_Env',
)# 加载本地图片
image_path = './jpg/000000000025.jpg'  # 替换为你的图片路径
# 加载本地框
box_path = './box/000000000025.json'
real_path = './real/000000000025.json'
with open(box_path, 'r') as f:box_dict_list = json.load(f)
box_list = []
for box_dict in box_dict_list:box = box_dict['box']box_list.append(box)
with open(real_path, 'r') as f:real_list = json.load(f)env = gym.make('detect_env-v0', render_mode='human', image_path=image_path)state,info = env.reset(options={'real_box':real_list, 'box':box_list})
print(0, state)
for epid in range(50):action = [env.action_space.sample() for _ in range(info['box_num'])]print('action', action)state, reward, _, _, _ = env.step(action)print(epid+1, state)env.render()