Laurent/booplaysgba
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

feat: websocket server

Browse source 
feat: emulator (virtualcam)
feat: basic web client
This commit is contained in:
Laureηt 2021-10-31 22:17:50 +01:00
parent 56d309e953
commit 7b7fbde1fe
No known key found for this signature in database
GPG key ID: D88C6B294FD40994
4 changed files with 254 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -1,3 +1,4 @@
roms/ roms/
.venv/ .venv/
.mypy_cache/ .mypy_cache/
states/

67
src/client.html Normal file
View file

@ -0,0 +1,67 @@
<!DOCTYPE html>
<html>
<head>
<title>Telecommande</title>
</head>
<body>
<button id="a">a</button>
<button id="b">b</button>
<button id="select">select</button>
<button id="start">start</button>
<button id="right">right</button>
<button id="left">left</button>
<button id="up">up</button>
<button id="down">down</button>
<button id="r">r</button>
<button id="l">l</button>
<script>
var websocket = new WebSocket("ws://127.0.0.1:6789/");
var a = document.getElementById('a'),
b = document.getElementById('b'),
select = document.getElementById('select'),
start = document.getElementById('start'),
right = document.getElementById('right'),
left = document.getElementById('left'),
up = document.getElementById('up'),
down = document.getElementById('down'),
r = document.getElementById('r'),
l = document.getElementById('l');
a.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'a' }));
}
b.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'b' }));
}
select.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'select' }));
}
start.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'start' }));
}
right.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'right' }));
}
left.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'left' }));
}
up.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'up' }));
}
down.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'down' }));
}
r.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'r' }));
}
l.onclick = function (event) {
websocket.send(JSON.stringify({ action: 'l' }));
}
</script>
</body>
</html>

67
src/emulator.py Normal file
View file

@ -0,0 +1,67 @@
import asyncio
import logging
import mgba.core
import mgba.image
import mgba.log
import numpy as np
import pyvirtualcam
import websockets
WIDTH = 240
HEIGHT = 160
URI: str = "ws://127.0.0.1:6789/"
KEYMAP: dict[str, int] = {
"a": 0,
"b": 1,
"select": 2,
"start": 3,
"right": 4,
"left": 5,
"up": 6,
"down": 7,
"r": 8,
"l": 9,
}
core = mgba.core.load_path("roms/pokemon.gba")
screen = mgba.image.Image(WIDTH, HEIGHT)
core.set_video_buffer(screen)
core.reset()
mgba.log.silence()
logging.basicConfig(level=logging.DEBUG)
async def main():
with pyvirtualcam.Camera(width=WIDTH, height=HEIGHT, fps=60) as cam:
logging.debug(f"Using virtual camera: {cam.device}")
async with websockets.connect(URI) as websocket:
await websocket.send('{"auth":"password"}')
logging.debug(f"connected to: {websocket}")
test = core.save_raw_state()
while True:
if core.frame_counter % 30 == 0:
await websocket.send('{"action":"get"}')
response = await websocket.recv()
if response in KEYMAP:
key = KEYMAP[response]
core.set_keys(key)
logging.debug(f"pressing: {key}")
# with open("pokemon.state", "w") as f:
# f.write(test)
core.load_raw_state(test)
print(str(test))
core.run_frame()
core.clear_keys(*KEYMAP.values())
frame = np.array(screen.to_pil().convert("RGB"), np.uint8)
cam.send(frame)
cam.sleep_until_next_frame()
asyncio.run(main())

119
src/server.py Normal file
View file

@ -0,0 +1,119 @@
import asyncio
import json
import logging
import time
from typing import Any
import websockets
logging.basicConfig(level=logging.DEBUG)
class User:
websocket: Any = None
last_message: float = time.time()
has_voted: bool = False
def register(self, websocket: Any):
self.websocket = websocket
USERS.add(self)
logging.debug(
f"user registered: {self}",
)
def unregister(self):
# self.websocket.close()
USERS.remove(self)
logging.debug(
f"user unregistered: {self}",
)
def __str__(self) -> str:
return f"{self.websocket.remote_address} ({self.websocket.id})"
USERS: set[User] = set()
EMULATOR: Any = None
PASSWORD: str = "password"
VOTES: dict[str, int] = {
"a": 0,
"b": 0,
"select": 0,
"start": 0,
"right": 0,
"left": 0,
"up": 0,
"down": 0,
"r": 0,
"l": 0,
}
def clear_votes():
for key in VOTES.keys():
VOTES[key] = 0
for user in USERS:
user.has_voted = False
logging.info(f"votes cleared: {VOTES}")
def next_move():
if any(VOTES.values()):
return max(VOTES, key=VOTES.get)
else:
return "null"
async def handler(websocket, path):
try:
global EMULATOR
# Register user
user = User()
user.register(websocket)
# Manage received messages
async for message in websocket:
data = json.loads(message)
if "auth" in data:
auth: str = data["auth"]
logging.debug(f"auth received: {auth}")
if not EMULATOR and auth == PASSWORD:
EMULATOR = websocket
user.unregister()
logging.debug(f"emulator authenticated: {EMULATOR}")
if "action" in data:
action: str = data["action"]
if action in VOTES:
VOTES[action] += 1
user.last_message = time.time()
user.has_voted = True
logging.debug(f"key received: {action} ({VOTES[action]}), from {user}")
elif action == "get":
if EMULATOR and websocket == EMULATOR:
move = next_move()
await websocket.send(move)
logging.info(f"vote sent: {move}")
clear_votes()
else:
logging.error(f"user is not EMULATOR: {user}")
else:
logging.error(f"unsupported action: {data}")
finally:
# Unregister user
user.unregister()
async def main():
async with websockets.serve(handler, "localhost", 6789):
await asyncio.Future() # run forever
if __name__ == "__main__":
asyncio.run(main())