GameboyServer/server.py at master · notdylanburns/GameboyServer · 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
import asyncio, websockets, json, hashlib, random, string, time, math, os
import pyboy
from pyboy import PyBoy, WindowEvent
import numpy as np
from io import BytesIO
import zlib

passphrase = "rickandmortyseason5"
validkeys = []

pyboys = {}
screens = {}
buttons = {
    "START_PRESS": WindowEvent.PRESS_BUTTON_START,
    "SELECT_PRESS": WindowEvent.PRESS_BUTTON_SELECT,
    "A_PRESS": WindowEvent.PRESS_BUTTON_A,
    "B_PRESS": WindowEvent.PRESS_BUTTON_B,
    "UP_PRESS": WindowEvent.PRESS_ARROW_UP,
    "DOWN_PRESS": WindowEvent.PRESS_ARROW_DOWN,
    "LEFT_PRESS": WindowEvent.PRESS_ARROW_LEFT,
    "RIGHT_PRESS": WindowEvent.PRESS_ARROW_RIGHT,
    "START_RELEASE": WindowEvent.RELEASE_BUTTON_START,
    "SELECT_RELEASE": WindowEvent.RELEASE_BUTTON_SELECT,
    "A_RELEASE": WindowEvent.RELEASE_BUTTON_A,
    "B_RELEASE": WindowEvent.RELEASE_BUTTON_B,
    "UP_RELEASE": WindowEvent.RELEASE_ARROW_UP,
    "DOWN_RELEASE": WindowEvent.RELEASE_ARROW_DOWN,
    "LEFT_RELEASE": WindowEvent.RELEASE_ARROW_LEFT,
    "RIGHT_RELEASE": WindowEvent.RELEASE_ARROW_RIGHT
}

async def invalidRequest(websocket):
    await websocket.send("Invalid Request")

async def generateKey(websocket, requestPassphrase):
    if requestPassphrase == passphrase:
        characterPool = string.ascii_letters + string.digits + string.punctuation
        salt = ''.join(random.choice(characterPool) for i in range(30))
        num = str(math.floor(time.time()))
        key = hashlib.sha1(bytes(requestPassphrase + salt + num, 'utf-8')).hexdigest()
        validkeys.append(key)

        await websocket.send(key)
    else:
        await invalidRequest(websocket)

def deAuthKey(key):
    if key in validkeys:
        validkeys.remove(key)

async def runCommand(websocket, cmd):
    if cmd["command"] == "stop":
        pyboys[cmd["authentication"]].stop()
        pyboys.pop(cmd["authentication"])
        screens.pop(cmd["authentication"])
        deAuthKey(cmd["authentication"])
        await websocket.send("Key Invalidated")
    elif cmd["command"] == "getRoms":
        path = "roms"
        files = os.listdir(path)
        roms = []
        for f in files:
            if f.endswith(".gbc"):
                roms.append(f[:-4])
        await websocket.send(json.dumps(roms))
    elif cmd["command"] == "start":
        filename = "roms/" + cmd["rom"] + ".gbc"
        if os.path.isfile(filename):
            pyboys[cmd["authentication"]] = PyBoy(filename, window_type="headless", disable_renderer=True)
            pyboys[cmd["authentication"]].set_emulation_speed(1)
            screens[cmd["authentication"]] = pyboys[cmd["authentication"]].botsupport_manager().screen()
            await websocket.send("PyBoy Instance Created")
        else:
            await invalidRequest(websocket);
    elif cmd["command"] == "getFrame":
        pyboys[cmd["authentication"]].tick()
        image = screens[cmd["authentication"]].screen_ndarray()
        screenBuffer = zlib.compress(json.dumps(image.tolist()).encode('utf-8'), level=-1)
        await websocket.send(screenBuffer)
    elif cmd["command"] == "sendInput":
        for button in cmd["buttons"]:
            pyboys[cmd["authentication"]].send_input(buttons[button])
        #await websocket.send("Buttons Pressed")


async def server(websocket, path):
    while True:
        requestRaw = await websocket.recv()
        try:
            request = json.loads(requestRaw)
            valid = False
            #if client requests to close the connection, break the loop
            if request["command"] == "close":
                valid = True
                if request["authentication"]:
                    deAuthKey(request["authentication"])
                break
            #authenticate commands do not require a valid key
            elif request["command"] == "authenticate":
                valid = True
                await generateKey(websocket, request["authentication"])
            #any other command requires authentication with a valid key
            elif request["authentication"] in validkeys:
                valid = True
                await runCommand(websocket, request)
            if not valid:
                await invalidRequest(websocket)
        except Exception as e:
            print(e)
            await invalidRequest(websocket)


start_server = websockets.serve(server, "192.168.1.114", 8765)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()