Adapt style of discordrp

src/prism/discordrp/__init__.py

@@ -22,56 +22,56 @@
 SOFTWARE.
 
 Copied from https://github.com/TenType/discord-rich-presence/
+
+Adapted for use in Prism
 """
 
 import json
 import os
 import socket
 import struct
 import sys
-
+from collections.abc import Mapping
 from enum import IntEnum
-from typing import Any
+from types import TracebackType
+from typing import Any, Self
 from uuid import uuid4
 
+
 class OpCode(IntEnum):
-    """
-    A list of valid opcodes that can be sent in packets to Discord.
-    """
+    """A list of valid opcodes that can be sent in packets to Discord."""
+
     HANDSHAKE = 0
     FRAME = 1
     CLOSE = 2
     PING = 3
     PONG = 4
 
-SOCKET_NAME = 'discord-ipc-{}'
 
-WINDOWS = 'win32'
+SOCKET_NAME = "discord-ipc-{}"
+
 
 class PresenceError(Exception):
-    """
-    Errors emitted by the Presence class.
-    """
+    """Errors emitted by the Presence class."""
+
 
 class Presence:
-    """
-    The main class used to connect to Discord for its rich presence API.
-    """
+    """The main class used to connect to Discord for its rich presence API."""
 
-    def __init__(self, client_id: str):
+    def __init__(self, client_id: str) -> None:
         self.client_id = client_id
-        self._platform = sys.platform
-        self._socket = None
+        self._socket: Any = None  # Either BufferedWriter or socket
 
         # Connect to Discord IPC
         self._connect()
 
         # Send a handshake request
         self._handshake()
 
-    def set(self, activity):
+    def set(self, activity: Mapping[str, object] | None) -> None:
         """
         Sends an activity payload to Discord.
+
         :param activity: A dictionary of this format:
 
         ```
@@ -103,30 +103,25 @@ def set(self, activity):
         One field of either 'state', 'details', or 'timestamps.start' is required.
         """
         payload = {
-            'cmd': 'SET_ACTIVITY',
-            'args': {
-                'pid': os.getpid(),
-                'activity': activity,
+            "cmd": "SET_ACTIVITY",
+            "args": {
+                "pid": os.getpid(),
+                "activity": activity,
             },
-            'nonce': str(uuid4()),
+            "nonce": str(uuid4()),
         }
         self._send(payload, OpCode.FRAME)
 
-    def clear(self):
-        """
-        Clears the current activity.
-        """
-        self.set(None)
-
-    def close(self):
+    def close(self) -> None:
         """
         Closes the current connection.
-        This method is automatically called when the program exits using the 'with' statement.
+
+        Automatically called when the program exits using the 'with' statement.
         """
         self._send({}, OpCode.CLOSE)
         self._socket.close()
 
-    def _connect(self):
+    def _connect(self) -> None:
         pipe = self._get_pipe()
 
         # Try to connect to a socket, starting from 0 up to 9
@@ -137,75 +132,82 @@ def _connect(self):
             except FileNotFoundError:
                 pass
         else:
-            raise PresenceError('Cannot find a socket to connect to Discord')
+            raise PresenceError("Cannot find a socket to connect to Discord")
 
     def _get_pipe(self) -> str:
-        if self._platform == WINDOWS:
+        if sys.platform == "win32":
             # Windows pipe
-            return R'\\?\pipe\\' + SOCKET_NAME
+            return R"\\?\pipe\\" + SOCKET_NAME
 
         # Unix pipe
-        for env in ('XDG_RUNTIME_DIR', 'TMPDIR', 'TMP', 'TEMP'):
+        for env in ("XDG_RUNTIME_DIR", "TMPDIR", "TMP", "TEMP"):
             path = os.environ.get(env)
             if path is not None:
                 return path + SOCKET_NAME
 
-        return '/tmp/' + SOCKET_NAME
+        return "/tmp/" + SOCKET_NAME
 
-    def _try_socket(self, pipe: str, i: int):
-        if self._platform == WINDOWS:
-            self._socket = open(pipe.format(i), 'wb')
+    def _try_socket(self, pipe: str, i: int) -> None:
+        if sys.platform == "win32":
+            self._socket = open(pipe.format(i), "wb")
         else:
             self._socket = socket.socket(socket.AF_UNIX)
             self._socket.connect(pipe.format(i))
 
-    def _handshake(self):
+    def _handshake(self) -> None:
         data = {
-            'v': 1,
-            'client_id': self.client_id,
+            "v": 1,
+            "client_id": self.client_id,
         }
         self._send(data, OpCode.HANDSHAKE)
-        _, data = self._read()
+        _, response = self._read()
 
-        if data.get('evt') != 'READY':
-            raise PresenceError('Discord returned an error response after a handshake request')
+        if response.get("evt") != "READY":
+            raise PresenceError(
+                "Discord returned an error response after a handshake request"
+            )
 
-    def _read(self) -> tuple[int, dict[str, Any]]:
+    def _read(self) -> tuple[int, Mapping[str, object]]:
         op, length = self._read_header()
         payload = self._read_bytes(length)
-        decoded = payload.decode('utf-8')
+        decoded = payload.decode("utf-8")
         data = json.loads(decoded)
         return op, data
 
     def _read_header(self) -> tuple[int, int]:
-        return struct.unpack('<ii', self._read_bytes(8))
+        return struct.unpack("<ii", self._read_bytes(8))  # type: ignore [return-value]
 
     def _read_bytes(self, size: int) -> bytes:
-        encoded = b''
+        encoded = b""
         while size > 0:
-            if self._platform == WINDOWS:
+            if sys.platform == "win32":
                 encoded += self._socket.read(size)
             else:
                 encoded += self._socket.recv(size)
 
             size -= len(encoded)
         return encoded
 
-    def _send(self, payload: dict[str, int], op: OpCode):
+    def _send(self, payload: Mapping[str, object], op: OpCode) -> None:
         data_json = json.dumps(payload)
-        encoded = data_json.encode('utf-8')
-        header = struct.pack('<ii', int(op), len(encoded))
+        encoded = data_json.encode("utf-8")
+        header = struct.pack("<ii", int(op), len(encoded))
         self._write(header + encoded)
 
-    def _write(self, data: bytes):
-        if self._platform == WINDOWS:
+    def _write(self, data: bytes) -> None:
+        if sys.platform == "win32":
             self._socket.write(data)
             self._socket.flush()
         else:
             self._socket.sendall(data)
 
-    def __enter__(self):
+    def __enter__(self) -> Self:
         return self
 
-    def __exit__(self, exc_type, exc_value, exc_traceback):
+    def __exit__(
+        self,
+        exc_type: type[BaseException],
+        exc_value: BaseException,
+        exc_traceback: TracebackType | None,
+    ) -> None:
         self.close()