competitive_tetris/competitive_tetris.py

#!/usr/bin/python3

"""
Competitive tetris server
"""
import random
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("comicsans", 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('comicsans', 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('comicsans', 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('comicsans', 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(-1)
    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(1)
    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_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):
    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.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 main(win):
    run = True
    frame_time = 0
    clock = pygame.time.Clock()

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

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

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

        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                run = False
                pygame.display.quit()

        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.Land:
                state, builder = handle_land(state)
            elif req.Command() == wire_protocol.Command.Command.GetGameState:
                state, builder = handle_get_game_state(state)
            else:
                state, builder = handle_other(state)

            socket.send(builder.Output())

        is_game_running = True
        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()

        if not is_game_running:
            draw_text_middle(win, "YOU LOST!", 80, (255,255,255))
            pygame.display.update()
            pygame.time.delay(1500)
            run = False


def main_menu(win):
    run = True
    while run:
        win.fill((0,0,0))
        draw_text_middle(win, 'Press Any Key To Play', 60, (255,255,255))
        pygame.display.update()
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                run = False
            if event.type == pygame.KEYDOWN:
                main(win)

    pygame.display.quit()


win = pygame.display.set_mode((S_WIDTH, S_HEIGHT))
pygame.display.set_caption('Tetris')
main_menu(win)