AI的二次开发

AI的二次开发中，有依赖AgentAPIMgr和ActionAPIMgr，AgentAPIMgr是Agent和GameReg进行交互的接口模块。ActionAPIMgr是和手机交互的动作接口模块。在本章我们先对这两部分进行介绍，再介绍AI插件。

1 AgentAPIMgr 介绍

1.1 整体流程

if __name__ == '__main__':
	# step1: AgentAPIMgr为对外接口类，用于和识别模块交互。
	agentAPI = AgentAPIMgr.AgentAPIMgr()

	#step2:初始化：初始化tbus,加载识别任务配置文件如task.json，配置文件的介绍详见三章节
    ret = agentAPI.Initialize("../cfg/task.json")
    if not ret:
        print('Agent API Init Failed')
        return

	#step3:发送识别命令到识别模块，参数介绍请详见二章节
    ret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_GROUP_ID, 1)
    if not ret:
        print('send message failed')
        
    while True:
		#step4:获取识别结果，参数介绍请详见第二章节
        msgDict = agentAPI.GetInfo(AgentAPIMgr.GAME_RESULT_INFO)

	#step5:资源释放，tbus资源释放
	agentAPI.Release()

1.2 agentAPI的介绍

AgentAPI.Initialize(confFile，referFile)

初始化操作，输入参数confFile为 json格式的任务配置文件。referFile为json格式的参考任务配置文件。此配置文件的详细说明请见第3章的场景识别配置中的任务配置文件说明。初始化的返回结果为True,表明初始化成功;False初始化失败，可根据屏幕打印的错误信息，定位原因。

AgentAPI.SendCmd(cmdID, cmdValue)

发送命令给识别模块，返回值为True,表示发送成功，False表示发送失败。识别模块会根据发送的命令，改变识别器。输入参数详情见下表：

cmdID	cmdValue	描述
MSG_SEND_GROUP_ID	groupID（python数字类型）	发送任务groupID，识别模块接受此group下的任务列表信息，创建识别器（groupID的具体数值请参考对应json配置文件中的groupID字段），如发送groupID=1的任务：ret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_GROUP_ID, 1)
MSG_SEND_TASK_FLAG	由taskid和对应操作组成的字典。	key值为taskID，为配置文件中对应group下的”taskID”项， value为True/False表示打开/关闭taskID对应的识别器。在后续的识别结果中，会增加或是减少此task的识别结果。如,打开taskID为1的任务，关闭taskID为2的任务。taskDict ={}taskDict[1] = TruetaskDict[2] = Falseret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_TASK_FLAG, taskDict )
MSG_SEND_ADD_TASK	task参数列表	增加一系列识别任务，识别模块接收到此消息后，根据参数列表创建一系列识别器。在后续的识别结果中，会增加此任务的识别结果。如增加taskID为10的任务：taskList =[]taskDict = dict()taskDict["taskID"] = 10taskDict["type"] = "fixed_obj"taskDict["elements"] = []elementDict = dict()elementDict["x"] = 946elementDict["y"] = 15elementDict["w"] = 220elementDict["h"] = 60elementDict["path"] = "../data/DNF/dnf_over.bmp"taskDict["elements"].append(elementDict)taskList.append(taskDict) agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_ADD_TASK, taskList)
MSG_SEND_DEL_TASK	taskID 的列表	删除一系列识别任务，识别模块接收到此消息后，根据task的设置，删除相应的识别器。在后续的识别结果中，不会存在此task识别结果。如：删除taskID为10的任务：taskList =[]taskList.append(10)ret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_DEL_TASK, taskList)
MSG_SEND_CHG_TASK	task参数列表	改变一系列识别任务，识别模块接收到此消息后，根据新的任务参数，修改识别器。在后续的识别中，根据新识别器对图像做识别。如改变taskID为10的任务：taskList = []taskDict = dict()taskDict["taskID"] = 10taskDict["type"] = "fixed_obj"taskDict["elements"] = []elementDict = dict()elementDict["x"] = 1000elementDict["y"] = 0elementDict["w"] = 200elementDict["h"] = 200elementDict["path"] = "../data/DNF/dnf_playing.bmp"taskDict["elements"].append(elementDict)taskList.append(taskDict)ret = agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_CHG_TASK, taskList)

AgentAPI.GetInfo(type)

获取结果，结果的类型type有以下几种。

type	Return	描述
CUR_GROUP_TASK_INFO	当前组任务参数列表	返回值为 python中的列表类型(list)如：groupTaskInfo = agentAPI.GetInfo(AgentAPIMgr.CUR_GROUP_TASK_INFO)for task in groupTaskInfo:taskID = task.get('taskID')type = task.get('type')elements = task.get('elements') ……
CUR_GROUP_INFO	当前组的参数	返回值为python中的字典类型(dict)如:curGroupInfo = agentAPI.GetInfo(AgentAPIMgr.CUR_GROUP_INFO)groupID = curGroupInfo.get('groupID')groupName = curGroupInfo.get('name')groupTask = curGroupInfo.get('task')……
GAME_RESULT_INFO	识别结果	返回值为python中的字典类型(dict)如:msgDict = agentAPI.GetInfo(AgentAPIMgr.GAME_RESULT_INFO)resultDic = msgDict['result'] groupID = msgDict['groupID']frameSeq = msgDict['frameSeq']deviceIndex = msgDict['deviceIndex']img = msgDict['image'].....
ALL_GROUP_INFO	所有组的参数	返回值为python中列表(list)类型。如：allGroupInfo = agentAPI.GetInfo(AgentAPIMgr.ALL_GROUP_INFO)for groupInfo in allGroupInfo:groupID = groupInfo.get('groupID')groupName = groupInfo.get('name')groupTask = groupInfo.get('task') ……

当调用AgentAPI.GetInfo(AgentAPIMgr.GAME_RESULT_INFO)接口去获取当前帧游戏图像的处理结果时，会返回每个任务(task)的结果项。不同task类型的结果说明，请见文档 Task识别结果说明。

AgentAPI.Release()

在程序的最后，调用此函数，释放tbus资源。

2 ActionAPIMgr介绍

2.1 整体流程

if __name__ == '__main__':
	#step1: ActionAPIMgr为对外接口类。
	actionAPI = ActionAPIMgr.ActionAPIMgr()

	#step2:初始化：初始化tbus 
    ret = actionAPI.Initialize()
    if not ret:
        print('Action API Init Failed')
        return

    #step3:发送动作
    actionAPI.SendAction(actionID=12, actionData={})  # 发送自定义12号动作
    actionAPI.Click(px=300, py=400, contact=0)  # 发送点击动作，点击（300, 400）位置

    #step4:资源释放，tbus资源释放
    actionAPI.Finish()

2.2 ActionAPI的介绍

ActionAPI.Initialize() 初始化操作，在其他方法之前调用。
ActionAPI.Finish() 结束操作，在进程退出时调用释放资源
ActionAPI.Reset(frameSeq=-1) 释放所有contact，frameSeq为帧序号
ActionAPI.SendAction(actionID, actionData , frameSeq=-1) 发送自定义动作，actionID为动作ID，actionData为对应的动作参数（字典）
ActionAPI.MovingInit(centerX, centerY, radius, contact=0, frameSeq=-1, waitTime=1000) 初始化移动摇杆中心点（centerX, centerY）和半径radius，触点ID为contact，之后等待时间waitTime毫秒
ActionAPI.MovingFinish() 释放移动摇杆初始化时的触点
ActionAPI.Moving(dirAngle，frameSeq=-1，durationMS=50) 控制移动摇杆向dirAngle方向滑动，滑动过程耗时durationMS毫秒
ActionAPI.Move(px, py, contact=0, frameSeq=-1, waitTime=0) 移动触点contact到（px, py）位置，瞬间移动不补充过渡点，之后等待时间waitTime毫秒
ActionAPI.Click(px, py, contact=0, frameSeq=-1, durationMS=-1) 点击触点contact在（px, py）位置，过程耗时durationMS毫秒
ActionAPI.Down(px, py, contact=0, frameSeq=-1, waitTime=0) 按压触点contact在（px, py）位置，之后等待时间waitTime毫秒
ActionAPI.Up(contact, frameSeq=-1, waitTime=0) 释放触点contact，之后等待时间waitTime毫秒
ActionAPI.Swipe(sx, sy, ex, ey, contact=0, frameSeq=-1, durationMS=50, needUp=True) 滑动触点contact从（sx, sy）到（ex, ey），中间会补充过渡点，过程耗时durationMS毫秒，根据needUp决定最后是否释放触点
ActionAPI.SwipeMove(px, py, contact=0, frameSeq=-1, durationMS=50) 移动触点contact到（px, py）位置，中间会补充过渡点，过程耗时durationMS毫秒

3. AI插件说明

3.1 AI插件介绍

平台中特定游戏的AI逻辑以AI插件的形式存在，AI插件主要包含两部分：Env插件和AI模型插件。 Env插件 Env插件通常实现与游戏业务相关的一些逻辑，如执行各种游戏动作输出、通过API设置图像识别任务并获取识别结果。
AI模型插件 AI模型插件通常用来实现特定的游戏AI算法，如强化学习算法、模仿学习算法、行为树等算法等。根据Env插件得到的游戏状态信息（游戏图像、游戏中技能状态、血条、敌人位置等）决策游戏动作，并执行对应的游戏动作。

平台定义了Env插件和AI模型插件的接口，继承并实现对应的插件接口，即可实现自己的AI插件，并由平台AI框架加载运行AI插件。Env插件和AI模型插件不是一一对应的关系，一个AI模型插件可以对应不同的Env插件。如果不同游戏使用了同样的AI算法，但游戏的一些业务逻辑存在差异，这种情况下每个游戏可以实现不同的Env插件，但可以使用同一个AI模型插件。

平台内置了DQN强化学习和模仿学习插件，可以通过配置cfg/task/agent/AgentAI.ini文件来直接使用内置AI插件，也可以通过配置cfg/task/agent/AgentAI.ini文件来使用用户开发的AI插件。

3.2 用户AI插件开发

下面以一个Speed游戏为例，说明用户该如何开发AI插件。开发后的Speed AI插件存放在src/PlugIn/ai/目录下，目录如下：

game_ai_sdk
└─src
    └─PlugIn
        └─ai
            └─speed
                   └─ SpeedAI.py
                   └─ SpeedEnv.py

其中SpeedEnv.py文件实现了Env插件，SpeedAI.py文件实现了AI模型插件。

修改cfg/task/agent/AgentAI.ini文件，让平台加载运行Speed AI插件，修改后的文件内容如下所示：

[AGENT_ENV]
UsePluginEnv = 1
EnvPackage = speed
EnvModule = SpeedEnv
EnvClass = SpeedEnv

[AI_MODEL]
UsePluginAIModel = 1
AIModelPackage = speed
AIModelModule = SpeedAI
AIModelClass = SpeedAI

[RUN_FUNCTION]
UseDefaultRunFunc = 1
;RunFuncPackage = MyPackage
;RunFuncModule = MyModule
;RunFuncName = MyRunFunc

UsePluginEnv = 1 表示使用PlugIn目录下的Env插件，而非平台内置的Env插件。
EnvPackage = speed 表示用户开发Env插件的Package名（Package名通常即为目录名）为speed
EnvModule = SpeedEnv 表示用户开发Env插件的Module名（Module名通常即为文件名）为SpeedEnv
EnvClass = SpeedEnv 表示用户开发的Env插件的Class名（Class名即为类名）为SpeedEnv
UsePluginAIModel = 1 表示使用PlugIn目录下的AI模型插件
AIModelPackage = speed 表示用户开发AI模型插件的Package名为speed
AIModelModule = SpeedAI 表示用户开发AI模型插件的Module名为SpeedAI

AIModelClass = SpeedAI 表示用户开发AI模型插件的Class名为SpeedAI

3.3 Env插件实现

Env插件需要继承平台中的GameEnv类，并实现GameEnv的接口。 SpeedEnv插件代码如下：

import time
import os
import cv2

from agentenv.GameEnv import GameEnv
from AgentAPI import AgentAPIMgr
from ActionAPI.ActionAPIMgr import ActionAPIMgr
from util import util

#游戏场景识别任务配置文件
TASK_CFG_FILE = 'cfg/task/gameReg/Task.json'

REG_GROUP_ID = 1

BEGIN_TASK_ID = 1 #游戏开始识别任务ID
WIN_TASK_ID = 2 #游戏结束识别任务ID
LOSE_TASK_ID = 3  #游戏失败识别任务ID
DATA_TASK_ID = 4  #游戏胜利识别任务ID

GAME_STATE_INVALID = 0
GAME_STATE_RUN = 1
GAME_STATE_WIN = 2
GAME_STATE_LOSE = 3

class SpeedEnv(GameEnv):
    #构造函数
    def __init__(self):
        GameEnv.__init__(self)
        self.__frameIndex = -1
        self.__actionMgr = ActionAPIMgr()  #创建执行动作对象
        self.__agentAPI = AgentAPIMgr.AgentAPIMgr()  #创建场景识别对象
        self.Reset()

    #初始化函数，通常初始化动作模块和识别模块
    def Init(self):
        taskCfgFile = util.ConvertToSDKFilePath(TASK_CFG_FILE)
        self.__agentAPI.Initialize(taskCfgFile)
        self.__agentAPI.SendCmd(AgentAPIMgr.MSG_SEND_GROUP_ID, REG_GROUP_ID)
        self.__actionMgr.Initialize()
        return True

    #退出函数
    def Finish(self):
        self.__agentAPI.Release()
        self.__actionMgr.Finish()

    #游戏动作个数
    def GetActionSpace(self):
        return 3

    #输出游戏动作
    def DoAction(self, action):
        if action == 0:
            #执行点击动作1
            self.__actionMgr.Click(140, 570, contact=0, frameSeq=self.__frameIndex, durationMS=100)
        elif action == 1:
            #执行点击动作2
            self.__actionMgr.Click(300, 570, contact=0, frameSeq=self.__frameIndex, durationMS=100)
        elif action == 2:
            #不进行任何动作
            pass

    #根据识别模块API获取游戏状态信息，并返回状态信息
    def GetState(self):
        #get game data , image and state
        gameInfo = self._GetGameInfo()
        self.__frameIndex = gameInfo['frameSeq']
        image = gameInfo['image']
        state = self.__gameState
        img = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)

        self.__isTerminal = True
        if state == GAME_STATE_RUN:
            self.__isTerminal = False

        return img

    #重置游戏状态，通常可以在每局游戏结束或开始时调用
    def Reset(self):
        self.__isTerminal = True
        self.__gameState = GAME_STATE_INVALID

    #根据识别模块获取的信息，判断游戏对局是否开始
    def IsEpisodeStart(self):
        gameInfo = self._GetGameInfo()
        if self.__gameState == GAME_STATE_RUN:
            self.__isTerminal = False
            return True

        return False

    #根据识别模块获取的信息，判断游戏对局是否结束
    def IsEpisodeOver(self):
        return self.__isTerminal

    #获取taskID对应的识别结果
    def _GetTemplateState(self, resultDict, taskID):
        state = False
        px = -1
        py = -1
        regResults = resultDict.get(taskID)
        if regResults is None:
            return (state, px, py)

        for item in regResults:
            flag = item['flag']
            if flag:
                x = item['boxes'][0]['x']
                y = item['boxes'][0]['y']
                w = item['boxes'][0]['w']
                h = item['boxes'][0]['h']

                state = True
                px = int(x + w/2)
                py = int(y + h/2)
                break

        return (state, px, py)

    #根据识别模块API获取识别的的游戏状态信息
    def _GetGameInfo(self):
        gameInfo = None

        while True:
            gameInfo = self.__agentAPI.GetInfo(AgentAPIMgr.GAME_RESULT_INFO)
            if gameInfo is None:
                time.sleep(0.002)
                continue

            result = gameInfo['result']
            if result is None:
                time.sleep(0.002)
                continue

            flag, _, _ = self._GetTemplateState(result, BEGIN_TASK_ID)
            if flag is True:
                self.__gameState = GAME_STATE_RUN

            flag, _, _ = self._GetTemplateState(result, WIN_TASK_ID)
            if flag is True:
                self.__gameState = GAME_STATE_WIN

            flag, _, _ = self._GetTemplateState(result, LOSE_TASK_ID)
            if flag is True:
                self.__gameState = GAME_STATE_LOSE

            data = None
            if result.get(DATA_TASK_ID) is not None:
                data = result.get(DATA_TASK_ID)[0]

            if data is None:
                time.sleep(0.002)
                continue
            else:
                break

        return gameInfo

3.4 AI模型插件实现

AI模型插件需要继承平台中的AIModel类，并实现AIModel的接口。实现QQSpeedAI AI模型插件举例：

import random

from aimodel.AIModel import AIModel
from agentenv.GameEnv import GameEnv

class SpeedAI(AIModel):
    #构造函数
    def __init__(self):
        AIModel.__init__(self)

    #初始化函数，参数agentEnv为 Env插件类实例对象
    def Init(self, agentEnv):
        self.__agentEnv = agentEnv
        return True

    #退出函数
    def Finish(self):
        pass

    #检测到每一局游戏开始后，AI算法进行的操作可以在此处实现，如一些变量的重置等
    def OnEpisodeStart(self):
        pass

    #检测到每一局游戏结束后，AI算法进行的操作可以在此处实现
    def OnEpisodeOver(self):
         self.__agentEnv.Reset()

    #当加载进入游戏场景时，需要进行的操作可以在此处实现
    def OnEnterEpisode(self):
        pass

    #当离开退出游戏场景时，需要进行的操作可以在此处实现
    def OnLeaveEpisode(self):
        pass

    #训练AI操作的每一个step实现,通常强化学习算法需要实现此接口,基于规则的AI无需训练,不需要实现此接口
    def TrainOneStep(self):
        image =  self.__agentEnv.GetState()

    #AI测试的每一个step实现，通常实现为agentEnv获取游戏状态数据，然后根据AI算法输出对应的游戏操作
    def TestOneStep(self):
        image =  self.__agentEnv.GetState()
        #本样例采用随机动作输出
        action = random.choice([0,1,2])
        self.__agentEnv.DoAction(action)

3.5 编译运行AI插件

编译运行AI插件时，请确保在game_ai_sdk虚拟环境下。进入game_ai_sdk/build目录，执行如下命令即可编译Speed AI插件：

(game_ai_sdk)./build_plugin.sh

进入game_ai_sdk/bin目录，运行如下命令即可运行Speed AI插件：

(game_ai_sdk)python agentai.py

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

DevelopAI.md

DevelopAI.md

AI的二次开发

1 AgentAPIMgr 介绍

1.1 整体流程

1.2 agentAPI的介绍

2 ActionAPIMgr介绍

2.1 整体流程

2.2 ActionAPI的介绍

3. AI插件说明

3.1 AI插件介绍

3.2 用户AI插件开发

3.3 Env插件实现

3.4 AI模型插件实现

3.5 编译运行AI插件

Files

DevelopAI.md

Latest commit

History

DevelopAI.md

File metadata and controls

AI的二次开发

1 AgentAPIMgr 介绍

1.1 整体流程

1.2 agentAPI的介绍

2 ActionAPIMgr介绍

2.1 整体流程

2.2 ActionAPI的介绍

3. AI插件说明

3.1 AI插件介绍

3.2 用户AI插件开发

3.3 Env插件实现

3.4 AI模型插件实现

3.5 编译运行AI插件