from typing import List, Dict, Union
import arrow
import datetime
from datapipelines import NotFoundError
from merakicommons.cache import lazy, lazy_property
from merakicommons.container import searchable, SearchableList
from ..data import Region, Platform, GameMode, GameType, Queue, Side
from .common import (
CoreData,
CoreDataList,
CassiopeiaObject,
CassiopeiaGhost,
CassiopeiaLazyList,
get_latest_version,
ghost_load_on,
)
from ..dto import spectator as dto
from .staticdata.profileicon import ProfileIcon
from .staticdata.champion import Champion
from .staticdata.rune import Rune
from .staticdata.summonerspell import SummonerSpell
from .staticdata.map import Map
from .summoner import Summoner
##############
# Data Types #
##############
class FeaturedGamesData(CoreDataList):
_dto_type = dto.FeaturedGamesDto
_renamed = {}
class CurrentGameParticipantData(CoreData):
_renamed = {
"spell1Id": "summonerSpellFId",
"spell2Id": "summonerSpellDId",
"teamId": "side",
}
def __call__(self, **kwargs):
if "perks" in kwargs:
self.runes = kwargs.pop("perks")["perkIds"]
super().__call__(**kwargs)
return self
class TeamData(CoreData):
_renamed = {"teamId": "side"}
class CurrentGameInfoData(CoreData):
_dto_type = dto.CurrentGameInfoDto
_renamed = {
"platformId": "platform",
"observers": "observerEncryptionKey",
"gameStartTime": "creation",
"gameLength": "duration",
"gameMode": "mode",
"mapId": "map",
"gameType": "type",
"gameQueueConfigId": "queue",
"gameId": "id",
}
def __call__(self, **kwargs):
if "observers" in kwargs:
self.observerEncryptionKey = kwargs.pop("observers")["encryptionKey"]
super().__call__(**kwargs)
return self
@property
def teams(self) -> List[TeamData]:
blue_team = {"participants": [], "bans": {}, "teamId": 100}
red_team = {"participants": [], "bans": {}, "teamId": 200}
for p in self.participants:
if Side(p["teamId"]) is Side.blue:
p = CurrentGameParticipantData(**p)
blue_team["participants"].append(p)
elif Side(p["teamId"]) is Side.red:
p = CurrentGameParticipantData(**p)
red_team["participants"].append(p)
for b in self.bannedChampions:
if Side(b["teamId"]) is Side.blue:
blue_team["bans"][b["pickTurn"]] = b["championId"]
elif Side(b["teamId"]) is Side.red:
red_team["bans"][b["pickTurn"]] = b["championId"]
return [TeamData(**blue_team), TeamData(**red_team)]
##############
# Core Types #
##############
[docs]class FeaturedMatches(CassiopeiaLazyList):
_data_types = {FeaturedGamesData}
def __init__(self, *, region: Union[Region, str]):
kwargs = {"region": region}
CassiopeiaObject.__init__(self, **kwargs)
@lazy_property
def region(self) -> Region:
return Region(self._data[FeaturedGamesData].region)
@lazy_property
def platform(self) -> Platform:
return self.region.platform
@property
def client_refresh_interval(self) -> int:
return self._data[FeaturedGamesData].clientRefreshInterval
[docs]@searchable(
{str: ["summoner", "champion"], Summoner: ["summoner"], Champion: ["champion"]}
)
class Participant(CassiopeiaObject):
_data_types = {CurrentGameParticipantData}
[docs] @classmethod
def from_data(cls, data: CoreData, match: "CurrentMatch"):
self = super().from_data(data)
self.__match = match
return self
@property
def champion(self) -> Champion:
return Champion(
id=self._data[CurrentGameParticipantData].championId,
region=self.__match.region,
version=get_latest_version(region=self.__match.region, endpoint="champion"),
)
@property
def summoner(self) -> Summoner:
ProfileIcon(
id=self._data[CurrentGameParticipantData].profileIconId,
region=self.__match.region,
)
if hasattr(self._data[CurrentGameParticipantData], "summonerId"):
return Summoner(
id=self._data[CurrentGameParticipantData].summonerId,
name=self._data[CurrentGameParticipantData].summonerName,
region=self.__match.region,
)
else:
return Summoner(
name=self._data[CurrentGameParticipantData].summonerName,
region=self.__match.region,
)
@property
def runes(self) -> List[Rune]:
return SearchableList(
[
Rune(
id=rune_id,
region=self.__match.region,
version=get_latest_version(
region=self.__match.region, endpoint="rune"
),
)
for rune_id in self._data[CurrentGameParticipantData].runes
]
)
@property
def is_bot(self) -> bool:
return self._data[CurrentGameParticipantData].isBot
@property
def side(self) -> Side:
return Side(self._data[CurrentGameParticipantData].side)
@property
def team(self) -> "Team":
if self.side == Side.blue:
return self.__match.blue_team
else:
return self.__match.red_team
@property
def summoner_spell_d(self) -> SummonerSpell:
return SummonerSpell(
id=self._data[CurrentGameParticipantData].summonerSpellDId,
region=self.__match.region,
version=get_latest_version(region=self.__match.region, endpoint="summoner"),
)
@property
def summoner_spell_f(self) -> SummonerSpell:
return SummonerSpell(
id=self._data[CurrentGameParticipantData].summonerSpellFId,
region=self.__match.region,
version=get_latest_version(region=self.__match.region, endpoint="summoner"),
)
[docs]@searchable({})
class Team(CassiopeiaObject):
_data_types = {TeamData}
[docs] @classmethod
def from_data(cls, data: CoreData, match: "CurrentMatch"):
self = super().from_data(data)
self.__match = match
return self
@lazy_property
def participants(self) -> List[Participant]:
return SearchableList(
[
Participant.from_data(p, match=self.__match)
for p in self._data[TeamData].participants
]
)
@lazy_property
def bans(self) -> Dict[int, Champion]:
return {
pick: Champion(
id=championId,
region=self.__match.region,
version=get_latest_version(
region=self.__match.region, endpoint="champion"
),
)
for pick, championId in self._data[TeamData].bans.items()
}
@property
def side(self) -> Side:
return Side(self._data[TeamData].side)
[docs]@searchable({})
class CurrentMatch(CassiopeiaGhost):
_data_types = {CurrentGameInfoData}
def __init__(
self,
*,
summoner: Union[Summoner, str] = None,
region: Union[Region, str] = None
):
kwargs = {"region": region}
if summoner is not None:
if isinstance(summoner, str):
if len(summoner) < 35:
summoner = Summoner(name=summoner, region=region)
else:
summoner = Summoner(id=summoner, region=region)
self.__summoner = summoner
super().__init__(**kwargs)
[docs] @classmethod
def from_data(cls, data: CurrentGameInfoData, summoner: Union[Summoner, str]):
self = super().from_data(data)
if isinstance(summoner, str):
if len(summoner) < 35:
summoner = Summoner(name=summoner, region=region)
else:
summoner = Summoner(id=summoner, region=region)
self.__summoner = summoner
return self
@classmethod
def __get_query_from_kwargs__(
cls, *, summoner: Union[Summoner, str], region: Union[Region, str]
) -> dict:
query = {"region": region}
if isinstance(summoner, Summoner):
query["summoner.id"] = summoner.id
elif isinstance(summoner, str):
if len(summoner) < 35:
query["summoner.id"] = Summoner(name=summoner, region=region).id
else:
query["summoner.id"] = summoner
assert "summoner.id" in query
return query
def __get_query__(self):
return {
"region": self.region,
"platform": self.platform,
"summoner.id": self.__summoner.id,
}
@property
def exists(self):
try:
if not self._Ghost__all_loaded:
self.__load__()
self.creation # Make sure we can access this attribute
return True
except (AttributeError, NotFoundError):
return False
@lazy_property
def region(self) -> Region:
return Region(self._data[CurrentGameInfoData].region)
@lazy_property
def platform(self) -> Platform:
return self.region.platform
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def id(self) -> int:
return self._data[CurrentGameInfoData].id
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
@lazy
def teams(self) -> List[Team]:
return SearchableList(
[
Team.from_data(team, match=self)
for team in self._data[CurrentGameInfoData].teams
]
)
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def blue_team(self) -> Team:
return self.teams[0]
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def red_team(self) -> Team:
return self.teams[1]
@property
def participants(self) -> List[Participant]:
return SearchableList(
[*self.blue_team.participants, *self.red_team.participants]
)
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def mode(self) -> GameMode:
return GameMode(self._data[CurrentGameInfoData].mode)
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def map(self) -> Map:
return Map(
id=self._data[CurrentGameInfoData].map,
region=self.region,
version=get_latest_version(self.region, endpoint="map"),
)
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def type(self) -> GameType:
return GameType(self._data[CurrentGameInfoData].type)
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def queue(self) -> Queue:
return Queue.from_id(self._data[CurrentGameInfoData].queue)
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def duration(self) -> datetime.timedelta:
return datetime.timedelta(seconds=self._data[CurrentGameInfoData].duration)
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def creation(self) -> arrow.Arrow:
return arrow.get(self._data[CurrentGameInfoData].creation / 1000)
@CassiopeiaGhost.property(CurrentGameInfoData)
@ghost_load_on
def observer_key(self) -> str:
return self._data[CurrentGameInfoData].observerEncryptionKey