competitive_tetris/competitive_tetris.py

#!/usr/bin/python3

"""
Competitive tetris server
"""
import random
import math
import pygame
import zmq
import flatbuffers

import tetrominos
import game_state
import wire_protocol.Command
import wire_protocol.CommandRequest
import wire_protocol.CommandResponse
import wire_protocol.Playfield
import wire_protocol.CommandResponseError
import wire_protocol.GameState

pygame.font.init()

# GLOBALS VARS
S_WIDTH = 800
S_HEIGHT = 700
PLAY_WIDTH = 300  # meaning 300 // 10 = 30 width per block
PLAY_HEIGHT = 600  # meaning 600 // 20 = 30 height per block
BLOCK_SIZE = 30

TOP_LEFT_X = (S_WIDTH - PLAY_WIDTH) // 2
TOP_LEFT_Y = S_HEIGHT - PLAY_HEIGHT


def draw_text_middle(surface, text, size, color):
    font = pygame.font.SysFont(
            pygame.font.get_default_font(), size, bold=True)
    label = font.render(text, 1, color)

    surface.blit(label, (TOP_LEFT_X + PLAY_WIDTH /2 - (label.get_width()/2), TOP_LEFT_Y + PLAY_HEIGHT/2 - label.get_height()/2))


def draw_grid(surface, grid):
    sx = TOP_LEFT_X
    sy = TOP_LEFT_Y

    for i in range(len(grid)):
        pygame.draw.line(surface, (128,128,128), (sx, sy + i*BLOCK_SIZE), (sx+PLAY_WIDTH, sy+ i*BLOCK_SIZE))
        for j in range(len(grid[i])):
            pygame.draw.line(surface, (128, 128, 128), (sx + j*BLOCK_SIZE, sy),(sx + j*BLOCK_SIZE, sy + PLAY_HEIGHT))


def draw_next_shape(shape, surface):
    font = pygame.font.SysFont(pygame.font.get_default_font(), 30)
    label = font.render("Next Shape", 1, (255,255,255))

    sx = TOP_LEFT_X + PLAY_WIDTH + 0
    sy = TOP_LEFT_Y + PLAY_HEIGHT/2 - 100
    shape_raw = tetrominos.SHAPES_RAW[shape][0]

    for i, line in enumerate(shape_raw):
        row = list(line)
        for j, column in enumerate(row):
            if column == "0":
                pygame.draw.rect(
                        surface, (255,0,0),
                        (sx + j*BLOCK_SIZE, sy + i*BLOCK_SIZE,
                         BLOCK_SIZE, BLOCK_SIZE),
                        0)
    surface.blit(label, (sx + 10, sy - 30))


def draw_game_state_to_window(surface, state):
    surface.fill((0, 0, 0))

    pygame.font.init()
    font = pygame.font.SysFont(pygame.font.get_default_font(), 60)
    label = font.render("Tetris", 1, (255, 255, 255))
    surface.blit(label, 
                 (TOP_LEFT_X + PLAY_WIDTH / 2 - (label.get_width() / 2), 30))

    # current score
    font = pygame.font.SysFont(pygame.font.get_default_font(), 30)
    label = font.render(f"Score: {state.data.score}", 1, (255,255,255))
    sx = TOP_LEFT_X + PLAY_WIDTH + 50
    sy = TOP_LEFT_Y + PLAY_HEIGHT/2 - 100
    surface.blit(label, (sx + 20, sy + 160))


    field = state.get_field()
    for row in range(field.shape[0]):
        for col in range(field.shape[1]):
            color = (0,0,0)
            tile = field[row,col]
            if tile == 1:
                color = (0,255,0)
            elif tile == 2:
                color = (255,0,0)

            pygame.draw.rect(surface, color,
                             (TOP_LEFT_X + col*BLOCK_SIZE,
                              TOP_LEFT_Y + row*BLOCK_SIZE,
                              BLOCK_SIZE, BLOCK_SIZE), 0)

    pygame.draw.rect(surface, (255, 0, 0),
                     (TOP_LEFT_X, TOP_LEFT_Y, PLAY_WIDTH, PLAY_HEIGHT), 5)

    draw_grid(surface, field)


def handle_shift_left(state):
    error_code = wire_protocol.CommandResponseError.CommandResponseError.OK
    next_state = state.shift_left()
    if next_state is None:
        error_code = wire_protocol.CommandResponseError.CommandResponseError.INVALID_COMMAND
    else:
        state = next_state

    builder = flatbuffers.Builder()
    wire_protocol.CommandResponse.Start(builder)
    wire_protocol.CommandResponse.AddError(builder, error_code)
    rsp = wire_protocol.CommandResponse.End(builder)
    builder.Finish(rsp)
    return state, builder


def handle_shift_right(state):
    error_code = wire_protocol.CommandResponseError.CommandResponseError.OK
    next_state = state.shift_right()
    if next_state is None:
        error_code = wire_protocol.CommandResponseError.CommandResponseError.INVALID_COMMAND
    else:
        state = next_state

    builder = flatbuffers.Builder()
    wire_protocol.CommandResponse.Start(builder)
    wire_protocol.CommandResponse.AddError(builder, error_code)
    rsp = wire_protocol.CommandResponse.End(builder)
    builder.Finish(rsp)
    return state, builder


def handle_rotate(state):
    error_code = wire_protocol.CommandResponseError.CommandResponseError.OK
    next_state = state.rotate()
    if next_state is None:
        error_code = wire_protocol.CommandResponseError.CommandResponseError.INVALID_COMMAND
    else:
        state = next_state

    builder = flatbuffers.Builder()
    wire_protocol.CommandResponse.Start(builder)
    wire_protocol.CommandResponse.AddError(builder, error_code)
    rsp = wire_protocol.CommandResponse.End(builder)
    builder.Finish(rsp)
    return state, builder


def handle_down(state):
    error_code = wire_protocol.CommandResponseError.CommandResponseError.OK
    next_state = state.down()
    if next_state is None:
        error_code = wire_protocol.CommandResponseError.CommandResponseError.INVALID_COMMAND
    else:
        state = next_state

    builder = flatbuffers.Builder()
    wire_protocol.CommandResponse.Start(builder)
    wire_protocol.CommandResponse.AddError(builder, error_code)
    rsp = wire_protocol.CommandResponse.End(builder)
    builder.Finish(rsp)
    return state, builder


def handle_land(state):
    error_code = wire_protocol.CommandResponseError.CommandResponseError.OK
    next_state = state.land()
    if next_state is None:
        error_code = wire_protocol.CommandResponseError.CommandResponseError.INVALID_COMMAND
    else:
        state = next_state

    builder = flatbuffers.Builder()
    wire_protocol.CommandResponse.Start(builder)
    wire_protocol.CommandResponse.AddError(builder, error_code)
    rsp = wire_protocol.CommandResponse.End(builder)
    builder.Finish(rsp)
    return state, builder


def handle_other(state):
    builder = flatbuffers.Builder()
    wire_protocol.CommandResponse.Start(builder)
    wire_protocol.CommandResponse.AddError(
            builder,
            wire_protocol.CommandResponseError.CommandResponseError.UNKNOWN_COMMAND)
    rsp = wire_protocol.CommandResponse.End(builder)
    builder.Finish(rsp)
    return state, builder


def handle_get_game_state(state, millis_before_down):
    error_code = wire_protocol.CommandResponseError.CommandResponseError.OK
    builder = flatbuffers.Builder()

    wire_protocol.GameState.StartGridVector(builder, state.data.grid.size)
    builder.head -= state.data.grid.size
    builder.Bytes[builder.head:(builder.head + state.data.grid.size)] = state.data.grid.tobytes()
    grid_vector = builder.EndVector(state.data.grid.size)

    wire_protocol.GameState.Start(builder)
    wire_protocol.GameState.AddRows(builder, state.data.grid.shape[0])
    wire_protocol.GameState.AddCols(builder, state.data.grid.shape[1])
    wire_protocol.GameState.AddNextPiece(builder, state.data.next_piece)
    wire_protocol.GameState.AddCurrentPiece(builder, state.data.current_piece)
    wire_protocol.GameState.AddPieceRotation(builder, state.data.rotation)
    wire_protocol.GameState.AddPieceRow(builder, state.data.row)
    wire_protocol.GameState.AddPieceCol(builder, state.data.col)
    wire_protocol.GameState.AddScore(builder, state.data.score)
    wire_protocol.GameState.AddMillisBeforeDown(builder, millis_before_down)
    wire_protocol.GameState.AddGrid(builder, grid_vector)
    game_state_msg = wire_protocol.GameState.End(builder)

    wire_protocol.CommandResponse.Start(builder)
    wire_protocol.CommandResponse.AddError(builder, error_code)
    wire_protocol.CommandResponse.AddGameState(builder, game_state_msg)
    rsp = wire_protocol.CommandResponse.End(builder)
    builder.Finish(rsp)

    return state, builder


def game_loop(win, poller):
    is_game_running = True
    frame_time = 0
    clock = pygame.time.Clock()

    def _next_piece_func():
        return random.choice(range(len(tetrominos.SHAPES)))
    state = game_state.create_initial_game_state(_next_piece_func)

    while is_game_running:
        clock.tick()
        dt_millis = clock.get_time()
        frame_time += dt_millis

        for event in pygame.event.get():
            if event.type == pygame.KEYDOWN and event.key == pygame.K_ESCAPE:
                return None, False
            if event.type == pygame.QUIT:
                return None, False

        for sock, _ in poller.poll(timeout=0):
            req_buf = sock.recv()
            req = wire_protocol.CommandRequest.CommandRequest.GetRootAs(req_buf)
            builder = None

            if req.Command() == wire_protocol.Command.Command.ShiftLeft:
                state, builder = handle_shift_left(state)
            elif req.Command() == wire_protocol.Command.Command.ShiftRight:
                state, builder = handle_shift_right(state)
            elif req.Command() == wire_protocol.Command.Command.Rotate:
                state, builder = handle_rotate(state)
            elif req.Command() == wire_protocol.Command.Command.Down:
                state, builder = handle_down(state)
            elif req.Command() == wire_protocol.Command.Command.Land:
                state, builder = handle_land(state)
            elif req.Command() == wire_protocol.Command.Command.GetGameState:
                state, builder = handle_get_game_state(
                        state, int(math.floor(max(0, 500 - frame_time))))
            else:
                state, builder = handle_other(state)

            sock.send(builder.Output())

        if frame_time >= 500:
            state, is_game_running = state.advance()
            frame_time = 0

        draw_game_state_to_window(win, state)
        draw_next_shape(state.data.next_piece, win)
        pygame.display.update()

    return state, True


def server_loop(win):
    zmq_context = zmq.Context()
    socket = zmq_context.socket(zmq.REP)
    socket.bind("tcp://*:5555")
    poller = zmq.Poller()
    poller.register(socket, zmq.POLLIN)

    state = None

    run = True
    is_game_over_sent = True
    while run:
        win.fill((0,0,0))
        if state is None:
            draw_text_middle(
                    win, "Waiting for connection.", 60, (255,255,255))
        else:
            draw_text_middle(
                    win,
                    f"Score = {state.data.score}. "
                    "Reconnect?", 60, (255,255,255))

        pygame.display.update()
        for event in pygame.event.get():
            if event.type == pygame.KEYDOWN and event.key == pygame.K_ESCAPE:
                run = False
            elif event.type == pygame.QUIT:
                run = False

        run_game_loop = False
        for sock, _ in poller.poll(timeout=0):
            req_buf = sock.recv()
            req = wire_protocol.CommandRequest.CommandRequest.GetRootAs(req_buf)
            builder = None

            if req.Command() == wire_protocol.Command.Command.StartGame:
                builder = flatbuffers.Builder()
                wire_protocol.CommandResponse.Start(builder)
                wire_protocol.CommandResponse.AddError(
                        builder, wire_protocol.CommandResponseError.CommandResponseError.OK)
                rsp = wire_protocol.CommandResponse.End(builder)
                builder.Finish(rsp)
                sock.send(builder.Output())
                run_game_loop = True
            elif not is_game_over_sent:
                is_game_over_sent = True

                builder = flatbuffers.Builder()
                wire_protocol.CommandResponse.Start(builder)
                wire_protocol.CommandResponse.AddError(
                        builder,
                        wire_protocol.CommandResponseError.CommandResponseError.GAME_OVER)
                rsp = wire_protocol.CommandResponse.End(builder)
                builder.Finish(rsp)
                sock.send(builder.Output())
            else:
                builder = flatbuffers.Builder()
                wire_protocol.CommandResponse.Start(builder)
                wire_protocol.CommandResponse.AddError(
                        builder,
                        wire_protocol.CommandResponseError.CommandResponseError.INVALID_COMMAND)
                rsp = wire_protocol.CommandResponse.End(builder)
                builder.Finish(rsp)
                sock.send(builder.Output())

        if run_game_loop:
            state, run = game_loop(win, poller)

    pygame.display.quit()


if __name__ == "__main__":
    #random.seed(0)
    win = pygame.display.set_mode((S_WIDTH, S_HEIGHT))
    pygame.display.set_caption("Tetris")
    server_loop(win)