Source code for arez.match

from __future__ import annotations

import logging
from itertools import count
from typing import Optional, Union, List, Dict, Iterable, Generator, TYPE_CHECKING

from .exceptions import NotFound
from .enums import Queue, Region, Rank
from .utils import chunk, _convert_map_name, _deduplicate
from .mixins import (
    CacheClient, CacheObject, MatchMixin, MatchPlayerMixin, Expandable, WinLoseMixin
)

if TYPE_CHECKING:
    from . import responses
    from .enums import Language
    from .champion import Champion, Skin
    from .cache import DataCache, CacheEntry
    from .player import PartialPlayer, Player


__all__ = [
    "PartialMatch",
    "MatchPlayer",
    "Match",
    "LivePlayer",
    "LiveMatch",
]
logger = logging.getLogger(__package__)


# this is a close duplicate of `PaladinsAPI.get_players`, modified for speed and its usage
async def _get_players(cache: DataCache, player_ids: Iterable[int]) -> Dict[int, Player]:
    ids_list: List[int] = _deduplicate(player_ids, 0)  # also remove private accounts
    if not ids_list:  # pragma: no cover
        return {}
    from .player import Player  # cyclic import
    players_dict: Dict[int, Player] = {}
    for chunk_ids in chunk(ids_list, 20):
        chunk_response = await cache.request("getplayerbatch", ','.join(map(str, chunk_ids)))
        for player_data in chunk_response:
            if player_data["ret_msg"]:  # pragma: no cover, skip private accounts
                continue
            player = Player(cache, player_data)
            players_dict[player.id] = player
    return players_dict


[docs]class PartialMatch(MatchPlayerMixin, MatchMixin, Expandable["Match"]): """ Represents a match from a single player's perspective only. This partial object is returned by the `PartialPlayer.get_match_history` player's method. To obtain an object with all match information, try awaiting on this object like so: .. code-block:: py match = await partial_match Attributes ---------- id : int The match ID. queue : Queue The queue this match was played in. region : Region The region this match was played in. timestamp : datetime.datetime A timestamp of when this match happened. duration : Duration The duration of the match. map_name : str The name of the map played. score : Tuple[int, int] The match's ending score.\n The first value is always the allied-team score, while the second one - enemy team score. winning_team : Literal[1, 2] The winning team of this match. player : Union[PartialPlayer, Player] The player who participated in this match.\n This is usually a new partial player object.\n All attributes, Name, ID and Platform, should be present. champion : Union[Champion, CacheObject] The champion used by the player in this match.\n With incomplete cache, this will be a `CacheObject` with the name and ID set. loadout : MatchLoadout The loadout used by the player in this match. items : List[MatchItem] A list of items bought by the player during this match. credits : int The amount of credits earned this match. experience : int The base amount of experience gained from this match. kills : int The amount of kills. deaths : int The amount of deaths. assists : int The amount of assists. damage_done : int The amount of damage dealt. damage_bot : int The amount of damage done by the player's bot after they disconnected. damage_taken : int The amount of damage taken. damage_mitigated : int The amount of damage mitigated (shielding). healing_done : int The amount of healing done to other players. healing_bot : int The amount of healing done by the player's bot after they disconnected. healing_self : int The amount of healing done to self (self-sustain). objective_time : int The amount of objective time the player got, in seconds. multikill_max : int The maximum multikill player did during the match. skin : Union[Skin, CacheObject] The skin the player had equipped for this match.\n With incomplete cache, this will be a `CacheObject` with the name and ID set. team_number : Literal[1, 2] The team this player belongs to. team_score : int The score of the player's team. winner : bool `True` if the player won this match, `False` otherwise. """ def __init__( self, player: Union[PartialPlayer, Player], language: Language, cache_entry: Optional[CacheEntry], match_data: responses.HistoryMatchObject, ): MatchPlayerMixin.__init__(self, player, cache_entry, match_data) MatchMixin.__init__(self, match_data) self._language = language async def _expand(self) -> Match: """ Upgrades this object into a full `Match` one, containing all match players and information. Uses up a single request. Returns ------- Match The full match object. Raises ------ NotFound The match could not be found. """ logger.info(f"PartialMatch(id={self.id}).expand()") response = await self._api.request("getmatchdetails", self.id) if not response: raise NotFound("Match") cache_entry = self._api.get_entry(self._language) return Match(self._api, cache_entry, response, {}) def __repr__(self) -> str: return f"{self.queue.name}: {self.champion.name}: {self.kda_text}" @property def disconnected(self) -> bool: """ Returns `True` if the player has disconnected during the match, `False` otherwise.\n This is done by checking if either `damage_bot` or `healing_bot` are non zero. :type: bool """ return self.damage_bot > 0 or self.healing_bot > 0
[docs]class MatchPlayer(MatchPlayerMixin): """ Represents a full match's player. Attributes ---------- match : Match The match this player belongs to. player : Union[PartialPlayer, Player] The player itself who participated in this match.\n This is usually a new partial player object.\n All attributes, Name, ID and Platform, should be present. rank : Optional[Rank] The player's rank. .. warning:: Due to API limitations, this is only available for matches played in ranked queues.\n For other queues, this attribute will be `None`. champion : Union[Champion, CacheObject] The champion used by the player in this match.\n With incomplete cache, this will be a `CacheObject` with the name and ID set. loadout : MatchLoadout The loadout used by the player in this match. items : List[MatchItem] A list of items bought by the player during this match. credits : int The amount of credits earned this match. experience : int The base amount of experience gained from this match. kills : int The amount of player kills. deaths : int The amount of deaths. assists : int The amount of assists. damage_done : int The amount of damage dealt. damage_bot : int The amount of damage done by the player's bot after they disconnected. damage_taken : int The amount of damage taken. damage_mitigated : int The amount of damage mitigated (shielding). healing_done : int The amount of healing done to other players. healing_bot : int The amount of healing done by the player's bot after they disconnected. healing_self : int The amount of healing done to self (self-sustain). objective_time : int The amount of objective time the player got, in seconds. multikill_max : int The maximum multikill player did during the match. skin : Union[Skin, CacheObject] The skin the player had equipped for this match.\n With incomplete cache, this will be a `CacheObject` with the name and ID set. team_number : Literal[1, 2] The team this player belongs to. team_score : int The score of the player's team. winner : bool `True` if the player won this match, `False` otherwise. points_captured : int The amount of times the player's team captured the point.\n This is ``0`` for non-Siege matches. push_successes : int The amount of times the player's team successfully pushed the payload to the end.\n This is ``0`` for non-Siege matches. kills_bot : int The amount of bot kills. account_level : int The player's account level. mastery_level : int The player's champion mastery level. party_number : int A number denoting the party the player belonged to.\n ``0`` means the player wasn't in a party. """ def __init__( self, match: Match, cache_entry: Optional[CacheEntry], player_data: responses.MatchPlayerObject, parties: Dict[int, int], players: Dict[int, Player], ): player: Optional[Union[PartialPlayer, Player]] = players.get(int(player_data["playerId"])) if player is None: # if no full player was found from .player import PartialPlayer # cyclic imports player = PartialPlayer( match._api, id=player_data["playerId"], name=player_data["playerName"], platform=player_data["playerPortalId"], ) super().__init__(player, cache_entry, player_data) self.rank: Optional[Rank] if match.queue.is_ranked(): self.rank = Rank(player_data["League_Tier"], _return_default=True) else: self.rank = None self.points_captured: int = player_data["Kills_Gold_Fury"] self.push_successes: int = player_data["Kills_Fire_Giant"] self.kills_bot: int = player_data["Kills_Bot"] self.account_level: int = player_data["Account_Level"] self.mastery_level: int = player_data["Mastery_Level"] self.party_number: int = parties.get(player_data["PartyId"], 0) @property def disconnected(self) -> bool: """ Returns `True` if the player has disconnected during the match, `False` otherwise.\n This is done by checking if either `damage_bot` or `healing_bot` are non zero. :type: bool """ return self.damage_bot > 0 or self.healing_bot > 0 def __repr__(self) -> str: return ( f"{self.player.name or 'Unknown'}({self.player.id}): {self.champion.name}: " f"({self.kda_text}, {self.damage_done}, {self.healing_done})" )
[docs]class Match(CacheClient, MatchMixin): """ Represents already-played full match information. You can get this from the `PaladinsAPI.get_match` and `PaladinsAPI.get_matches` methods, as well as from upgrading a `PartialMatch` object. Attributes ---------- id : int The match ID. queue : Queue The queue this match was played in. region : Region The region this match was played in. timestamp : datetime.datetime A timestamp of when this match happened. duration : Duration The duration of the match. map_name : str The name of the map played. score : Tuple[int, int] The match's ending score.\n The first value is the ``team1`` score, while the second value - ``team2`` score. winning_team : Literal[1, 2] The winning team of this match. replay_available : bool `True` if this match has a replay that you can watch, `False` otherwise. bans : List[Optional[Union[Champion, CacheObject]]] A list of champions banned in this match.\n With incomplete cache, the list will contain `CacheObject` objects with the name and ID set.\n This will be an empty list for non-ranked matches.\n `None` indicates there was no ban. team1 : List[MatchPlayer] A list of players in the first team. team2 : List[MatchPlayer] A list of players in the second team. players : Generator[MatchPlayer] A generator that iterates over all match players in the match. """ def __init__( self, api: DataCache, cache_entry: Optional[CacheEntry], match_data: List[responses.MatchPlayerObject], players: Dict[int, Player], ): CacheClient.__init__(self, api) first_player = match_data[0] MatchMixin.__init__(self, first_player) logger.debug(f"Match(id={self.id}) -> creating...") self.replay_available: bool = first_player["hasReplay"] == "y" self.bans: List[Optional[Union[Champion, CacheObject]]] = [] if self.queue.is_ranked(): for i in range(1, 7): ban_id: int = first_player[f"BanId{i}"] # type: ignore[misc] if not ban_id: # zero indicates no ban has happened - use None self.bans.append(None) continue ban_champ: Optional[Union[Champion, CacheObject]] = None if cache_entry is not None: ban_champ = cache_entry.champions.get(ban_id) if ban_champ is None: ban_champ = CacheObject( id=ban_id, name=first_player[f"Ban_{i}"] # type: ignore[misc] ) self.bans.append(ban_champ) self.team1: List[MatchPlayer] = [] self.team2: List[MatchPlayer] = [] # Determine party numbers # We need to do this here because apparently one-man parties are a thing party_count = count(1) parties: Dict[int, int] = {} for player_data in match_data: pid = player_data["PartyId"] # process only non-0 parties if pid: if pid not in parties: # haven't seen this one yet, assign zero parties[pid] = 0 elif parties[pid] == 0: # we've seen this one, and it doesn't have a number assigned - assign one parties[pid] = next(party_count) # iterate over a second time, now that we have the party numbers sorted out for player_data in match_data: match_player = MatchPlayer(self, cache_entry, player_data, parties, players) team_number = player_data["TaskForce"] if team_number == 1: self.team1.append(match_player) elif team_number == 2: # pragma: no branch self.team2.append(match_player) logger.debug(f"Match(id={self.id}) -> created") @property def players(self) -> Generator[MatchPlayer, None, None]: for p in self.team1: yield p for p in self.team2: yield p def __repr__(self) -> str: return f"{self.queue.name}({self.id}): {self.score}"
[docs] async def expand_players(self): """ Makes partial player objects in the containing match player objects be expanded into full `Player` objects, if possible. Uses up a single request to do the expansion. """ players_dict = await _get_players(self._api, (p.player.id for p in self.players)) for mp in self.players: pid = mp.player.id # skip 0s if pid == 0: continue if (p := players_dict.get(pid)) is not None: # pragma: no branch mp.player = p
[docs]class LivePlayer(WinLoseMixin, CacheClient): """ Represents a live match player. You can find these on the `LiveMatch.team1` and `LiveMatch.team2` attributes. Attributes ---------- match: LiveMatch The match this player belongs to. player : Union[PartialPlayer, Player] The actual player playing in this match. champion : Union[Champion, CacheObject] The champion the player is using in this match.\n With incomplete cache, this will be a `CacheObject` with the name and ID set. skin : Union[Skin, CacheObject] The skin the player has equipped for this match.\n With incomplete cache, this will be a `CacheObject` with the name and ID set. rank : Optional[Rank] The player's rank. .. warning:: Due to API limitations, this is only available for matches played in ranked queues.\n For other queues, this attribute will be `None`. account_level : int The player's account level. mastery_level : int The player's champion mastery level. wins : int The amount of wins. losses : int The amount of losses. """ def __init__( self, match: LiveMatch, cache_entry: Optional[CacheEntry], player_data: responses.LivePlayerObject, players: Dict[int, Player], ): CacheClient.__init__(self, match._api) WinLoseMixin.__init__( self, wins=player_data["tierWins"], losses=player_data["tierLosses"], ) self.match: LiveMatch = match # Player player: Optional[Union[PartialPlayer, Player]] = players.get(int(player_data["playerId"])) if player is None: # if no full player was found from .player import PartialPlayer # cyclic imports player = PartialPlayer( self._api, id=player_data["playerId"], name=player_data["playerName"], platform=player_data["playerPortalId"], ) self.player: Union[PartialPlayer, Player] = player # Champion champion_id: int = player_data["ChampionId"] champion: Optional[Union[Champion, CacheObject]] = None if cache_entry is not None: champion = cache_entry.champions.get(champion_id) if champion is None: champion = CacheObject(id=champion_id, name=player_data["ChampionName"]) self.champion: Union[Champion, CacheObject] = champion # Skin skin_id = player_data["SkinId"] skin: Optional[Union[Skin, CacheObject]] = None if cache_entry is not None: skin = cache_entry.skins.get(skin_id) if skin is None: # pragma: no cover skin = CacheObject(id=skin_id, name=player_data["Skin"]) self.skin: Union[Skin, CacheObject] = skin # Other self.rank: Optional[Rank] if match.queue.is_ranked(): # pragma: no cover self.rank = Rank(player_data["Tier"], _return_default=True) else: self.rank = None self.account_level: int = player_data["Account_Level"] self.mastery_level: int = player_data["Mastery_Level"] def __repr__(self) -> str: return ( f"{self.player.name or 'Unknown'}({self.player.id}): " f"{self.account_level} level: " f"{self.champion.name}({self.mastery_level})" )
[docs]class LiveMatch(CacheClient): """ Represents an on-going live match. You can get this from the `PlayerStatus.get_live_match` method. Attributes ---------- id : int The match ID. map_name : str The name of the map played. queue : Queue The queue the match is being played in. region : Region The region this match is being played in. team1 : List[LivePlayer] A list of live players in the first team. team2 : List[LivePlayer] A list of live players in the second team. players : Generator[LivePlayer] A generator that iterates over all live match players in the match. """ def __init__( self, api: DataCache, cache_entry: Optional[CacheEntry], match_data: List[responses.LivePlayerObject], players: Dict[int, Player], ): super().__init__(api) first_player = match_data[0] self.id: int = first_player["Match"] self.map_name: str = _convert_map_name(first_player["mapGame"]) self.queue = Queue(int(first_player["Queue"]), _return_default=True) self.region = Region(first_player["playerRegion"], _return_default=True) self.team1: List[LivePlayer] = [] self.team2: List[LivePlayer] = [] for player_data in match_data: live_player = LivePlayer(self, cache_entry, player_data, players) if player_data["taskForce"] == 1: self.team1.append(live_player) elif player_data["taskForce"] == 2: # pragma: no branch self.team2.append(live_player) def __repr__(self) -> str: return f"{self.__class__.__name__}({self.queue.name}): {self.map_name}" @property def players(self) -> Generator[LivePlayer, None, None]: for p in self.team1: yield p for p in self.team2: yield p
[docs] async def expand_players(self): """ Makes partial player objects in the containing match player objects be expanded into full `Player` objects, if possible. Uses up a single request to do the expansion. """ players_dict = await _get_players(self._api, (p.player.id for p in self.players)) for mp in self.players: pid = mp.player.id # skip 0s if pid == 0: # pragma: no cover continue if (p := players_dict.get(pid)) is not None: # pragma: no branch mp.player = p