--- /dev/null
+#!/usr/bin/env python3
+import curses
+import plom_socket
+import socket
+import threading
+from parser import ArgError, Parser
+from plomrogue2 import (MapBase, WorldBase, ThingBase, cmd_MAP, cmd_THING_TYPE,
+ cmd_THING_POS)
+import types
+
+
+class Map(MapBase):
+
+ def y_cut(self, map_lines, center_y, view_height):
+ map_height = len(map_lines)
+ if map_height > view_height and center_y > view_height / 2:
+ if center_y > map_height - view_height / 2:
+ map_lines[:] = map_lines[map_height - view_height:]
+ else:
+ start = center_y - int(view_height / 2) - 1
+ map_lines[:] = map_lines[start:start + view_height]
+
+ def x_cut(self, map_lines, center_x, view_width, map_width):
+ if map_width > view_width and center_x > view_width / 2:
+ if center_x > map_width - view_width / 2:
+ cut_start = map_width - view_width
+ cut_end = None
+ else:
+ cut_start = center_x - int(view_width / 2)
+ cut_end = cut_start + view_width
+ map_lines[:] = [line[cut_start:cut_end] for line in map_lines]
+
+ def format_to_view(self, map_string, center, size):
+
+ def map_string_to_lines(map_string):
+ map_view_chars = ['0']
+ x = 0
+ y = 0
+ for c in map_string:
+ map_view_chars += [c, ' ']
+ x += 1
+ if x == self.size[1]:
+ map_view_chars += ['\n']
+ x = 0
+ y += 1
+ if y % 2 == 0:
+ map_view_chars += ['0']
+ if y % 2 == 0:
+ map_view_chars = map_view_chars[:-1]
+ map_view_chars = map_view_chars[:-1]
+ return ''.join(map_view_chars).split('\n')
+
+ map_lines = map_string_to_lines(map_string)
+ self.y_cut(map_lines, center[0], size[0])
+ map_width = self.size[1] * 2 + 1
+ self.x_cut(map_lines, center[1] * 2, size[1], map_width)
+ return map_lines
+
+
+class World(WorldBase):
+
+ def __init__(self, *args, **kwargs):
+ """Extend original with local classes and empty default map.
+
+ We need the empty default map because we draw the map widget
+ on any update, even before we actually receive map data.
+ """
+ super().__init__(*args, **kwargs)
+ self.map_ = Map()
+ self.player_position = (0, 0)
+
+ def new_map(self, yx):
+ self.map_ = Map(yx)
+
+
+def cmd_LAST_PLAYER_TASK_RESULT(self, msg):
+ if msg != "success":
+ self.log(msg)
+cmd_LAST_PLAYER_TASK_RESULT.argtypes = 'string'
+
+def cmd_TURN_FINISHED(self, n):
+ """Do nothing. (This may be extended later.)"""
+ pass
+cmd_TURN_FINISHED.argtypes = 'int:nonneg'
+
+def cmd_TURN(self, n):
+ """Set self.turn to n, empty self.things."""
+ self.world.turn = n
+ self.world.things = []
+ self.to_update['turn'] = False
+ self.to_update['map'] = False
+cmd_TURN.argtypes = 'int:nonneg'
+
+def cmd_VISIBLE_MAP_LINE(self, y, terrain_line):
+ self.world.map_.set_line(y, terrain_line)
+cmd_VISIBLE_MAP_LINE.argtypes = 'int:nonneg string'
+
+def cmd_PLAYER_POS(self, yx):
+ self.world.player_position = yx
+cmd_PLAYER_POS.argtypes = 'yx_tuple:pos'
+
+def cmd_GAME_STATE_COMPLETE(self):
+ self.to_update['turn'] = True
+ self.to_update['map'] = True
+
+
+class Game:
+
+ def __init__(self):
+ self.parser = Parser(self)
+ self.world = World(self)
+ self.thing_type = ThingBase
+ self.commands = {'LAST_PLAYER_TASK_RESULT': cmd_LAST_PLAYER_TASK_RESULT,
+ 'TURN_FINISHED': cmd_TURN_FINISHED,
+ 'TURN': cmd_TURN,
+ 'VISIBLE_MAP_LINE': cmd_VISIBLE_MAP_LINE,
+ 'PLAYER_POS': cmd_PLAYER_POS,
+ 'GAME_STATE_COMPLETE': cmd_GAME_STATE_COMPLETE,
+ 'MAP': cmd_MAP,
+ 'THING_TYPE': cmd_THING_TYPE,
+ 'THING_POS': cmd_THING_POS}
+ self.log_text = ''
+ self.to_update = {
+ 'log': True,
+ 'map': True,
+ 'turn': True,
+ }
+ self.do_quit = False
+
+ def get_command(self, command_name):
+ from functools import partial
+ if command_name in self.commands:
+ f = partial(self.commands[command_name], self)
+ if hasattr(self.commands[command_name], 'argtypes'):
+ f.argtypes = self.commands[command_name].argtypes
+ return f
+ return None
+
+ def get_string_options(self, string_option_type):
+ return None
+
+ def handle_input(self, msg):
+ if msg == 'BYE':
+ self.do_quit = True
+ return
+ try:
+ command, args = self.parser.parse(msg)
+ if command is None:
+ self.log('UNHANDLED INPUT: ' + msg)
+ self.to_update['log'] = True
+ else:
+ command(*args)
+ except ArgError as e:
+ self.log('ARGUMENT ERROR: ' + msg + '\n' + str(e))
+ self.to_update['log'] = True
+
+ def log(self, msg):
+ """Prefix msg plus newline to self.log_text."""
+ self.log_text = msg + '\n' + self.log_text
+ self.to_update['log'] = True
+
+ def symbol_for_type(self, type_):
+ symbol = '?'
+ if type_ == 'human':
+ symbol = '@'
+ elif type_ == 'monster':
+ symbol = 'm'
+ return symbol
+
+
+ASCII_printable = ' !"#$%&\'\(\)*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWX'\
+ 'YZ[\\]^_\`abcdefghijklmnopqrstuvwxyz{|}~'
+
+
+def recv_loop(plom_socket, game):
+ for msg in plom_socket.recv():
+ game.handle_input(msg)
+
+
+class Widget:
+
+ def __init__(self, tui, start, size, check_game=[], check_tui=[]):
+ self.check_game = check_game
+ self.check_tui = check_tui
+ self.tui = tui
+ self.start = start
+ self.win = curses.newwin(1, 1, self.start[0], self.start[1])
+ self.size_def = size # store for re-calling .size on SIGWINCH
+ self.size = size
+ self.do_update = True
+
+ @property
+ def size(self):
+ return self.win.getmaxyx()
+
+ @size.setter
+ def size(self, size):
+ """Set window size. Size be y,x tuple. If y or x None, use legal max."""
+ n_lines, n_cols = size
+ if n_lines is None:
+ n_lines = self.tui.stdscr.getmaxyx()[0] - self.start[0]
+ if n_cols is None:
+ n_cols = self.tui.stdscr.getmaxyx()[1] - self.start[1]
+ self.win.resize(n_lines, n_cols)
+
+ def __len__(self):
+ return self.win.getmaxyx()[0] * self.win.getmaxyx()[1]
+
+ def safe_write(self, foo):
+
+ def to_chars_with_attrs(part):
+ attr = curses.A_NORMAL
+ part_string = part
+ if not type(part) == str:
+ part_string = part[0]
+ attr = part[1]
+ if len(part_string) > 0:
+ return [(char, attr) for char in part_string]
+ elif len(part_string) == 1:
+ return [part]
+ return []
+
+ chars_with_attrs = []
+ if type(foo) == str or len(foo) == 2 and type(foo[1]) == int:
+ chars_with_attrs += to_chars_with_attrs(foo)
+ else:
+ for part in foo:
+ chars_with_attrs += to_chars_with_attrs(part)
+ self.win.move(0, 0)
+ if len(chars_with_attrs) < len(self):
+ for char_with_attr in chars_with_attrs:
+ self.win.addstr(char_with_attr[0], char_with_attr[1])
+ else: # workaround to <https://stackoverflow.com/q/7063128>
+ cut = chars_with_attrs[:len(self) - 1]
+ last_char_with_attr = chars_with_attrs[len(self) - 1]
+ self.win.addstr(self.size[0] - 1, self.size[1] - 2,
+ last_char_with_attr[0], last_char_with_attr[1])
+ self.win.insstr(self.size[0] - 1, self.size[1] - 2, ' ')
+ self.win.move(0, 0)
+ for char_with_attr in cut:
+ self.win.addstr(char_with_attr[0], char_with_attr[1])
+
+ def ensure_freshness(self, do_refresh=False):
+ if not do_refresh:
+ for key in self.check_game:
+ if self.tui.game.to_update[key]:
+ do_refresh = True
+ break
+ if not do_refresh:
+ for key in self.check_tui:
+ if self.tui.to_update[key]:
+ do_refresh = True
+ break
+ if do_refresh:
+ self.win.erase()
+ self.draw()
+ self.win.refresh()
+
+
+class EditWidget(Widget):
+
+ def draw(self):
+ self.safe_write((''.join(self.tui.to_send), curses.color_pair(1)))
+
+
+class LogWidget(Widget):
+
+ def draw(self):
+ line_width = self.size[1]
+ log_lines = self.tui.game.log_text.split('\n')
+ to_join = []
+ for line in log_lines:
+ to_pad = line_width - (len(line) % line_width)
+ if to_pad == line_width:
+ to_pad = 0
+ to_join += [line + ' '*to_pad]
+ self.safe_write((''.join(to_join), curses.color_pair(3)))
+
+
+class MapWidget(Widget):
+
+ def draw(self):
+
+ def terrain_with_objects():
+ terrain_as_list = list(self.tui.game.world.map_.terrain[:])
+ for t in self.tui.game.world.things:
+ pos_i = self.tui.game.world.map_.get_position_index(t.position)
+ terrain_as_list[pos_i] = self.tui.game.symbol_for_type(t.type_)
+ return ''.join(terrain_as_list)
+
+ def pad_or_cut_x(lines):
+ line_width = self.size[1]
+ for y in range(len(lines)):
+ line = lines[y]
+ if line_width > len(line):
+ to_pad = line_width - (len(line) % line_width)
+ lines[y] = line + '0' * to_pad
+ else:
+ lines[y] = line[:line_width]
+
+ def pad_y(lines):
+ if len(lines) < self.size[0]:
+ to_pad = self.size[0] - len(lines)
+ lines += to_pad * ['0' * self.size[1]]
+
+ def lines_to_colored_chars(lines):
+ chars_with_attrs = []
+ for c in ''.join(lines):
+ if c in {'@', 'm'}:
+ chars_with_attrs += [(c, curses.color_pair(1))]
+ elif c == '.':
+ chars_with_attrs += [(c, curses.color_pair(2))]
+ elif c in {'x', 'X', '#'}:
+ chars_with_attrs += [(c, curses.color_pair(3))]
+ else:
+ chars_with_attrs += [c]
+ return chars_with_attrs
+
+ if self.tui.game.world.map_.terrain == '':
+ lines = []
+ pad_y(lines)
+ self.safe_write(''.join(lines))
+ return
+
+ terrain_with_objects = terrain_with_objects()
+ center = self.tui.game.world.player_position
+ lines = self.tui.game.world.map_.format_to_view(terrain_with_objects,
+ center, self.size)
+ pad_or_cut_x(lines)
+ pad_y(lines)
+ self.safe_write(lines_to_colored_chars(lines))
+
+
+class TurnWidget(Widget):
+
+ def draw(self):
+ self.safe_write((str(self.tui.game.world.turn), curses.color_pair(2)))
+
+
+class TUI:
+
+ def __init__(self, plom_socket, game):
+ self.socket = plom_socket
+ self.game = game
+ self.parser = Parser(self.game)
+ self.to_update = {'edit': False}
+ curses.wrapper(self.loop)
+
+ def setup_screen(self, stdscr):
+ self.stdscr = stdscr
+ self.stdscr.refresh() # will be called by getkey else, clearing screen
+ self.stdscr.timeout(10)
+ self.stdscr.addstr(0, 0, 'SEND:')
+ self.stdscr.addstr(2, 0, 'TURN:')
+
+ def loop(self, stdscr):
+ self.setup_screen(stdscr)
+ curses.init_pair(1, curses.COLOR_BLACK, curses.COLOR_RED)
+ curses.init_pair(2, curses.COLOR_BLACK, curses.COLOR_GREEN)
+ curses.init_pair(3, curses.COLOR_BLACK, curses.COLOR_BLUE)
+ curses.curs_set(False) # hide cursor
+ self.to_send = []
+ self.edit = EditWidget(self, (0, 6), (1, 14), check_tui = ['edit'])
+ self.turn = TurnWidget(self, (2, 6), (1, 14), ['turn'])
+ self.log = LogWidget(self, (4, 0), (None, 20), ['log'])
+ self.map_ = MapWidget(self, (0, 21), (None, None), ['map'])
+ widgets = (self.edit, self.turn, self.log, self.map_)
+ map_mode = False
+ while True:
+ for w in widgets:
+ w.ensure_freshness()
+ for key in self.game.to_update:
+ self.game.to_update[key] = False
+ for key in self.to_update:
+ self.to_update[key] = False
+ try:
+ key = self.stdscr.getkey()
+ if key == 'KEY_RESIZE':
+ curses.endwin()
+ self.setup_screen(curses.initscr())
+ for w in widgets:
+ w.size = w.size_def
+ w.ensure_freshness(True)
+ elif key == '\t': # Tabulator key.
+ map_mode = False if map_mode else True
+ elif map_mode:
+ if key == 'w':
+ self.socket.send('TASK:MOVE UPLEFT')
+ elif key == 'e':
+ self.socket.send('TASK:MOVE UPRIGHT')
+ if key == 's':
+ self.socket.send('TASK:MOVE LEFT')
+ elif key == 'd':
+ self.socket.send('TASK:MOVE RIGHT')
+ if key == 'x':
+ self.socket.send('TASK:MOVE DOWNLEFT')
+ elif key == 'c':
+ self.socket.send('TASK:MOVE DOWNRIGHT')
+ else:
+ if len(key) == 1 and key in ASCII_printable and \
+ len(self.to_send) < len(self.edit):
+ self.to_send += [key]
+ self.to_update['edit'] = True
+ elif key == 'KEY_BACKSPACE':
+ self.to_send[:] = self.to_send[:-1]
+ self.to_update['edit'] = True
+ elif key == '\n': # Return key
+ self.socket.send(''.join(self.to_send))
+ self.to_send[:] = []
+ self.to_update['edit'] = True
+ except curses.error:
+ pass
+ if self.game.do_quit:
+ break
+
+
+s = socket.create_connection(('127.0.0.1', 5000))
+plom_socket = plom_socket.PlomSocket(s)
+game = Game()
+t = threading.Thread(target=recv_loop, args=(plom_socket, game))
+t.start()
+TUI(plom_socket, game)
--- /dev/null
+#!/usr/bin/env python3
+import socketserver
+import threading
+import queue
+import sys
+import parser
+
+
+class GameError(Exception):
+ pass
+
+
+# Avoid "Address already in use" errors.
+socketserver.TCPServer.allow_reuse_address = True
+
+
+class Server(socketserver.ThreadingTCPServer):
+ """Bind together threaded IO handling server and message queue."""
+
+ def __init__(self, queue, port, *args, **kwargs):
+ super().__init__(('localhost', port), IO_Handler, *args, **kwargs)
+ self.queue_out = queue
+ self.daemon_threads = True # Else, server's threads have daemon=False.
+
+
+class IO_Handler(socketserver.BaseRequestHandler):
+
+ def handle(self):
+ """Move messages between network socket and game IO loop via queues.
+
+ On start (a new connection from client to server), sets up a
+ new queue, sends it via self.server.queue_out to the game IO
+ loop thread, and from then on receives messages to send back
+ from the game IO loop via that new queue.
+
+ At the same time, loops over socket's recv to get messages
+ from the outside into the game IO loop by way of
+ self.server.queue_out into the game IO. Ends connection once a
+ 'QUIT' message is received from socket, and then also calls
+ for a kill of its own queue.
+
+ All messages to the game IO loop are tuples, with the first
+ element a meta command ('ADD_QUEUE' for queue creation,
+ 'KILL_QUEUE' for queue deletion, and 'COMMAND' for everything
+ else), the second element a UUID that uniquely identifies the
+ thread (so that the game IO loop knows whom to send replies
+ back to), and optionally a third element for further
+ instructions.
+
+ """
+
+ def send_queue_messages(plom_socket, queue_in, thread_alive):
+ """Send messages via socket from queue_in while thread_alive[0]."""
+ while thread_alive[0]:
+ try:
+ msg = queue_in.get(timeout=1)
+ except queue.Empty:
+ continue
+ plom_socket.send(msg, True)
+
+ import uuid
+ import plom_socket
+ plom_socket = plom_socket.PlomSocket(self.request)
+ print('CONNECTION FROM:', str(self.client_address))
+ connection_id = uuid.uuid4()
+ queue_in = queue.Queue()
+ self.server.queue_out.put(('ADD_QUEUE', connection_id, queue_in))
+ thread_alive = [True]
+ t = threading.Thread(target=send_queue_messages,
+ args=(plom_socket, queue_in, thread_alive))
+ t.start()
+ for message in plom_socket.recv():
+ if message is None:
+ plom_socket.send('BAD MESSAGE', True)
+ elif 'QUIT' == message:
+ plom_socket.send('BYE', True)
+ break
+ else:
+ self.server.queue_out.put(('COMMAND', connection_id, message))
+ self.server.queue_out.put(('KILL_QUEUE', connection_id))
+ thread_alive[0] = False
+ print('CONNECTION CLOSED FROM:', str(self.client_address))
+ plom_socket.socket.close()
+
+
+class GameIO():
+
+ def __init__(self, game_file_name, game):
+ self.game_file_name = game_file_name
+ self.queues_out = {}
+ self.parser = parser.Parser(game)
+
+ def loop(self, q):
+ """Handle commands coming through queue q, send results back.
+
+ Commands from q are expected to be tuples, with the first element
+ either 'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element
+ a UUID, and an optional third element of arbitrary type. The UUID
+ identifies a receiver for replies.
+
+ An 'ADD_QUEUE' command should contain as third element a queue
+ through which to send messages back to the sender of the
+ command. A 'KILL_QUEUE' command removes the queue for that
+ receiver from the list of queues through which to send replies.
+
+ A 'COMMAND' command is specified in greater detail by a string
+ that is the tuple's third element. The game_command_handler takes
+ care of processing this and sending out replies.
+
+ """
+ while True:
+ x = q.get()
+ command_type = x[0]
+ connection_id = x[1]
+ content = None if len(x) == 2 else x[2]
+ if command_type == 'ADD_QUEUE':
+ self.queues_out[connection_id] = content
+ elif command_type == 'KILL_QUEUE':
+ del self.queues_out[connection_id]
+ elif command_type == 'COMMAND':
+ self.handle_input(content, connection_id)
+
+ def run_loop_with_server(self):
+ """Run connection of server talking to clients and game IO loop.
+
+ We have the TCP server (an instance of Server) and we have the
+ game IO loop, a thread running self.loop. Both communicate with
+ each other via a queue.Queue. While the TCP server may spawn
+ parallel threads to many clients, the IO loop works sequentially
+ through game commands received from the TCP server's threads (=
+ client connections to the TCP server). A processed command may
+ trigger messages to the commanding client or to all clients,
+ delivered from the IO loop to the TCP server via the queue.
+
+ """
+ q = queue.Queue()
+ c = threading.Thread(target=self.loop, daemon=True, args=(q,))
+ c.start()
+ server = Server(q, 5000)
+ try:
+ server.serve_forever()
+ except KeyboardInterrupt:
+ pass
+ finally:
+ print('Killing server')
+ server.server_close()
+
+ def handle_input(self, input_, connection_id=None, store=True):
+ """Process input_ to command grammar, call command handler if found."""
+ from inspect import signature
+
+ def answer(connection_id, msg):
+ if connection_id:
+ self.send(msg, connection_id)
+ else:
+ print(msg)
+
+ try:
+ command, args = self.parser.parse(input_)
+ if command is None:
+ answer(connection_id, 'UNHANDLED_INPUT')
+ else:
+ if 'connection_id' in list(signature(command).parameters):
+ command(*args, connection_id=connection_id)
+ else:
+ command(*args)
+ if store and not hasattr(command, 'dont_save'):
+ with open(self.game_file_name, 'a') as f:
+ f.write(input_ + '\n')
+ except parser.ArgError as e:
+ answer(connection_id, 'ARGUMENT_ERROR ' + quote(str(e)))
+ except GameError as e:
+ answer(connection_id, 'GAME_ERROR ' + quote(str(e)))
+
+ def send(self, msg, connection_id=None):
+ """Send message msg to server's client(s) via self.queues_out.
+
+ If a specific client is identified by connection_id, only
+ sends msg to that one. Else, sends it to all clients
+ identified in self.queues_out.
+
+ """
+ if connection_id:
+ self.queues_out[connection_id].put(msg)
+ else:
+ for connection_id in self.queues_out:
+ self.queues_out[connection_id].put(msg)
+
+
+class MapBase:
+
+ def __init__(self, size=(0, 0)):
+ self.size = size
+ self.terrain = '?'*self.size_i
+
+ @property
+ def size_i(self):
+ return self.size[0] * self.size[1]
+
+ def set_line(self, y, line):
+ height_map = self.size[0]
+ width_map = self.size[1]
+ if y >= height_map:
+ raise ArgError('too large row number %s' % y)
+ width_line = len(line)
+ if width_line > width_map:
+ raise ArgError('too large map line width %s' % width_line)
+ self.terrain = self.terrain[:y * width_map] + line +\
+ self.terrain[(y + 1) * width_map:]
+
+ def get_position_index(self, yx):
+ return yx[0] * self.size[1] + yx[1]
+
+
+class Map(MapBase):
+
+ def __getitem__(self, yx):
+ return self.terrain[self.get_position_index(yx)]
+
+ def __setitem__(self, yx, c):
+ pos_i = self.get_position_index(yx)
+ if type(c) == str:
+ self.terrain = self.terrain[:pos_i] + c + self.terrain[pos_i + 1:]
+ else:
+ self.terrain[pos_i] = c
+
+ def __iter__(self):
+ """Iterate over YX position coordinates."""
+ for y in range(self.size[0]):
+ for x in range(self.size[1]):
+ yield [y, x]
+
+ def lines(self):
+ width = self.size[1]
+ for y in range(self.size[0]):
+ yield (y, self.terrain[y * width:(y + 1) * width])
+
+ def get_fov_map(self, yx):
+ return self.fov_map_type(self, yx)
+
+ def get_directions(self):
+ directions = []
+ for name in dir(self):
+ if name[:5] == 'move_':
+ directions += [name[5:]]
+ return directions
+
+ def get_neighbors(self, pos):
+ neighbors = {}
+ if not hasattr(self, 'neighbors_to'):
+ self.neighbors_to = {}
+ if pos in self.neighbors_to:
+ return self.neighbors_to[pos]
+ for direction in self.get_directions():
+ neighbors[direction] = None
+ try:
+ neighbors[direction] = self.move(pos, direction)
+ except GameError:
+ pass
+ self.neighbors_to[pos] = neighbors
+ return neighbors
+
+ def new_from_shape(self, init_char):
+ import copy
+ new_map = copy.deepcopy(self)
+ for pos in new_map:
+ new_map[pos] = init_char
+ return new_map
+
+ def move(self, start_pos, direction):
+ mover = getattr(self, 'move_' + direction)
+ new_pos = mover(start_pos)
+ if new_pos[0] < 0 or new_pos[1] < 0 or \
+ new_pos[0] >= self.size[0] or new_pos[1] >= self.size[1]:
+ raise GameError('would move outside map bounds')
+ return new_pos
+
+ def move_LEFT(self, start_pos):
+ return [start_pos[0], start_pos[1] - 1]
+
+ def move_RIGHT(self, start_pos):
+ return [start_pos[0], start_pos[1] + 1]
+
+
+
+class MapHex(Map):
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.fov_map_type = FovMapHex
+
+ def move_UPLEFT(self, start_pos):
+ if start_pos[0] % 2 == 1:
+ return [start_pos[0] - 1, start_pos[1] - 1]
+ else:
+ return [start_pos[0] - 1, start_pos[1]]
+
+ def move_UPRIGHT(self, start_pos):
+ if start_pos[0] % 2 == 1:
+ return [start_pos[0] - 1, start_pos[1]]
+ else:
+ return [start_pos[0] - 1, start_pos[1] + 1]
+
+ def move_DOWNLEFT(self, start_pos):
+ if start_pos[0] % 2 == 1:
+ return [start_pos[0] + 1, start_pos[1] - 1]
+ else:
+ return [start_pos[0] + 1, start_pos[1]]
+
+ def move_DOWNRIGHT(self, start_pos):
+ if start_pos[0] % 2 == 1:
+ return [start_pos[0] + 1, start_pos[1]]
+ else:
+ return [start_pos[0] + 1, start_pos[1] + 1]
+
+
+
+class FovMap:
+
+ def __init__(self, source_map, yx):
+ self.source_map = source_map
+ self.size = self.source_map.size
+ self.terrain = '?' * self.size_i
+ self[yx] = '.'
+ self.shadow_cones = []
+ self.circle_out(yx, self.shadow_process_hex)
+
+ def shadow_process_hex(self, yx, distance_to_center, dir_i, dir_progress):
+ # Possible optimization: If no shadow_cones yet and self[yx] == '.',
+ # skip all.
+ CIRCLE = 360 # Since we'll float anyways, number is actually arbitrary.
+
+ def correct_arm(arm):
+ if arm < 0:
+ arm += CIRCLE
+ return arm
+
+ def in_shadow_cone(new_cone):
+ for old_cone in self.shadow_cones:
+ if old_cone[0] >= new_cone[0] and \
+ new_cone[1] >= old_cone[1]:
+ #print('DEBUG shadowed by:', old_cone)
+ return True
+ # We might want to also shade hexes whose middle arm is inside a
+ # shadow cone for a darker FOV. Note that we then could not for
+ # optimization purposes rely anymore on the assumption that a
+ # shaded hex cannot add growth to existing shadow cones.
+ return False
+
+ def merge_cone(new_cone):
+ import math
+ for old_cone in self.shadow_cones:
+ if new_cone[0] > old_cone[0] and \
+ (new_cone[1] < old_cone[0] or
+ math.isclose(new_cone[1], old_cone[0])):
+ #print('DEBUG merging to', old_cone)
+ old_cone[0] = new_cone[0]
+ #print('DEBUG merged cone:', old_cone)
+ return True
+ if new_cone[1] < old_cone[1] and \
+ (new_cone[0] > old_cone[1] or
+ math.isclose(new_cone[0], old_cone[1])):
+ #print('DEBUG merging to', old_cone)
+ old_cone[1] = new_cone[1]
+ #print('DEBUG merged cone:', old_cone)
+ return True
+ return False
+
+ def eval_cone(cone):
+ #print('DEBUG CONE', cone, '(', step_size, distance_to_center, number_steps, ')')
+ if in_shadow_cone(cone):
+ return
+ self[yx] = '.'
+ if self.source_map[yx] != '.':
+ #print('DEBUG throws shadow', cone)
+ unmerged = True
+ while merge_cone(cone):
+ unmerged = False
+ if unmerged:
+ self.shadow_cones += [cone]
+
+ #print('DEBUG', yx)
+ step_size = (CIRCLE/len(self.circle_out_directions)) / distance_to_center
+ number_steps = dir_i * distance_to_center + dir_progress
+ left_arm = correct_arm(-(step_size/2) - step_size*number_steps)
+ right_arm = correct_arm(left_arm - step_size)
+ # Optimization potential: left cone could be derived from previous
+ # right cone. Better even: Precalculate all cones.
+ if right_arm > left_arm:
+ eval_cone([left_arm, 0])
+ eval_cone([CIRCLE, right_arm])
+ else:
+ eval_cone([left_arm, right_arm])
+
+ def basic_circle_out_move(self, pos, direction):
+ """Move position pos into direction. Return whether still in map."""
+ mover = getattr(self, 'move_' + direction)
+ pos[:] = mover(pos)
+ if pos[0] < 0 or pos[1] < 0 or \
+ pos[0] >= self.size[0] or pos[1] >= self.size[1]:
+ return False
+ return True
+
+ def circle_out(self, yx, f):
+ # Optimization potential: Precalculate movement positions. (How to check
+ # circle_in_map then?)
+ # Optimization potential: Precalculate what hexes are shaded by what hex
+ # and skip evaluation of already shaded hexes. (This only works if hex
+ # shading implies they completely lie in existing shades; otherwise we
+ # would lose shade growth through hexes at shade borders.)
+
+ # TODO: Start circling only in earliest obstacle distance.
+ circle_in_map = True
+ distance = 1
+ yx = yx[:]
+ #print('DEBUG CIRCLE_OUT', yx)
+ while circle_in_map:
+ circle_in_map = False
+ self.basic_circle_out_move(yx, 'RIGHT')
+ for dir_i in range(len(self.circle_out_directions)):
+ for dir_progress in range(distance):
+ direction = self.circle_out_directions[dir_i]
+ if self.circle_out_move(yx, direction):
+ f(yx, distance, dir_i, dir_progress)
+ circle_in_map = True
+ distance += 1
+
+
+
+class FovMapHex(FovMap, MapHex):
+ circle_out_directions = ('DOWNLEFT', 'LEFT', 'UPLEFT',
+ 'UPRIGHT', 'RIGHT', 'DOWNRIGHT')
+
+ def circle_out_move(self, yx, direction):
+ return self.basic_circle_out_move(yx, direction)
+
+
+
+class ThingBase:
+
+ def __init__(self, world, id_, type_='?', position=[0,0]):
+ self.world = world
+ self.id_ = id_
+ self.type_ = type_
+ self.position = position
+
+
+class Thing(ThingBase):
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.set_task('WAIT')
+ self._last_task_result = None
+ self._stencil = None
+
+ def move_towards_target(self, target):
+ dijkstra_map = type(self.world.map_)(self.world.map_.size)
+ n_max = 256
+ dijkstra_map.terrain = [n_max for i in range(dijkstra_map.size_i)]
+ dijkstra_map[target] = 0
+ shrunk = True
+ visible_map = self.get_visible_map()
+ while shrunk:
+ shrunk = False
+ for pos in dijkstra_map:
+ if visible_map[pos] != '.':
+ continue
+ neighbors = dijkstra_map.get_neighbors(tuple(pos))
+ for direction in neighbors:
+ yx = neighbors[direction]
+ if yx is not None and dijkstra_map[yx] < dijkstra_map[pos] - 1:
+ dijkstra_map[pos] = dijkstra_map[yx] + 1
+ shrunk = True
+ #with open('log', 'a') as f:
+ # f.write('---------------------------------\n')
+ # for y, line in dijkstra_map.lines():
+ # for val in line:
+ # if val < 10:
+ # f.write(str(val))
+ # elif val == 256:
+ # f.write('x')
+ # else:
+ # f.write('~')
+ # f.write('\n')
+ neighbors = dijkstra_map.get_neighbors(tuple(self.position))
+ n = n_max
+ #print('DEBUG', self.position, neighbors)
+ #dirs = dijkstra_map.get_directions()
+ #print('DEBUG dirs', dirs)
+ #print('DEBUG neighbors', neighbors)
+ #debug_scores = []
+ #for pos in neighbors:
+ # if pos is None:
+ # debug_scores += [9000]
+ # else:
+ # debug_scores += [dijkstra_map[pos]]
+ #print('DEBUG debug_scores', debug_scores)
+ target_direction = None
+ for direction in neighbors:
+ yx = neighbors[direction]
+ if yx is not None:
+ n_new = dijkstra_map[yx]
+ if n_new < n:
+ n = n_new
+ target_direction = direction
+ #print('DEBUG result', direction)
+ if target_direction:
+ self.set_task('MOVE', (target_direction,))
+
+ def decide_task(self):
+ # TODO: Check if monster can follow player too well (even when they should lose them)
+ visible_things = self.get_visible_things()
+ target = None
+ for t in visible_things:
+ if t.type_ == 'human':
+ target = t.position
+ break
+ if target is not None:
+ try:
+ self.move_towards_target(target)
+ return
+ except GameError:
+ pass
+ self.set_task('WAIT')
+
+ def set_task(self, task_name, args=()):
+ task_class = self.world.game.tasks[task_name]
+ self.task = task_class(self, args)
+ self.task.check() # will throw GameError if necessary
+
+ def proceed(self, is_AI=True):
+ """Further the thing in its tasks.
+
+ Decrements .task.todo; if it thus falls to <= 0, enacts method
+ whose name is 'task_' + self.task.name and sets .task =
+ None. If is_AI, calls .decide_task to decide a self.task.
+
+ Before doing anything, ensures an empty map visibility stencil
+ and checks that task is still possible, and aborts it
+ otherwise (for AI things, decides a new task).
+
+ """
+ self._stencil = None
+ try:
+ self.task.check()
+ except GameError as e:
+ self.task = None
+ self._last_task_result = e
+ if is_AI:
+ try:
+ self.decide_task()
+ except GameError:
+ self.set_task('WAIT')
+ return
+ self.task.todo -= 1
+ if self.task.todo <= 0:
+ self._last_task_result = self.task.do()
+ self.task = None
+ if is_AI and self.task is None:
+ try:
+ self.decide_task()
+ except GameError:
+ self.set_task('WAIT')
+
+ def get_stencil(self):
+ if self._stencil is not None:
+ return self._stencil
+ self._stencil = self.world.map_.get_fov_map(self.position)
+ return self._stencil
+
+ def get_visible_map(self):
+ stencil = self.get_stencil()
+ m = self.world.map_.new_from_shape(' ')
+ for pos in m:
+ if stencil[pos] == '.':
+ m[pos] = self.world.map_[pos]
+ return m
+
+ def get_visible_things(self):
+ stencil = self.get_stencil()
+ visible_things = []
+ for thing in self.world.things:
+ if stencil[thing.position] == '.':
+ visible_things += [thing]
+ return visible_things
+
+
+
+class Task:
+ argtypes = ''
+
+ def __init__(self, thing, args=()):
+ self.thing = thing
+ self.args = args
+ self.todo = 3
+
+ @property
+ def name(self):
+ prefix = 'Task_'
+ class_name = self.__class__.__name__
+ return class_name[len(prefix):]
+
+ def check(self):
+ pass
+
+ def get_args_string(self):
+ stringed_args = []
+ for arg in self.args:
+ if type(arg) == str:
+ stringed_args += [quote(arg)]
+ else:
+ raise GameError('stringifying arg type not implemented')
+ return ' '.join(stringed_args)
+
+
+
+class Task_WAIT(Task):
+
+ def do(self):
+ return 'success'
+
+
+
+class Task_MOVE(Task):
+ argtypes = 'string:direction'
+
+ def check(self):
+ test_pos = self.thing.world.map_.move(self.thing.position, self.args[0])
+ if self.thing.world.map_[test_pos] != '.':
+ raise GameError('%s would move into illegal terrain' % self.thing.id_)
+ for t in self.thing.world.things:
+ if t.position == test_pos:
+ raise GameError('%s would move into other thing' % self.thing.id_)
+
+ def do(self):
+ self.thing.position = self.thing.world.map_.move(self.thing.position,
+ self.args[0])
+
+
+
+class WorldBase:
+
+ def __init__(self, game):
+ self.turn = 0
+ self.things = []
+ self.game = game
+
+ def get_thing(self, id_, create_unfound=True):
+ for thing in self.things:
+ if id_ == thing.id_:
+ return thing
+ if create_unfound:
+ t = self.game.thing_type(self, id_)
+ self.things += [t]
+ return t
+ return None
+
+
+class World(WorldBase):
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.player_id = 0
+
+ def new_map(self, yx):
+ self.map_ = self.game.map_type(yx)
+
+ def proceed_to_next_player_turn(self):
+ """Run game world turns until player can decide their next step.
+
+ Iterates through all non-player things, on each step
+ furthering them in their tasks (and letting them decide new
+ ones if they finish). The iteration order is: first all things
+ that come after the player in the world things list, then
+ (after incrementing the world turn) all that come before the
+ player; then the player's .proceed() is run, and if it does
+ not finish his task, the loop starts at the beginning. Once
+ the player's task is finished, the loop breaks.
+ """
+ while True:
+ player = self.get_player()
+ player_i = self.things.index(player)
+ for thing in self.things[player_i+1:]:
+ thing.proceed()
+ self.turn += 1
+ for thing in self.things[:player_i]:
+ thing.proceed()
+ player.proceed(is_AI=False)
+ if player.task is None:
+ break
+
+ def get_player(self):
+ return self.get_thing(self.player_id)
+
+ def make_new(self, yx, seed):
+ import random
+ random.seed(seed)
+ self.turn = 0
+ self.new_map(yx)
+ for pos in self.map_:
+ if 0 in pos or (yx[0] - 1) == pos[0] or (yx[1] - 1) == pos[1]:
+ self.map_[pos] = '#'
+ continue
+ self.map_[pos] = random.choice(('.', '.', '.', '.', 'x'))
+ player = self.game.thing_type(self, 0)
+ player.type_ = 'human'
+ player.position = [random.randint(0, yx[0] -1),
+ random.randint(0, yx[1] - 1)]
+ npc = self.game.thing_type(self, 1)
+ npc.type_ = 'monster'
+ npc.position = [random.randint(0, yx[0] -1),
+ random.randint(0, yx[1] -1)]
+ self.things = [player, npc]
+ return 'success'
+
+
+
+def cmd_GEN_WORLD(self, yx, seed):
+ self.world.make_new(yx, seed)
+cmd_GEN_WORLD.argtypes = 'yx_tuple:pos string'
+
+def cmd_GET_GAMESTATE(self, connection_id):
+ """Send game state to caller."""
+ self.send_gamestate(connection_id)
+
+def cmd_MAP(self, yx):
+ """Create new map of size yx and only '?' cells."""
+ self.world.new_map(yx)
+cmd_MAP.argtypes = 'yx_tuple:pos'
+
+def cmd_THING_TYPE(self, i, type_):
+ t = self.world.get_thing(i)
+ t.type_ = type_
+cmd_THING_TYPE.argtypes = 'int:nonneg string'
+
+def cmd_THING_POS(self, i, yx):
+ t = self.world.get_thing(i)
+ t.position = list(yx)
+cmd_THING_POS.argtypes = 'int:nonneg yx_tuple:nonneg'
+
+def cmd_TERRAIN_LINE(self, y, terrain_line):
+ self.world.map_.set_line(y, terrain_line)
+cmd_TERRAIN_LINE.argtypes = 'int:nonneg string'
+
+def cmd_PLAYER_ID(self, id_):
+ # TODO: test whether valid thing ID
+ self.world.player_id = id_
+cmd_PLAYER_ID.argtypes = 'int:nonneg'
+
+def cmd_TURN(self, n):
+ self.world.turn = n
+cmd_TURN.argtypes = 'int:nonneg'
+
+def cmd_SWITCH_PLAYER(self):
+ player = self.world.get_player()
+ player.set_task('WAIT')
+ thing_ids = [t.id_ for t in self.world.things]
+ player_index = thing_ids.index(player.id_)
+ if player_index == len(thing_ids) - 1:
+ self.world.player_id = thing_ids[0]
+ else:
+ self.world.player_id = thing_ids[player_index + 1]
+ self.proceed()
+
+def cmd_SAVE(self):
+
+ def write(f, msg):
+ f.write(msg + '\n')
+
+ save_file_name = self.io.game_file_name + '.save'
+ with open(save_file_name, 'w') as f:
+ write(f, 'TURN %s' % self.world.turn)
+ write(f, 'MAP ' + stringify_yx(self.world.map_.size))
+ for y, line in self.world.map_.lines():
+ write(f, 'TERRAIN_LINE %5s %s' % (y, quote(line)))
+ for thing in self.world.things:
+ write(f, 'THING_TYPE %s %s' % (thing.id_, thing.type_))
+ write(f, 'THING_POS %s %s' % (thing.id_,
+ stringify_yx(thing.position)))
+ task = thing.task
+ if task is not None:
+ task_args = task.get_args_string()
+ write(f, 'SET_TASK:%s %s %s %s' % (task.name, thing.id_,
+ task.todo, task_args))
+ write(f, 'PLAYER_ID %s' % self.world.player_id)
+cmd_SAVE.dont_save = True
+
+
+class Game:
+
+ def __init__(self, game_file_name):
+ self.io = GameIO(game_file_name, self)
+ self.map_type = MapHex
+ self.tasks = {'WAIT': Task_WAIT, 'MOVE': Task_MOVE}
+ self.commands = {'GEN_WORLD': cmd_GEN_WORLD,
+ 'GET_GAMESTATE': cmd_GET_GAMESTATE,
+ 'MAP': cmd_MAP,
+ 'THING_TYPE': cmd_THING_TYPE,
+ 'THING_POS': cmd_THING_POS,
+ 'TERRAIN_LINE': cmd_TERRAIN_LINE,
+ 'PLAYER_ID': cmd_PLAYER_ID,
+ 'TURN': cmd_TURN,
+ 'SWITCH_PLAYER': cmd_SWITCH_PLAYER,
+ 'SAVE': cmd_SAVE}
+ self.world_type = World
+ self.world = self.world_type(self)
+ self.thing_type = Thing
+
+ def get_string_options(self, string_option_type):
+ if string_option_type == 'direction':
+ return self.world.map_.get_directions()
+ return None
+
+ def send_gamestate(self, connection_id=None):
+ """Send out game state data relevant to clients."""
+
+ self.io.send('TURN ' + str(self.world.turn))
+ self.io.send('MAP ' + stringify_yx(self.world.map_.size))
+ visible_map = self.world.get_player().get_visible_map()
+ for y, line in visible_map.lines():
+ self.io.send('VISIBLE_MAP_LINE %5s %s' % (y, quote(line)))
+ visible_things = self.world.get_player().get_visible_things()
+ for thing in visible_things:
+ self.io.send('THING_TYPE %s %s' % (thing.id_, thing.type_))
+ self.io.send('THING_POS %s %s' % (thing.id_,
+ stringify_yx(thing.position)))
+ player = self.world.get_player()
+ self.io.send('PLAYER_POS %s' % (stringify_yx(player.position)))
+ self.io.send('GAME_STATE_COMPLETE')
+
+ def proceed(self):
+ """Send turn finish signal, run game world, send new world data.
+
+ First sends 'TURN_FINISHED' message, then runs game world
+ until new player input is needed, then sends game state.
+ """
+ self.io.send('TURN_FINISHED ' + str(self.world.turn))
+ self.world.proceed_to_next_player_turn()
+ msg = str(self.world.get_player()._last_task_result)
+ self.io.send('LAST_PLAYER_TASK_RESULT ' + quote(msg))
+ self.send_gamestate()
+
+ def get_command(self, command_name):
+ from functools import partial
+
+ def cmd_TASK_colon(task_name, game, *args):
+ game.world.get_player().set_task(task_name, args)
+ game.proceed()
+
+ def cmd_SET_TASK_colon(task_name, game, thing_id, todo, *args):
+ t = game.world.get_thing(thing_id, False)
+ if t is None:
+ raiseArgError('No such Thing.')
+ task_class = game.tasks[task_name]
+ t.task = task_class(t, args)
+ t.task.todo = todo
+
+ def task_prefixed(command_name, task_prefix, task_command,
+ argtypes_prefix=None):
+ if command_name[:len(task_prefix)] == task_prefix:
+ task_name = command_name[len(task_prefix):]
+ if task_name in self.tasks:
+ f = partial(task_command, task_name, self)
+ task = self.tasks[task_name]
+ if argtypes_prefix:
+ f.argtypes = argtypes_prefix + ' ' + task.argtypes
+ else:
+ f.argtypes = task.argtypes
+ return f
+ return None
+
+ command = task_prefixed(command_name, 'TASK:', cmd_TASK_colon)
+ if command:
+ return command
+ command = task_prefixed(command_name, 'SET_TASK:', cmd_SET_TASK_colon,
+ 'int:nonneg int:nonneg ')
+ if command:
+ return command
+ if command_name in self.commands:
+ f = partial(self.commands[command_name], self)
+ if hasattr(self.commands[command_name], 'argtypes'):
+ f.argtypes = self.commands[command_name].argtypes
+ return f
+ return None
+
+
+
+def quote(string):
+ """Quote & escape string so client interprets it as single token."""
+ quoted = []
+ quoted += ['"']
+ for c in string:
+ if c in {'"', '\\'}:
+ quoted += ['\\']
+ quoted += [c]
+ quoted += ['"']
+ return ''.join(quoted)
+
+
+def stringify_yx(tuple_):
+ """Transform tuple (y,x) into string 'Y:'+str(y)+',X:'+str(x)."""
+ return 'Y:' + str(tuple_[0]) + ',X:' + str(tuple_[1])
+
+
+
+if __name__ == "__main__":
+ import sys
+ import os
+ if len(sys.argv) != 2:
+ print('wrong number of arguments, expected one (game file)')
+ exit(1)
+ game_file_name = sys.argv[1]
+ game = Game(game_file_name)
+ if os.path.exists(game_file_name):
+ if not os.path.isfile(game_file_name):
+ print('game file name does not refer to a valid game file')
+ else:
+ with open(game_file_name, 'r') as f:
+ lines = f.readlines()
+ for i in range(len(lines)):
+ line = lines[i]
+ print("FILE INPUT LINE %5s: %s" % (i, line), end='')
+ game.io.handle_input(line, store=False)
+ else:
+ game.io.handle_input('GEN_WORLD Y:16,X:16 bar')
+ game.io.run_loop_with_server()
projects / plomrogue2-experiments / commitdiff