valconomy/controller/valconomy.py

import base64
from dataclasses import dataclass
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()

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

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]):
    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

    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.players[info.uuid] = info
        self.callback(info)

    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():
        self.port, self.password = None, None

      time.sleep(self.poll_interval)

  def stop(self):
    self.running = False