Initial draft of HypeTrainStatus

Author: chillymosh

twitchio/http.py

@@ -123,6 +123,7 @@
         GlobalChatBadgesResponse,
         GlobalEmotesResponse,
         HypeTrainEventsResponseData,
+        HypeTrainStatusResponse,
         ModeratedChannelsResponseData,
         ModeratorsResponseData,
         PollsResponse,
@@ -1609,6 +1610,12 @@ def converter(data: HypeTrainEventsResponseData, *, raw: Any) -> HypeTrainEvent:
 
         return iterator
 
+    async def get_hype_train_status(self, broadcaster_id: str | int, token_for: str) -> HypeTrainStatusResponse:
+        params = {"broadcaster_id": broadcaster_id}
+
+        route: Route = Route("GET", "hypetrain/status", params=params, token_for=token_for)
+        return await self.request_json(route)
+
     ### Moderation ###
 
     @handle_user_ids()

twitchio/models/hype_train.py

@@ -24,7 +24,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Literal, NamedTuple
 
 from twitchio.user import PartialUser
 from twitchio.utils import parse_timestamp
@@ -34,9 +34,16 @@
     import datetime
 
     from twitchio.http import HTTPClient
-    from twitchio.types_.responses import HypeTrainEventsResponseContributions, HypeTrainEventsResponseData
+    from twitchio.types_.responses import (
+        HypeTrainEventsResponseContributions,
+        HypeTrainEventsResponseData,
+        HypeTrainStatusResponseData,
+        HypeTrainStatusTopContributions,
+    )
+
+    CombinedContributionType = Literal["BITS", "SUBS", "OTHER", "bits", "subscriptions", "other"]
 
-__all__ = ("HypeTrainContribution", "HypeTrainEvent")
+__all__ = ("HypeTrainAllTimeHigh", "HypeTrainContribution", "HypeTrainEvent", "HypeTrainStatus")
 
 
 class HypeTrainEvent:
@@ -119,18 +126,157 @@ class HypeTrainContribution:
     total: int
         The total amount contributed. If type is ``BITS``, total represents the amount of Bits used.
         If type is ``SUBS``, total is 500, 1000, or 2500 to represent tier 1, 2, or 3 subscriptions, respectively.
-    type: typing.Literal["BITS", "SUBS", "OTHER"]
-        Identifies the contribution method, either BITS, SUBS or OTHER.
+    type: typing.Literal["BITS", "SUBS", "OTHER", "bits", "subscription", "other"]
+        Identifies the contribution method.
+
+        **HypeTrainEvent:**
+
+        - **BITS**: cheering with Bits.
+        - **SUBS**: subscription activity like subscribing or gifting subscriptions.
+        - **OTHER**: covers other contribution methods not listed.
+
+        **HypeTrainStatus:**
+
+        - **bits**: cheering with Bits.
+        - **subscription**: subscription activity like subscribing or gifting subscriptions.
+        - **other**: covers other contribution methods not listed.
+
     user: PartialUser
         The user making the contribution.
     """
 
-    __slots__ = "total", "type", "user"
+    __slots__ = ("total", "type", "user")
 
-    def __init__(self, data: HypeTrainEventsResponseContributions, *, http: HTTPClient) -> None:
+    def __init__(
+        self, data: HypeTrainEventsResponseContributions | HypeTrainStatusTopContributions, *, http: HTTPClient
+    ) -> None:
         self.total: int = int(data["total"])
-        self.type: Literal["BITS", "SUBS", "OTHER"] = data["type"]
-        self.user = PartialUser(data["user"], None, http=http)
+        self.type: CombinedContributionType = data["type"]
+
+        if "user" in data:
+            self.user = PartialUser(data["user"], http=http)
+        else:
+            self.user = PartialUser(data["user_id"], data["user_login"], data["user_name"], http=http)
 
     def __repr__(self) -> str:
         return f"<HypeTrainContribution total={self.total} type={self.type} user={self.user}>"
+
+
+class HypeTrainAllTimeHigh(NamedTuple):
+    """The all time high data for a Hype Train.
+
+    Attributes
+    -----------
+    level: int
+        The level of the record Hype Train.
+    total: int
+        The total amount contributed to the record Hype Train.
+    achieved_at: datetime.datetime
+        The datetime when the record was achieved.
+    """
+
+    level: int
+    total: int
+    achieved_at: datetime.datetime
+
+
+class HypeTrainStatus:
+    """Represents the current status of a Hype Train.
+
+    Attributes
+    -----------
+    id: str
+        The ID of the Hype Train.
+    broadcaster: PartialUser
+        The user whose channel the Hype Train is occurring on.
+    level: int
+        The current level of the Hype Train.
+    total: int1
+        The total amount contributed to the Hype Train.
+    progress: int
+        The number of points contributed to the Hype Train at the current level.
+    goal: int
+        The value needed to reach the next level.
+    top_contributions: list[HypeTrainContribution]
+        The contributors with the most points contributed.
+    started_at: datetime.datetime
+        The time the Hype Train started.
+    expires_at: datetime.datetime
+        The time the Hype Train expires.
+    type: Literal["treasure", "golden_kappa", "regular"]
+        The type of Hype Train. Can be one of `treasure`, `golden_kappa`, or `regular`.
+    all_time_high: HypeTrainAllTimeHigh | None
+        Information about the channel's hype train records.
+    shared_train: bool
+        Whether this Hype Train is a shared train.
+    shared_train_participants: list[PartialUser]
+        A list containing the broadcasters participating in the shared hype train.
+    shared_all_time_high: HypeTrainAllTimeHigh | None
+        Information about the channel's shared hype train records.
+    """
+
+    __slots__ = (
+        "all_time_high",
+        "broadcaster",
+        "expires_at",
+        "goal",
+        "id",
+        "level",
+        "progress",
+        "shared_all_time_high",
+        "shared_train",
+        "shared_train_participants",
+        "started_at",
+        "top_contributions",
+        "total",
+        "type",
+    )
+
+    def __init__(self, data: HypeTrainStatusResponseData, *, http: HTTPClient) -> None:
+        current = data.get("current")
+        if current is None:
+            raise ValueError("HypeTrainStatus requires 'current'")
+        all_time_high = data.get("all_time_high")
+        shared_all_time_high = data.get("shared_all_time_high")
+
+        self.id: str = current["id"]
+        self.broadcaster = PartialUser(
+            current["broadcaster_user_id"], current["broadcaster_user_id"], current["broadcaster_user_name"], http=http
+        )
+        self.level: int = current["level"]
+        self.total: int = current["total"]
+        self.progress: int = current["progress"]
+        self.goal: int = current["goal"]
+        self.started_at: datetime.datetime = parse_timestamp(current["started_at"])
+        self.expires_at: datetime.datetime = parse_timestamp(current["expires_at"])
+        self.type: Literal["treasure", "golden_kappa", "regular"] = current["type"]
+        self.top_contributions: list[HypeTrainContribution] = [
+            HypeTrainContribution(c, http=http) for c in current["top_contributions"]
+        ]
+        self.all_time_high: HypeTrainAllTimeHigh | None = (
+            HypeTrainAllTimeHigh(
+                level=all_time_high["level"],
+                total=all_time_high["total"],
+                achieved_at=parse_timestamp(all_time_high["achieved_at"]),
+            )
+            if all_time_high is not None
+            else None
+        )
+        self.shared_train: bool = current["is_shared_train"]
+        self.shared_train_participants: list[PartialUser] = (
+            [
+                PartialUser(u["broadcaster_user_id"], u["broadcaster_user_login"], u["broadcaster_user_name"], http=http)
+                for u in current["shared_train_participants"]
+            ]
+            if self.shared_train
+            else []
+        )
+        self.shared_all_time_high: HypeTrainAllTimeHigh | None = (
+            HypeTrainAllTimeHigh(
+                level=shared_all_time_high["level"],
+                total=shared_all_time_high["total"],
+                achieved_at=parse_timestamp(shared_all_time_high["achieved_at"]),
+            )
+            if shared_all_time_high is not None
+            else None
+        )

twitchio/types_/responses.py

@@ -66,6 +66,10 @@
     "GlobalChatBadgesResponseVersions",
     "GlobalEmotesResponse",
     "GlobalEmotesResponseData",
+    "HypeTrainEventsResponse",
+    "HypeTrainStatusResponse",
+    "HypeTrainStatusResponseData",
+    "HypeTrainStatusTopContributions",
     "OAuthResponses",
     "RawResponse",
     "RefreshTokenResponse",
@@ -1088,6 +1092,53 @@ class HypeTrainEventsResponse(TypedDict):
     pagination: Pagination
 
 
+class HypeTrainStatusTopContributions(TypedDict):
+    user_id: str
+    user_login: str
+    user_name: str
+    type: Literal["bits", "subscriptions", "other"]
+    total: int
+
+
+class HypeTrainStatusSharedParticipantsData(TypedDict):
+    broadcaster_user_id: str
+    broadcaster_user_login: str
+    broadcaster_user_name: str
+
+
+class HypeTrainStatusAllTimeHighData(TypedDict):
+    level: int
+    total: int
+    achieved_at: str
+
+
+class HypeTrainStatusCurrentData(TypedDict):
+    id: str
+    broadcaster_user_id: str
+    broadcaster_user_login: str
+    broadcaster_user_name: str
+    level: int
+    total: int
+    progress: int
+    goal: int
+    top_contributions: list[HypeTrainStatusTopContributions]
+    shared_train_participants: list[HypeTrainStatusSharedParticipantsData]
+    started_at: str
+    expires_at: str
+    type: Literal["treasure", "golden_kappa", "regular"]
+    is_shared_train: bool
+
+
+class HypeTrainStatusResponseData(TypedDict):
+    current: HypeTrainStatusCurrentData | None
+    all_time_high: HypeTrainStatusAllTimeHighData | None
+    shared_all_time_high: HypeTrainStatusAllTimeHighData | None
+
+
+class HypeTrainStatusResponse(TypedDict):
+    data: list[HypeTrainStatusResponseData]
+
+
 class CheckAutomodStatusResponseData(TypedDict):
     msg_id: str
     is_permitted: bool

twitchio/user.py

@@ -55,7 +55,7 @@
     from .models.clips import Clip, CreatedClip
     from .models.eventsub_ import ChannelChatMessageEvent, ChatMessageBadge
     from .models.goals import Goal
-    from .models.hype_train import HypeTrainEvent
+    from .models.hype_train import HypeTrainEvent, HypeTrainStatus
     from .models.moderation import (
         AutomodCheckMessage,
         AutomodSettings,
@@ -1377,6 +1377,27 @@ def fetch_hype_train_events(
             max_results=max_results,
         )
 
+    async def fetch_hype_train_status(self) -> HypeTrainStatus | None:
+        """|coro|
+
+        Fetches the current Hype Train status for the broadcaster.
+
+        .. note::
+            Requires a user access token that includes the ``channel:read:hype_train`` scope.
+
+        Returns
+        -------
+        HypeTrainStatus | None
+            HypeTrainStatus object if a Hype Train is currently active, otherwise None.
+
+        """
+        data = await self._http.get_hype_train_status(broadcaster_id=self.id, token_for=self.id)
+
+        if data["data"][0]["current"] is None:
+            return None
+
+        return HypeTrainStatus(data["data"][0], http=self._http)
+
     async def start_raid(self, to_broadcaster: str | int | PartialUser) -> Raid:
         """|coro|