dev/valconomy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

controller: Implement economy suggestions and add test

Browse Source
This commit is contained in:
Jack O'Sullivan
2024-12-12 23:07:29 +00:00
parent bb11a1607b
commit 0607f08b83
5 changed files with 288 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
controller/.gitignore vendored
View File

@@ -2,3 +2,4 @@ __pycache__/
/build/
/dist/
Valconomy.spec
.pytest_cache/

1
controller/pyproject.toml
View File

@@ -12,4 +12,5 @@ dependencies = [
[dependency-groups]
dev = [
"pyinstaller>=6.11.1",
"pytest>=8.3.4",
]

170
controller/test_sm.py Normal file
View File

@@ -0,0 +1,170 @@
import valconomy
from valconomy import ValorantPlayerInfo, RiotPlayerInfo, EconomyDecision
class DummyValHandler:
def __init__(self):
self.seq = []
def none(self):
self.seq.append('none')
def menu(self, info: RiotPlayerInfo, was_idle: bool=False):
self.seq.append(('menu', was_idle))
def idle(self, info: RiotPlayerInfo):
self.seq.append('idle')
def queue_start(self, info: RiotPlayerInfo):
self.seq.append(('queue_start', info.valorant.is_party_owner and info.valorant.queue_type == 'unrated'))
def match_found(self, info: RiotPlayerInfo):
self.seq.append(('match_found', info.valorant.queue_type == 'premier-seasonmatch'))
def pregame(self, info: RiotPlayerInfo):
self.seq.append(('pregame', info.valorant.map == '/Game/Maps/Bonsai/Bonsai'))
def game_generic(self, info: RiotPlayerInfo):
t = info.valorant.queue_type
if info.valorant.max_party_size == 12:
t = 'custom'
elif not t:
t = 'unknown'
self.seq.append(('game_generic', t))
def game_start(self, info: RiotPlayerInfo):
self.seq.append('game_start')
def round_start(self, info: RiotPlayerInfo, won: bool, economy: EconomyDecision):
self.seq.append(('round_start', won, info.valorant.score, info.valorant.enemy_score, economy))
def game_over(self, info: RiotPlayerInfo, won: bool):
self.seq.append(('game_over', won))
class TestSm:
def setup_method(self, method):
self.mock = DummyValHandler()
self.sm = valconomy.ValconomyStateMachine('dummy', self.mock)
def do(self, **kwargs):
gone = kwargs.pop('gone', False)
info = RiotPlayerInfo(kwargs.pop('uuid', 'dummy'), 'Team Player', 'gamer', 'unknown')
info.valorant = ValorantPlayerInfo(**kwargs)
self.sm.handle_presence(info, gone)
def test_comp_normal(self):
self.do(game_state='MENUS')
self.do(game_state='MENUS', queue_type='competitive')
for _ in range(2):
self.do(game_state='MENUS', party_state='MATCHMAKING', queue_type='competitive')
for _ in range(3):
self.do(game_state='MENUS', party_state='MATCHMADE_GAME_STARTING', queue_type='competitive')
self.do(queue_type='competitive', game_state='PREGAME', map='/Game/Maps/Bonsai/Bonsai')
self.do(queue_type='competitive', game_state='INGAME', map='/Game/Maps/Bonsai/Bonsai')
for _ in range(2):
self.do(queue_type='competitive', game_state='INGAME', map='/Game/Maps/Bonsai/Bonsai', score=1, enemy_score=0)
self.do(queue_type='competitive', game_state='INGAME', map='/Game/Maps/Bonsai/Bonsai', score=2, enemy_score=0)
for i in range(14):
self.do(queue_type='competitive', game_state='INGAME', map='/Game/Maps/Bonsai/Bonsai', score=2, enemy_score=i)
self.do(queue_type='competitive', game_state='INGAME', map='/Game/Maps/Bonsai/Bonsai', score=0, enemy_score=0)
assert self.mock.seq == [
('menu', False),
('queue_start', False),
('match_found', False),
('pregame', True),
'game_start',
('round_start', None, 0, 0, EconomyDecision.BUY),
('round_start', True, 1, 0, EconomyDecision.BUY),
('round_start', True, 2, 0, EconomyDecision.BONUS),
('round_start', False, 2, 1, EconomyDecision.MATCH_TEAM),
] + [
('round_start', False, 2, 2 + i, EconomyDecision.MATCH_TEAM) for i in range(8)] + [
('round_start', False, 2, 10, EconomyDecision.BUY),
('round_start', False, 2, 11, EconomyDecision.SAVE),
('round_start', False, 2, 12, EconomyDecision.BUY),
('game_over', False)
]
def test_comp_overtime(self):
for _ in range(2):
self.do(queue_type='competitive', game_state='INGAME')
self.do(queue_type='competitive', game_state='INGAME', score=1)
for i in range(13):
self.do(queue_type='competitive', game_state='INGAME', score=1, enemy_score=i)
for i in range(12):
self.do(queue_type='competitive', game_state='INGAME', score=1 + i, enemy_score=12)
self.do(queue_type='competitive', game_state='INGAME', score=13, enemy_score=12)
self.do(queue_type='competitive', game_state='INGAME', score=13, enemy_score=13)
self.do(queue_type='competitive', game_state='INGAME', score=13, enemy_score=14)
self.do(queue_type='competitive', game_state='INGAME', score=14, enemy_score=14)
self.do(queue_type='competitive', game_state='INGAME', score=15, enemy_score=14)
self.do(queue_type='competitive', game_state='INGAME', score=16, enemy_score=14)
self.do(queue_type='competitive', game_state='INGAME')
assert self.mock.seq == [
'game_start',
('round_start', None, 0, 0, EconomyDecision.BUY),
('round_start', True, 1, 0, EconomyDecision.BUY),
('round_start', False, 1, 1, EconomyDecision.SAVE),
] + [
('round_start', False, 1, 2 + i, EconomyDecision.MATCH_TEAM) for i in range(9)] + [
('round_start', False, 1, 11, EconomyDecision.BUY),
('round_start', False, 1, 12, EconomyDecision.BUY),
] + [
('round_start', True, 2 + i, 12, EconomyDecision.BUY) for i in range(12)] + [
('round_start', False, 13, 13, EconomyDecision.BUY),
('round_start', False, 13, 14, EconomyDecision.BUY),
('round_start', True, 14, 14, EconomyDecision.BUY),
('round_start', True, 15, 14, EconomyDecision.BUY),
('game_over', True),
]
def test_swiftplay(self):
self.do(game_state='MENUS')
self.do(game_state='MENUS', queue_type='swiftplay')
self.do(game_state='MENUS', party_state='MATCHMAKING', queue_type='swiftplay')
self.do(game_state='MENUS', party_state='MATCHMADE_GAME_STARTING', queue_type='swiftplay')
self.do(queue_type='swiftplay', game_state='PREGAME')
self.do(queue_type='swiftplay', game_state='INGAME')
for i in range(5):
self.do(queue_type='swiftplay', game_state='INGAME', enemy_score=1 + i)
self.do(queue_type='swiftplay', game_state='INGAME')
assert self.mock.seq == [
('menu', False),
('queue_start', False),
('match_found', False),
('pregame', False),
'game_start',
('round_start', None, 0, 0, EconomyDecision.BUY),
] + [
('round_start', False, 0, 1 + i, EconomyDecision.BUY) for i in range(4)] + [
('game_over', False),
]
def test_tdm(self):
self.do(game_state='MENUS')
self.do(game_state='MENUS', queue_type='hurm')
self.do(game_state='MENUS', party_state='MATCHMAKING', queue_type='hurm')
self.do(game_state='MENUS', party_state='MATCHMADE_GAME_STARTING', queue_type='hurm')
self.do(queue_type='hurm', game_state='PREGAME')
self.do(queue_type='hurm', game_state='INGAME')
for i in range(100):
self.do(queue_type='hurm', game_state='INGAME', enemy_score=1 + i)
self.do(queue_type='hurm', game_state='INGAME')
assert self.mock.seq == [
('menu', False),
('queue_start', False),
('match_found', False),
('game_generic', 'hurm'),
]

48
controller/uv.lock generated
View File

@@ -43,6 +43,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/bf/9b/08c0432272d77b04803958a4598a51e2a4b51c06640af8b8f0f908c18bf2/charset_normalizer-3.4.0-py3-none-any.whl", hash = "sha256:fe9f97feb71aa9896b81973a7bbada8c49501dc73e58a10fcef6663af95e5079", size = 49446 },
]
[[package]]
name = "colorama"
version = "0.4.6"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d8/53/6f443c9a4a8358a93a6792e2acffb9d9d5cb0a5cfd8802644b7b1c9a02e4/colorama-0.4.6.tar.gz", hash = "sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44", size = 27697 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335 },
]
[[package]]
name = "hidapi"
version = "0.14.0.post4"
@@ -85,6 +94,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e5/de/e97c6111025d5a95753bcdab2db30846e1a9eb3cc57aafec98c088b1f649/infi.systray-0.1.12-py3-none-any.whl", hash = "sha256:4041ab3693f7f00a6b2b2d7b553bf60fce8941708b83e32261ca40926952ed0d", size = 9234 },
]
[[package]]
name = "iniconfig"
version = "2.0.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d7/4b/cbd8e699e64a6f16ca3a8220661b5f83792b3017d0f79807cb8708d33913/iniconfig-2.0.0.tar.gz", hash = "sha256:2d91e135bf72d31a410b17c16da610a82cb55f6b0477d1a902134b24a455b8b3", size = 4646 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ef/a6/62565a6e1cf69e10f5727360368e451d4b7f58beeac6173dc9db836a5b46/iniconfig-2.0.0-py3-none-any.whl", hash = "sha256:b6a85871a79d2e3b22d2d1b94ac2824226a63c6b741c88f7ae975f18b6778374", size = 5892 },
]
[[package]]
name = "macholib"
version = "1.16.3"
@@ -115,6 +133,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/55/26/d0ad8b448476d0a1e8d3ea5622dc77b916db84c6aa3cb1e1c0965af948fc/pefile-2023.2.7-py3-none-any.whl", hash = "sha256:da185cd2af68c08a6cd4481f7325ed600a88f6a813bad9dea07ab3ef73d8d8d6", size = 71791 },
]
[[package]]
name = "pluggy"
version = "1.5.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/96/2d/02d4312c973c6050a18b314a5ad0b3210edb65a906f868e31c111dede4a6/pluggy-1.5.0.tar.gz", hash = "sha256:2cffa88e94fdc978c4c574f15f9e59b7f4201d439195c3715ca9e2486f1d0cf1", size = 67955 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/88/5f/e351af9a41f866ac3f1fac4ca0613908d9a41741cfcf2228f4ad853b697d/pluggy-1.5.0-py3-none-any.whl", hash = "sha256:44e1ad92c8ca002de6377e165f3e0f1be63266ab4d554740532335b9d75ea669", size = 20556 },
]
[[package]]
name = "pyinstaller"
version = "6.11.1"
@@ -156,6 +183,21 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/a9/64/445861ee7a5fd32874c0f6cfe8222aacc8feda22539332e0d8ff50dadec6/pyinstaller_hooks_contrib-2024.10-py3-none-any.whl", hash = "sha256:ad47db0e153683b4151e10d231cb91f2d93c85079e78d76d9e0f57ac6c8a5e10", size = 338417 },
]
[[package]]
name = "pytest"
version = "8.3.4"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
{ name = "iniconfig" },
{ name = "packaging" },
{ name = "pluggy" },
]
sdist = { url = "https://files.pythonhosted.org/packages/05/35/30e0d83068951d90a01852cb1cef56e5d8a09d20c7f511634cc2f7e0372a/pytest-8.3.4.tar.gz", hash = "sha256:965370d062bce11e73868e0335abac31b4d3de0e82f4007408d242b4f8610761", size = 1445919 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/11/92/76a1c94d3afee238333bc0a42b82935dd8f9cf8ce9e336ff87ee14d9e1cf/pytest-8.3.4-py3-none-any.whl", hash = "sha256:50e16d954148559c9a74109af1eaf0c945ba2d8f30f0a3d3335edde19788b6f6", size = 343083 },
]
[[package]]
name = "pywin32-ctypes"
version = "0.2.3"
@@ -211,6 +253,7 @@ dependencies = [
[package.dev-dependencies]
dev = [
{ name = "pyinstaller" },
{ name = "pytest" },
]
[package.metadata]
@@ -221,4 +264,7 @@ requires-dist = [
]
[package.metadata.requires-dev]
dev = [{ name = "pyinstaller", specifier = ">=6.11.1" }]
dev = [
{ name = "pyinstaller", specifier = ">=6.11.1" },
{ name = "pytest", specifier = ">=8.3.4" },
]

85
controller/valconomy.py
View File

@@ -23,9 +23,9 @@ log = logging.getLogger('valconomy')
class ValorantPlayerInfo:
is_idle: bool = False
is_party_owner: bool = False
max_party_size: int = 0
party_size: int = 0
is_party_owner: bool = True
max_party_size: int = 5
party_size: int = 1
party_state: str = None
queue_type: str = None
@@ -144,7 +144,8 @@ class ValorantLocalClient:
class EconomyDecision(Enum):
BUY = 0
SAVE = 1
MATCH_TEAM = 2
BONUS = 2
MATCH_TEAM = 3
class ValconomyHandler:
# Valorant isn't open
@@ -204,12 +205,19 @@ class ValconomyHandler:
log.info('OK best of luck!')
# Round started
def round_start(self, info: RiotPlayerInfo, economy: EconomyDecision):
def round_start(self, info: RiotPlayerInfo, won: bool, economy: EconomyDecision):
msg = f'Start round {info.valorant.score}-{info.valorant.enemy_score}'
if won is not None:
msg += f' ({"won" if won else "lost"})'
log.info(msg)
match economy:
case EconomyDecision.BUY:
log.info('You should buy!')
case EconomyDecision.SAVE:
log.info('Time to save...')
case EconomyDecision.BONUS:
log.info('Keep your gun from last round!')
case EconomyDecision.MATCH_TEAM:
log.info('Follow the team economy!')
@@ -229,9 +237,10 @@ class GameState(Enum):
IN_QUEUE = 3
GAME_FOUND = 4
PRE_GAME = 5
IN_GAME = 6
GAME_OVER = 7
IN_GAME_GENERIC = 8
GAME_START = 6
IN_GAME = 7
GAME_OVER = 8
IN_GAME_GENERIC = 9
class ValconomyStateMachine:
def __init__(self, player_uuid: str, handler: ValconomyHandler):
@@ -241,23 +250,61 @@ class ValconomyStateMachine:
self.last = None
self.score = -1
self.score_enemy = -1
self.round_history = []
def handle_state(self, s: GameState, p: RiotPlayerInfo):
log.info(f'{self.last} -> {s}')
if s == GameState.IN_GAME:
if p.valorant.score == self.score and p.valorant.enemy_score == self.score_enemy:
# Same score line
return
eco = EconomyDecision.MATCH_TEAM
if p.valorant.score == 0 and self.score == -1:
# First round
eco = EconomyDecision.BUY
# TODO: ......
won_last = None
if self.score != -1:
won_last = True if p.valorant.score > self.score else False
self.round_history.append(won_last)
self.handler.round_start(p, eco)
if p.valorant.queue_type == 'swiftplay':
if p.valorant.score == 5 or p.valorant.enemy_score == 5:
# Game is over
return
eco = EconomyDecision.BUY
else:
if p.valorant.score > 12 or p.valorant.enemy_score > 12:
if p.valorant.queue_type == 'unrated' or abs(p.valorant.score - p.valorant.enemy_score) >= 2:
# Match is over
return
eco = EconomyDecision.MATCH_TEAM
rounds_played = p.valorant.score + p.valorant.enemy_score
if p.valorant.enemy_score == 12:
# Enemy about to win!
eco = EconomyDecision.BUY
elif rounds_played in (0, 12):
# First round (of half)
eco = EconomyDecision.BUY
elif rounds_played in (1, 13):
# Second round (of half)
eco = EconomyDecision.BUY if won_last else EconomyDecision.SAVE
elif rounds_played in (2, 14):
# Third round (of half)
match self.round_history[-2:]:
case [True, True]:
eco = EconomyDecision.BONUS
case [True, False]:
eco = EconomyDecision.SAVE
case [False, _]:
eco = EconomyDecision.BUY
elif rounds_played >= 24:
# Sudden death or overtime (buy either way)
eco = EconomyDecision.BUY
self.handler.round_start(p, won_last, eco)
self.score = p.valorant.score
self.score_enemy = p.valorant.enemy_score
self.last = GameState.IN_GAME
return
if s == self.last:
@@ -265,15 +312,18 @@ class ValconomyStateMachine:
if s != GameState.GAME_OVER:
self.score = self.score_enemy = -1
self.round_history = []
match s:
case GameState.IN_GAME_GENERIC:
self.handler.game_generic(p)
case GameState.GAME_START:
self.handler.game_start(p)
case GameState.PRE_GAME:
self.handler.pregame(p)
case GameState.GAME_FOUND:
self.handler.match_found(p)
case GameState.IN_QUEUE:
self.handler.match_found(p)
self.handler.queue_start(p)
case GameState.IDLE:
self.handler.idle(p)
@@ -309,7 +359,10 @@ class ValconomyStateMachine:
self.handle_state(GameState.PRE_GAME, p)
return
if p.valorant.game_state == 'INGAME':
if self.score > 0 or self.score_enemy > 0 and p.valorant.score == p.valorant.enemy_score == 0:
if self.score == -1 and p.valorant.score == p.valorant.enemy_score == 0:
self.handle_state(GameState.GAME_START, p)
if (self.score > 0 or self.score_enemy > 0) and p.valorant.score == p.valorant.enemy_score == 0:
self.handle_state(GameState.GAME_OVER, p)
else:
self.handle_state(GameState.IN_GAME, p)