bsd-runner/discord.py at master · spacevx/bsd-runner · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from typing import Any, Final
from enum import Enum, auto
import time
import sys

from strings import rpcGameName, rpcInMenu, rpcPlaying, rpcGameOver


_BROWSER_ENV: Final[bool] = sys.platform == "emscripten"

AioPresence: Any = None
DiscordNotFound: Any = Exception
PipeClosed: Any = Exception
_PYPRESENCE_AVAILABLE: bool = False

if not _BROWSER_ENV:
    try:
        from pypresence import AioPresence as _AioPresence, DiscordNotFound as _DiscordNotFound, PipeClosed as _PipeClosed  # type: ignore
        AioPresence = _AioPresence
        DiscordNotFound = _DiscordNotFound
        PipeClosed = _PipeClosed
        _PYPRESENCE_AVAILABLE = True
    except ImportError:
        pass


clientId: Final[str] = "1468228058472386624"

class PresenceState(Enum):
    MENU = auto()
    PLAYING = auto()
    GAME_OVER = auto()


class DiscordRPC:
    def __init__(self) -> None:
        self.rpc: Any = None
        self.bConnected: bool = False
        self.currentState: PresenceState | None = None
        self.currentScore: int = 0
        self.startTime: int = 0

    async def connect(self) -> None:
        if not _PYPRESENCE_AVAILABLE or AioPresence is None:
            self.bConnected = False
            return
        try:
            self.rpc = AioPresence(clientId)
            await self.rpc.connect()
            self.bConnected = True
            self.startTime = int(time.time())
        except (DiscordNotFound, PipeClosed, ConnectionRefusedError, FileNotFoundError):
            self.bConnected = False
        except Exception:
            self.bConnected = False

    async def updateMenu(self) -> None:
        if not self.bConnected or self.rpc is None:
            return
        if self.currentState == PresenceState.MENU:
            return
        self.currentState = PresenceState.MENU
        try:
            await self.rpc.update(
                state=rpcInMenu,
                large_image="game_logo",
                large_text=rpcGameName,
                start=self.startTime
            )
        except (PipeClosed, ConnectionRefusedError, BrokenPipeError, RuntimeError):
            self.bConnected = False

    async def updatePlaying(self, score: int) -> None:
        if not self.bConnected or self.rpc is None:
            return
        if self.currentState == PresenceState.PLAYING and self.currentScore == score:
            return
        self.currentState = PresenceState.PLAYING
        self.currentScore = score
        try:
            await self.rpc.update(
                state=rpcPlaying.format(score=score),
                large_image="game_logo",
                large_text=rpcGameName,
                start=self.startTime
            )
        except (PipeClosed, ConnectionRefusedError, BrokenPipeError, RuntimeError):
            self.bConnected = False

    async def updateGameOver(self, finalScore: int) -> None:
        if not self.bConnected or self.rpc is None:
            return
        if self.currentState == PresenceState.GAME_OVER and self.currentScore == finalScore:
            return
        self.currentState = PresenceState.GAME_OVER
        self.currentScore = finalScore
        try:
            await self.rpc.update(
                state=rpcGameOver.format(score=finalScore),
                large_image="game_logo",
                large_text=rpcGameName,
                start=self.startTime
            )
        except (PipeClosed, ConnectionRefusedError, BrokenPipeError, RuntimeError):
            self.bConnected = False

    async def close(self) -> None:
        if self.rpc is not None:
            try:
                await self.rpc.clear()
            except (PipeClosed, ConnectionRefusedError, BrokenPipeError, Exception):
                pass
            try:
                self.rpc.close()
            except (PipeClosed, ConnectionRefusedError, BrokenPipeError, Exception):
                pass
        self.bConnected = False
        self.rpc = None