#!/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__": win = pygame.display.set_mode((S_WIDTH, S_HEIGHT)) pygame.display.set_caption("Tetris") server_loop(win)