テスト実行プログラム
1. 概要
テスト実行プログラムは、あらかじめ生成したチェック対象となる衝突目標地点(目標点)の配置データファイル(コリジョンファイル)を読み込み、 これらの衝突目標地点に対する自動テストを複数ゲームプロセスで並列実行できるようにしたものです。
このプログラムは全体を統括するサーバープログラムと、 サーバーからの指示で衝突目標地点に対してコリジョンチェックを行うクライアントプログラムとで構成されています。
図のように、サーバー稼働 PC の上で動作する 1 つのサーバープロセスに対して、 複数のクライアント稼働 PC の上で動作する複数のクライアントプロセスが接続するという構成になっています。
関連ファイル
テスト実行プログラムに関連するファイルは以下です。
Python
├── base +--------------------------+ 全ゲーム共通
│ ├── base_container.py +---------+ ゲームから送信された情報を Python で扱いやすいように加工
│ ├── base_debug_launcher.py +----+ タイトル画面 / デバッグ機能等の操作モジュール群
│ ├── base_state_machines.py +----+ ステートマシンモジュール群
│ └── base_wrapper.py +-----------+ ゲーム環境との通信に関わるクラス
│
├── custom +------------------------+ ゲーム固有
│ ├── alfort +--------------------+ UE サンプルゲーム(Alfort)向けコード
│ │ ├── custom_container.py +---+ ゲームから送信された情報を Python で扱いやすいように加工
│ │ ├── custom_util.py +--------+ ゲーム固有の汎用関数
│ │ ├── custom_wrapper.py +-----+ ゲーム環境との通信に関わるクラス
│ │ ├── debug_launcher.py +-----+ タイトル画面 / デバッグ機能等の操作モジュール群
│ │ └── state_machines.py +-----+ ステートマシンモジュール群
│ ├── alfort +--------------------+ Unity サンプルゲーム(unity_demo)向けコード
│ │ ├── custom_container.py +---+ ゲームから送信された情報を Python で扱いやすいように加工
│ │ ├── custom_util.py +--------+ ゲーム固有の汎用関数
│ │ ├── custom_wrapper.py +-----+ ゲーム環境との通信に関わるクラス
│ │ ├── debug_launcher.py +-----+ タイトル画面 / デバッグ機能等の操作モジュール群
│ │ └── state_machines.py +-----+ ステートマシンモジュール群
│ │
│ └── __init__.py +----------------+ 対象のゲームを指定
│
├── game_modules +------------------+ ゲーム非依存処理
│ ├── game_modules.py +-----------+ ゲーム操作に関連するモジュール群
│ └── state_machines.py +---------+ ステートマシンモジュール群
│
├── map_data +-----------------------+ スキャン結果や地図データを扱うための処理
│ └── map_data.py +----------------+ スキャン結果や地図データを扱うためのモジュール群
│
├── utils +-------------------------+ ユーティリティ
│ ├── assert_window_watcher.py +--+ ゲームプログラムの Assert 監視
│ ├── logger.py +-----------------+ ログ出力モジュール群
│ ├── profiler.py +---------------+ 時間プロファイルモジュール
│ ├── shared_data_server.py +-----+ 通信モジュール群
│ └── util_modules.py +-----------+ 汎用関数群
│
├── res +---------------------------+ フォントなどのゲームに関連しないリソースフォルダ
├── result +------------------------+ 結果の格納先フォルダ
│
├── data +--------------------------+ ゲームごとのリソース/設定ファイル用フォルダ
│ ├── alfort +--------------------+ UE サンプルゲーム(Alfort)向けデータ
│ │ ├── _map_list.yaml +--------+ 設定ファイル
│ │ └── *.png +-----------------+ 各マップの地図画像
│ └── unity_demo +----------------+ Unity サンプルゲーム(unity_demo)向けコード
│ ├── _map_list.yaml +--------+ 設定ファイル
│ └── *.png +-----------------+ 各マップの地図画像
│
├── coltest_checker.py +------------+ make_colfile.py で作成した結果を地図上に表示するデータチェック用プログラム
├── make_colfile.py +---------------+ Map Scanner で作成したデータを用いてコリジョンチェックの目標点をリストアップするプログラム
├── result_viewer.py +--------------+ コリジョンチェックの結果表示プログラム
├── svrun_client.py +---------------+ クライアントプロセスを複数起動するプログラム
├── svrun_client_logic.py +---------+ クライアントプログラム本体
├── svrun_protocol.py +-------------+ 通信プロトコル関連のモジュール群
└── svrun_server.py +---------------+ サーバープログラム
2. クライアントプロセス
メインループ
メインループでは、プレイヤーキャラクターの行動ロジックを扱う agent 変数、
ゲームプロセスとの通信を扱う env 変数、サーバープログラムとの通信を扱う client の 3 変数を動かすことでシステムを実行しています。
# svrun_client_logic.py
def run(self):
"""コリジョンチェックメインループ
"""
# ゲームを起動
game_data = self.__boot()
while True:
try:
# 前フレームでゲームから受け取った情報を元に、次フレームのエージェントの行動（パッド入力内容）を決定
pad_input, user_dict, reset, restart, reboot, finish = self.agent.step(game_data)
# 上記で決定したパッド入力内容をゲームへ送信
game_data = self.env.step(pad_input, user_dict)
if reset:
game_data = self.__reset()
elif restart:
game_data = self.__restart()
elif reboot:
self.__shutdown()
game_data = self.__boot()
elif finish:
self.__shutdown()
break
except: # 実行中にエラーが発生した場合はゲーム終了かつ通信切断を必ず実行する
# エラーログを出力
log(traceback.format_exc())
# サーバー切断&ゲーム終了
self.__shutdown()
# 再びサーバー接続&ゲーム起動
game_data = self.__boot()
テストエージェント
テストエージェント動作コード
コリジョン抜け検出を行う際、プレイヤーがよく行う行動からプレイヤーが殆ど行わないであろう行動に至るまで、様々な衝突方式を用意する必要性が出てくると考えられます。 こうした技術的要件を満たすために、ステートマシンを利用して挙動をモジュール化しています。
メインループと合わせて、CollisionTestAgent クラスの内部では以下のようなコードでステートマシンが動作します。
# svrun_client_logic.py
# テストエージェント動作コード
class CollisionTestAgent:
def step(self, game_data: CustomContainer) -> Tuple[bridge.PseudoPadInput, Dict[str, Any], bool, bool, bool, bool]:
"""エージェントの更新関数. 関数内で次の目標点の選択と衝突等を実行する
Args:
game_data (CustomContainer): Game-Python Bridge経由でゲームから送信されたゲームデータ
Returns:
Tuple[bridge.PseudoPadInput, Dict[str, Any], bool, bool, bool, bool]:
パッド入力情報, ユーザー定義情報, エージェントをリセットするか,
前述+ゲームを再起動するか, 前述+ゲームとの接続をリセットするか, テストを終了させるか
"""
# 各種状態判定
pad_input = None
user_dict = None
restart = time.time() - self.reset_time > self.reset_limit
reboot = False
finish = False
drop = False
# デバッグランチャーの操作委託
if CustomDebugLauncher.instance.has_order():
pad_input = CustomDebugLauncher.instance.step(game_data)
else:
# 衝突チェック中のみ行われる処理
if "Check" == self.status:
# プレイヤーが奈落落ち状態になっているかの判定
drop = self.does_player_drop(game_data)
# 衝突チェック中に一定以上座標が移動した場合は、都度サーバーに座標送信
player_pos = np.array([game_data.pos_x, game_data.pos_y, game_data.pos_z])
dis = np.linalg.norm(player_pos - self.__recent_pos)
if self.__param_send_interval < dis:
self.__recent_pos = player_pos
CollisionTestProtocol.c2s_notice_pos(self.client, self.id, self.current_target_id, player_pos, game_data.player_hit)
# データ処理負荷軽減のため、game_data["IllegalPoints"][idx]["IsDetect"] が任意のidx(値域は0~16)でTrueの時だけサーバーにコリジョン抜け検出を通知
lackcol_data = game_data.illegal_points
detected = False
for ray_dir in lackcol_data: # 検出用のレイは周囲16方向+鉛直方向の17本
if ray_dir["IsDetect"]:
detected = True
break
if detected:
CollisionTestProtocol.c2s_notice_lackcol(self.client, self.id, self.current_target_id, player_pos, lackcol_data)
# 衝突チェック中に落下判定が有効になりリセットする場合は、サーバーに衝突終了を通知してからリセット
if drop:
CollisionTestProtocol.c2s_notice_end_check(self.client, self.id, self.current_target_id, is_drop=True)
self.reset_time = time.time()
restart = True
# ステートマシン処理
pad_input, user_dict, next_state = self.__state.step(game_data)
if self.__state != next_state:
self.__state = next_state
return pad_input, user_dict, drop, restart, reboot, finish
備考
ステートマシン
ステートベース AI はキャラクターなどの挙動を状態毎に分ける手法です。
各状態は別の状態へ遷移する条件があり、ある状態から別の状態へ遷移することで、
キャラクターが思考しているようにユーザーに思わせることができます。
このような状態が遷移するステートベースの手法をステートマシンと呼びます。
引用元: https://yttm-work.jp/game_ai/game_ai_0003.html
ステートマシンのクラス図
衝突方式のバリエーションを増やす場合は、ステート基底クラスを継承します。
現在、各ステートマシンクラスは以下 2 つのファイル内に定義されています。
game_modules/state_machines.py にあるクラスはゲーム非依存のステート、
alfort/state_machines.py にあるクラスはゲーム依存のステートとなっています。
Python
├── custom +----------------------+ ゲーム固有
│ └── alfort +------------------+ UE サンプルゲーム(Alfort)向けコード
│ └── state_machines.py +---+ ステートマシンモジュール群
└── game_modules +----------------+ ゲーム非依存処理
└── state_machines.py +-------+ ステートマシンモジュール群
ステートマシンのフローチャート
現在は以下のような制御フローで動いています。
3. サーバープロセス
別スレッドで動くデータサーバーからデータを取得し、取得したデータをコリジョンチェック全体の管理クラスCollisionTestManagerに渡しています。
# svrun_server.py
# サーバーメインループ (簡略コード)
if __name__ == "__main__":
# サーバー listen 開始
data_server = TwoWaySharedDataServer(args.listennum, args.addr, args.port)
data_server.start()
# コリジョンチェック全体の管理クラス (CollisionTestManager) インスタンスの生成と初期化
# テスト対象マップの名称リストを管理クラスに渡す
test_manager = CollisionTestManager(args.map)
# 対象マップ毎にコリジョンチェックの事前準備を行う
test_manager.init(args.datafolder, result_folder_path, args.reset)
while True:
# クライアントから送られてきたデータを取得
results = data_server.receive()
# データ毎にコリジョンチェック管理クラスへ送信
for index, data in results:
test_manager.on_recv_protocol_data(data_server, index, data)
メインループで渡されたデータは、エージェント ID ごとに割り振られたテスト管理クラスCollisionTestに渡されます。
この仕組みによりシームレスにテストマップを切り替えることができます。
# svrun_server.py
# コリジョンチェック全体の管理クラス (簡略コード)
class CollisionTestManager:
def __init__(self, map_list):
# テスト対象マップの名称リストを受け取る
self.__map_list: List[str] = map_list
def init(self, data_folder_path, result_folder_path, reset_progress):
# 対象マップ毎にコリジョンチェックの事前準備を行う
for test_id, map_name in enumerate(self.__map_list):
# 対象マップのテストを管理するクラス (CollisionTest) インスタンスを生成
test = CollisionTest(data_folder_path, result_folder_path, test_id, map_name)
# テストの事前準備を実行 (コリジョンファイルの読み込みと pickle データの読み込み)
test.init(reset_progress)
# テスト配列にテスト管理インスタンスを追加
self.__test_list.append(test)
def on_recv_protocol_data(self, server, client_index, data):
# クライアントから送られてきたデータを処理する
# プロトコルの識別キーがエージェント ID 要求の場合
if data["PKEY"] == C2S_REQUEST_AGENT_ID:
# テストが終了していないテストを __test_list 変数の中から取得
test = self.__get_test()
# client_index は重複なし加算のためエージェント ID として流用
agent_id = client_index
# エージェントに割り振ったテスト管理変数をデータベースに記録
self.__agent2test_database[agent_id] = test
# エージェント ID をクライアントに返信
CollisionTestProtocol.s2c_response_agentid(server, client_index, agent_id)
# エージェント ID 要求以外の場合はエージェント毎に割り振られたテスト管理変数にデータ受信処理を委譲
else:
# エージェントに割り振られたテスト管理変数を取得
test = self.__agent2test_database[data["AgentID"]]
# テスト管理変数にデータ受信を通知
test.on_recv_protocol_data(server, client_index, data)
担当マップのテスト管理するクラスCollisionTestでは、受け取ったデータの識別キーに応じたデータ受信時の処理を実行します。
# svrun_server.py
# 担当マップのテスト管理するクラス (簡略コード)
class CollisionTest:
def __init__(self, data_folder_path, result_folder_path, id, map_name):
# 担当マップの ID とマップ名を保存
self.id = id
self.map_name: str = map_name
def init(self, reset: bool):
# 担当マップのコリジョンファイルを読み込み
self.__load_collision_data()
# 実行時コマンドで pickle データ (テストの途中経過データ) が指定されていた場合は読み込み
self.__load_pickle_instances()
def on_recv_protocol_data(self, server, client_index, data):
# 受信した識別キーに応じて処理を実行
# 例: 識別キーが「目標点情報を要求」の場合
if data["PKEY"] == C2S_REQUEST_TARGET_INFO:
# データ受信時の処理を記述
# 例: 識別キーが「マップ情報を要求」の場合
elif data["PKEY"] == C2S_REQUEST_MAP_INFO:
# データ受信時の処理を記述
4. 連携シーケンス
以下にサーバーとクライアントの連携シーケンスを示します。
- [PC] - PC 操作
- [S] - サーバー処理
- [C] - クライアント処理
- [C2S] - クライアント から サーバー へデータ送信
- [S2C] - サーバー から クライアント へデータ返信
1. [PC] サーバー稼働 PC でサーバープロセス起動コマンド実行
# クライアント上限数 150 で PL_010VIL と PL_0202RIV を順次テストする場合
$ python svrun_server.py -n 150 -m PL_010VIL PL_0202RIV
2. [S] サーバー起動
foreach(クライアント稼働 PC in 全クライアント稼働 PC)
|
| 3. [PC] クライアント稼働 PC で複数のクライアントプロセスを起動するコマンド実行
| # 5 つのクライアントプロセスを起動する場合
| $ python svrun_client.py -n 5
|
| foreach(クライアントプロセス in 全クライアントプロセス)
| |
| | 4. [C] クライアントがサーバーに接続
| |
| | while ([S] 未完了のテスト対象マップがある間)
| | |
| | | 5. [C2S] エージェント ID の発行を要求
| | | {"PKEY": C2S_REQUEST_AGENT_ID}
| | | 6. [S] エージェント ID の発行後にエージェント別でテスト対象マップを割り当て
| | | 7. [S2C] エージェント ID を返信 (client-idx を流用)
| | | {"PKEY": S2C_RESPONSE_AGENT_ID, "AgentID": <id>}
| | | 8. [C2S] マップ情報を要求
| | | {"PKEY": C2S_REQUEST_MAP_INFO, "AgentID": <id>}
| | | 9. [S2C] マップ情報を返信
| | | {"PKEY": S2C_RESPONSE_MAP_INFO, "MapName": <map_name>}
| | | 10. [C] 対象マップのセーブデータを読み込みパスにコピー
| | | 11. [C] ゲーム起動
| | | 12. [C] ゲームセットアップ
| | |
| | | while ([S] テスト対象マップ内に未完了の目標点がある間)
| | | |
| | | | 13. [C2S] 目標点情報を要求
| | | | {"PKEY": C2S_REQUEST_TARGET_INFO, "Pos": <xyz>}
| | | | 14. [S2C] 目標点情報を返信
| | | | {"PKEY": S2C_RESPONSE_TARGET_INFO, "TarID": <id>, "Pos": <xyz>, "Dir": <xyz>}
| | | | 15. [C2S] 衝突開始を通知
| | | | {"PKEY": C2S_NOTICE_START_CHECK, "AgentID": <id>, "TarID": <id>}
| | | |
| | | | while ([C] プレイヤー移動中の間)
| | | | |
| | | | | 16. [C2S] 現在の座標情報を通知
| | | | | {"PKEY": C2S_NOTICE_POSITION, "AgentID": <id>, "TarID": <id>, "Pos": <xyz>}
| | | | | 17. [C2S] 怪しい箇所の検出情報を通知
| | | | | {"PKEY": C2S_NOTICE_LACK_COLLISION, "AgentID": <id>, "TarID": <id>, "Pos": <xyz>, "LackCol": <lackcol_data>}
| | | | |
| | | | 18. [C2S] 衝突終了を通知
| | | | {"PKEY": C2S_NOTICE_END_CHECK, "AgentID": <id>, "TarID": <id>}
| | | |
| | | 19. [S] テスト結果データを json ファイルに出力
| | | 20. [C2S] 目標点情報を要求
| | | {"PKEY": C2S_REQUEST_TARGET_INFO, "Pos": <xyz>}
| | | 21. [S2C] 対象マップのテスト終了を返信
| | | {"PKEY": S2C_RESPONSE_TARGET_INFO, "TarID": -1, "Pos": [-1,-1,-1], "Dir": [-1,-1,-1]}
| | | 22. [C] ゲームをシャットダウン & 再起動
| | |
| | 23. [C2S] エージェント ID の発行を要求
| | {"PKEY": C2S_REQUEST_AGENT_ID}
| | 24. [S2C] 全対象マップのテスト終了を通知
| | {"PKEY": S2C_RESPONSE_AGENT_ID, "AgentID": -1}
|
25. [S] サーバープロセス終了
5. 通信プロトコル
サーバーとクライアント間の双方向通信は TCP ソケットを利用しています。 全ての通信データを json 形式に変換してから送受信しています。
クライアント To サーバー
エージェント ID の発行を要求
- PKEY: C2S_REQUEST_AGENT_ID (1:int)
目標点情報を要求
- PKEY: C2S_REQUEST_TARGET_INFO (2:int)
- AgentID: エージェント ID (int)
- Pos: 座標_xyz (float*3)
マップ情報を要求
- PKEY: C2S_REQUEST_MAP_INFO (3:int)
- AgentID: エージェント ID (int)
衝突開始を通知
- PKEY: C2S_NOTICE_START_CHECK (4:int)
- AgentID: エージェント ID (int)
- TarID: 目標点番号 (int)
現在の座標情報を通知
- PKEY: C2S_NOTICE_POSITION (5:int)
- AgentID: エージェント ID (int)
- TarID: 目標点番号 (int)
- Pos: 座標_xyz (float*3)
- Hit: 接触フラグ (bool)
衝突終了を通知
- PKEY: C2S_NOTICE_END_CHECK (6:int)
- AgentID: エージェント ID (int)
- TarID: 目標点番号 (int)
- Drop: 奈落落ち発生フラグ (bool)
ゲームリセットを通知
- PKEY: C2S_NOTICE_RESTART (7:int)
- AgentID: エージェント ID (int)
怪しい箇所の検出情報を通知
- PKEY: C2S_NOTICE_LACK_COLLISION (8:int)
- AgentID: エージェント ID (int)
- TarID: 目標点番号 (int)
- Pos: 座標_xyz (float*3)
- LackCol: game_data["IllegalPoints"] を送信
サーバー To クライアント
エージェント ID を返信
- PKEY: S2C_RESPONSE_AGENT_ID (101:int)
- AgentID: エージェント ID (int)
目標点情報を返信
- PKEY: S2C_RESPONSE_TARGET_INFO (102:int)
- TarID: 目標点番号 (int)
- Pos: 座標_xyz (float*3)
- Dir: 衝突方向ベクトル_xy (float*3)
マップ情報を返信
- PKEY: S2C_RESPONSE_MAP_INFO (104:int)
- MapName: マップ ID (int)
6. テスト結果データ
テスト完了後に結果データがファイル保存されます。
データフォーマット
ReachData: 到達点情報 (list)
- idx: 要素番号 (int)
- X: 到達点座標_Y (float)
- Y: 到達点座標_Y (float)
- Z: 到達点座標_Z (float)
- link: この到達点を始点としてリンクしている到達点の要素番号リスト (list(int))
ReachTargetData: 各目標点のテスト中にどの到達点に訪問したかの情報 (list)
- TarIdx: 目標点番号 (int)
- list: 上記目標点のテスト中に訪問した到達点の要素番号リスト (list(int))
TouchData: 接触点情報 (list)
- idx: 要素番号 (int)
- X: 到達点座標_Y (float)
- Y: 到達点座標_Y (float)
- Z: 到達点座標_Z (float)
- cnt: 接触回数 (int)
TouchTargetData: 各目標点のテスト中にどの接触点に訪問したかの情報 (list)
- TarIdx: 目標点番号 (int)
- list: 上記目標点のテスト中に訪問した接触点の要素番号リスト (list(int))
LackColData: 怪しい箇所情報 (list)
- idx: 要素番号 (int)
- X: 到達点座標_Y (float)
- Y: 到達点座標_Y (float)
- Z: 到達点座標_Z (float)
- IllegalPoints: 17 本分のレイキャスト情報 (list)
- IsDetect: レイキャスト判定結果 (float)
- Location: レイキャスト接触座標 (dict)
- X: レイキャスト接触座標_X (float)
- Y: レイキャスト接触座標_Y (float)
- Z: レイキャスト接触座標_Z (float)
LackcolTargetData: 各目標点のテスト中にどの怪しい箇所に訪問したかの情報 (list)
- TarIdx: 目標点番号 (int)
- list: 上記目標点のテスト中に訪問した怪しい箇所の要素番号リスト (list(int))