From 7c8b2d8cf1d0ad265093ed54b4d7acb44be36c66 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Laure=CE=B7t?= Date: Wed, 3 Nov 2021 18:40:11 +0100 Subject: [PATCH] feat: created (partially) an admin interface --- src/admin.html | 79 +++++++++++++++++++++++++++++++++++++++++++++++++ src/emulator.py | 11 +++---- src/server.py | 38 ++++++++++++++++++++---- 3 files changed, 117 insertions(+), 11 deletions(-) create mode 100644 src/admin.html diff --git a/src/admin.html b/src/admin.html new file mode 100644 index 0000000..d18ef0b --- /dev/null +++ b/src/admin.html @@ -0,0 +1,79 @@ + + + + + Admin + + + + + + + +
+ + +
+ +
+ + +
+ + + + + diff --git a/src/emulator.py b/src/emulator.py index af41c08..debde8d 100644 --- a/src/emulator.py +++ b/src/emulator.py @@ -1,6 +1,7 @@ import asyncio import json import logging +import time import mgba.core import mgba.image @@ -10,8 +11,8 @@ import pyvirtualcam import websockets from mgba._pylib import ffi -WIDTH = 240 -HEIGHT = 160 +WIDTH: int = 240 +HEIGHT: int = 160 URI: str = "ws://127.0.0.1:6789/" FPS: int = 60 HZ: int = 10 @@ -36,7 +37,7 @@ core.set_video_buffer(screen) core.reset() mgba.log.silence() -logging.basicConfig(level=logging.DEBUG) +# logging.basicConfig(level=logging.DEBUG) def parse_message(message: dict[str, str]): @@ -60,7 +61,7 @@ def parse_message(message: dict[str, str]): data = message["admin"] if data == "save": # voodoo magic incomming state = core.save_raw_state() - with open("states/test.state", "wb") as state_file: + with open(f"states/{time.strftime('%Y-%m-%dT%H:%M:%S')}.state", "wb") as state_file: for byte in state: state_file.write(byte.to_bytes(4, byteorder="big", signed=False)) elif data == "load": # black magic incomming @@ -78,7 +79,7 @@ async def main(): with pyvirtualcam.Camera(width=WIDTH, height=HEIGHT, fps=FPS) as cam: logging.debug(f"Using virtual camera: {cam.device}") async with websockets.connect(URI) as websocket: - await websocket.send('{"auth":"password"}') + await websocket.send('{"auth":"emulator_password"}') logging.debug(f"connected to: {websocket}") while True: diff --git a/src/server.py b/src/server.py index 593291f..04cb30a 100644 --- a/src/server.py +++ b/src/server.py @@ -1,6 +1,7 @@ import asyncio import json import logging +import os import time from typing import Any @@ -8,15 +9,22 @@ import websockets from utils import User, Users, Votes -logging.basicConfig(level=logging.DEBUG) +# logging.basicConfig(level=logging.DEBUG) -PASSWORD_ADMIN: str = "password" -PASSWORD_EMU: str = "password" +PASSWORD_ADMIN: str = "admin_password" +PASSWORD_EMU: str = "emulator_password" VOTES: Votes = Votes() USERS: Users = Users() -async def parse_message(user: User, message: dict[str, str]): +async def send_states(user: User) -> None: + files = os.listdir("states") + states = list(filter(lambda x: x.endswith(".state"), files)) + message = json.dumps({"state": states}) + await user.send(message) + + +async def parse_message(user: User, message: dict[str, str]) -> None: """Parse the user's message. Args: @@ -25,12 +33,14 @@ async def parse_message(user: User, message: dict[str, str]): """ if "auth" in message: data = message["auth"] - if USERS.emulator is None and data == PASSWORD_EMU: + if USERS.emulator is None and data == PASSWORD_EMU and user != USERS.admin: USERS.emulator = user logging.debug(f"emulator authenticated: {user}") - elif USERS.admin is None and data == PASSWORD_ADMIN: + await user.send('{"auth":"success"}') + elif USERS.admin is None and data == PASSWORD_ADMIN and user != USERS.emulator: USERS.admin = user logging.debug(f"admin authenticated: {user}") + await user.send('{"auth":"success"}') if "action" in message: data = message["action"] @@ -65,6 +75,16 @@ async def parse_message(user: User, message: dict[str, str]): else: logging.error(f"user is not emulator: {user}") + if "state" in message: + data = message["state"] + if user == USERS.admin: + if data == "get": + await send_states(user) + else: + logging.error(f"unsupported state action: {data}") + else: + logging.error(f"user is not admin: {user}") + async def handler(websocket: Any, path: str): """Handle the messages sent by a user. @@ -84,6 +104,12 @@ async def handler(websocket: Any, path: str): finally: # Unregister user USERS.unregister(user) + if user == USERS.admin: + USERS.admin = None + logging.debug(f"admin disconnected: {user}") + elif user == USERS.emulator: + logging.debug(f"emulator disconnected: {user}") + USERS.emulator = None async def main():