Add first design

main
Bram van den Heuvel 3 months ago
parent ae07e26e03
commit 753d1f31a5
  1. 31
      config-sample.yml
  2. 238
      main.py
  3. 30
      requirements.txt

@ -0,0 +1,31 @@
# Config that determines the file location of all relevant files.
# It is recommended to place them in a shared file location.
files:
# Filename of the game that you're going to load.
game_save: './game.gb'
# If running in a Docker container, you should use something like
# game_save: '/config/game.gb'
# Filename of the latest state. This is the game state when a user last gave
# an input. If the program unexpectedly closes, the game will continue from
# this state.
latest_state: './latest.state'
# If running in a Docker container, you should use something like
# latest_state: '/config/latest.state'
# Location to temporarily store the screenshots that are sent to Matrix.
# The file permanently shows the last taken screenshot.
screenshot: './screenshot.png'
# If running in a Docker container, you should use something like
# screenshot: '/config/screenshot.png'
# Config that connects the program to Matrix
matrix:
# BaseURL of the homeserver.
homeserver: https://matrix.example.com
# Access token that the bot can use.
access_token: abcdefghijklmnopqrstuvwxyz
# Room ID that the bot uses to host the game in.
game_room: '!some-matrix-room:noordstar.me'

@ -0,0 +1,238 @@
import asyncio
import os
import re
import aiofiles.os
from nio import AsyncClient, MatrixRoom, RoomMessageText, UploadResponse
from PIL import Image
from pyboy import PyBoy, WindowEvent
import yaml
# --------------------------
# Constant definitions
with open('config.yml', 'r', encoding='utf-8') as open_file:
config = yaml.safe_load(open_file)
GAME_FILE_NAME = config['files']['game_save']
STATE_FILE_NAME = config['files']['latest_state']
SCREENSHOT_FILE_NAME = config['files']['screenshot']
HOMESERVER = config['matrix']['homeserver']
ACCESS_TOKEN = config['matrix']['access_token']
GAME_ROOM = config['matrix']['game_room']
# --------------------------
# Program setup
# Start emulator
pyboy = PyBoy(GAME_FILE_NAME)
progress = dict(still_running=True)
# Load old state
try:
pyboy.load_state(open(STATE_FILE_NAME, 'rb'))
except FileNotFoundError:
pass
# Set up a Matrix client
client = AsyncClient(HOMESERVER)
client.access_token = ACCESS_TOKEN
# --------------------------
# GameBoy Emulation
async def run_simulation():
"""
Keep the game running while no-one's giving any input.
"""
while not pyboy.tick():
# print("Another tick - maybe it's stuck?")
await asyncio.sleep(0)
progress['still_running'] = False
pyboy.stop()
raise RuntimeError("We are done here.")
def press_button_shortly(press, release, ticks = 5):
"""
Press a button for a certain amount of ticks.
"""
pyboy.send_input(press)
for _ in range(ticks):
pyboy.tick()
pyboy.send_input(release)
def press_button(button : str) -> str | None:
"""
Press the button that's given as a string.
"""
button = button.replace(' ', '').lower().strip()
if not re.fullmatch(
r'a|b|up?|d(own)?|l(eft)?|r(ight)?|start|select|screen(shot)?', button
):
return None
match button[0]:
case 'a':
press_button_shortly(
WindowEvent.PRESS_BUTTON_A,
WindowEvent.RELEASE_BUTTON_A
)
return 'a'
case 'b':
press_button_shortly(
WindowEvent.PRESS_BUTTON_B,
WindowEvent.RELEASE_BUTTON_B
)
return 'b'
case 'u':
press_button_shortly(
WindowEvent.PRESS_ARROW_UP,
WindowEvent.RELEASE_ARROW_UP
)
return 'up'
case 'd':
press_button_shortly(
WindowEvent.PRESS_ARROW_DOWN,
WindowEvent.RELEASE_ARROW_DOWN
)
return 'down'
case 'l':
press_button_shortly(
WindowEvent.PRESS_ARROW_LEFT,
WindowEvent.RELEASE_ARROW_LEFT
)
return 'left'
case 'r':
press_button_shortly(
WindowEvent.PRESS_ARROW_RIGHT,
WindowEvent.RELEASE_ARROW_RIGHT
)
return 'right'
case 's':
match button[1]:
# start
case 't':
press_button_shortly(
WindowEvent.PRESS_BUTTON_START,
WindowEvent.RELEASE_BUTTON_START
)
return 'start'
# select
case 'e':
press_button_shortly(
WindowEvent.PRESS_BUTTON_SELECT,
WindowEvent.RELEASE_BUTTON_SELECT
)
return 'select'
# screenshot
case 'c':
img = pyboy.screen_image()
img.save(SCREENSHOT_FILE_NAME)
return 'screenshot'
# --------------------------
# Matrix API
async def send_image(client, room_id, image):
"""Send image to toom.
Arguments:
---------
client : Client
room_id : str
image : str, file name of image
https://matrix-nio.readthedocs.io/en/latest/examples.html#sending-an-image
"""
im = Image.open(image)
(width, height) = im.size # im.size returns (width,height) tuple
# first do an upload of image, then send URI of upload to room
file_stat = await aiofiles.os.stat(image)
async with aiofiles.open(image, "r+b") as f:
resp, maybe_keys = await client.upload(
f,
content_type="image/png", # image/jpeg
filename=os.path.basename(image),
filesize=file_stat.st_size,
)
if isinstance(resp, UploadResponse):
print("Image was uploaded successfully to server. ")
else:
print(f"Failed to upload image. Failure response: {resp}")
content = {
"body": os.path.basename(image), # descriptive title
"info": {
"size": file_stat.st_size,
"mimetype": 'image/png',
"thumbnail_info": None, # TODO
"w": width, # width in pixel
"h": height, # height in pixel
"thumbnail_url": None, # TODO
},
"msgtype": "m.image",
"url": resp.content_uri,
}
try:
await client.room_send(room_id, message_type="m.room.message", content=content)
print("Sent screenshot!")
except Exception:
print(f"Failed to send screenshot.")
async def message_callback(room : MatrixRoom, event : RoomMessageText) -> None:
"""
Process a message if it was sent
"""
if room.room_id != GAME_ROOM:
return
exec = press_button(event.body)
if exec is not None:
# Save current state
pyboy.save_state(open(STATE_FILE_NAME, 'wb'))
# Mark accepted button message as read
await client.room_read_markers(
room_id=GAME_ROOM,
fully_read_event=event.event_id,
read_event=event.event_id
)
# Send screenshot if requested
if exec == 'screenshot':
await send_image(client, GAME_ROOM, SCREENSHOT_FILE_NAME)
async def setup_matrix_client():
client.add_event_callback(message_callback, RoomMessageText)
await client.sync_forever(timeout=30000)
# --------------------------
# Run the program
async def main():
await asyncio.gather(
# Emulate the game
run_simulation(),
# Gather input from Matrix, then handle it
setup_matrix_client()
)
if __name__ == '__main__':
asyncio.run(main())

@ -0,0 +1,30 @@
aiofiles==0.6.0
aiohttp==3.8.3
aiohttp-socks==0.7.1
aiosignal==1.3.1
async-timeout==4.0.2
attrs==22.1.0
charset-normalizer==2.1.1
Cython==0.29.32
frozenlist==1.3.3
future==0.18.2
h11==0.12.0
h2==4.1.0
hpack==4.0.0
hyperframe==6.0.1
idna==3.4
jsonschema==4.17.0
Logbook==1.5.3
matrix-nio==0.20.1
multidict==6.0.2
numpy==1.23.4
Pillow==9.3.0
pyboy==1.5.3
pycryptodome==3.15.0
pyrsistent==0.19.2
PySDL2==0.9.14
pysdl2-dll==2.24.1
python-socks==2.0.3
PyYAML==6.0
unpaddedbase64==2.1.0
yarl==1.8.1
Loading…
Cancel
Save