# 配置环境变量，添加miniconda环境的路径到系统PATH中，以便可以使用该环境中的Python及其库 # export PATH="/Users/puyuan/miniconda3/envs/arm64-py38/bin:$PATH" # 设置Flask应用相关的环境变量并启动Flask应用 # FLASK_APP=app.py FLASK_ENV=development FLASK_DEBUG=1 flask run --port 5001 import time # 导入time模块用于时间操作 from flask import Flask, request, jsonify # 导入Flask用于创建Web应用 from flask_restx import Api, Resource, fields # 导入Flask-RESTx扩展用于创建REST API app = Flask(__name__) # 初始化Flask应用 api = Api( # 初始化REST API app=app, version="0.0.1", # API版本 title="gomoku_server_ui App", # API标题 description="Play Gomoku with LightZero Agent, Powered by OpenDILab" # API描述 ) @app.after_request # Flask装饰器，在每个请求之后运行 def after_request(response): # 设置CORS，允许所有源访问 response.headers.add('Access-Control-Allow-Origin', '*') # 允许跨源请求包含的头部字段 response.headers.add('Access-Control-Allow-Headers', 'Content-Type,Authorization') # 允许的HTTP方法 response.headers.add('Access-Control-Allow-Methods', 'GET,PUT,POST,DELETE,OPTIONS') return response # 返回修改后的响应 # 定义REST API的命名空间 name_space = api.namespace('gomoku_server_ui', description='gomoku_server_ui APIs') # 定义传入API的数据模型 model = api.model( 'gomoku_server_ui params', { 'command': fields.String(required=False, description="Command Field", help="可选参数为：reset, step"), 'argument': fields.Integer(required=False, description="Argument Field", help="如果输入 cmd 是 reset，则 argument 表示 agent type, 如果输入 cmd 是 step，则 argument 表示 [action_x, action_y, agent_type]"), } ) MAX_ENV_NUM = 50 # 最大环境数限制 ENV_TIMEOUT_SECOND = 6000 # 环境超时时间（秒） import sys from easydict import EasyDict # 导入EasyDict，用于创建类似字典的对象，但可以像访问属性一样访问其元素 from zoo.board_games.gomoku.envs.gomoku_env import GomokuEnv # 导入Gomoku环境 from agent import Agent # 导入Agent类 # 配置Gomoku环境的参数 cfg = EasyDict( prob_random_agent=0, board_size=15, battle_mode='self_play_mode', channel_last=False, scale=False, agent_vs_human=False, bot_action_type='v1', prob_random_action_in_bot=0., check_action_to_connect4_in_agent_v0=False, render_mode='state_realtime_mode', replay_path=None, screen_scaling=9, alphazero_mcts_ctree=False, ) env = GomokuEnv(cfg) # 创建一个Gomoku环境实例 obs = env.reset() # 重置环境并获取初始观察 agent = Agent() # 创建一个Agent实例 envs = {} # 初始化环境字典 envs['127.0.0.1:1'] = {'env': env, 'update_time': time.time()} # 假设这是类外面的全局变量 # envs = {} from threading import Thread, Lock envs_lock = Lock() # 监控游戏环境的函数，用于清理超时的游戏环境 def env_monitor(): while True: # 无限循环 with envs_lock: # 确保在修改envs字典时不会有其他线程干扰 cur_time = time.time() # 获取当前时间 pop_keys = [] # 准备一个列表来记录超时的环境键 for k, v in envs.items(): # 遍历所有游戏环境 if cur_time - v['update_time'] >= ENV_TIMEOUT_SECOND: # 如果当前时间与环境的最后更新时间差大于超时时间 pop_keys.append(k) # 将该环境的键加入到pop_keys列表中 for k in pop_keys: # 遍历需要清理的环境键 envs.pop(k) # 从envs字典中移除该环境 time.sleep(1) # 休眠1秒，减少CPU占用 # 创建一个守护线程运行env_monitor函数 api.env_thread = Thread(target=env_monitor, daemon=True) api.env_thread.start() # 启动线程 # 定义API的路由，即当POST请求发送到"/"时，执行MainClass内的方法 @name_space.route("/", methods=['POST']) class MainClass(Resource): # 定义一个资源类 @api.expect(model) # 指定预期的输入模型 def post(self): # 定义处理POST请求的方法 try: t_start = time.time() # 记录开始处理请求的时间 data = request.json # 获取请求的JSON数据 cmd, arg, uid = data['command'], data['argument'], data['uid'] # 从数据中提取命令、参数和用户ID print("请求来源的IP地址:", request.remote_addr) # 打印请求来源的IP地址 ip = request.remote_addr + uid # 将IP地址和用户ID组合作为唯一标识 print("命令、参数、用户ID: ", cmd, arg, uid) # 打印命令、参数、用户ID print('envs:', envs) # 打印当前所有的游戏环境 # 如果组合的IP不在envs字典中，即用户的游戏环境不存在 if ip not in envs: print('ip not in envs') if cmd == 'reset': # 如果命令是重置 if len(envs) >= MAX_ENV_NUM: # 如果当前环境数量已达到最大限制 # 返回一个错误响应，告知资源不足 response = jsonify( { "statusCode": 501, "status": "No enough env resource, please wait a moment", } ) return response else: # 创建一个新的游戏环境并记录当前时间 env = GomokuEnv(cfg) envs[ip] = {'env': env, 'update_time': time.time()} else: # 返回一个错误响应，告知用户长时间无响应，需要重置游戏 response = jsonify( { "statusCode": 501, "status": "No response for too long time, please reset the game", } ) return response else: # 如果环境已存在，更新环境的最后活跃时间 env = envs[ip]['env'] envs[ip]['update_time'] = time.time() # 根据不同的命令，处理游戏逻辑 if cmd == 'reset': # cmd == 'reset' 表示前端玩家设置了 AI 先手，需要重置游戏环境, 并返回一个 Agent action observation = env.reset() # 重置游戏环境 agent_type = arg print('agent type: {}'.format(agent_type)) if agent_type == -2: # 'Random' agent_action = env.random_action() # 获取一个随机动作 elif agent_type == -1: # 'RuleBot': agent_action = env.bot_action() # 让规则玩家计算动作 elif agent_type == 0: # 'AlphaZero' agent_action = agent.compute_action(observation) # 让智能体计算动作 # 更新游戏环境 observation, reward, done, info = env.step(agent_action) # print('original agent action: {}'.format(agent_action)) agent_action = {'i': int(agent_action // 15), 'j': int(agent_action % 15)} print('agent action: {}'.format(agent_action)) # 返回一个响应，包含游戏板状态、智能体动作、游戏是否结束和其他信息 response = jsonify( { "statusCode": 200, "status": "Execution action", "result": { 'board': env.board.tolist(), 'action': agent_action, 'done': done, 'info': info, } } ) elif cmd == 'step': # 使用之前存储的键来检索环境 data = request.json action = data.get('argument')[:-1] # 前端发送的动作 action: [i, j] 从0开始的，表示下在第i+1行，第j+1列 print(f'前端发送过来的动作: {action}') agent_type = data.get('argument')[-1] print('agent type: {}'.format(agent_type)) action = action[0] * 15 + action[1] # 更新游戏环境 observation, reward, done, info = env.step(action) env.render() # 如果游戏没有结束，获取 agent 的动作 if not done: if agent_type == -2: # 'Random' agent_action = env.random_action() # 获取一个随机动作 elif agent_type == -1: # 'RuleBot': agent_action = env.bot_action() # 让规则玩家计算动作 elif agent_type == 0: # 'AlphaZero' agent_action = agent.compute_action(observation) # 让智能体计算动作 # 更新环境状态 _, _, done, _ = env.step(agent_action) # 准备响应数据 print('original agent action: {}'.format(agent_action)) agent_action = {'i': int(agent_action // 15), 'j': int(agent_action % 15)} print('agent action: {}'.format(agent_action)) else: # agent_action = None # agent 的动作置为 None 会在前端中报错 agent_action = {'i': -1, 'j': -1} # 如果游戏结束，agent 的动作置为特殊的值，表示游戏结束 observation = env.reset() # 重置游戏环境 # print(type(agent_action), type(done), type(info)) response = { "statusCode": 200, "status": "Execution action", "result": { 'board': None, 'action': agent_action, # agent action格式为 {'i': x, 'j': y } 'done': done, } } else: response = jsonify({ "statusCode": 500, "status": "Invalid command: {}".format(cmd), }) return response print('backend process time: {}'.format(time.time() - t_start)) print('current env number: {}'.format(len(envs))) return response except Exception as e: import traceback print(repr(e)) print(traceback.format_exc()) response = jsonify({ "statusCode": 500, "status": "Could not execute action", }) return response