mch2022.badge.team Hatchery

backend.py

raw

Content

import uasyncio
import uso_client
import utils
import uso_protocol
import lobby
import easydraw
import static_pages
import display
import game
import main
import buttons
import mch22
import sys
import time
import random

class GameClient:
    """SocketIO transport."""

def __init__(self, event_loop, create_game=True, namespace="/", game_code=None):
        print("Network instance created")
        self.namespace = namespace
        self.socketio = uso_client.connect(utils.BACKEND_URL)
        self.host = create_game
        self.uid = None
        self.lobby = None
        self.game_id = game_code
        self.loop = event_loop
        self.joining_game = False
        self.game_in_progress = False
        self.current_level = 1
        self.instruction_task = None
        print("After GameClient __init__ run")
        server = self

@self.socketio.on("welcome")
        async def set_uid_on_welcome(message):
            server.uid = message["uid"]
            print("received welcome", message)
            if self.host:
                await self.create_game("TestGame", public=False)
            else:
                await self.join_game(server.game_id)

@self.socketio.on("game_info")
        async def game_joined(game_info):
            # Someone else joined
            if server.lobby:
                server.lobby.update_game_info(game_info)
                server.lobby.draw_lobby()
            else:  # We just joined
                game_code = utils.code_string_to_code(game_info["konami"])
                server.lobby = lobby.Lobby(game_code, server.uid, game_info)
                server.lobby.draw_lobby()
                server.joining_game = False
                # No signal ready functionality for now,
                # so just mark as ready right away
                await server.ready()

@self.socketio.on("game_started")
        async def call_print_on_game_started(message):
            print("Game Starting")
            server.instruction_task = server.loop.create_task(
                server.instructions_task()
            )

@self.socketio.on("game_over")
        async def game_over_screen(message):
            print("Game Over")
            await server.exit_game(cancel_socketio_task=False)
            WIDTH = display.width() - 50
            utils.draw_background(logo=False)

easydraw.lineCentered(
                30, "GAME OVER", utils.ARCADE_FONT[18], utils.TEXT_HIGHLIGHT
            )

utils.draw_lines(
                random.choice(utils.GAME_OVER_PRIMARY),
                y_start=65,
                vertical_padding=3,
                width=WIDTH,
                font=utils.ARCADE_FONT[9],
            )

utils.draw_lines(
                random.choice(utils.GAME_OVER_SECONDARY),
                y_start=100,
                vertical_padding=3,
                width=WIDTH,
                font=utils.ARCADE_FONT[9],
            )

easydraw.lineCentered(
                150, "YOU ACHIEVED:", utils.ARCADE_FONT[8], utils.TEXT_MAIN
            )
            easydraw.lineCentered(
                165,
                "LEVEL %d" % server.current_level,
                utils.ARCADE_FONT[18],
                utils.TEXT_HIGHLIGHT,
            )

utils.draw_lines(
                "Press MENU to return to the main menu",
                y_start=195,
                vertical_padding=3,
                width=WIDTH,
                font=utils.ARCADE_FONT[8],
            )

display.flush()

mch22.set_handler(buttons.hwButtons.__callback)

@self.socketio.on("player_disconnected")
        async def connection_lost(message):
            print("Someone disconnected, game over")
            await server.player_disconnected()

@self.socketio.on("command")
        async def set_command(command):
            print("Command received:", command)
            server.game.update_assignment(command)

@self.socketio.on("grid")
        async def set_grid(grid):
            if not server.game_in_progress:
                server.game_in_progress = True
                server.loop.create_task(server.timer_update_loop())
            print("Grid received:", grid)
            server.game = game.GameState(grid)

@self.socketio.on("health_info")
        async def set_health_on_health_info(message):
            print("received health_info", message)
            server.game.update_health_info(message)

@self.socketio.on("next_level")
        async def next_level(message):
            print("Achieved next level")
            server.game_in_progress = False
            server.current_level += 1

# Recap
            utils.draw_background(logo=False)
            easydraw.lineCentered(
                80,
                random.choice(utils.LEVEL_TRANSITION_ENCOURAGEMENTS).upper(),
                utils.ARCADE_FONT[18],
                utils.TEXT_HIGHLIGHT,
            )

for i, line in enumerate(random.choice(utils.LEVEL_TRANSITION_NARRATIVES)):
                easydraw.lineCentered(
                    120 + i * 20, line.upper(), utils.ARCADE_FONT[12], utils.TEXT_MAIN
                )

display.flush()
            await uasyncio.sleep(3)

# Prepare
            utils.draw_background(logo=False)
            easydraw.lineCentered(
                100,
                f"Level {str(server.current_level)}".upper(),
                utils.ARCADE_FONT[18],
                utils.TEXT_HIGHLIGHT,
            )
            easydraw.lineCentered(
                135,
                random.choice(utils.LEVEL_TRANSITION_PREPARE).upper(),
                utils.ARCADE_FONT[12],
                utils.TEXT_MAIN,
            )

display.flush()
            await uasyncio.sleep(2)

await server.intro_done()

@self.socketio.on("game_join_fail")
        async def join_failed(message):
            if server.joining_game:
                utils.draw_background()
                easydraw.lineCentered(
                    180, "Failed to join game", utils.ARCADE_FONT[12], utils.TEXT_MAIN
                )
                easydraw.lineCentered(
                    200, "Please try again", utils.ARCADE_FONT[12], utils.TEXT_MAIN
                )
                display.flush()
                await uasyncio.sleep(3)
                await server.socketio.close()
                server.loop.stop()
                server.loop.close()
                uasyncio.new_event_loop()
                mch22.set_handler(buttons.hwButtons.__callback)
                static_pages.join_lobby_input()

self.task = self.loop.create_task(self.receive_event_loop())

async def receive_event_loop(self):
        print("Starting SocketIO receive_event_loop")
        await self.socketio.run_forever()

async def connect(self, namespace=None):
        print("Sending connect message")
        await self.socketio._send_message(uso_protocol.MESSAGE_CONNECT, data=None)

async def create_game(self, name, public=False):
        await self.socketio.emit(
            "create_game",
            data={
                "name": name,
                "public": public,
            },
        )

async def instructions_task(self):
        try:
            MAIN_FONT = utils.ARCADE_FONT[9]
            LINE_SPACING = 4
            width = display.width() - 50

utils.draw_background(logo=True)
            easydraw.lineCentered(
                160, "Level 1 Starting", utils.ARCADE_FONT[12], utils.TEXT_HIGHLIGHT
            )
            easydraw.lineCentered(
                185, "Press START to", utils.ARCADE_FONT[9], utils.TEXT_MAIN
            )
            easydraw.lineCentered(
                200, "skip instructions", utils.ARCADE_FONT[9], utils.TEXT_MAIN
            )
            display.flush()
            await uasyncio.sleep(3)

utils.draw_background(logo=False)
            easydraw.lineCentered(
                35, "INSTRUCTIONS", utils.ARCADE_FONT[18], utils.TEXT_HIGHLIGHT
            )

y = 55
            y = utils.draw_lines(
                "+ Every player has their own unique set of controls",
                y,
                LINE_SPACING,
                width,
                MAIN_FONT,
            )
            y += 5
            y = utils.draw_lines(
                "+ Every player receives unique instructions",
                y,
                LINE_SPACING,
                width,
                MAIN_FONT,
            )
            y += 5
            y = utils.draw_lines(
                "+ The player with the control in their city dashboard has to carry out the instruction",
                y,
                LINE_SPACING,
                width,
                MAIN_FONT,
            )
            y += 5
            utils.draw_lines(
                "+ Instructions have to be carried out within the time limit (top left)",
                y,
                LINE_SPACING,
                width,
                MAIN_FONT,
            )

display.flush()
            await uasyncio.sleep(15)

utils.draw_background(logo=False)
            easydraw.lineCentered(
                80, "ARE YOU READY?", utils.ARCADE_FONT[18], utils.TEXT_HIGHLIGHT
            )

y = utils.draw_lines(
                "Work as a team to stay ahead of the algorithm",
                110,
                LINE_SPACING,
                width,
                utils.ARCADE_FONT[12],
            )
            utils.draw_lines(
                "Good luck!", y + 10, LINE_SPACING, width, utils.ARCADE_FONT[12]
            )

display.flush()
            await uasyncio.sleep(7)

finally:
            # Irrespective of if the task gets cancelled or finishes,
            # we need to signal intro_done to the server
            print("Sending intro_done from instructions_task")
            utils.draw_background(logo=True)
            easydraw.lineCentered(
                170, "Waiting for", utils.ARCADE_FONT[12], utils.TEXT_HIGHLIGHT
            )
            easydraw.lineCentered(
                190, "other players...", utils.ARCADE_FONT[12], utils.TEXT_HIGHLIGHT
            )
            display.flush()
            await self.intro_done()

async def join_game(self, konami_code):
        await self.socketio.emit("join_game", data={"konami": konami_code})

async def ready(self):
        await self.socketio.emit("ready")

async def start_game(self):
        if self.lobby and self.lobby.can_start:
            self.lobby.can_start = False
            await self.socketio.emit("start_game")

async def intro_done(self):
        await self.socketio.emit("intro_done")

async def game_button_pressed(self, button):
        command_data = self.game.handle_button_press(button)
        if command_data:
            await self.socketio.emit("command", command_data)
        else:
            print("Handled button press, but did not get command data")

async def report_exception(self, exception):
        try:
            await self.socketio.emit("exception", utils.get_exception_string(exception))
        except Exception:
            print("Failed to report exception")

async def timer_update_loop(self):
        last_ticks = time.ticks_ms() - 1000
        while self.game_in_progress:
            if self.game_in_progress and self.game:
                try:
                    self.game.trigger_counter()
                except Exception as exception:
                    print("Failed to update, received exception:")
                    sys.print_exception(exception)
                    await self.report_exception(exception)
            cur_ticks = time.ticks_ms()
            drift = time.ticks_diff(cur_ticks, last_ticks) - 1000
            ms_to_sleep = 1000 - drift
            last_ticks = cur_ticks
            await uasyncio.sleep_ms(ms_to_sleep)

async def player_disconnected(self):
        await self.exit_game(cancel_socketio_task=False)
        WIDTH = display.width() - 50
        utils.draw_background(logo=False)
        easydraw.lineCentered(
            20, "LOST", utils.ARCADE_FONT[18], utils.TEXT_HIGHLIGHT
        )
        easydraw.lineCentered(
            40, "CONNECTION", utils.ARCADE_FONT[18], utils.TEXT_HIGHLIGHT
        )
        y = utils.draw_lines(
            "Someone lost connection to the server or exited the game.",
            y_start=65,
            vertical_padding=3,
            width=WIDTH,
            font=utils.ARCADE_FONT[12],
        )

easydraw.lineCentered(
            y + 20, "YOU ACHIEVED:", utils.ARCADE_FONT[8], utils.TEXT_MAIN
        )
        easydraw.lineCentered(
            y + 35,
            "LEVEL %d" % self.current_level,
            utils.ARCADE_FONT[18],
            utils.TEXT_HIGHLIGHT,
        )

utils.draw_lines(
            "Press MENU to return to the main menu",
            y_start=195,
            vertical_padding=3,
            width=WIDTH,
            font=utils.ARCADE_FONT[8],
        )

display.flush()
        mch22.set_handler(buttons.hwButtons.__callback)

async def exit_game(self, cancel_socketio_task=True):
        self.game_in_progress = False
        if cancel_socketio_task:
            self.task.cancel()
        if self.socketio and self.socketio.websocket.open:
            await self.socketio.close()
        self.loop.stop()
        self.loop.close()
        uasyncio.new_event_loop()