}
self.do_quit = False
+ def get_command_signature(self, command_name):
+ method_candidate = 'cmd_' + command_name
+ method = None
+ argtypes = ''
+ if hasattr(self, method_candidate):
+ method = getattr(self, method_candidate)
+ if hasattr(method, 'argtypes'):
+ argtypes = method.argtypes
+ return method, argtypes
+
+ def get_string_options(self, string_option_type):
+ if string_option_type == 'geometry':
+ return self.map_manager.get_map_geometries()
+ return None
+
def handle_input(self, msg):
if msg == 'BYE':
self.do_quit = True
else:
command()
except ArgError as e:
- self.log('ARGUMENT ERROR: ' + msg + '\n' + str(e))
- self.to_update['log'] = True
+ 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."""
elif map_mode:
if type(self.game.world.map_) == MapSquare:
if key == 'a':
- plom_socket_io.send(self.socket, 'MOVE LEFT')
+ plom_socket_io.send(self.socket, 'TASK:MOVE LEFT')
elif key == 'd':
- plom_socket_io.send(self.socket, 'MOVE RIGHT')
+ plom_socket_io.send(self.socket, 'TASK:MOVE RIGHT')
elif key == 'w':
- plom_socket_io.send(self.socket, 'MOVE UP')
+ plom_socket_io.send(self.socket, 'TASK:MOVE UP')
elif key == 's':
- plom_socket_io.send(self.socket, 'MOVE DOWN')
+ plom_socket_io.send(self.socket, 'TASK:MOVE DOWN')
elif type(self.game.world.map_) == MapHex:
if key == 'w':
- plom_socket_io.send(self.socket, 'MOVE UPLEFT')
+ plom_socket_io.send(self.socket, 'TASK:MOVE UPLEFT')
elif key == 'e':
- plom_socket_io.send(self.socket, 'MOVE UPRIGHT')
+ plom_socket_io.send(self.socket, 'TASK:MOVE UPRIGHT')
if key == 's':
- plom_socket_io.send(self.socket, 'MOVE LEFT')
+ plom_socket_io.send(self.socket, 'TASK:MOVE LEFT')
elif key == 'd':
- plom_socket_io.send(self.socket, 'MOVE RIGHT')
+ plom_socket_io.send(self.socket, 'TASK:MOVE RIGHT')
if key == 'x':
- plom_socket_io.send(self.socket, 'MOVE DOWNLEFT')
+ plom_socket_io.send(self.socket, 'TASK:MOVE DOWNLEFT')
elif key == 'c':
- plom_socket_io.send(self.socket, 'MOVE DOWNRIGHT')
+ plom_socket_io.send(self.socket, 'TASK:MOVE DOWNRIGHT')
else:
if len(key) == 1 and key in ASCII_printable and \
len(self.to_send) < len(self.edit):
def cmd_MAP(self, geometry, yx):
"""Create new map of grid geometry, size yx and only '?' cells."""
- legal_grids = self.map_manager.get_map_geometries()
- if geometry not in legal_grids:
- raise ArgError('First map argument must be one of: ' +
- ', '.join(legal_grids))
self.world.new_map(geometry, yx)
- cmd_MAP.argtypes = 'string yx_tuple:pos'
+ cmd_MAP.argtypes = 'string:geometry yx_tuple:pos'
def cmd_THING_TYPE(self, i, type_):
t = self.world.get_thing(i)
return tokens
def parse(self, msg):
- """Parse msg as call to self.game method, return method with arguments.
+ """Parse msg as call to method, return method with arguments.
Respects method signatures defined in methods' .argtypes attributes.
"""
tokens = self.tokenize(msg)
if len(tokens) == 0:
return None
- method_candidate = 'cmd_' + tokens[0]
- if not hasattr(self.game, method_candidate):
+ method, argtypes = self.game.get_command_signature(tokens[0])
+ if method is None:
return None
- method = getattr(self.game, method_candidate)
+ if len(argtypes) == 0:
+ if len(tokens) > 1:
+ raise ArgError('Command expects no argument(s).')
+ return method
if len(tokens) == 1:
- if not hasattr(method, 'argtypes'):
- return method
- else:
- raise ArgError('Command expects argument(s).')
+ raise ArgError('Command expects argument(s).')
args_candidates = tokens[1:]
- if not hasattr(method, 'argtypes'):
- raise ArgError('Command expects no argument(s).')
- args, kwargs = self.argsparse(method.argtypes, args_candidates)
- return partial(method, *args, **kwargs)
+ args = self.argsparse(argtypes, args_candidates)
+ return partial(method, *args)
def parse_yx_tuple(self, yx_string, range_):
"""Parse yx_string as yx_tuple:nonneg argtype, return result.
return (y, x)
def argsparse(self, signature, args_tokens):
- """Parse into / return args_tokens as args/kwargs defined by signature.
+ """Parse into / return args_tokens as args defined by signature.
Expects signature to be a ' '-delimited sequence of any of the strings
'int:nonneg', 'yx_tuple:nonneg', 'yx_tuple:pos', 'string',
- 'seq:int:nonneg', defining the respective argument types.
+ 'seq:int:nonneg', 'string:' + an option type string accepted by
+ self.game.get_string_options, defining the respective argument types.
"""
tmpl_tokens = signature.split()
if len(tmpl_tokens) != len(args_tokens):
') not expected number (' + str(len(tmpl_tokens))
+ ').')
args = []
+ string_string = 'string'
for i in range(len(tmpl_tokens)):
tmpl = tmpl_tokens[i]
arg = args_tokens[i]
args += [self.parse_yx_tuple(arg, 'nonneg')]
elif tmpl == 'yx_tuple:pos':
args += [self.parse_yx_tuple(arg, 'pos')]
- elif tmpl == 'string':
- args += [arg]
elif tmpl == 'seq:int:nonneg':
sub_tokens = arg.split(',')
if len(sub_tokens) < 1:
'non-negative integers.')
seq += [int(tok)]
args += [seq]
+ elif tmpl == string_string:
+ args += [arg]
+ elif tmpl[:len(string_string) + 1] == string_string + ':':
+ if not hasattr(self.game, 'get_string_options'):
+ raise ArgError('No string option directory.')
+ string_option_type = tmpl[len(string_string) + 1:]
+ options = self.game.get_string_options(string_option_type)
+ if options is None:
+ raise ArgError('Unknown string option type.')
+ if arg not in options:
+ msg = 'Argument #%s must be one of: %s' % (i + 1, options)
+ raise ArgError(msg)
+ args += [arg]
else:
raise ArgError('Unknown argument type.')
- return args, {}
+ return args
class TestParser(unittest.TestCase):
self.assertEqual(p.parse('x'), None)
def test_argsparse(self):
- from functools import partial
p = Parser()
assertErr = partial(self.assertRaises, ArgError, p.argsparse)
assertErr('', ['foo'])
class Task:
- def __init__(self, thing, name, args=(), kwargs={}):
+ def __init__(self, thing, name, args=()):
self.name = name
self.thing = thing
self.args = args
- self.kwargs = kwargs
self.todo = 3
def check(self):
- if self.name == 'move':
- if len(self.args) > 0:
- direction = self.args[0]
- else:
- direction = self.kwargs['direction']
- test_pos = self.thing.world.map_.move(self.thing.position, direction)
+ if self.name == 'MOVE':
+ test_pos = self.thing.world.map_.move(self.thing.position, self.args[0])
if self.thing.world.map_[test_pos] != '.':
raise GameError(str(self.thing.id_) +
' would move into illegal terrain')
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
- self.task = Task(self, 'wait')
+ self.task = Task(self, 'WAIT')
self.last_task_result = None
self._stencil = None
- def task_wait(self):
+ def task_WAIT(self):
return 'success'
- def task_move(self, direction):
+ def task_MOVE(self, direction):
self.position = self.world.map_.move(self.position, direction)
return 'success'
+ task_MOVE.argtypes = 'string:direction'
def move_towards_target(self, target):
dijkstra_map = type(self.world.map_)(self.world.map_.size)
direction = dirs[i_dir]
#print('DEBUG result', direction)
if direction:
- self.set_task('move', direction=direction)
+ self.set_task('MOVE', (direction,))
#self.world.game.io.send('would move ' + direction)
def decide_task(self):
return
except GameError:
pass
- self.set_task('wait')
+ self.set_task('WAIT')
- def set_task(self, task_name, *args, **kwargs):
- self.task = Task(self, task_name, args, kwargs)
+ def set_task(self, task_name, args=()):
+ self.task = Task(self, task_name, args)
self.task.check() # will throw GameError if necessary
def proceed(self, is_AI=True):
try:
self.decide_task()
except GameError:
- self.set_task('wait')
+ self.set_task('WAIT')
return
self.task.todo -= 1
if self.task.todo <= 0:
task = getattr(self, 'task_' + self.task.name)
- self.last_task_result = task(*self.task.args, **self.task.kwargs)
+ self.last_task_result = task(*self.task.args)
self.task = None
if is_AI and self.task is None:
try:
self.decide_task()
except GameError:
- self.set_task('wait')
+ self.set_task('WAIT')
def get_stencil(self):
if self._stencil is not None:
self.send_gamestate()
self.pool_result = self.pool.map_async(fib, (35, 35))
- def cmd_MOVE(self, direction):
- """Set player task to 'move' with direction arg, finish player turn."""
- import parser
- legal_directions = self.world.map_.get_directions()
- if direction not in legal_directions:
- raise parser.ArgError('Move argument must be one of: ' +
- ', '.join(legal_directions))
- self.world.get_player().set_task('move', direction=direction)
- self.proceed()
- cmd_MOVE.argtypes = 'string'
-
def cmd_SWITCH_PLAYER(self):
player = self.world.get_player()
- player.set_task('wait')
+ 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[player_index + 1]
self.proceed()
- def cmd_WAIT(self):
- """Set player task to 'wait', finish player turn."""
- self.world.get_player().set_task('wait')
- self.proceed()
-
def cmd_GET_GAMESTATE(self, connection_id):
"""Send game state to caller."""
self.send_gamestate(connection_id)
cmd_TERRAIN_LINE.argtypes = 'int:nonneg string'
def cmd_GEN_WORLD(self, geometry, yx, seed):
- legal_grids = self.map_manager.get_map_geometries()
- if geometry not in legal_grids:
- raise ArgError('First map argument must be one of: ' +
- ', '.join(legal_grids))
self.world.make_new(geometry, yx, seed)
- cmd_GEN_WORLD.argtypes = 'string yx_tuple:pos string'
+ cmd_GEN_WORLD.argtypes = 'string:geometry yx_tuple:pos string'
+
+ def get_command_signature(self, command_name):
+ from functools import partial
+
+ def cmd_TASK_colon(task_name, *args):
+ self.world.get_player().set_task(task_name, args)
+ self.proceed()
+
+ method = None
+ argtypes = ''
+ task_prefix = 'TASK:'
+ if command_name[:len(task_prefix)] == task_prefix:
+ task_name = command_name[len(task_prefix):]
+ task_method_candidate = 'task_' + task_name
+ if hasattr(Thing, task_method_candidate):
+ method = partial(cmd_TASK_colon, task_name)
+ task_method = getattr(Thing, task_method_candidate)
+ if hasattr(task_method, 'argtypes'):
+ argtypes = task_method.argtypes
+ return method, argtypes
+ method_candidate = 'cmd_' + command_name
+ if hasattr(self, method_candidate):
+ method = getattr(self, method_candidate)
+ if hasattr(method, 'argtypes'):
+ argtypes = method.argtypes
+ return method, argtypes
+
+ def get_string_options(self, string_option_type):
+ if string_option_type == 'geometry':
+ return self.map_manager.get_map_geometries()
+ elif string_option_type == 'direction':
+ return self.world.map_.get_directions()
+ return None
projects / plomrogue2-experiments / commitdiff