En esta entrega me gustaría terminar con la conversión de Dungeon a una aplicación basada en eventos.
Vamos a seguir profundizando en el modelo de eventos para Dungeon. Esta vez, partiendo de una aproximación un poco distinta.
Métodos que son eventos
Player tiene algunos métodos como is_alive o has_won que nos comunican cosas sobre su estado. El objeto Application tiene que consultarlos en cada ciclo para saber si ha acabado el juego.
while player.is_alive() and not player.has_won():
Ese estado ha cambiado como resultado de que algo ha pasado. is_alive es siempre True hasta que la energía de la jugadora se agota. Por su parte, has_won es False hasta que la jugadora cruza una puerta Exit.
Application tiene que consultar esos métodos en cada ciclo porque el evento causante puede haber sucedido en cualquier momento. Sabemos que esos sucesos cambian el estado de Player y por eso miramos a ver si ese estado ha cambiado.
Pero ya sabemos que Player podría notificar esos eventos, haciendo innecesario acceder a su estado. Y en la entrega anterior pusimos los medios para que eso fuese posible. Lo único que necesitamos es que el objeto interesado se registre como observer o subscriptor.
Solo que Application no es un buen candidato a observer. Tenemos que traer a Game de vuelta.
En su momento eliminamos Game por considerarlo una lazy class, que únicamente se ocupaba de delegar en otro objeto lo que se le pedía hacer.
Sin embargo, Application se apropió del bucle de juego, lo que en realidad implica ocuparse de distintas responsabilidades. Application tendría que centrarse en preparar las cosas para poder jugar, gestionar si la jugadora quiere hacer una nueva partida, etc. Por otro lado, Game gestionaría el bucle del juego en sí, su inicio y su final.
Por eso, tendría sentido que Game atienda a eventos relacionados con lo que ocurre en el juego en lugar de Application.
Esto también tiene efectos en cómo se relaciona Player con los otros objetos del juego. Por ejemplo, con Dungeon. Cada vez que la jugadora emite un comando, Player lo ejecuta con el método do, pero necesita que le pasemos Dungeon cada vez, y cada vez es la misma instancia.
¿No tendría más sentido que Player y Dungeon se asociaran una única vez?
La historia del juego es más o menos así “Te despiertas en el interior de una misteriosa mazmorra. A tu alrededor puedes ver puertas hacia el norte…”
player.awakedIn(dungeon)
De hecho, esto daría lugar al evento PlayerAwaked. Otro evento más que nos podría interesar controlar en otras partes de la aplicación.
Application podría encargarse de preparar los objetos Player, Dungeon y Game, registrar sus observers y lanzar el Game.
Como vemos, son varios cambios y bastante profundos.
Lo primero que voy a hacer es intentar introducir el objeto Game para que se ocupe del bucle de juego sin tocar todavía otros objetos. Creo que es un refactor que puedo hacer con los tests que tengo ahora mismo.
class Game:
def __init__(self, player, input, printer, dungeon):
self._dungeon = dungeon
self._player = player
self._input = input
self._printer = printer
def run(self):
while self._player.is_alive() and not self._player.has_won():
command = self._input.command()
self._player.do(command, self._dungeon)
self._printer.draw()
Así quedaría Application:
class Application:
def __init__(self, obtain_user_command, show_output, factory, toggles):
self._toggles = toggles
self._obtain_user_command = obtain_user_command
self._show_output = show_output
self._printer = Printer(show_output)
self._factory = factory
def run(self, dungeon_name='game'):
self._show_scene(Scene(title="Welcome to the Dungeon", command="", description="", energy="100"))
dungeon = self._build_dungeon(dungeon_name)
player = Player.awake()
player.register(self._printer)
dungeon.register(self._printer)
game = Game(player=player, printer=self._printer, input=self._obtain_user_command, dungeon=dungeon)
game.run()
def _show_scene(self, scene):
self._show_output.put(scene)
def _build_dungeon(self, dungeon):
return self._factory.make(dungeon)
Con esto ya se puede apreciar la idea básica. Ahora hay que mejorar algunos detalles. Por ejemplo, encapsulemos la detección del final del juego:
class Game:
def __init__(self, player, input, printer, dungeon):
self._dungeon = dungeon
self._player = player
self._input = input
self._printer = printer
def run(self):
while self.not_finished():
command = self._input.command()
self._player.do(command, self._dungeon)
self._printer.draw()
def not_finished(self):
return self._player.is_alive() and not self._player.has_won()
Esto nos servirá para cambiarla fácilmente cuando Game comience a escuchar eventos.
Otra cuestión que queremos arreglar es la relación entre Player y Dungeon por lo que hemos comentado antes. Player se mueve dentro de Dungeon, pero es quien recibe los comandos de la jugadora. A primera vista, creo que prefiero que Player sea, por así decir, la raíz del agregado.
Para ello, necesito asociarlo con Dungeon, de tal forma que no necesite pasarla cada vez que ejecute un comando.
class Player():
# Removed code
def awake_in(self, dungeon):
self._receiver = dungeon
def do(self, command, receiver):
if self._receiver is None:
self._receiver = receiver
self._execute_command(command, self._receiver)
self._update_energy()
De este modo, podemos asociarlo en Application:
def run(self, dungeon_name='game'):
self._show_scene(Scene(title="Welcome to the Dungeon", command="", description="", energy="100"))
dungeon = self._build_dungeon(dungeon_name)
player = Player.awake()
player.register(self._printer)
dungeon.register(self._printer)
player.awake_in(dungeon)
game = Game(player=player, printer=self._printer, input=self._obtain_user_command, dungeon=dungeon)
game.run()
Consolido estos cambios en un nuevo commit, ya que los tests están pasando. Ahora nos quedaría modificar los tests para verificar que ya no es necesario que sigamos pasando Dungeon al método do.
El caso es que si uso el refactor automático Change Signature no ha dejado de funcionar ni un solo test. Así que, subimos este cambio también.
Aparte de eso, ya no necesito pasar Dungeon a Game. Otra ganancia más.
class Game:
def __init__(self, player, input, printer):
self._player = player
self._input = input
self._printer = printer
def run(self):
while self.not_finished():
command = self._input.command()
self._player.do(command)
self._printer.draw()
def not_finished(self):
return self._player.is_alive() and not self._player.has_won()
Por último, creo que voy a reorganizar un poco el código de Application.run para que se vea más claro.
def run(self, dungeon_name='game'):
self._show_scene(Scene(title="Welcome to the Dungeon", command="", description="", energy="100"))
player = self.set_up_player(dungeon_name)
game = Game(player=player, input=self._obtain_user_command, printer=self._printer)
game.run()
def set_up_player(self, dungeon_name):
dungeon = self._build_dungeon(dungeon_name)
player = Player.awake()
player.register(self._printer)
dungeon.register(self._printer)
player.awake_in(dungeon)
return player
Ahora ya deberíamos poder centrarnos en otros cambios. Por ejemplo, que Game atienda a los eventos de Player.
Más eventos en Player
Como hemos visto, Player podría perfectamente notificar eventos como PlayerAwaked y PlayerDied. Otro evento importante es PlayerExited aunque aún no tengo claro qué objeto debería publicarlo. Luego podremos suscribir otros objetos para que los puedan escuchar.
Vamos al test. Empecemos por el último:
def test_notifies_player_died_event_when_energy_is_0(self):
fake_observer = FakeObserver()
player = Player.awake_with_energy(EnergyUnit(100))
player.register(fake_observer)
player.do(TestCommand(EnergyUnit(100)))
self.assertTrue(fake_observer.is_aware_of("player_died"))
El evento sería:
class PlayerDied:
def __init__(self):
pass
def name(self):
return "player_died"
Y en Player se notifica cuando el nivel de energía es demasiado bajo.
def _update_energy(self):
self._energy.decrease(self._last_action_cost())
self._notify_observers(PlayerEnergyChanged(self._energy.current()))
if not self._energy.is_alive():
self._notify_observers(PlayerDied())
self._last_result.set("energy", str(self._energy))
Me molesta un poco el método is_alive en Energy, así que lo voy a cambiar:
class Energy:
# Removed code
def is_dead(self):
return self._energy.is_lower_than(EnergyUnit(1))
# Removed code
De modo que queda todo un poco más claro:
def _update_energy(self):
self._energy.decrease(self._last_action_cost())
self._notify_observers(PlayerEnergyChanged(self._energy.current()))
if self._energy.is_dead():
self._notify_observers(PlayerDied())
self._last_result.set("energy", str(self._energy))
Veo también que la última línea del método ya no debería ser necesaria.
El siguiente evento que querría tratar es PlayerExited ya que me interesan para que Game se suscriba a ellos y controle el final del bucle de juego.
Ese evento debería producirse cuando la jugadora cruza la puerta Exit. Y tendría toda la lógica del mundo este objeto fuese el emisor del evento. Pero esto introduce algunas complicaciones, ya que la construcción de la mazmorra se produce lejos de donde podríamos registrar a Game como observer de Exit.
Ahora mismo Dungeon es un Subject. Me pregunto si sería buena idea que el register de Dungeon extienda el registro a Rooms y Walls que lo ofrezcan. De este modo, al registrar un observer a Dungeon, se registraría automáticamente a cualquier elemento de la mazmorra que notifique eventos.
Este test debería servirnos. En él construimos una mazmorra simple, con una puerta de salida Exit, que será la emisora del evento.
def test_supports_player_exited_event(self):
fake_observer = FakeObserver()
builder = DungeonBuilder()
builder.add('start')
builder.set('start', Dir.N, Exit())
dungeon = builder.build()
dungeon.register(fake_observer)
dungeon.go(Dir.N)
self.assertTrue(fake_observer.is_aware_of("player_exited"))
Exit quedaría así:
class Exit(Wall):
def __init__(self):
self._subject = Subject()
def go(self):
self._notify_observers(PlayerExited())
return ActionResult.player_exited("Congrats. You're out")
def look(self):
return ActionResult.player_acted("There is a door")
def register(self, observer):
self._subject.register(observer)
def _notify_observers(self, event):
self._subject.notify_observers(event)
Ahora nos quedaría desarrollar una forma en la que Dungeon vaya preguntando a habitaciones y paredes si registran observers.
class Dungeon:
def __init__(self, rooms):
self._rooms = rooms
self._current = 'start'
self._subject = Subject()
# Removed code
def register(self, observer):
self._subject.register(observer)
self._rooms.register(observer)
Como Rooms encapsula una colección, no tenemos más que iterarla:
class Rooms:
def __init__(self):
self._rooms = dict()
# Removed code
def register(self, observer):
for name, room in self._rooms.items():
room.register(observer)
Tenemos que forzar a cada Room a tener un método register, ya que aunque nunca llegue a ser emisora de eventos por sí misma, contendrá otros objetos que sí.
class Room:
def __init__(self, walls):
self._walls = walls
# Removed code
def register(self, observer):
self._walls.register(observer)
Y la colección de paredes hace lo siguiente para registrar solo los tipos de paredes que notifiquen eventos:
class Walls:
def __init__(self):
self._walls = {
Dir.N: Wall(),
Dir.E: Wall(),
Dir.S: Wall(),
Dir.W: Wall()
}
# Removed code
def register(self, observer):
for d, wall in self._walls.items():
if hasattr(wall, "register"):
wall.register(observer)
Y con estos cambios el test pasa, lo que significa que estamos listas para que Game escuche eventos que controlen el bucle del juego. Antes de pasar a eso, consolido los cambios que he hecho y, por otra parte, reorganizo el código ya que tengo los eventos dispersos por varios archivos.
Este cambio rompe el test de Printer porque no nos aseguramos de gestionar eventos conocidos. Basta
class Printer:
def __init__(self, show_output):
self.show_output = show_output
self._command = ""
self._description = ""
self._energy = ""
self._title = ""
def notify(self, event):
if event.name() == "player_energy_changed":
self._energy = str(event.energy().value())
elif event.name() == "player_got_description":
self._description = event.description()
elif event.name() == "player_moved":
self._title = event.room()
elif event.name() == "player_sent_command":
self._command = "{} {}".format(event.command(), event.argument())
# Removed code
Ahora vamos a hacer que Game se registre como observer de Dungeon. Implementamos un método notify que pueda atender a los eventos player_died y player_exited, poniendo a True un flag finished en Game.
No tengo muy claro qué test hacer para desarrollar esto, pero creo que este podría ser un comienzo:
class GameTestCase(unittest.TestCase):
def test_game_handles_player_exited_event(self):
dungeon = DungeonFactory().make('test')
player = Player.awake()
player.awake_in(dungeon)
game = Game(player=Player, obtain_input=FixedObtainUserCommand("go north"), printer=Printer(FakeShowOutput()))
game.notify(PlayerExited())
self.assertTrue(game.finished())
Aquí lo tenemos:
class Game:
def __init__(self, player, obtain_input, printer):
self._finished = False
self._player = player
self._input = obtain_input
self._printer = printer
# Code removed
def notify(self, event):
if event.name() == "player_exited":
self._finished = True
def finished(self):
return self._finished
Lo mismo para “player_died”.
def test_game_handles_player_died_event(self):
dungeon = DungeonFactory().make('test')
player = Player.awake()
player.awake_in(dungeon)
game = Game(player=Player, obtain_input=FixedObtainUserCommand("go north"), printer=Printer(FakeShowOutput()))
game.notify(PlayerDied())
self.assertTrue(game.finished())
class Game:
def __init__(self, player, obtain_input, printer):
self._finished = False
self._player = player
self._input = obtain_input
self._printer = printer
# Code removed
def notify(self, event):
if event.name() == "player_exited":
self._finished = True
if event.name() == "player_died":
self._finished = True
def finished(self):
return self._finished
Ahora cedemos el control de Game a los eventos:
class Game:
# Removed code
def run(self):
while not self.finished():
self._player.do(self._input.command())
self._printer.draw()
# Removed code
Pero antes tenemos que registrar a Game, como observer de Dungeon, de lo contrario nunca sabrá que han ocurrido los eventos. Tendríamos que cambiar un poco el código en Application para eso:
class Application:
def __init__(self, obtain_user_command, show_output, factory, toggles):
self._toggles = toggles
self._obtain_user_command = obtain_user_command
self._show_output = show_output
self._printer = Printer(show_output)
self._factory = factory
def run(self, dungeon_name='game'):
self._show_scene(Scene(title="Welcome to the Dungeon", command="", description="", energy="100"))
dungeon = self._build_dungeon(dungeon_name)
player = self._setup_player(dungeon)
game = Game(player=player, obtain_input=self._obtain_user_command, printer=self._printer)
dungeon.register(game)
game.run()
def _setup_player(self, dungeon):
player = Player.awake()
player.register(self._printer)
dungeon.register(self._printer)
player.awake_in(dungeon)
return player
Con esto los tests pasan y ganaremos al salir de la mazmorra.
Únicamente nos faltaría limpiar el código no usado. En Game es el método not_finished que ya no necesitamos. Pero Player tampoco necesita exponer los métodos is_alive y has_won. Estos ahora solo se emplean en tests, con lo que habría que valorar si es posible cubrir el mismo comportamiento con otros tests.
Ahora mismo tenemos una cobertura de un 96% de líneas. Vamos a ver qué ocurre eliminando los tests y el código que ya no necesitamos.
Al hacerlo, la cobertura se mantiene prácticamente igual, por lo que podemos suprimir todo eso código, puesto que lo que nos está indicando la métrica es que el código está siendo ejercitado por otros tests.
Finalmente, aprovecho para limpiar algo de código no usado. El resultado está en este commit.
Próximos pasos
En este momento, la mayor parte de la aplicación se comunica mediante el patrón observer. Esto nos ayuda a construir clases con mejor encapsulación que exponen menos detalles de sí mismas.
Con todo, quedarían algunos aspectos mejorables. Del mismo modo que Exit es capaz de emitir un evento cuando la jugadora sale de la mazmorra, podríamos resolver temas como este, ya que el evento PlayerMoved perfectamente puede ser emitido por Door, en cuyo caso, Dungeon también podría escucharlo.
def go(self, direction):
result = self._current_room().go(Dir(direction))
if result.get("destination") is not None:
self._current = result.get("destination")
self._notify_observers(PlayerMoved(self._current))
return result
Con ello, estaríamos quitando responsabilidades a ActionResult.
Pero este trabajo lo vamos a dejar para la próxima entrega.