valconomy/controller/valconomy.py

import base64
from dataclasses import dataclass
from enum import Enum
import json
import logging
from pprint import pprint
import os
import time
from typing import Callable

import hid
import requests
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')

@dataclass
class ValorantPlayerInfo:
  is_idle: bool = False

  is_party_owner: bool = False
  max_party_size: int = 0
  party_size: int = 0
  party_state: str = None

  queue_type: str = None
  game_state: str = None

  map: str = None
  team: str = None
  score: int = 0
  enemy_score: int = 0

@dataclass
class RiotPlayerInfo:
  uuid: str
  name: str
  tag: str
  state: str

  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, bool], None]):
    self.callback = callback

    self.port, self.password = None, None
    self.running = True
    self.players = {}

  def _load_credentials(self):
    if not os.path.exists(self.lockfile_path):
      return False

    try:
      with open(self.lockfile_path) as f:
        for line in f:
          if line.startswith('Riot Client'):
            client, pid, self.port, self.password, proto = line.rstrip().split(':')
            assert proto == 'https'
            return True
    except FileNotFoundError:
      return False

    return False

  def _do_poll(self):
    try:
      resp = requests.get(
        f'https://127.0.0.1:{self.port}/chat/v4/presences',
        auth=('riot', self.password), verify=False)
      resp.raise_for_status()
      data = resp.json()
    except (requests.HTTPError, requests.ConnectionError) as ex:
      # Most likely one of these means the Riot Client shut down
      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

      v = json.loads(base64.b64decode(p.pop('private'))) # Valorant-specfic
      # pprint(p)
      # pprint(v)

      val_info = ValorantPlayerInfo(
        is_idle=v['isIdle'], is_party_owner=v['isPartyOwner'],
        max_party_size=v['maxPartySize'], party_size=v['partySize'],
        party_state=v['partyState'],
        queue_type=v['queueId'] or None, game_state=v['sessionLoopState'],
        map=v['matchMap'] or None,
        score=v['partyOwnerMatchScoreAllyTeam'], enemy_score=v['partyOwnerMatchScoreEnemyTeam'])
      match v['partyOwnerMatchCurrentTeam']:
        case 'Red':
          val_info.team = 'attackers'
        case 'Blue':
          val_info.team = 'defenders'

      info = RiotPlayerInfo(p['puuid'], p['game_name'], p['game_tag'], p['state'], valorant=val_info)

      last = self.players.get(info.uuid)
      if info != last:
        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):
    while self.running:
      if self.password is None:
        if not self._load_credentials():
          time.sleep(self.poll_interval)
          continue
        log.info('Detected Riot Client credentials, starting polling')

      if not self._do_poll():
        for p in self.players.values():
          self.callback(p, True)
        self.players = {}
        self.port, self.password = None, None

      time.sleep(self.poll_interval)

  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)