EESAST
/
THUAI6
Not watched
1
0
Fork 0
Code
Releases 3 Wiki evaluate Activity
Issues 0 Pull Requests 0 Datasets Model Cloudbrain HPC
Browse Source

feat(PyAPI): ✨ complete most parts of PyAPI except logger

tags/0.1.0
DragonAura 2 years ago
parent
b1db196351
commit
f42646da8e
16 changed files with 1130 additions and 826 deletions
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. +0
    -237
      PyAPI/API/API.py
  2. +0
    -240
      PyAPI/API/DebugAPI.py
  3. +0
    -164
      PyAPI/API/logic.py
  4. +0
    -26
      PyAPI/API/main.py
  5. +0
    -105
      PyAPI/API/structures.py
  6. +6
    -3
      PyAPI/PyAPI/AI.py
  7. +250
    -0
      PyAPI/PyAPI/API.py
  8. +21
    -19
      PyAPI/PyAPI/Communication.py
  9. +250
    -0
      PyAPI/PyAPI/DebugAPI.py
  10. +25
    -24
      PyAPI/PyAPI/Interface.py
  11. +1
    -1
      PyAPI/PyAPI/State.py
  12. +374
    -0
      PyAPI/PyAPI/logic.py
  13. +30
    -0
      PyAPI/PyAPI/main.py
  14. +159
    -0
      PyAPI/PyAPI/structures.py
  15. +8
    -7
      PyAPI/PyAPI/utils.py
  16. +6
    -0
      PyAPI/logs/logic-log.txt

+ 0
- 237
PyAPI/API/API.py View File

@@ -1,237 +0,0 @@
from Interface import ILogic, IHumanAPI, IButcherAPI, IGameTimer, IAI
from typing import List, Union
import structures as THUAI6


class HumanAPI(IHumanAPI, IGameTimer):

# 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴

def __init__(self, logic: ILogic):
self.__logic = logic

def Move(self, timeInMilliseconds: int, angle: float) -> bool:
pass

# 向特定方向移动

def MoveRight(self, timeInMilliseconds: int) -> bool:
pass

def MoveLeft(self, timeInMilliseconds: int) -> bool:
pass

def MoveUp(self, timeInMilliseconds: int) -> bool:
pass

def MoveDown(self, timeInMilliseconds: int) -> bool:
pass

# 道具和技能相关

def PickProp(self, propType: THUAI6.PropType) -> bool:
pass

def UseProp(self) -> bool:
pass

def UseSkill(self) -> bool:
pass

# 消息相关，接收消息时无消息则返回(-1, '')

def SendMessage(self, toID: int, message: str) -> bool:
pass

def HaveMessage(self) -> bool:
pass

def GetMessage(self) -> tuple[int, str]:
pass

# 等待下一帧

def Wait(self) -> bool:
pass

# 获取各类游戏中的消息

def GetFrameCount(self) -> int:
pass

def GetPlayerGUIDs(self) -> List[int]:
pass

def GetButchers(self) -> List[THUAI6.Butcher]:
pass

def GetHumans(self) -> List[THUAI6.Human]:
pass

def GetProps(self) -> List[THUAI6.Prop]:
pass

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
pass

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
pass

def GetPlaceType(self, cellX: int, cellY: int) -> THUAI6.PlaceType:
pass

# 用于DEBUG的输出函数，仅在DEBUG模式下有效

def PrintHuman(self) -> None:
pass

def PrintButcher(self) -> None:
pass

def PrintProp(self) -> None:
pass

def PrintSelfInfo(self) -> None:
pass

# 人类阵营的特殊函数

def Escape(self) -> bool:
pass

def StartFixMachine(self) -> bool:
pass

def EndFixMachine(self) -> bool:
pass

def StartSaveHuman(self) -> bool:
pass

def EndSaveHuman(self) -> bool:
pass

# Timer用

def StartTimer(self) -> None:
pass

def EndTimer(self) -> None:
pass

def Play(self, ai: IAI) -> None:
pass


class ButcherAPI(IButcherAPI, IGameTimer):

# 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴

def Move(self, timeInMilliseconds: int, angle: float) -> bool:
pass

# 向特定方向移动

def MoveRight(self, timeInMilliseconds: int) -> bool:
pass

def MoveLeft(self, timeInMilliseconds: int) -> bool:
pass

def MoveUp(self, timeInMilliseconds: int) -> bool:
pass

def MoveDown(self, timeInMilliseconds: int) -> bool:
pass

# 道具和技能相关

def PickProp(self, propType: THUAI6.PropType) -> bool:
pass

def UseProp(self) -> bool:
pass

def UseSkill(self) -> bool:
pass

# 消息相关，接收消息时无消息则返回(-1, '')

def SendMessage(self, toID: int, message: str) -> bool:
pass

def HaveMessage(self) -> bool:
pass

def GetMessage(self) -> tuple[int, str]:
pass

# 等待下一帧

def Wait(self) -> bool:
pass

# 获取各类游戏中的消息

def GetFrameCount(self) -> int:
pass

def GetPlayerGUIDs(self) -> List[int]:
pass

def GetButchers(self) -> List[THUAI6.Butcher]:
pass

def GetHumans(self) -> List[THUAI6.Human]:
pass

def GetProps(self) -> List[THUAI6.Prop]:
pass

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
pass

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
pass

def GetPlaceType(self, cellX: int, cellY: int) -> THUAI6.PlaceType:
pass

# 用于DEBUG的输出函数，仅在DEBUG模式下有效

def PrintHuman(self) -> None:
pass

def PrintButcher(self) -> None:
pass

def PrintProp(self) -> None:
pass

def PrintSelfInfo(self) -> None:
pass

# 屠夫阵营的特殊函数

def Attack(self, angle: float) -> bool:
pass

def CarryHuman(self) -> bool:
pass

def ReleaseHuman(self) -> bool:
pass

def HangHuman(self) -> bool:
pass

# Timer用

def StartTimer(self) -> None:
pass

def EndTimer(self) -> None:
pass

def Play(self, ai: IAI) -> None:
pass

+ 0
- 240
PyAPI/API/DebugAPI.py View File

@@ -1,240 +0,0 @@
from Interface import IHumanAPI, IButcherAPI, IGameTimer, IAI
from logic import ILogic

from Interface import ILogic, IHumanAPI, IButcherAPI, IGameTimer
from typing import List, Union
import structures as THUAI6


class HumanDebugAPI(IHumanAPI, IGameTimer):

# 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴

def __init__(self, logic: ILogic):
self.__logic = logic

def Move(self, timeInMilliseconds: int, angle: float) -> bool:
pass

# 向特定方向移动

def MoveRight(self, timeInMilliseconds: int) -> bool:
pass

def MoveLeft(self, timeInMilliseconds: int) -> bool:
pass

def MoveUp(self, timeInMilliseconds: int) -> bool:
pass

def MoveDown(self, timeInMilliseconds: int) -> bool:
pass

# 道具和技能相关

def PickProp(self, propType: THUAI6.PropType) -> bool:
pass

def UseProp(self) -> bool:
pass

def UseSkill(self) -> bool:
pass

# 消息相关，接收消息时无消息则返回(-1, '')

def SendMessage(self, toID: int, message: str) -> bool:
pass

def HaveMessage(self) -> bool:
pass

def GetMessage(self) -> tuple[int, str]:
pass

# 等待下一帧

def Wait(self) -> bool:
pass

# 获取各类游戏中的消息

def GetFrameCount(self) -> int:
pass

def GetPlayerGUIDs(self) -> List[int]:
pass

def GetButchers(self) -> List[THUAI6.Butcher]:
pass

def GetHumans(self) -> List[THUAI6.Human]:
pass

def GetProps(self) -> List[THUAI6.Prop]:
pass

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
pass

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
pass

def GetPlaceType(self, cellX: int, cellY: int) -> THUAI6.PlaceType:
pass

# 用于DEBUG的输出函数，仅在DEBUG模式下有效

def PrintHuman(self) -> None:
pass

def PrintButcher(self) -> None:
pass

def PrintProp(self) -> None:
pass

def PrintSelfInfo(self) -> None:
pass

# 人类阵营的特殊函数

def Escape(self) -> bool:
pass

def StartFixMachine(self) -> bool:
pass

def EndFixMachine(self) -> bool:
pass

def StartSaveHuman(self) -> bool:
pass

def EndSaveHuman(self) -> bool:
pass

# Timer用

def StartTimer(self) -> None:
pass

def EndTimer(self) -> None:
pass

def Play(self, ai: IAI) -> None:
pass


class ButcherDebugAPI(IButcherAPI, IGameTimer):

# 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴

def Move(self, timeInMilliseconds: int, angle: float) -> bool:
pass

# 向特定方向移动

def MoveRight(self, timeInMilliseconds: int) -> bool:
pass

def MoveLeft(self, timeInMilliseconds: int) -> bool:
pass

def MoveUp(self, timeInMilliseconds: int) -> bool:
pass

def MoveDown(self, timeInMilliseconds: int) -> bool:
pass

# 道具和技能相关

def PickProp(self, propType: THUAI6.PropType) -> bool:
pass

def UseProp(self) -> bool:
pass

def UseSkill(self) -> bool:
pass

# 消息相关，接收消息时无消息则返回(-1, '')

def SendMessage(self, toID: int, message: str) -> bool:
pass

def HaveMessage(self) -> bool:
pass

def GetMessage(self) -> tuple[int, str]:
pass

# 等待下一帧

def Wait(self) -> bool:
pass

# 获取各类游戏中的消息

def GetFrameCount(self) -> int:
pass

def GetPlayerGUIDs(self) -> List[int]:
pass

def GetButchers(self) -> List[THUAI6.Butcher]:
pass

def GetHumans(self) -> List[THUAI6.Human]:
pass

def GetProps(self) -> List[THUAI6.Prop]:
pass

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
pass

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
pass

def GetPlaceType(self, cellX: int, cellY: int) -> THUAI6.PlaceType:
pass

# 用于DEBUG的输出函数，仅在DEBUG模式下有效

def PrintHuman(self) -> None:
pass

def PrintButcher(self) -> None:
pass

def PrintProp(self) -> None:
pass

def PrintSelfInfo(self) -> None:
pass

# 屠夫阵营的特殊函数

def Attack(self, angle: float) -> bool:
pass

def CarryHuman(self) -> bool:
pass

def ReleaseHuman(self) -> bool:
pass

def HangHuman(self) -> bool:
pass

# Timer用

def StartTimer(self) -> None:
pass

def EndTimer(self) -> None:
pass

def Play(self, ai: IAI) -> None:
pass

+ 0
- 164
PyAPI/API/logic.py View File

@@ -1,164 +0,0 @@
from abc import abstractmethod
from typing import List, Union, Callable
import threading

from Interface import ILogic, IGameTimer
from State import State
import structures as THUAI6
import proto.Message2Clients_pb2 as Message2Clients
import proto.Message2Server_pb2 as Message2Server
import proto.MessageType_pb2 as MessageType


class Logic(ILogic):
def __init__(self, playerType: THUAI6.PlayerType, playerID: int, humanType: THUAI6.HumanType, butcherType: THUAI6.ButcherType) -> None:

# 类型、ID
self.__playerType: THUAI6.PlayerType = playerType
self.__playerID: int = playerID
self.__humanType: THUAI6.HumanType = humanType
self.__butcherType: THUAI6.ButcherType = butcherType

# 存储状态
self.__currentState: State
self.__bufferState: State

# timer，用于实际运行AI
self.__timer: IGameTimer

# AI线程
self.__threadAI: threading.Thread

# 互斥锁
self.__mtxAI: threading.Lock
self.__mtxBuffer: threading.Lock
self.__mtxTimer: threading.Lock

# 条件变量
self.__cvBuffer: threading.Condition
self.__cvAI: threading.Condition

# 保存缓冲区数
self.__counterState: int
self.__counterBuffer: int

# 记录游戏状态
self.__gameState: THUAI6.GameState = THUAI6.GameState.NullGameState

# 是否应该执行player()
self.__AILoop: bool = True

# buffer是否更新完毕
self.__bufferUpdated: bool = False

# 是否应当启动AI
self.__AIStart: bool = False

# asynchronous为true时控制内容更新的变量
self.__freshed: bool = False

# IAPI统一可用的接口

def GetButchers(self) -> List[THUAI6.Butcher]:
pass

def GetHumans(self) -> List[THUAI6.Human]:
pass

def GetProps(self) -> List[THUAI6.Prop]:
pass

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
pass

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
pass

def GetPlaceType(self, x: int, y: int) -> THUAI6.PlaceType:
pass

def Move(self, time: int, angle: float) -> bool:
pass

def PickProp(self, propType: THUAI6.PropType) -> bool:
pass

def UseProp(self) -> bool:
pass

def UseSkill(self) -> bool:
pass

def SendMessage(self, toID: int, message: str) -> bool:
pass

def HaveMessage(self) -> bool:
pass

def GetMessage(self) -> tuple[int, str]:
pass

def WaitThread(self) -> bool:
pass

def GetCounter(self) -> int:
pass

def GetPlayerGUIDs(self) -> List[int]:
pass

# IHumanAPI使用的接口

def Escape(self) -> bool:
pass

def StartFixMachine(self) -> bool:
pass

def EndFixMachine(self) -> bool:
pass

def StartSaveHuman(self) -> bool:
pass

def EndSaveHuman(self) -> bool:
pass

# Butcher使用的接口

def Attack(self, angle: float) -> bool:
pass

def CarryHuman(self) -> bool:
pass

def ReleaseHuman(self) -> bool:
pass

def HangHuman(self) -> bool:
pass

# Logic内部逻辑
def __TryConnection(self) -> bool:
pass

def __ProcessMessage(self) -> None:
pass

def __LoadBuffer(self) -> None:
pass

def __UnBlockBuffer(self) -> None:
pass

def __UnBlockAI(self) -> None:
pass

def __Update(self) -> None:
pass

def __Wait(self) -> None:
pass

def Main(self, createAI: Callable, IP: str, port: str, file: bool, print: bool, warnOnly: bool) -> None:
pass

+ 0
- 26
PyAPI/API/main.py View File

@@ -1,26 +0,0 @@
import sys
from typing import List, Callable
from logic import Logic
from AI import AI, Setting
from Interface import IAI
import structures as THUAI6


def THUAI6Main(argv: List[str], AIBuilder: Callable) -> None:
pID: int = 114514
sIP: str = "114.51.41.91"
sPort: str = "9810"
file: bool = False
print: bool = False
warnOnly: bool = False
logic = Logic(Setting.playerType(), pID,
Setting.humanType(), Setting.butcherType())
logic.Main(AIBuilder, sIP, sPort, file, print, warnOnly)


def CreateAI() -> IAI:
return AI()


if __name__ == '__main__':
THUAI6Main(sys.argv, CreateAI)

+ 0
- 105
PyAPI/API/structures.py View File

@@ -1,105 +0,0 @@
from enum import Enum
from typing import List


class GameState(Enum):
NullGameState = 0
GameStart = 1
GameRunning = 2
GameEnd = 3


class PlaceType(Enum):
NullPlaceType = 0
Land = 1
Wall = 2
Grass = 3
Machine = 4
Gate = 5
HiddenGate = 6


class ShapeType(Enum):
NullShapeType = 0
Square = 1
Circle = 2


class PlayerType(Enum):
NullPlayerType = 0
HumanPlayer = 1
ButcherPlayer = 2


class PropType(Enum):
NullPropType = 0
PropType1 = 1


class HumanType(Enum):
NullHumanType = 0
HumanType1 = 1


class ButcherType(Enum):
NullButcherType = 0
ButcherType1 = 1


class HumanBuffType(Enum):
NullHumanBuffType = 0
HumanBuffType1 = 1


class ButcherBuffType(Enum):
NullButcherBuffType = 0
ButcherBuffType1 = 1


class HumanState(Enum):
NullHumanState = 0
Idle = 1
Fixing = 2
Dying = 3
OnChair = 4
Dead = 5


class Player:
x: int
y: int
speed: int
viewRange: int
playerID: int
guid: int
radius: int
timeUntilSkillAvailable: float
playerType: PlayerType
prop: PropType
place: PlaceType


class Human(Player):
state: HumanState
life: int
hangedTime: int
humanType: HumanType
buff: List[HumanBuffType]


class Butcher(Player):
damage: int
movable: bool
butcherType: ButcherType
buff: List[ButcherBuffType]


class Prop:
x: int
y: int
size: int
guid: int
type: PropType
place: PlaceType
facingDirection: float
isMoving: bool

PyAPI/API/AI.py → PyAPI/PyAPI/AI.py View File

@@ -1,22 +1,25 @@
from abc import abstractmethod
import PyAPI.structures as THUAI6
from PyAPI.Interface import IHumanAPI, IButcherAPI, IAI
from typing import Union from typing import Union
from Interface import IHumanAPI, IButcherAPI, IAI
import structures as THUAI6




class Setting: class Setting:
# 为假则play()期间确保游戏状态不更新，为真则只保证游戏状态在调用相关方法时不更新 # 为假则play()期间确保游戏状态不更新，为真则只保证游戏状态在调用相关方法时不更新
@staticmethod
def asynchronous() -> bool: def asynchronous() -> bool:
return False return False


# 选手必须修改该函数的返回值来选择自己的阵营 # 选手必须修改该函数的返回值来选择自己的阵营
@staticmethod
def playerType() -> THUAI6.PlayerType: def playerType() -> THUAI6.PlayerType:
return THUAI6.PlayerType.HumanPlayer return THUAI6.PlayerType.HumanPlayer


# 选手需要将两个都定义，本份代码中不选择的阵营任意定义即可 # 选手需要将两个都定义，本份代码中不选择的阵营任意定义即可
@staticmethod
def humanType() -> THUAI6.HumanType: def humanType() -> THUAI6.HumanType:
return THUAI6.HumanType.HumanType1 return THUAI6.HumanType.HumanType1


@staticmethod
def butcherType() -> THUAI6.ButcherType: def butcherType() -> THUAI6.ButcherType:
return THUAI6.ButcherType.ButcherType1 return THUAI6.ButcherType.ButcherType1



+ 250
- 0
PyAPI/PyAPI/API.py View File

@@ -0,0 +1,250 @@
import PyAPI.structures as THUAI6
from PyAPI.Interface import ILogic, IHumanAPI, IButcherAPI, IGameTimer, IAI
from math import pi
from concurrent.futures import ProcessPoolExecutor, Future
from typing import List, Union


class HumanAPI(IHumanAPI, IGameTimer):

def __init__(self, logic: ILogic) -> None:
self.__logic = logic
self.__pool = ProcessPoolExecutor(20)

# 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴

def Move(self, timeInMilliseconds: int, angle: float) -> Future[bool]:
return self.__pool.submit(self.__logic.Move, timeInMilliseconds, angle)

# 向特定方向移动

def MoveRight(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi * 0.5)

def MoveLeft(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi * 1.5)

def MoveUp(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, 0)

def MoveDown(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi)

# 道具和技能相关

def PickProp(self, propType: THUAI6.PropType) -> Future[bool]:
return self.__pool.submit(self.__logic.PickProp, propType)

def UseProp(self) -> Future[bool]:
return self.__pool.submit(self.__logic.UseProp)

def UseSkill(self) -> Future[bool]:
return self.__pool.submit(self.__logic.UseSkill)

# 消息相关，接收消息时无消息则返回(-1, '')

def SendMessage(self, toID: int, message: str) -> Future[bool]:
return self.__pool.submit(self.__logic.SendMessage, toID, message)

def HaveMessage(self) -> Future[bool]:
return self.__pool.submit(self.__logic.HaveMessage)

def GetMessage(self) -> Future[tuple[int, str]]:
return self.__pool.submit(self.__logic.GetMessage)

# 等待下一帧

def Wait(self) -> Future[bool]:
if self.__logic.GetCounter() == -1:
return self.__pool.submit(lambda: False)
else:
return self.__pool.submit(self.__logic.WaitThread)

# 获取各类游戏中的消息

def GetFrameCount(self) -> int:
return self.__logic.GetCounter()

def GetPlayerGUIDs(self) -> List[int]:
return self.__logic.GetPlayerGUIDs()

def GetButchers(self) -> List[THUAI6.Butcher]:
return self.__logic.GetButchers()

def GetHumans(self) -> List[THUAI6.Human]:
return self.__logic.GetHumans()

def GetProps(self) -> List[THUAI6.Prop]:
return self.__logic.GetProps()

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
return self.__logic.GetSelfInfo()

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
return self.__logic.GetFullMap()

def GetPlaceType(self, cellX: int, cellY: int) -> THUAI6.PlaceType:
return self.__logic.GetPlaceType(cellX, cellY)

# 用于DEBUG的输出函数，仅在DEBUG模式下有效

def PrintHuman(self) -> None:
pass

def PrintButcher(self) -> None:
pass

def PrintProp(self) -> None:
pass

def PrintSelfInfo(self) -> None:
pass

# 人类阵营的特殊函数

def Escape(self) -> Future[bool]:
return self.__pool.submit(self.__logic.Escape)

def StartFixMachine(self) -> Future[bool]:
return self.__pool.submit(self.__logic.StartFixMachine)

def EndFixMachine(self) -> Future[bool]:
return self.__pool.submit(self.__logic.EndFixMachine)

def StartSaveHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.StartSaveHuman)

def EndSaveHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.EndSaveHuman)

# Timer用

def StartTimer(self) -> None:
pass

def EndTimer(self) -> None:
pass

def Play(self, ai: IAI) -> None:
pass


class ButcherAPI(IButcherAPI, IGameTimer):

def __init__(self, logic: ILogic) -> None:
self.__logic = logic
self.__pool = ProcessPoolExecutor(20)

# 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴

def Move(self, timeInMilliseconds: int, angle: float) -> Future[bool]:
return self.__pool.submit(self.__logic.Move, timeInMilliseconds, angle)

# 向特定方向移动

def MoveRight(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi * 0.5)

def MoveLeft(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi * 1.5)

def MoveUp(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, 0)

def MoveDown(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi)

# 道具和技能相关

def PickProp(self, propType: THUAI6.PropType) -> Future[bool]:
return self.__pool.submit(self.__logic.PickProp, propType)

def UseProp(self) -> Future[bool]:
return self.__pool.submit(self.__logic.UseProp)

def UseSkill(self) -> Future[bool]:
return self.__pool.submit(self.__logic.UseSkill)

# 消息相关，接收消息时无消息则返回(-1, '')

def SendMessage(self, toID: int, message: str) -> Future[bool]:
return self.__pool.submit(self.__logic.SendMessage, toID, message)

def HaveMessage(self) -> Future[bool]:
return self.__pool.submit(self.__logic.HaveMessage)

def GetMessage(self) -> Future[tuple[int, str]]:
return self.__pool.submit(self.__logic.GetMessage)

# 等待下一帧

def Wait(self) -> Future[bool]:
if self.__logic.GetCounter() == -1:
return self.__pool.submit(lambda: False)
else:
return self.__pool.submit(self.__logic.WaitThread)

# 获取各类游戏中的消息

def GetFrameCount(self) -> int:
return self.__logic.GetCounter()

def GetPlayerGUIDs(self) -> List[int]:
return self.__logic.GetPlayerGUIDs()

def GetButchers(self) -> List[THUAI6.Butcher]:
return self.__logic.GetButchers()

def GetHumans(self) -> List[THUAI6.Human]:
return self.__logic.GetHumans()

def GetProps(self) -> List[THUAI6.Prop]:
return self.__logic.GetProps()

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
return self.__logic.GetSelfInfo()

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
return self.__logic.GetFullMap()

def GetPlaceType(self, cellX: int, cellY: int) -> THUAI6.PlaceType:
return self.__logic.GetPlaceType(cellX, cellY)

# 用于DEBUG的输出函数，仅在DEBUG模式下有效

def PrintHuman(self) -> None:
pass

def PrintButcher(self) -> None:
pass

def PrintProp(self) -> None:
pass

def PrintSelfInfo(self) -> None:
pass

# 屠夫阵营的特殊函数

def Attack(self, angle: float) -> Future[bool]:
return self.__pool.submit(self.__logic.Attack, angle)

def CarryHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.CarryHuman)

def ReleaseHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.ReleaseHuman)

def HangHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.HangHuman)

# Timer用

def StartTimer(self) -> None:
pass

def EndTimer(self) -> None:
pass

def Play(self, ai: IAI) -> None:
pass

PyAPI/API/Communication.py → PyAPI/PyAPI/Communication.py View File

@@ -1,14 +1,12 @@
from queue import Queue
import grpc
import threading
import proto.Message2Clients_pb2 as Message2Clients
import proto.Message2Server_pb2 as Message2Server
import proto.MessageType_pb2 as MessageType
import PyAPI.structures as THUAI6
from PyAPI.AI import Setting
from PyAPI.utils import THUAI62Proto
from PyAPI.Interface import IErrorHandler
import proto.Services_pb2_grpc as Services import proto.Services_pb2_grpc as Services
from logic import Logic
from Interface import IErrorHandler
from utils import THUAI62Proto
import structures as THUAI6
import proto.Message2Clients_pb2 as Message2Clients
import threading
import grpc
from queue import Queue




# 使用gRPC的异步来减少通信对于选手而言损失的时间，而gRPC的return值有result()方法，故若连接错误时也应当返回一个具有result()方法的对象，使用此处的ErrorHandler类来实现 # 使用gRPC的异步来减少通信对于选手而言损失的时间，而gRPC的return值有result()方法，故若连接错误时也应当返回一个具有result()方法的对象，使用此处的ErrorHandler类来实现
@@ -24,6 +22,8 @@ class Communication:
__haveNewMessage: bool __haveNewMessage: bool
__message2Client: Message2Clients.MessageToClient __message2Client: Message2Clients.MessageToClient
__messageQueue: Queue # Python的Queue是线程安全的，故无须自己实现Queue __messageQueue: Queue # Python的Queue是线程安全的，故无须自己实现Queue
__mtxMessage: threading.Lock
__cvMessage: threading.Condition


def __init__(self, sIP: str, sPort: str): def __init__(self, sIP: str, sPort: str):
aim = sIP + ':' + sPort aim = sIP + ':' + sPort
@@ -31,6 +31,7 @@ class Communication:
self.__THUAI6Stub = Services.AvailableServiceStub(channel) self.__THUAI6Stub = Services.AvailableServiceStub(channel)
self.__haveNewMessage = False self.__haveNewMessage = False
self.__messageQueue = Queue() self.__messageQueue = Queue()
self.__cvMessage = threading.Condition()


def Move(self, time: int, angle: float, playerID: int) -> bool: def Move(self, time: int, angle: float, playerID: int) -> bool:
try: try:
@@ -190,21 +191,22 @@ class Communication:
else: else:
return True return True


def HaveMessage2Client(self) -> bool:
return self.__haveNewMessage

def GetMessage2Client(self) -> Message2Clients.MessageToClient: def GetMessage2Client(self) -> Message2Clients.MessageToClient:
self.__haveNewMessage = False
return self.__message2Client
with self.__cvMessage:
self.__cvMessage.wait_for(lambda: self.__haveNewMessage)
self.__haveNewMessage = False
return self.__message2Client


def AddPlayer(self, playerID: int, playerType: THUAI6.PlayerType, humanType: THUAI6.HumanType, butcherType: THUAI6.ButcherType) -> None:
def AddPlayer(self, playerID: int) -> None:
def tMessage(): def tMessage():
try: try:
playerMsg = THUAI62Proto.THUAI62ProtobufPlayer( playerMsg = THUAI62Proto.THUAI62ProtobufPlayer(
playerID, playerType, humanType, butcherType)
playerID, Setting.playerType(), Setting.humanType(), Setting.butcherType())
for msg in self.__THUAI6Stub.AddPlayer(playerMsg): for msg in self.__THUAI6Stub.AddPlayer(playerMsg):
self.__haveNewMessage = True
self.__message2Client = msg
with self.__cvMessage:
self.__haveNewMessage = True
self.__message2Client = msg
self.__cvMessage.notify()
except grpc.RpcError as e: except grpc.RpcError as e:
return return



+ 250
- 0
PyAPI/PyAPI/DebugAPI.py View File

@@ -0,0 +1,250 @@
import PyAPI.structures as THUAI6
from PyAPI.Interface import ILogic, IHumanAPI, IButcherAPI, IGameTimer, IAI
from math import pi
from concurrent.futures import ProcessPoolExecutor, Future
from typing import List, Union


class HumanDebugAPI(IHumanAPI, IGameTimer):

def __init__(self, logic: ILogic, file: bool, screen: bool, warnOnly: bool, playerID: int) -> None:
self.__logic = logic
self.__pool = ProcessPoolExecutor(20)

# 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴

def Move(self, timeInMilliseconds: int, angle: float) -> Future[bool]:
return self.__pool.submit(self.__logic.Move, timeInMilliseconds, angle)

# 向特定方向移动

def MoveRight(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi * 0.5)

def MoveLeft(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi * 1.5)

def MoveUp(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, 0)

def MoveDown(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi)

# 道具和技能相关

def PickProp(self, propType: THUAI6.PropType) -> Future[bool]:
return self.__pool.submit(self.__logic.PickProp, propType)

def UseProp(self) -> Future[bool]:
return self.__pool.submit(self.__logic.UseProp)

def UseSkill(self) -> Future[bool]:
return self.__pool.submit(self.__logic.UseSkill)

# 消息相关，接收消息时无消息则返回(-1, '')

def SendMessage(self, toID: int, message: str) -> Future[bool]:
return self.__pool.submit(self.__logic.SendMessage, toID, message)

def HaveMessage(self) -> Future[bool]:
return self.__pool.submit(self.__logic.HaveMessage)

def GetMessage(self) -> Future[tuple[int, str]]:
return self.__pool.submit(self.__logic.GetMessage)

# 等待下一帧

def Wait(self) -> Future[bool]:
if self.__logic.GetCounter() == -1:
return self.__pool.submit(lambda: False)
else:
return self.__pool.submit(self.__logic.WaitThread)

# 获取各类游戏中的消息

def GetFrameCount(self) -> int:
return self.__logic.GetCounter()

def GetPlayerGUIDs(self) -> List[int]:
return self.__logic.GetPlayerGUIDs()

def GetButchers(self) -> List[THUAI6.Butcher]:
return self.__logic.GetButchers()

def GetHumans(self) -> List[THUAI6.Human]:
return self.__logic.GetHumans()

def GetProps(self) -> List[THUAI6.Prop]:
return self.__logic.GetProps()

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
return self.__logic.GetSelfInfo()

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
return self.__logic.GetFullMap()

def GetPlaceType(self, cellX: int, cellY: int) -> THUAI6.PlaceType:
return self.__logic.GetPlaceType(cellX, cellY)

# 用于DEBUG的输出函数，仅在DEBUG模式下有效

def PrintHuman(self) -> None:
pass

def PrintButcher(self) -> None:
pass

def PrintProp(self) -> None:
pass

def PrintSelfInfo(self) -> None:
pass

# 人类阵营的特殊函数

def Escape(self) -> Future[bool]:
return self.__pool.submit(self.__logic.Escape)

def StartFixMachine(self) -> Future[bool]:
return self.__pool.submit(self.__logic.StartFixMachine)

def EndFixMachine(self) -> Future[bool]:
return self.__pool.submit(self.__logic.EndFixMachine)

def StartSaveHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.StartSaveHuman)

def EndSaveHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.EndSaveHuman)

# Timer用

def StartTimer(self) -> None:
pass

def EndTimer(self) -> None:
pass

def Play(self, ai: IAI) -> None:
pass


class ButcherDebugAPI(IButcherAPI, IGameTimer):

def __init__(self, logic: ILogic, file: bool, screen: bool, warnOnly: bool, playerID: int) -> None:
self.__logic = logic
self.__pool = ProcessPoolExecutor(20)

# 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴

def Move(self, timeInMilliseconds: int, angle: float) -> Future[bool]:
return self.__pool.submit(self.__logic.Move, timeInMilliseconds, angle)

# 向特定方向移动

def MoveRight(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi * 0.5)

def MoveLeft(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi * 1.5)

def MoveUp(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, 0)

def MoveDown(self, timeInMilliseconds: int) -> Future[bool]:
return self.Move(timeInMilliseconds, pi)

# 道具和技能相关

def PickProp(self, propType: THUAI6.PropType) -> Future[bool]:
return self.__pool.submit(self.__logic.PickProp, propType)

def UseProp(self) -> Future[bool]:
return self.__pool.submit(self.__logic.UseProp)

def UseSkill(self) -> Future[bool]:
return self.__pool.submit(self.__logic.UseSkill)

# 消息相关，接收消息时无消息则返回(-1, '')

def SendMessage(self, toID: int, message: str) -> Future[bool]:
return self.__pool.submit(self.__logic.SendMessage, toID, message)

def HaveMessage(self) -> Future[bool]:
return self.__pool.submit(self.__logic.HaveMessage)

def GetMessage(self) -> Future[tuple[int, str]]:
return self.__pool.submit(self.__logic.GetMessage)

# 等待下一帧

def Wait(self) -> Future[bool]:
if self.__logic.GetCounter() == -1:
return self.__pool.submit(lambda: False)
else:
return self.__pool.submit(self.__logic.WaitThread)

# 获取各类游戏中的消息

def GetFrameCount(self) -> int:
return self.__logic.GetCounter()

def GetPlayerGUIDs(self) -> List[int]:
return self.__logic.GetPlayerGUIDs()

def GetButchers(self) -> List[THUAI6.Butcher]:
return self.__logic.GetButchers()

def GetHumans(self) -> List[THUAI6.Human]:
return self.__logic.GetHumans()

def GetProps(self) -> List[THUAI6.Prop]:
return self.__logic.GetProps()

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
return self.__logic.GetSelfInfo()

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
return self.__logic.GetFullMap()

def GetPlaceType(self, cellX: int, cellY: int) -> THUAI6.PlaceType:
return self.__logic.GetPlaceType(cellX, cellY)

# 用于DEBUG的输出函数，仅在DEBUG模式下有效

def PrintHuman(self) -> None:
pass

def PrintButcher(self) -> None:
pass

def PrintProp(self) -> None:
pass

def PrintSelfInfo(self) -> None:
pass

# 屠夫阵营的特殊函数

def Attack(self, angle: float) -> Future[bool]:
return self.__pool.submit(self.__logic.Attack, angle)

def CarryHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.CarryHuman)

def ReleaseHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.ReleaseHuman)

def HangHuman(self) -> Future[bool]:
return self.__pool.submit(self.__logic.HangHuman)

# Timer用

def StartTimer(self) -> None:
pass

def EndTimer(self) -> None:
pass

def Play(self, ai: IAI) -> None:
pass

PyAPI/API/Interface.py → PyAPI/PyAPI/Interface.py View File

@@ -1,6 +1,7 @@
from abc import abstractmethod, ABCMeta
from typing import List, Union from typing import List, Union
import structures as THUAI6
from concurrent.futures import Future
from abc import abstractmethod, ABCMeta
import PyAPI.structures as THUAI6




class ILogic(metaclass=ABCMeta): class ILogic(metaclass=ABCMeta):
@@ -112,65 +113,65 @@ class ILogic(metaclass=ABCMeta):
pass pass




class IAPI(metacls=ABCMeta):
class IAPI(metaclass=ABCMeta):


# 选手可执行的操作 # 选手可执行的操作
# 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴 # 指挥本角色进行移动，`timeInMilliseconds` 为移动时间，单位为毫秒；`angleInRadian` 表示移动的方向，单位是弧度，使用极坐标——竖直向下方向为 x 轴，水平向右方向为 y 轴


@abstractmethod @abstractmethod
def Move(self, timeInMilliseconds: int, angle: float) -> bool:
def Move(self, timeInMilliseconds: int, angle: float) -> Future[bool]:
pass pass


# 向特定方向移动 # 向特定方向移动


@abstractmethod @abstractmethod
def MoveRight(self, timeInMilliseconds: int) -> bool:
def MoveRight(self, timeInMilliseconds: int) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def MoveLeft(self, timeInMilliseconds: int) -> bool:
def MoveLeft(self, timeInMilliseconds: int) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def MoveUp(self, timeInMilliseconds: int) -> bool:
def MoveUp(self, timeInMilliseconds: int) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def MoveDown(self, timeInMilliseconds: int) -> bool:
def MoveDown(self, timeInMilliseconds: int) -> Future[bool]:
pass pass


# 道具和技能相关 # 道具和技能相关


@abstractmethod @abstractmethod
def PickProp(self, propType: THUAI6.PropType) -> bool:
def PickProp(self, propType: THUAI6.PropType) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def UseProp(self) -> bool:
def UseProp(self) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def UseSkill(self) -> bool:
def UseSkill(self) -> Future[bool]:
pass pass


# 消息相关，接收消息时无消息则返回(-1, '') # 消息相关，接收消息时无消息则返回(-1, '')


@abstractmethod @abstractmethod
def SendMessage(self, toID: int, message: str) -> bool:
def SendMessage(self, toID: int, message: str) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def HaveMessage(self) -> bool:
def HaveMessage(self) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def GetMessage(self) -> tuple[int, str]:
def GetMessage(self) -> Future[tuple[int, str]]:
pass pass


# 等待下一帧 # 等待下一帧


@abstractmethod @abstractmethod
def Wait(self) -> bool:
def Wait(self) -> Future[bool]:
pass pass


# 获取各类游戏中的消息 # 获取各类游戏中的消息
@@ -231,23 +232,23 @@ class IHumanAPI(IAPI, metaclass=ABCMeta):
# 人类阵营的特殊函数 # 人类阵营的特殊函数


@abstractmethod @abstractmethod
def Escape(self) -> bool:
def Escape(self) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def StartFixMachine(self) -> bool:
def StartFixMachine(self) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def EndFixMachine(self) -> bool:
def EndFixMachine(self) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def StartSaveHuman(self) -> bool:
def StartSaveHuman(self) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def EndSaveHuman(self) -> bool:
def EndSaveHuman(self) -> Future[bool]:
pass pass




@@ -256,19 +257,19 @@ class IButcherAPI(IAPI, metaclass=ABCMeta):
# 屠夫阵营的特殊函数 # 屠夫阵营的特殊函数


@abstractmethod @abstractmethod
def Attack(self, angle: float) -> bool:
def Attack(self, angle: float) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def CarryHuman(self) -> bool:
def CarryHuman(self) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def ReleaseHuman(self) -> bool:
def ReleaseHuman(self) -> Future[bool]:
pass pass


@abstractmethod @abstractmethod
def HangHuman(self) -> bool:
def HangHuman(self) -> Future[bool]:
pass pass





PyAPI/API/State.py → PyAPI/PyAPI/State.py View File

@@ -1,5 +1,5 @@
from typing import List, Union from typing import List, Union
import structures as THUAI6
import PyAPI.structures as THUAI6




class State: class State:

+ 374
- 0
PyAPI/PyAPI/logic.py View File

@@ -0,0 +1,374 @@
import os
from abc import abstractmethod
from typing import List, Union, Callable
import threading
import logging
import proto.MessageType_pb2 as MessageType
import proto.Message2Server_pb2 as Message2Server
import proto.Message2Clients_pb2 as Message2Clients
import PyAPI.structures as THUAI6
from PyAPI.utils import Proto2THUAI6, AssistFunction
from PyAPI.DebugAPI import HumanDebugAPI, ButcherDebugAPI
from PyAPI.API import HumanAPI, ButcherAPI
from PyAPI.AI import Setting
from PyAPI.Communication import Communication
from PyAPI.State import State
from PyAPI.Interface import ILogic, IGameTimer


class Logic(ILogic):
def __init__(self, playerID: int) -> None:

# ID
self.__playerID: int = playerID
self.__playerGUIDs: List[int] = []

# 通信
self.__comm: Communication

# 存储状态
self.__currentState: State
self.__bufferState: State

# timer，用于实际运行AI
self.__timer: IGameTimer

# AI线程
self.__threadAI: threading.Thread

# 互斥锁
self.__mtxState: threading.Lock = threading.Lock()

# 条件变量
self.__cvBuffer: threading.Condition = threading.Condition()
self.__cvAI: threading.Condition = threading.Condition()

# 保存缓冲区数
self.__counterState: int
self.__counterBuffer: int

# 记录游戏状态
self.__gameState: THUAI6.GameState = THUAI6.GameState.NullGameState

# 是否应该执行player()
self.__AILoop: bool = True

# buffer是否更新完毕
self.__bufferUpdated: bool = False

# 是否应当启动AI
self.__AIStart: bool = False

# asynchronous为True时控制内容更新的变量
self.__freshed: bool = False

self.__logger: logging.Logger = logging.getLogger("logic")

# IAPI统一可用的接口

def GetButchers(self) -> List[THUAI6.Butcher]:
with self.__mtxState:
return self.__currentState.butchers

def GetHumans(self) -> List[THUAI6.Human]:
with self.__mtxState:
return self.__currentState.humans

def GetProps(self) -> List[THUAI6.Prop]:
with self.__mtxState:
return self.__currentState.props

def GetSelfInfo(self) -> Union[THUAI6.Human, THUAI6.Butcher]:
with self.__mtxState:
return self.__currentState.self

def GetFullMap(self) -> List[List[THUAI6.PlaceType]]:
with self.__mtxState:
return self.__currentState.map

def GetPlaceType(self, x: int, y: int) -> THUAI6.PlaceType:
with self.__mtxState:
return self.__currentState.map[x][y]

def Move(self, time: int, angle: float) -> bool:
return self.__comm.Move(time, angle, self.__playerID)

def PickProp(self, propType: THUAI6.PropType) -> bool:
return self.__comm.PickProp(propType, self.__playerID)

def UseProp(self) -> bool:
return self.__comm.UseProp(self.__playerID)

def UseSkill(self) -> bool:
return self.__comm.UseSkill(self.__playerID)

def SendMessage(self, toID: int, message: str) -> bool:
return self.__comm.SendMessage(toID, message, self.__playerID)

def HaveMessage(self) -> bool:
return self.__comm.HaveMessage()

def GetMessage(self) -> tuple[int, str]:
return self.__comm.GetMessage()

def WaitThread(self) -> bool:
self.__Update()
return True

def GetCounter(self) -> int:
with self.__mtxState:
return self.__counterState

def GetPlayerGUIDs(self) -> List[int]:
with self.__mtxState:
return self.__playerGUIDs

# IHumanAPI使用的接口

def Escape(self) -> bool:
return self.__comm.Escape(self.__playerID)

def StartFixMachine(self) -> bool:
return self.__comm.StartFixMachine(self.__playerID)

def EndFixMachine(self) -> bool:
return self.__comm.EndFixMachine(self.__playerID)

def StartSaveHuman(self) -> bool:
return self.__comm.StartSaveHuman(self.__playerID)

def EndSaveHuman(self) -> bool:
return self.__comm.EndSaveHuman(self.__playerID)

# Butcher使用的接口

def Attack(self, angle: float) -> bool:
return self.__comm.Attack(angle, self.__playerID)

def CarryHuman(self) -> bool:
return self.__comm.CarryHuman(self.__playerID)

def ReleaseHuman(self) -> bool:
return self.__comm.ReleaseHuman(self.__playerID)

def HangHuman(self) -> bool:
return self.__comm.HangHuman(self.__playerID)

# Logic内部逻辑
def __TryConnection(self) -> bool:
return self.__comm.TryConnection(self.__playerID)

def __ProcessMessage(self) -> None:
def messageThread():
self.__comm.AddPlayer(self.__playerID)
self.__comm.ReadMessage(self.__playerID)

while self.__gameState != THUAI6.GameState.GameEnd:
# 读取消息，无消息时此处阻塞
clientMsg = self.__comm.GetMessage2Client()
self.__gameState = Proto2THUAI6.gameStateDict[
clientMsg.game_state]

if self.__gameState == THUAI6.GameState.GameStart:
# 读取玩家的GUID
self.__playerGUIDs.clear()
for human in clientMsg.human_message:
self.__playerGUIDs.append(human.guid)
for butcher in clientMsg.butcher_message:
self.__playerGUIDs.append(butcher.guid)
self.__currentState.guids = self.__playerGUIDs
self.__bufferState.guids = self.__playerGUIDs

self.__LoadBuffer(clientMsg)
self.__AILoop = True
self.__UnBlockAI()

elif self.__gameState == THUAI6.GameState.GameRunning:
# 读取玩家的GUID
self.__playerGUIDs.clear()
for human in clientMsg.human_message:
self.__playerGUIDs.append(human.guid)
for butcher in clientMsg.butcher_message:
self.__playerGUIDs.append(butcher.guid)
self.__currentState.guids = self.__playerGUIDs
self.__bufferState.guids = self.__playerGUIDs
self.__LoadBuffer(clientMsg)
else:
continue
self.__AILoop = False
with self.__cvBuffer:
self.__bufferUpdated = True
self.__counterBuffer = -1
self.__cvBuffer.notify()

threading.Thread(target=messageThread).start()

def __LoadBuffer(self, message: Message2Clients.MessageToClient) -> None:
with self.__cvBuffer:
self.__bufferState.humans.clear()
self.__bufferState.butchers.clear()
self.__bufferState.props.clear()
self.__bufferState.map = Proto2THUAI6.Protobuf2THUAI6Map(
message.map_message)
if Setting.playerType() == THUAI6.PlayerType.HumanPlayer:
for human in message.human_message:
if human.player_id == self.__playerID:
self.__bufferState.self = Proto2THUAI6.Protobuf2THUAI6Human(
human)
self.__bufferState.humans.append(
Proto2THUAI6.Protobuf2THUAI6Human(human))
for butcher in message.butcher_message:
viewRange: int = self.__bufferState.self.viewRange
deltaX: int = butcher.x - self.__bufferState.self.x
deltaY: int = butcher.y - self.__bufferState.self.y
if deltaX * deltaX + deltaY * deltaY <= viewRange * viewRange:
divide: int = max(abs(deltaX), abs(deltaY)) // 100
dx: float = deltaX / divide
dy: float = deltaY / divide
selfX: float = self.__bufferState.self.x
selfY: float = self.__bufferState.self.y
for i in range(divide):
selfX += dx
selfY += dy
if self.__bufferState.map[AssistFunction.GridToCell(int(selfX))][AssistFunction.GridToCell(int(selfY))]:
break
else:
self.__bufferState.butchers.append(
Proto2THUAI6.Protobuf2THUAI6Butcher(butcher))
else:
for butcher in message.butcher_message:
if butcher.player_id == self.__playerID:
self.__bufferState.self = Proto2THUAI6.Protobuf2THUAI6Butcher(
butcher)
self.__bufferState.butchers.append(
Proto2THUAI6.Protobuf2THUAI6Butcher(butcher))
for human in message.human_message:
viewRange: int = self.__bufferState.self.viewRange
deltaX: int = human.x - self.__bufferState.self.x
deltaY: int = human.y - self.__bufferState.self.y
if deltaX * deltaX + deltaY * deltaY <= viewRange * viewRange:
divide: int = max(abs(deltaX), abs(deltaY)) // 100
dx: float = deltaX / divide
dy: float = deltaY / divide
selfX: float = self.__bufferState.self.x
selfY: float = self.__bufferState.self.y
for i in range(divide):
selfX += dx
selfY += dy
if self.__bufferState.map[AssistFunction.GridToCell(int(selfX))][AssistFunction.GridToCell(int(selfY))]:
break
else:
self.__bufferState.humans.append(
Proto2THUAI6.Protobuf2THUAI6Human(human))
for prop in message.prop_message:
self.__bufferState.props.append(
Proto2THUAI6.Protobuf2THUAI6Prop(prop))
if Setting.asynchronous():
with self.__mtxState:
self.__currentState, self.__bufferState = self.__bufferState, self.__currentState
self.__freshed = True
else:
self.__bufferUpdated = True
self.__counterBuffer += 1
self.__cvBuffer.notify()

def __UnBlockBuffer(self) -> None:
with self.__cvBuffer:
self.__bufferUpdated = True
self.__cvBuffer.notify()

def __UnBlockAI(self) -> None:
with self.__cvAI:
self.__AIStart = True
self.__cvAI.notify()

def __Update(self) -> None:
if not Setting.asynchronous():
with self.__cvBuffer:
self.__cvBuffer.wait_for(lambda: self.__bufferUpdated)
self.__bufferState, self.__currentState = self.__currentState, self.__bufferState
self.__bufferUpdated = False
self.__counterState = self.__counterBuffer

def __Wait(self) -> None:
self.__freshed = False
with self.__cvBuffer:
self.__cvBuffer.wait_for(lambda: self.__freshed)

def Main(self, createAI: Callable, IP: str, port: str, file: bool, screen: bool, warnOnly: bool) -> None:

# 建立日志组件
self.__logger.setLevel(logging.DEBUG)
formatter = logging.Formatter(
"[%(asctime)s][%(name)s][%(levelname)s] %(message)s", "%H:%M:%S.%e")
# 确保文件存在
if not os.path.exists(os.path.dirname(os.path.dirname(os.path.realpath(__file__))) + "/logs"):
os.makedirs(os.path.dirname(os.path.dirname(
os.path.realpath(__file__))) + "/logs")

fileHandler = logging.FileHandler(os.path.dirname(
os.path.dirname(os.path.realpath(__file__))) + "/logs/logic-log.txt")
screenHandler = logging.StreamHandler()
if file:
fileHandler.setLevel(logging.DEBUG)
fileHandler.setFormatter(formatter)
self.__logger.addHandler(fileHandler)
if screen:
if warnOnly:
screenHandler.setLevel(logging.WARNING)
else:
screenHandler.setLevel(logging.INFO)
screenHandler.setFormatter(formatter)
self.__logger.addHandler(screenHandler)

self.__logger.info("*********Basic Info*********")
self.__logger.info("asynchronous: %s", Setting.asynchronous())
self.__logger.info("server: %s:%s", IP, port)
self.__logger.info("playerID: %s", self.__playerID)
self.__logger.info("player type: %s",
THUAI6.playerTypeDict[Setting.playerType()])
self.__logger.info("****************************")

# 建立通信组件
self.__comm = Communication(IP, port)

# 构造timer
if Setting.playerType() == THUAI6.PlayerType.HumanPlayer:
if not file and not screen:
self.__timer = HumanAPI(self)
else:
self.__timer = HumanDebugAPI(
self, file, screen, warnOnly, self.__playerID)
elif Setting.playerType() == THUAI6.PlayerType.ButcherPlayer:
if not file and not screen:
self.__timer = ButcherAPI(self)
else:
self.__timer = ButcherDebugAPI(
self, file, screen, warnOnly, self.__playerID)

# 构建AI线程
def AIThread():
with self.__cvAI:
self.__cvAI.wait_for(lambda: self.__AIStart)

ai = createAI()
while self.__AILoop:
if Setting.asynchronous():
self.__Wait()
self.__timer.StartTimer()
self.__timer.Play(ai)
self.__timer.EndTimer()
else:
self.__Update()
self.__timer.StartTimer()
self.__timer.Play(ai)
self.__timer.EndTimer()

if self.__TryConnection():
self.__threadAI = threading.Thread(target=AIThread)
self.__threadAI.start()
self.__ProcessMessage()
self.__threadAI.join()
else:
self.__AILoop = False
return

+ 30
- 0
PyAPI/PyAPI/main.py View File

@@ -0,0 +1,30 @@
import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.realpath(__file__))))
sys.path.append(os.path.dirname(os.path.dirname(
os.path.realpath(__file__))) + '/proto')

from PyAPI.Interface import IAI
from PyAPI.AI import AI
from PyAPI.logic import Logic
from typing import List, Callable


def THUAI6Main(argv: List[str], AIBuilder: Callable) -> None:
pID: int = 114514
sIP: str = "114.51.41.91"
sPort: str = "9810"
file: bool = True
screen: bool = True
warnOnly: bool = False
logic = Logic(pID)
print("Welcome to THUAI6")
logic.Main(AIBuilder, sIP, sPort, file, screen, warnOnly)


def CreateAI() -> IAI:
return AI()


if __name__ == '__main__':
THUAI6Main(sys.argv, CreateAI)

+ 159
- 0
PyAPI/PyAPI/structures.py View File

@@ -0,0 +1,159 @@
from enum import Enum
from typing import List, Dict


class GameState(Enum):
NullGameState = 0
GameStart = 1
GameRunning = 2
GameEnd = 3


class PlaceType(Enum):
NullPlaceType = 0
Land = 1
Wall = 2
Grass = 3
Machine = 4
Gate = 5
HiddenGate = 6


class ShapeType(Enum):
NullShapeType = 0
Square = 1
Circle = 2


class PlayerType(Enum):
NullPlayerType = 0
HumanPlayer = 1
ButcherPlayer = 2


class PropType(Enum):
NullPropType = 0
PropType1 = 1


class HumanType(Enum):
NullHumanType = 0
HumanType1 = 1


class ButcherType(Enum):
NullButcherType = 0
ButcherType1 = 1


class HumanBuffType(Enum):
NullHumanBuffType = 0
HumanBuffType1 = 1


class ButcherBuffType(Enum):
NullButcherBuffType = 0
ButcherBuffType1 = 1


class HumanState(Enum):
NullHumanState = 0
Idle = 1
Fixing = 2
Dying = 3
OnChair = 4
Dead = 5


class Player:
x: int
y: int
speed: int
viewRange: int
playerID: int
guid: int
radius: int
timeUntilSkillAvailable: float
playerType: PlayerType
prop: PropType
place: PlaceType


class Human(Player):
state: HumanState
life: int
hangedTime: int
humanType: HumanType
buff: List[HumanBuffType]


class Butcher(Player):
damage: int
movable: bool
butcherType: ButcherType
buff: List[ButcherBuffType]


class Prop:
x: int
y: int
size: int
guid: int
type: PropType
place: PlaceType
facingDirection: float
isMoving: bool


gameStateDict: Dict[GameState, str] = {
GameState.NullGameState: "NullGameState",
GameState.GameStart: "GameStart",
GameState.GameRunning: "GameRunning",
GameState.GameEnd: "GameEnd"}

humanStateDict: Dict[HumanState, str] = {
HumanState.NullHumanState: "NullHumanState",
HumanState.Idle: "Idle",
HumanState.Fixing: "Fixing",
HumanState.Dying: "Dying",
HumanState.OnChair: "OnChair",
HumanState.Dead: "Dead"}

playerTypeDict: Dict[PlayerType, str] = {
PlayerType.NullPlayerType: "NullPlayerType",
PlayerType.HumanPlayer: "HumanPlayer",
PlayerType.ButcherPlayer: "ButcherPlayer"}

propTypeDict: Dict[PropType, str] = {
PropType.NullPropType: "NullPropType",
PropType.PropType1: "PropType1"}

humanTypeDict: Dict[HumanType, str] = {
HumanType.NullHumanType: "NullHumanType",
HumanType.HumanType1: "HumanType1"}

butcherTypeDict: Dict[ButcherType, str] = {
ButcherType.NullButcherType: "NullButcherType",
ButcherType.ButcherType1: "ButcherType1"}

humanBuffTypeDict: Dict[HumanBuffType, str] = {
HumanBuffType.NullHumanBuffType: "NullHumanBuffType",
HumanBuffType.HumanBuffType1: "HumanBuffType1"}

butcherBuffTypeDict: Dict[ButcherBuffType, str] = {
ButcherBuffType.NullButcherBuffType: "NullButcherBuffType",
ButcherBuffType.ButcherBuffType1: "ButcherBuffType1"}

placeTypeDict: Dict[PlaceType, str] = {
PlaceType.NullPlaceType: "NullPlaceType",
PlaceType.Land: "Land",
PlaceType.Wall: "Wall",
PlaceType.Grass: "Grass",
PlaceType.Machine: "Machine",
PlaceType.Gate: "Gate",
PlaceType.HiddenGate: "HiddenGate"}

shapeTypeDict: Dict[ShapeType, str] = {
ShapeType.NullShapeType: "NullShapeType",
ShapeType.Square: "Square",
ShapeType.Circle: "Circle"}

PyAPI/API/utils.py → PyAPI/PyAPI/utils.py View File

@@ -1,15 +1,16 @@
import structures as THUAI6
from typing import Final, List
import proto.Message2Clients_pb2 as Message2Clients
import proto.Message2Server_pb2 as Message2Server
import proto.MessageType_pb2 as MessageType import proto.MessageType_pb2 as MessageType
import proto.Message2Server_pb2 as Message2Server
import proto.Message2Clients_pb2 as Message2Clients
import PyAPI.structures as THUAI6
from typing import Final, List



numOfGridPerCell: Final[int] = 1000 numOfGridPerCell: Final[int] = 1000




# 起到NameSpace的作用 # 起到NameSpace的作用
class NoInstance: class NoInstance:
def __call__(self, *args, **kwargs):
def __call__(self):
raise TypeError("This class cannot be instantiated.") raise TypeError("This class cannot be instantiated.")




@@ -17,11 +18,11 @@ class AssistFunction(NoInstance):
# 辅助函数 # 辅助函数
@staticmethod @staticmethod
def CellToGrid(cell: int) -> int: def CellToGrid(cell: int) -> int:
return cell*numOfGridPerCell+numOfGridPerCell//2
return cell * numOfGridPerCell + numOfGridPerCell // 2


@staticmethod @staticmethod
def GridToCell(grid: int) -> int: def GridToCell(grid: int) -> int:
return grid//numOfGridPerCell
return grid // numOfGridPerCell




class Proto2THUAI6(NoInstance): class Proto2THUAI6(NoInstance):

+ 6
- 0
PyAPI/logs/logic-log.txt View File

@@ -0,0 +1,6 @@
[21:02:47.26][logic][INFO] *********Basic Info*********
[21:02:47.26][logic][INFO] asynchronous: False
[21:02:47.26][logic][INFO] server: 114.51.41.91:9810
[21:02:47.26][logic][INFO] playerID: 114514
[21:02:47.26][logic][INFO] player type: HumanPlayer
[21:02:47.26][logic][INFO] ****************************

Write Preview
Loading…
Cancel
Save