Moving to AoE4World for some data

Author: FluffyMaguro

src/AoE4_Overlay.py

@@ -16,7 +16,7 @@
 
 logger = get_logger(__name__)
 
-VERSION = "1.2.7"
+VERSION = "1.2.8"
 
 # Might or might not help
 os.environ["PYTHONIOENCODING"] = "utf-8"

src/overlay/api_checking.py

@@ -15,6 +15,7 @@
 
 import json
 import time
+from datetime import datetime
 from typing import Any, Dict, List, Optional, Tuple
 
 import requests
@@ -249,13 +250,13 @@ class Api_checker:
 
     def __init__(self):
         self.force_stop = False
-        self.last_match_timestamp = -1
-        self.last_rating_timestamp = -1
+        self.last_match_timestamp = datetime(1900, 1, 1, 0, 0, 0)
+        self.last_rating_timestamp = datetime(1900, 1, 1, 0, 0, 0)
 
     def reset(self):
         """ Resets last timestamps"""
-        self.last_match_timestamp = -1
-        self.last_rating_timestamp = -1
+        self.last_match_timestamp = datetime(1900, 1, 1, 0, 0, 0)
+        self.last_rating_timestamp = datetime(1900, 1, 1, 0, 0, 0)
 
     def sleep(self, seconds: int) -> bool:
         """ Sleeps while checking for force_stop
@@ -284,133 +285,56 @@ def check_for_new_game(self,
                 return
 
     def get_data(self) -> Optional[Dict[str, Any]]:
-        """ Returns match data if there is a new game"""
         if self.force_stop:
             return
 
-        # Match history
+        # Get last match from aoe4world.com
         try:
-            match_history = get_match_history(amount=1, raise_exception=True)
-        except json.decoder.JSONDecodeError:
-            # AoEIV.net down
-            return {'server_down': True}
+            url = f"https://aoe4world.com/api/v0/players/{settings.profile_id}/games/last"
+            resp = session.get(url)
+            data = json.loads(resp.text)
         except Exception:
-            return
-
-        if not match_history:
-            return
-
-        match = match_history[0]
-        leaderboard_id = match_mode(match)
-
-        # No players in the game
-        if not match['players']:
+            logger.exception("")
             return
 
         if self.force_stop:
             return
-
-        # Show the last game
-        if match['started'] > self.last_match_timestamp:
-            self.last_match_timestamp = match['started']
-
-            # Remove duplicated players
-            match['players'] = self.get_unique_players(match['players'])
-            # Gets additional player data from leaderboards stats (in-place)
-            for player in match['players']:
-                self.get_player_data(leaderboard_id, player)
-            return match
-
-        # Rating history
-        if not quickmatch_game(match):
-            return
-        rating_history = get_rating_history(leaderboard_id, amount=1)
-        if not rating_history:
+        if "error" in data:
             return
-        rating = rating_history[0]
-
-        # Check for new rating data
-        if self.last_match_timestamp != -1 and \
-            rating['timestamp'] > self.last_rating_timestamp:
 
-            self.last_rating_timestamp = rating['timestamp']
-            return {"new_rating": True, 'timestamp': rating['timestamp']}
+        # Calc old leaderboard id
+        data['leaderboard_id'] = 0
+        try:
+            data['leaderboard_id'] = int(data['kind'][-1]) + 16
+        except Exception:
+            logger.exception("")
 
-    @staticmethod
-    def get_unique_players(
-            players: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """ Create a new list of players that has only unique players
+        # Calc started time
+        started = datetime.strptime(data['started_at'],
+                                    "%Y-%m-%dT%H:%M:%S.000Z")
+        data['started_sec'] = started.timestamp()
 
-        AoEIV.net sometimes returns players multiple times in player list """
-        ids = set()
-        filtered = list()
+        # Show the last game
+        if started > self.last_match_timestamp:  # and data['ongoing']:
+            self.last_match_timestamp = started
+            return data
 
-        for p in players:
-            playerid = p["profile_id"]
-            if playerid in ids:
-                continue
-            ids.add(playerid)
-            filtered.append(p)
-        return filtered
+        if not "qm_" in data['kind']:
+            return
 
-    @staticmethod
-    def get_player_data(leaderboard_id: int, player_dict: Dict[str, Any]):
-        """ Updates player data inplace"""
+        # When a game finished
+        if started > self.last_rating_timestamp and not data['ongoing']:
 
-        try:
-            url = f"https://aoe4world.com/api/v0/players/{player_dict['profile_id']}"
-            data = json.loads(session.get(url).text)
-            player_dict['name'] = data['name']
-
-            data = data['modes'][f'qm_{mode_data[leaderboard_id]}']
-            player_dict['rank'] = zeroed(data["rank"])
-            player_dict['rating'] = zeroed(data["rating"])
-            player_dict['wins'] = zeroed(data["wins_count"])
-            player_dict['losses'] = zeroed(data["games_count"]) - zeroed(
-                data["wins_count"])
-            player_dict['streak'] = zeroed(data["streak"])
-
-            # AoE4World.com civ stats currently work only for 1v1
-            if mode_data[leaderboard_id] == "1v1":
-                civ_name = net_to_world.get(player_dict['civ'])
-                for civ in data['civilizations']:
-                    if civ['civilization'] == civ_name:
-                        player_dict['civ_games'] = civ['games_count']
-                        player_dict['civ_winrate'] = civ['win_rate']
-                        player_dict['civ_win_length_median'] = civ[
-                            'game_length']['wins_median']
-                        return
-                logger.warning(
-                    f"Didn't find civ: {civ_name} in aoe4world.com player civ list"
-                )
-            return
-        except Exception:
-            logger.exception(
-                f"AoE4wWorld.com failed for player {player_dict.get('profile_id')}"
-            )
+            # Check for new ratings
+            if not data['leaderboard_id']:
+                return
 
-        # If AoE4World.com fails, default to aoeiv.net
-        try:
-            url = f"https://aoeiv.net/api/leaderboard?game=aoe4&leaderboard_id={leaderboard_id}&profile_id={player_dict['profile_id']}&count=1"
-            data = json.loads(session.get(url).text)
-
-            if data['leaderboard']:
-                data = data["leaderboard"][0]
-                player_dict['rank'] = data["rank"]
-                player_dict['rating'] = data["rating"]
-                player_dict['wins'] = data["wins"]
-                player_dict['losses'] = data["losses"]
-                player_dict['streak'] = data["streak"]
-
-                # Sometimes AoEIV.net returns no name for the player
-                if player_dict['name'] is None:
-                    player_dict['name'] = data['name']
-            else:
-                player_dict['rating'] = "–"
+            rating_history = get_rating_history(data['leaderboard_id'],
+                                                amount=1)
 
-        except Exception:
-            logger.exception("Failed to get player data from aoeiv.net")
+            if not rating_history:
+                return
 
-        # Last check on AoE4World for player name
-        if player_dict['name'] is None:
-            player_dict['name'] = get_player_name(player_dict['profile_id'])
+            self.last_rating_timestamp = started
+            rating = rating_history[0]
+            return {"new_rating": True, 'timestamp': rating['timestamp']}

src/overlay/helper_func.py

@@ -91,55 +91,64 @@ def process_game(game_data: Dict[str, Any]) -> Dict[str, Any]:
     Sorts players to main is at the top. Calculates winrates. 
     Gets text for civs and maps. Apart from `team`, all player data returned as string."""
     result = {}
-    result['map'] = map_data.get(game_data["map_type"], "Unknown map")
-    result['mode'] = match_mode(game_data)
-    result['started'] = game_data['started']
-    result['ranked'] = game_data['ranked']
+    result['map'] = game_data['map']
+    result['mode'] = game_data['leaderboard_id']
+    result['started'] = game_data['started_at']
+    result['ranked'] = 'qm_' in game_data['kind']
     result['server'] = game_data['server']
-    result['version'] = game_data['version']
-    result['match_id'] = game_data['match_id']
+    result['match_id'] = game_data['game_id']
+    mode = game_data['kind']
 
     # Sort players so the main player team is first
-    team = None
-    for player in game_data['players']:
-        if player['profile_id'] == settings.profile_id:
-            team = player['team']
-            break
+    players = []
+    main_team = None
+    for idx, team in enumerate(game_data['teams']):
+        for player in team:
+            player['team'] = idx
+            players.append(player)
+            if player['profile_id'] == settings.profile_id:
+                main_team = idx
 
     def sortingf(player: Dict[str, Any]) -> int:
         if player['team'] is None:
             return 99
-        if player['team'] == team:
+        if player['team'] == main_team:
             return -1
         return player['team']
 
-    game_data['players'] = sorted(game_data['players'], key=sortingf)
+    players = sorted(players, key=sortingf)
 
     # Add player data
     result['players'] = []
-    for player in game_data['players']:
-        wins = zeroed(player.get('wins'))
-        losses = zeroed(player.get('losses'))
-        games = wins + losses
-        winrate = wins / games if games else 0
-        civ_win_median = ''
-        if 'civ_win_length_median' in player:
-            civ_win_median = time.strftime(
-                "%M:%S", time.gmtime(player['civ_win_length_median']))
-        civ_winrate = ''
-        if 'civ_winrate' in player:
-            civ_winrate = f"{player['civ_winrate']/100:.1%}"
+    for player in players:
+        current_civ = player['civilization']
+        name = player['name'] if player['name'] is not None else "?"
+
+        civ_games = ""
+        civ_winrate = ""
+        civ_win_median = ""
+        try:
+            for civ in player['modes'][mode]['civilizations']:
+                if civ['civilization'] == current_civ:
+                    civ_games = str(civ['games_count'])
+                    civ_winrate = f"{civ['win_rate']/100:.1%}"
+                    med = civ['game_length']['wins_median']
+                    civ_win_median = time.strftime("%M:%S", time.gmtime(med))
+        except Exception:
+            ...
+
+        mode_data = player.get('modes', {}).get(mode, {})
 
         data = {
-            'civ': civ_data.get(player['civ'], "Unknown civ"),
-            'name': player['name'],
-            'team': zeroed(player['team']),
-            'rating': str(player.get('rating', '')),
-            'rank': f"#{player.get('rank', '')}",
-            'wins': str(wins) if wins else '',
-            'losses': str(losses) if losses else '',
-            'winrate': f"{winrate:.1%}",
-            'civ_games': str(player.get('civ_games', '')),
+            'civ': current_civ.replace("_", " ").title(),
+            'name': name,
+            'team': zeroed(player['team'] + 1),
+            'rating': str(mode_data.get('rating', 0)),
+            'rank': f"#{mode_data.get('rank',0)}",
+            'wins': str(mode_data.get('wins_count', 0)),
+            'losses': str(mode_data.get('losses_count', 0)),
+            'winrate': f"{mode_data.get('winrate',0)}%",
+            'civ_games': civ_games,
             'civ_winrate': civ_winrate,
             'civ_win_length_median': civ_win_median
         }

src/overlay/tab_main.py

@@ -116,10 +116,8 @@ def new_game(self, game_data: Optional[Dict[str, Any]]):
             if settings.log_matches:
                 log_match(game_data)
             processed = hf.process_game(game_data)
-            start = time.strftime("%Y-%m-%d %H:%M:%S",
-                                  time.localtime(processed['started']))
             logger.info(
-                f"New live game (match_id: {processed['match_id']} | mode: {processed['mode']-16} | started: {start})"
+                f"New live game (game_id: {game_data['game_id']} | mode: {game_data['kind']} | started: {game_data['started_at']})"
             )
             self.override_tab.update_data(processed)
             if not self.prevent_overlay_update: