feat: better settings.py

src/emulator.py

@@ -7,17 +7,33 @@ import mgba.image
 import mgba.log
 import redis
 
-from settings import FPS, HEIGHT, KEYS, MGBA_KEYS, POLLING_RATE, REDIS_INIT, SPF, WIDTH
+from settings import (
+    EMULATOR_FPS,
+    EMULATOR_HEIGHT,
+    EMULATOR_POLLING_RATE,
+    EMULATOR_ROM_PATH,
+    EMULATOR_SPF,
+    EMULATOR_WIDTH,
+    FFMPEG_BITRATE,
+    FFMPEG_FPS,
+    FFMPEG_HEIGHT,
+    FFMPEG_WIDTH,
+    KEYS_ID,
+    KEYS_MGBA,
+    KEYS_RESET,
+    REDIS_HOST,
+    REDIS_PORT,
+    RTMP_STREAM_URI,
+)
 
-core = mgba.core.load_path("roms/pokemon.gba")
-# core = mgba.core.load_path("roms/BtnTest.gba")
-screen = mgba.image.Image(WIDTH, HEIGHT)
+core = mgba.core.load_path(EMULATOR_ROM_PATH)
+screen = mgba.image.Image(EMULATOR_WIDTH, EMULATOR_HEIGHT)
 core.set_video_buffer(screen)
 core.reset()
 
 logging.basicConfig(level=logging.DEBUG)
 mgba.log.silence()
-r = redis.Redis(host="redis", port=6379, db=0)
+r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)
 
 
 def next_action():
@@ -26,9 +42,9 @@ def next_action():
     Returns:
         int: key used by mgba
     """
-    votes = list(map(int, r.mget(KEYS)))
+    votes = list(map(int, r.mget(KEYS_ID)))
     if any(votes):
-        r.mset(REDIS_INIT)
+        r.mset(KEYS_RESET)
         return votes.index(max(votes))
     else:
         return -1
@@ -43,26 +59,26 @@ stream = Popen(
         "-vcodec",
         "png",
         "-r",
-        f"{FPS}",
+        f"{EMULATOR_FPS}",
         "-s",
-        f"{WIDTH}x{HEIGHT}",
+        f"{EMULATOR_WIDTH}x{EMULATOR_HEIGHT}",
         "-i",
         "-",
         "-f",
         "flv",
         "-s",
-        f"{WIDTH}x{HEIGHT}",
+        f"{FFMPEG_WIDTH}x{FFMPEG_HEIGHT}",
         "-r",
-        "30",
+        f"{FFMPEG_FPS}",
         "-b:v",
-        "2M",
+        FFMPEG_BITRATE,
         "-fflags",
         "nobuffer",
         "-flags",
         "low_delay",
         "-strict",
         "experimental",
-        "rtmp://rtmp:1935/live/test",
+        RTMP_STREAM_URI,
     ],
     stdin=PIPE,
 )
@@ -71,8 +87,8 @@ while True:
 
     last_frame_t = time.time()
 
-    if not (core.frame_counter % POLLING_RATE):
-        core.clear_keys(*MGBA_KEYS)
+    if not (core.frame_counter % EMULATOR_POLLING_RATE):
+        core.clear_keys(*KEYS_MGBA)
         next_key = next_action()
         if next_key != -1:
             core.set_keys(next_key)
@@ -82,6 +98,6 @@ while True:
     image = screen.to_pil().convert("RGB")
     image.save(stream.stdin, "PNG")
 
-    sleep_t = last_frame_t - time.time() + SPF
+    sleep_t = last_frame_t - time.time() + EMULATOR_SPF
     if sleep_t > 0:
         time.sleep(sleep_t)

src/server.py

@@ -6,13 +6,22 @@ import time
 import redis
 import websockets
 
-from settings import KEYS, PASSWORD_ADMIN, REDIS_INIT, USER_TIMEOUT
+from settings import (
+    KEYS_ID,
+    KEYS_RESET,
+    PASSWORD_ADMIN,
+    REDIS_HOST,
+    REDIS_PORT,
+    USER_TIMEOUT,
+    WEBSOCKET_LISTEN,
+    WEBSOCKET_PORT,
+)
 from utils import User, Users
 
 logging.basicConfig(level=logging.DEBUG)
 
-r = redis.Redis(host="redis", port=6379, db=0)
-r.mset(REDIS_INIT)
+r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=0)
+r.mset(KEYS_RESET)
 
 USERS: Users = Users()
 
@@ -37,7 +46,7 @@ async def parse_message(user: User, message: dict[str, str]) -> None:
         if user.last_message + USER_TIMEOUT > time.time():
             logging.debug(f"dropping action: {data}")
             return None
-        elif data in KEYS:
+        elif data in KEYS_ID:
             r.incr(data)
             user.last_message = time.time()
             user.has_voted = True
@@ -67,7 +76,7 @@ async def handler(websocket, path: str):
 
 async def main():
     """Start the websocket server."""
-    async with websockets.serve(handler, "0.0.0.0", 6789):  # nosec
+    async with websockets.serve(handler, WEBSOCKET_LISTEN, WEBSOCKET_PORT):  # nosec
         await asyncio.Future()  # run forever
 
 

src/settings.py

@@ -1,13 +1,35 @@
-WIDTH: int = 240
-HEIGHT: int = 160
+from os import getenv
 
-URI: str = "ws://127.0.0.1:6789/"
-PASSWORD_ADMIN: str = "password_admin"
+WEBSOCKET_HOST: str = getenv("WEBSOCKET_HOST", "localhost")
+WEBSOCKET_PORT: int = int(getenv("WEBSOCKET_PORT", 6789))
+WEBSOCKET_URI: str = f"ws://{WEBSOCKET_HOST}:{WEBSOCKET_PORT}/"
+WEBSOCKET_LISTEN: str = getenv("WEBSOCKET_LISTEN", "localhost")
+
+RTMP_HOST: str = getenv("WEBSOCKET_HOST", "localhost")
+RTMP_PORT: int = int(getenv("WEBSOCKET_PORT", 1935))
+RTMP_URI: str = f"rtmp://{RTMP_HOST}:{RTMP_PORT}/"
+RTMP_STREAM_PATH: str = getenv("RTMP_STREAM_PATH", "live")
+RTMP_STREAM_KEY: str = getenv("RTMP_STREAM_KEY", "test")
+RTMP_STREAM_URI: str = RTMP_URI + f"{RTMP_STREAM_PATH}/{RTMP_STREAM_KEY}"
+
+REDIS_HOST: str = getenv("REDIS_HOST", "localhost")
+REDIS_PORT: int = int(getenv("REDIS_PORT", 6379))
+
+EMULATOR_WIDTH: int = int(getenv("EMULATOR_WIDTH", 240))
+EMULATOR_HEIGHT: int = int(getenv("EMULATOR_HEIGHT", 160))
+EMULATOR_FPS: int = int(getenv("EMULATOR_FPS", 60))
+EMULATOR_SPF: float = 1.0 / EMULATOR_FPS
+EMULATOR_INPUT_HZ: int = int(getenv("EMULATOR_INPUT_HZ", 10))
+EMULATOR_POLLING_RATE: int = EMULATOR_FPS // EMULATOR_INPUT_HZ
+EMULATOR_ROM_PATH: str = getenv("EMULATOR_ROM_PATH", "roms/pokemon.gba")
+
+FFMPEG_WIDTH: int = int(getenv("FFMPEG_WIDTH", EMULATOR_WIDTH))
+FFMPEG_HEIGHT: int = int(getenv("FFMPEG_HEIGHT", EMULATOR_HEIGHT))
+FFMPEG_FPS: int = int(getenv("FFMPEG_FPS", 30))
+FFMPEG_BITRATE: str = getenv("FFMPEG_BIRATE", "2M")
+
+PASSWORD_ADMIN: str = getenv("PASSWORD_ADMIN", "password_admin")
 
-FPS: int = 60
-SPF: float = 1.0 / FPS
-HZ: int = 10
-POLLING_RATE: int = FPS // HZ
 USER_TIMEOUT: float = 0.5
 
 KEYMAP: dict[str, int] = {
@@ -22,6 +44,6 @@ KEYMAP: dict[str, int] = {
     "r": 8,
     "l": 9,
 }
-KEYS: tuple = tuple(KEYMAP.keys())
-MGBA_KEYS: tuple = tuple(KEYMAP.values())
-REDIS_INIT: dict = dict([(x, 0) for x in KEYS])
+KEYS_ID: tuple = tuple(KEYMAP.keys())
+KEYS_MGBA: tuple = tuple(KEYMAP.values())
+KEYS_RESET: dict = dict([(x, 0) for x in KEYS_ID])