mirror of
https://github.com/phin05/discord-rich-presence-plex
synced 2024-11-22 09:33:07 +00:00
92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
# type: ignore
|
|
|
|
from store.constants import isUnix, processID
|
|
from utils.logs import logger
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import struct
|
|
import time
|
|
|
|
class DiscordRpcService:
|
|
|
|
clientID = "413407336082833418"
|
|
ipcPipe = ((os.environ.get("XDG_RUNTIME_DIR", None) or os.environ.get("TMPDIR", None) or os.environ.get("TMP", None) or os.environ.get("TEMP", None) or "/tmp") + "/discord-ipc-0") if isUnix else r"\\?\pipe\discord-ipc-0"
|
|
|
|
def __init__(self):
|
|
self.loop = None
|
|
self.pipeReader = None
|
|
self.pipeWriter = None
|
|
self.connected = False
|
|
|
|
def connect(self):
|
|
if self.connected:
|
|
logger.debug("Attempt to connect Discord IPC Pipe while already connected")
|
|
return
|
|
logger.info("Connecting Discord IPC Pipe")
|
|
self.loop = asyncio.new_event_loop() if isUnix else asyncio.ProactorEventLoop()
|
|
self.loop.run_until_complete(self.handshake())
|
|
|
|
async def handshake(self):
|
|
try:
|
|
if isUnix:
|
|
self.pipeReader, self.pipeWriter = await asyncio.open_unix_connection(self.ipcPipe)
|
|
else:
|
|
self.pipeReader = asyncio.StreamReader()
|
|
self.pipeWriter, _ = await self.loop.create_pipe_connection(lambda: asyncio.StreamReaderProtocol(self.pipeReader), self.ipcPipe)
|
|
self.write(0, { "v": 1, "client_id": self.clientID })
|
|
if await self.read():
|
|
self.connected = True
|
|
except:
|
|
logger.exception("An unexpected error occured during a RPC handshake operation")
|
|
|
|
async def read(self):
|
|
try:
|
|
dataBytes = await self.pipeReader.read(1024)
|
|
data = json.loads(dataBytes[8:].decode("utf-8"))
|
|
logger.debug("[READ] %s", data)
|
|
return data
|
|
except:
|
|
logger.exception("An unexpected error occured during a RPC read operation")
|
|
self.connected = False
|
|
|
|
def write(self, op, payload):
|
|
try:
|
|
logger.debug("[WRITE] %s", payload)
|
|
payload = json.dumps(payload)
|
|
self.pipeWriter.write(struct.pack("<ii", op, len(payload)) + payload.encode("utf-8"))
|
|
except:
|
|
logger.exception("An unexpected error occured during a RPC write operation")
|
|
self.connected = False
|
|
|
|
def disconnect(self):
|
|
if not self.connected:
|
|
logger.debug("Attempt to disconnect Discord IPC Pipe while not connected")
|
|
return
|
|
logger.info("Disconnecting Discord IPC Pipe")
|
|
try:
|
|
self.pipeWriter.close()
|
|
except:
|
|
logger.exception("An unexpected error occured while closing an IPC pipe writer")
|
|
try:
|
|
self.loop.run_until_complete(self.pipeReader.read())
|
|
except:
|
|
logger.exception("An unexpected error occured while closing an IPC pipe reader")
|
|
try:
|
|
self.loop.close()
|
|
except:
|
|
logger.exception("An unexpected error occured while closing an asyncio event loop")
|
|
self.connected = False
|
|
|
|
def sendActivity(self, activity):
|
|
logger.info("Activity update: %s", activity)
|
|
payload = {
|
|
"cmd": "SET_ACTIVITY",
|
|
"args": {
|
|
"pid": processID,
|
|
"activity": activity,
|
|
},
|
|
"nonce": "{0:.2f}".format(time.time()),
|
|
}
|
|
self.write(1, payload)
|
|
self.loop.run_until_complete(self.read())
|