🪥Some Brush-ups

Erm

steam/ext/dota2/client.py

@@ -4,7 +4,7 @@
 
 import asyncio
 from functools import partial
-from typing import TYPE_CHECKING, Final
+from typing import TYPE_CHECKING, Any, Final, overload
 
 from ..._const import DOCS_BUILDING, timeout
 from ..._gc import Client as Client_
@@ -14,7 +14,7 @@
     MISSING,
     cached_property,
 )
-from .models import ClientUser, LiveMatch, MatchDetails, MatchMinimal, PartialUser
+from .models import ClientUser, LiveMatch, MatchMinimal, PartialUser
 from .protobufs import client_messages, watch
 from .state import GCState  # noqa: TCH001
 
@@ -46,21 +46,55 @@ class Client(Client_):
         @cached_property
         def user(self) -> ClientUser: ...
 
+    # TODO: maybe this should exist as a part of the whole lib (?)
+    def instantiate_partial_user(self, id: Intable) -> PartialUser:
+        return PartialUser(self._state, id)
+
+    async def _fetch_find_top_source_tv_games(self, *, limit: int = 100, **kwargs: Any) -> list[LiveMatch]:
+        """Private helper method to use `watch.ClientToGCFindTopSourceTVGames` requests.
+
+        Valid kwargs are fields from `watch.ClientToGCFindTopSourceTVGames` definition.
+        Some combinations are not working together.
+        This is why `Client` offers three methods below with offered prepared kwargs options that make sense.
+        """
+        if limit < 1 or limit > 100:
+            raise ValueError("limit value should be between 1 and 100 inclusively.")
+
+        # mini-math: limit 100 -> start_game 90, 91 -> 90, 90 -> 80
+        start_game = (limit - 1) // 10 * 10
+
+        def check(start_game: int, msg: watch.GCToClientFindTopSourceTVGamesResponse) -> bool:
+            return msg.start_game == start_game
+
+        futures = [
+            self._state.ws.gc_wait_for(
+                watch.GCToClientFindTopSourceTVGamesResponse,
+                check=partial(check, start_game),
+            )
+            for start_game in range(0, start_game + 1, 10)
+        ]
+        await self._state.ws.send_gc_message(watch.ClientToGCFindTopSourceTVGames(start_game=start_game, **kwargs))
+        async with timeout(15.0):
+            responses = await asyncio.gather(*futures)
+        # each response.game_list is 10 games (except possibly last one if filtered by hero)
+        live_matches = [LiveMatch(self._state, match) for response in responses for match in response.game_list]
+        # still need to slice the list, i.e. limit = 85, but live_matches above will have 90 matches
+        return live_matches[:limit]
+
     async def top_live_matches(self, *, hero: Hero = MISSING, limit: int = 100) -> list[LiveMatch]:
-        """Fetch top live matches
+        """Fetch top live matches.
 
         This is similar to game list in the Watch Tab of Dota 2 game app.
         "Top matches" in this context means
-
-        * featured tournament matches
-        * highest average MMR matches
+            * featured tournament matches
+            * highest average MMR matches
 
         Parameters
         ----------
         hero
-            Filter matches by Hero. Note, in this case Game Coordinator will still use only current top100 live matches,
-            i.e. requesting "filter by Muerta" will return only subset of those matches in which
-            Muerta is currently being played. It will not look into lower MMR match than top100 to extend the return
+            Filter matches by Hero. Note, in this case Game Coordinator still only uses current top100 live matches,
+            i.e. requesting "filter by Muerta" results only in subset of those matches in which
+            Muerta is currently being played. It does not look into lower MMR match than top100 to extend the return
             list to number of games from `limit` argument. This behavior is consistent with how Watch Tab works.
         limit
             Maximum amount of matches to fetch. This works rather as a boundary limit than "number of matches" to
@@ -79,42 +113,10 @@ async def top_live_matches(self, *, hero: Hero = MISSING, limit: int = 100) -> l
             Request time-outed. The reason is usually Dota 2 Game Coordinator lagging or being down.
         """
 
-        if limit < 1 or limit > 100:
-            raise ValueError("limit value should be between 1 and 100 inclusively.")
-
-        # mini-math: limit 100 -> start_game 90, 91 -> 90, 90 -> 80
-        start_game = (limit - 1) // 10 * 10
-
-        def callback(start_game: int, msg: watch.GCToClientFindTopSourceTVGamesResponse) -> bool:
-            return msg.start_game == start_game
-
-        futures = [
-            self._state.ws.gc_wait_for(
-                watch.GCToClientFindTopSourceTVGamesResponse,
-                check=partial(callback, start_game),
-            )
-            for start_game in range(0, start_game + 1, 10)
-        ]
-
-        if hero is MISSING:
-            await self._state.ws.send_gc_message(watch.ClientToGCFindTopSourceTVGames(start_game=start_game))
-        else:
-            await self._state.ws.send_gc_message(
-                watch.ClientToGCFindTopSourceTVGames(start_game=start_game, hero_id=hero.value)
-            )
+        hero_id = hero.value if hero else 0
+        return await self._fetch_find_top_source_tv_games(limit=limit, hero_id=hero_id)
 
-        async with timeout(15.0):
-            responses = await asyncio.gather(*futures)
-        # each response.game_list is 10 games (except possibly last one if filtered by hero)
-        live_matches = [LiveMatch(self._state, match) for response in responses for match in response.game_list]
-        # still need to slice the list, i.e. limit = 85, but live_matches above will have 90 matches
-        return live_matches[:limit]
-
-    async def tournament_live_matches(
-        self,
-        # todo: league_id as integer is not human-readable/gettable thing, introduce methods to easily find those
-        league_id: int,
-    ) -> list[LiveMatch]:
+    async def tournament_live_matches(self, league_id: int) -> list[LiveMatch]:
         """Fetch currently live tournament matches
 
         Parameters
@@ -142,11 +144,13 @@ async def tournament_live_matches(
             response = await future
         return [LiveMatch(self._state, match) for match in response.game_list]
 
-    async def live_matches(
-        self,
-        # todo: lobby_ids is not easy to get by the user. Introduce methods to get it, i.e. from Rich Presence
-        lobby_ids: list[int],
-    ) -> list[LiveMatch]:
+    @overload
+    async def live_matches(self, *, lobby_id: int = ...) -> LiveMatch: ...
+
+    @overload
+    async def live_matches(self, *, lobby_ids: list[int] = ...) -> list[LiveMatch]: ...
+
+    async def live_matches(self, *, lobby_id: int = MISSING, lobby_ids: list[int] = MISSING):
         """Fetch currently live matches by lobby_ids
 
         Parameters
@@ -163,28 +167,23 @@ async def live_matches(
         asyncio.TimeoutError
             Request time-outed. The reason is usually Dota 2 Game Coordinator lagging or being down.
         """
+        if lobby_id is not MISSING and lobby_ids is not MISSING:
+            raise TypeError("Cannot mix lobby_id and lobby_ids keyword arguments.")
 
-        response = await self._state.fetch_live_matches(lobby_ids)
-        return [LiveMatch(self._state, match) for match in response.game_list]
-
-    async def match_details(self, match_id: int) -> MatchDetails:
-        proto = await self._state.fetch_match_details(match_id)
-        if proto.eresult == 1:
-            return MatchDetails(self._state, proto.match)
-        else:
-            msg = f"Failed to get match_details for {match_id}"
-            raise ValueError(msg)
+        lobby_ids = [lobby_id] if lobby_id else lobby_ids
+        live_matches = await self._fetch_find_top_source_tv_games(limit=len(lobby_ids), lobby_ids=[lobby_ids])
+        # ig live_matches[0] is IndexError safe because if lobby_id is not valid then TimeoutError occurs above;
+        return live_matches[0] if lobby_id else live_matches
 
-    async def match_minimal(self, match_id: int) -> MatchMinimal:
-        proto = await self._state.fetch_match_minimal(match_ids=[match_id])
-        return MatchMinimal(self._state, proto.matches[0])
+    async def matches_minimal(self, match_id: int) -> list[MatchMinimal]:
+        proto = await self._state.fetch_matches_minimal(match_ids=[match_id])
+        return [MatchMinimal(self._state, match) for match in proto.matches]
 
     async def matchmaking_stats(self):
         future = self._state.ws.gc_wait_for(client_messages.MatchmakingStatsResponse)
         await self._state.ws.send_gc_message(client_messages.MatchmakingStatsRequest())
         async with timeout(15.0):
             return await future
-        # return BehaviorSummary(behavior_score=resp.rank_value, communication_score=resp.rank_data1)
 
     if TYPE_CHECKING or DOCS_BUILDING:
 

steam/ext/dota2/enums.py

@@ -15,6 +15,7 @@
     "Hero",
     "GameMode",
     "LobbyType",
+    "Outcome",
     "RankTier",
 )
 
@@ -291,6 +292,9 @@ def display_name(self) -> str:
     def id(self) -> int:
         return self.value
 
+    def __bool__(self) -> bool:  # type: ignore # idk I need Hero.NONE to be False
+        return bool(self.value)
+
 
 class GameMode(IntEnum):  # source: dota_shared_enums.proto
     NONE                = 0
@@ -435,22 +439,20 @@ class RankTier(IntEnum):
 
     @property
     def division(self) -> str:
-        if self.name[-1].isdigit():
-            return self.name[:-1]
-        else:
+        if self.value % 10 == 0:
             return self.name
+        else:
+            return self.name[:-1]
 
     @property
     def stars(self) -> str:
         return self.value % 10
 
-
-    # @property
-    # def display_name(self) -> str:
-    #     if self.value % 10 == 0:
-    #         return self.name
-    #     else:
-    #         return self.name[:-1] + ' ' + self.name[-1]
+    # do we need it as a factory helper method?
+    @property
+    def display_name(self) -> str:
+        suffix = f' {self.stars}' if self.stars else ''
+        return self.division + suffix
 
 
 class Outcome(IntEnum):  # source: dota_shared_enums.proto

steam/ext/dota2/state.py

@@ -2,15 +2,15 @@
 
 from __future__ import annotations
 
+from asyncio import timeout
 from typing import TYPE_CHECKING, Any
 
-from ..._const import timeout
 from ..._gc import GCState as GCState_
 from ...app import DOTA2
 from ...id import _ID64_TO_ID32
 from ...state import parser
 from .models import PartialUser, User
-from .protobufs import client_messages, common, sdk, watch
+from .protobufs import sdk, watch
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Sequence
@@ -21,7 +21,7 @@
     from .client import Client
 
 
-class GCState(GCState_[Any]):  # todo: implement basket-analogy for dota2
+class GCState(GCState_[Any]):  # TODO: implement basket-analogy for dota2
     client: Client  # type: ignore  # PEP 705
     _users: WeakValueDictionary[ID32, User]
     _APP = DOTA2  # type: ignore
@@ -70,49 +70,13 @@ async def parse_gc_client_connect(self, msg: sdk.ClientWelcome) -> None:
             self.dispatch("gc_ready")
 
     # dota fetch proto calls
-    # the difference between these and the functions in `client`/`models` is that
-    # these give raw proto responses while the latter modelize/structure/refine them.
+    # shortcuts to proto calls that are used more than once in `client`/`models`
 
-    async def fetch_user_dota2_profile_card(self, user_id: int) -> common.ProfileCard:
-        """Fetch User's Dota 2 Profile Card.
+    async def fetch_matches_minimal(self, match_ids: list[int]) -> watch.ClientToGCMatchesMinimalResponse:
+        """Fetch Matches Minimal.
 
-        Contains basic info about the account. Kinda mirrors old profile page.
-        """
-        await self.ws.send_gc_message(client_messages.ClientToGCGetProfileCard(account_id=user_id))
-        return await self.ws.gc_wait_for(
-            common.ProfileCard,
-            check=lambda msg: msg.account_id == user_id,
-        )
-
-    async def fetch_match_details(self, match_id: int) -> client_messages.MatchDetailsResponse:
-        """Fetch Match Details.
-
-        This call is for already finished games. Contains most of the info that can be found in post-match stats.
-        """
-        await self.ws.send_gc_message(client_messages.MatchDetailsRequest(match_id=match_id))
-        async with timeout(15.0):
-            return await self.ws.gc_wait_for(
-                client_messages.MatchDetailsResponse,
-                check=lambda msg: msg.match.match_id == match_id,
-            )
-
-    async def fetch_match_minimal(self, match_ids: list[int]) -> watch.ClientToGCMatchesMinimalResponse:
-        """Fetch Match Minimal.
-
-        This call is for already finished games. Contains basic data about the match.
+        This call is for already finished games. Contains basic data about the matches.
         """
         await self.ws.send_gc_message(watch.ClientToGCMatchesMinimalRequest(match_ids=match_ids))
         async with timeout(15.0):
-            return await self.ws.gc_wait_for(
-                watch.ClientToGCMatchesMinimalResponse,
-            )
-
-    async def fetch_live_matches(self, lobby_ids: list[int]) -> watch.GCToClientFindTopSourceTVGamesResponse:
-        """Fetch Live Match by lobby ids."""
-        await self.ws.send_gc_message(watch.ClientToGCFindTopSourceTVGames(lobby_ids=lobby_ids))
-        async with timeout(15.0):
-            # todo: test with more than 10 lobby_ids, Game Coordinator will probably chunk it wrongly or fail at all
-            return await self.ws.gc_wait_for(
-                watch.GCToClientFindTopSourceTVGamesResponse,
-                check=lambda msg: msg.specific_games == True,
-            )
+            return await self.ws.gc_wait_for(watch.ClientToGCMatchesMinimalResponse)