Add rich presence helper and Client.set_rich_presence

discord/__init__.py

@@ -68,6 +68,7 @@
 from .read_state import *
 from .reaction import *
 from .relationship import *
+from .rich_presence import RichPresence, get_external_assets # import discord.richPresence works too.
 from .role import *
 from .scheduled_event import *
 from .settings import *

discord/client.py

@@ -24,6 +24,8 @@
 
 from __future__ import annotations
 
+
+import time
 import asyncio
 from datetime import datetime
 import logging
@@ -65,6 +67,7 @@
 from .gateway import *
 from .gateway import ConnectionClosed
 from .activity import ActivityTypes, BaseActivity, Session, Spotify, create_activity
+from .rich_presence import RichPresence
 from .voice_client import VoiceClient
 from .http import HTTPClient
 from .state import ConnectionState
@@ -1870,6 +1873,46 @@ async def change_presence(
             if payload:
                 await self.settings.edit(**payload)
 
+    async def set_rich_presence(
+        self,
+        rp: RichPresence,
+        *,
+        status: str = "online",
+        afk: bool = False,
+    ) -> None:
+        """
+        @mikasa: Send a raw OP 3 PRESENCE_UPDATE with a RichPresence activity.
+        """
+        activity_payload = rp.to_dict()
+
+        d = {
+            "since": int(time.time() * 1000),
+            "activities": [activity_payload],
+            "status": status,  # "online", "idle", "dnd", "invisible"
+            "afk": afk,
+        }
+
+        conn = self._connection
+        ws = None
+
+        # Depending on version, this could be _get_websocket() or .ws
+        if hasattr(conn, "_get_websocket") and callable(conn._get_websocket):
+            ws = conn._get_websocket()
+        elif hasattr(conn, "ws"):
+            ws = conn.ws
+
+        if ws is None:
+            raise RuntimeError("Could not find websocket on client._connection")
+
+        payload = {"op": 3, "d": d}
+
+        if hasattr(ws, "send_json"):
+            await ws.send_json(payload)
+        else:
+            import json
+            await ws.send(json.dumps(payload))
+
+
     async def change_voice_state(
         self,
         *,