diff --git a/controller/app.py b/controller/app.py index 07f0e01..363b0c1 100644 --- a/controller/app.py +++ b/controller/app.py @@ -1,3 +1,4 @@ +import configparser import logging import os import sys @@ -9,9 +10,23 @@ import valconomy DATA_DIR = os.path.join(os.environ['LOCALAPPDATA'], 'Valconomy') LOG_FILE = os.path.join(DATA_DIR, 'app.log') +CONFIG_FILE = os.path.join(DATA_DIR, 'config.ini') log = logging.getLogger('valconomy-app') +def parse_log_level(level: str) -> int: + match level.lower(): + case 'debug': + return logging.DEBUG + case 'info': + return logging.INFO + case 'warning': + return logging.WARNING + case 'error': + return logging.ERROR + case _: + return logging.INFO + def data_file(fname: str) -> str: return os.path.abspath(os.path.join(os.path.dirname(__file__), fname)) @@ -25,15 +40,29 @@ def main(): # running in console-less mode, redirect to log file sys.stderr = open(LOG_FILE, 'a') + conf = configparser.ConfigParser() + conf.read_dict({ + 'general': {'log_level': 'info'}, + 'valorant': {'player_uuid': ''}, + }) + conf.read(CONFIG_FILE) + with open(CONFIG_FILE, 'w') as f: + conf.write(f) + + log_level = parse_log_level(conf['general']['log_level']) logging.basicConfig( - format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO, + format='%(asctime)s %(name)s %(levelname)s %(message)s', level=log_level, stream=sys.stderr) - log.info('Starting up') - def cb(i: valconomy.RiotPlayerInfo): - print(repr(i)) - val_client = valconomy.ValorantLocalClient(cb) + if not conf['valorant']['player_uuid']: + log.error(f'No player UUID set, exiting...') + sys.exit(1) + + val_handler = valconomy.ValconomyHandler() + val_sm = valconomy.ValconomyStateMachine(conf['valorant']['player_uuid'], val_handler) + + val_client = valconomy.ValorantLocalClient(val_sm.handle_presence) def do_quit(tray: infi.systray.SysTrayIcon): log.info('Shutting down') val_client.stop() diff --git a/controller/valconomy.py b/controller/valconomy.py index 7ab7927..8928335 100644 --- a/controller/valconomy.py +++ b/controller/valconomy.py @@ -1,5 +1,6 @@ import base64 from dataclasses import dataclass +from enum import Enum import json import logging from pprint import pprint @@ -13,6 +14,8 @@ import urllib3 # Local HTTPS connection, we need to not verify the cert urllib3.disable_warnings() +# Disable so we can debug without a log for every request +logging.getLogger('urllib3.connectionpool').disabled = True log = logging.getLogger('valconomy') @@ -42,11 +45,14 @@ class RiotPlayerInfo: valorant: ValorantPlayerInfo = None + def full_name(self) -> str: + return f'{self.name}#{self.tag}' + class ValorantLocalClient: lockfile_path = os.path.join(os.environ['LOCALAPPDATA'], r'Riot Games\Riot Client\Config\lockfile') poll_interval = 0.5 - def __init__(self, callback: Callable[[RiotPlayerInfo], None]): + def __init__(self, callback: Callable[[RiotPlayerInfo, bool], None]): self.callback = callback self.port, self.password = None, None @@ -81,6 +87,7 @@ class ValorantLocalClient: log.error(f'Failed to make request to Riot Client: {ex}') return False + next_players = {} for p in data['presences']: if p['product'] != 'valorant': continue @@ -106,9 +113,13 @@ class ValorantLocalClient: last = self.players.get(info.uuid) if info != last: - self.players[info.uuid] = info - self.callback(info) + self.callback(info, False) + next_players[info.uuid] = info + for uuid in self.players.keys() - next_players.keys(): + self.callback(self.players[uuid], True) + + self.players = next_players return True def run(self): @@ -126,3 +137,188 @@ class ValorantLocalClient: def stop(self): self.running = False + +class EconomyDecision(Enum): + BUY = 0 + SAVE = 1 + MATCH_TEAM = 2 + +class ValconomyHandler: + # Valorant isn't open + def none(self): + log.info('Val Time soon?') + + # Welcome the user (back) to the game! + def menu(self, info: RiotPlayerInfo, was_idle: bool=False): + if was_idle: + log.info(f"Welcome back, {info.name}!") + else: + log.info(f"It's Val Time, {info.name}!") + + # The user is idle in the menu... + def idle(self, info: RiotPlayerInfo): + log.info(f'Come back soon, {info.name}...') + + # Just entered queue + def queue_start(self, info: RiotPlayerInfo): + if info.valorant.is_party_owner and info.valorant.queue_type == 'unrated': + log.info('Uhhhh should that be comp, MoonStar?') + else: + log.info(f'Hope you find a game quickly, {info.name}!') + + def match_found(self, info: RiotPlayerInfo): + if info.valorant.queue_type == 'premier-seasonmatch': + log.info('Do the Cosmonauts proud!') + else: + log.info("I hope it's not Split...") + + # Loaded into the agent select + def pregame(self, info: RiotPlayerInfo): + if info.valorant.map == '/Game/Maps/Bonsai/Bonsai': + log.info('Ewwwww, Split....') + else: + log.info('Pick a good agent!') + + # Game where we're not providing economy help + def game_generic(self, info: RiotPlayerInfo): + match info.valorant.queue_type: + case 'hurm': + log.info('Have a good TDM!') + case 'deathmatch': + log.info('Have a good deathmatch!') + case 'ggteam': + log.info('Have a good Escalation!') + case 'spikerush': + log.info('Have a good Spike Rush!') + case _: + if info.valorant.max_party_size == 12: + log.info('Have a good custom!') + else: + log.info('Have a good game!') + + # Loaded into the game + def game_start(self, info: RiotPlayerInfo): + log.info('OK best of luck!') + + # Round started + def round_start(self, info: RiotPlayerInfo, economy: EconomyDecision): + match economy: + case EconomyDecision.BUY: + log.info('You should buy!') + case EconomyDecision.SAVE: + log.info('Time to save...') + case EconomyDecision.MATCH_TEAM: + log.info('Follow the team economy!') + + def game_over(self, info: RiotPlayerInfo, won: bool): + if won: + log.info('Well played!') + else: + log.info('Hard luck...') + +class HIDValconomyHandler(ValconomyHandler): + pass + +class GameState(Enum): + NONE = 0 + MENU = 1 + IDLE = 2 + IN_QUEUE = 3 + GAME_FOUND = 4 + PRE_GAME = 5 + IN_GAME = 6 + GAME_OVER = 7 + IN_GAME_GENERIC = 8 + +class ValconomyStateMachine: + def __init__(self, player_uuid: str, handler: ValconomyHandler): + self.uuid = player_uuid + self.handler = handler + + self.last = None + self.score = -1 + self.score_enemy = -1 + + 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: + return + + eco = EconomyDecision.MATCH_TEAM + if p.valorant.score == 0 and self.score == -1: + # First round + eco = EconomyDecision.BUY + # TODO: ...... + + self.handler.round_start(p, eco) + self.score = p.valorant.score + self.score_enemy = p.valorant.enemy_score + return + + if s == self.last: + return + + if s != GameState.GAME_OVER: + self.score = self.score_enemy = -1 + match s: + case GameState.IN_GAME_GENERIC: + self.handler.game_generic(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) + + case GameState.IDLE: + self.handler.idle(p) + case GameState.MENU: + self.handler.menu(p, self.last == GameState.IDLE) + + case GameState.NONE: + self.handler.none() + + case GameState.GAME_OVER: + self.handler.game_over(p, self.score > self.score_enemy) + + self.last = s + + def handle_presence(self, p: RiotPlayerInfo, gone: bool): + log.debug(f'{repr(p)}, gone: {gone}') + if p.uuid != self.uuid: + return + + if gone: + self.handle_state(GameState.NONE, p) + return + + if p.valorant.party_state == 'MATCHMADE_GAME_STARTING': + self.handle_state(GameState.GAME_FOUND, p) + return + if p.valorant.party_state == 'MATCHMAKING': + self.handle_state(GameState.IN_QUEUE, p) + return + + if p.valorant.queue_type in ('unrated', 'competitive', 'premier-seasonmatch', 'swiftplay'): + if p.valorant.game_state == 'PREGAME': + 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: + self.handle_state(GameState.GAME_OVER, p) + else: + self.handle_state(GameState.IN_GAME, p) + return + + if p.valorant.game_state in ('PREGAME', 'INGAME'): + self.handle_state(GameState.IN_GAME_GENERIC, p) + return + + if p.valorant.game_state == 'MENUS': + if p.valorant.is_idle: + self.handle_state(GameState.IDLE, p) + return + + self.handle_state(GameState.MENU, p)