From 78aa8005bdc895986b425d71e24faa061605b7db Mon Sep 17 00:00:00 2001
From: Jack O'Sullivan <jackos1998@gmail.com>
Date: Thu, 12 Dec 2024 17:12:13 +0000
Subject: [PATCH] controller: Initial polling and parsing of Valorant presence
---
controller/app.py | 38 +++++++-----
controller/valconomy.py | 127 ++++++++++++++++++++++++++++++++++++++++
2 files changed, 149 insertions(+), 16 deletions(-)
diff --git a/controller/app.py b/controller/app.py
index 4baf269..07f0e01 100644
--- a/controller/app.py
+++ b/controller/app.py
@@ -1,7 +1,6 @@
import logging
import os
import sys
-import time
import infi.systray
@@ -11,33 +10,40 @@ import valconomy
DATA_DIR = os.path.join(os.environ['LOCALAPPDATA'], 'Valconomy')
LOG_FILE = os.path.join(DATA_DIR, 'app.log')
-log = logging.getLogger('valconomy')
+log = logging.getLogger('valconomy-app')
def data_file(fname: str) -> str:
return os.path.abspath(os.path.join(os.path.dirname(__file__), fname))
-running = True
-def do_quit(tray: infi.systray.SysTrayIcon):
- global running
- log.info('Shutting down')
- running = False
-
-def do_view_logs(tray: infi.systray.SysTrayIcon):
+def do_open_datadir(tray: infi.systray.SysTrayIcon):
os.startfile(DATA_DIR)
def main():
os.makedirs(DATA_DIR, exist_ok=True)
+
+ if sys.stderr is None:
+ # running in console-less mode, redirect to log file
+ sys.stderr = open(LOG_FILE, 'a')
+
logging.basicConfig(
format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO,
- handlers=(
- logging.StreamHandler(sys.stderr), # log to stderr as well as file
- logging.FileHandler(LOG_FILE),
- ))
+ stream=sys.stderr)
log.info('Starting up')
- with infi.systray.SysTrayIcon(data_file('icon.ico'), 'Valconomy', menu_options=(('View logs', None, do_view_logs),), on_quit=do_quit) as tray:
- while running:
- time.sleep(0.5)
+ def cb(i: valconomy.RiotPlayerInfo):
+ print(repr(i))
+
+ val_client = valconomy.ValorantLocalClient(cb)
+ def do_quit(tray: infi.systray.SysTrayIcon):
+ log.info('Shutting down')
+ val_client.stop()
+
+ with infi.systray.SysTrayIcon(
+ data_file('icon.ico'), 'Valconomy', on_quit=do_quit,
+ menu_options=(
+ ('Open data directory', None, do_open_datadir),
+ )) as tray:
+ val_client.run()
if __name__ == '__main__':
main()
diff --git a/controller/valconomy.py b/controller/valconomy.py
index b1206e6..7ab7927 100644
--- a/controller/valconomy.py
+++ b/controller/valconomy.py
@@ -1 +1,128 @@
+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