Reload KnowledgeBase object saved with a game.

Author: MarcCote

textworld/generator/game.py

@@ -339,8 +339,7 @@ class Game:
     _SERIAL_VERSION = 1
 
     def __init__(self, world: World, grammar: Optional[Grammar] = None,
-                 quests: Iterable[Quest] = (),
-                 kb: Optional[KnowledgeBase] = None) -> None:
+                 quests: Iterable[Quest] = ()) -> None:
         """
         Args:
             world: The world to use for the game.
@@ -352,7 +351,7 @@ def __init__(self, world: World, grammar: Optional[Grammar] = None,
         self.metadata = {}
         self._objective = None
         self._infos = self._build_infos()
-        self.kb = kb or KnowledgeBase.default()
+        self.kb = world.kb
         self.extras = {}
 
         # Check if we can derive a global winning policy from the quests.
@@ -379,7 +378,7 @@ def _build_infos(self) -> Dict[str, EntityInfo]:
 
     def copy(self) -> "Game":
         """ Make a shallow copy of this game. """
-        game = Game(self.world, self.grammar, self.quests, self.kb)
+        game = Game(self.world, self.grammar, self.quests)
         game._infos = dict(self.infos)
         game._objective = self._objective
         game.metadata = dict(self.metadata)
@@ -435,12 +434,12 @@ def deserialize(cls, data: Mapping) -> "Game":
         if version != cls._SERIAL_VERSION:
             raise ValueError("Cannot deserialize a TextWorld version {} game, expected version {}".format(version, cls._SERIAL_VERSION))
 
-        world = World.deserialize(data["world"])
+        kb = KnowledgeBase.deserialize(data["KB"])
+        world = World.deserialize(data["world"], kb=kb)
         game = cls(world)
         game.grammar = Grammar(data["grammar"])
         game.quests = tuple([Quest.deserialize(d) for d in data["quests"]])
         game._infos = {k: EntityInfo.deserialize(v) for k, v in data["infos"]}
-        game.kb = KnowledgeBase.deserialize(data["KB"])
         game.metadata = data.get("metadata", {})
         game._objective = data.get("objective", None)
         game.extras = data.get("extras", {})

textworld/generator/maker.py

@@ -24,11 +24,12 @@
 from textworld.envs.wrappers import Recorder
 
 
-def get_failing_constraints(state):
+def get_failing_constraints(state, kb: Optional[KnowledgeBase] = None):
+    kb = kb or KnowledgeBase.default()
     fail = Proposition("fail", [])
 
     failed_constraints = []
-    constraints = state.all_applicable_actions(KnowledgeBase.default().constraints.values())
+    constraints = state.all_applicable_actions(kb.constraints.values())
     for constraint in constraints:
         if state.is_applicable(constraint):
             # Optimistically delay copying the state
@@ -53,6 +54,10 @@ class PlayerAlreadySetError(ValueError):
     pass
 
 
+class QuestError(ValueError):
+    pass
+
+
 class FailedConstraintsError(ValueError):
     """
     Thrown when a constraint has failed during generation.
@@ -76,7 +81,8 @@ class WorldEntity:
     """
 
     def __init__(self, var: Variable, name: Optional[str] = None,
-                 desc: Optional[str] = None) -> None:
+                 desc: Optional[str] = None,
+                 kb: Optional[KnowledgeBase] = None) -> None:
         """
         Args:
             var: The underlying variable for the entity which is used
@@ -93,6 +99,7 @@ def __init__(self, var: Variable, name: Optional[str] = None,
         self.infos.desc = desc
         self.content = []
         self.parent = None
+        self._kb = kb or KnowledgeBase.default()
 
     @property
     def id(self) -> str:
@@ -158,11 +165,11 @@ def remove_property(self, name: str) -> None:
 
     def add(self, *entities: List["WorldEntity"]) -> None:
         """ Add children to this entity. """
-        if KnowledgeBase.default().types.is_descendant_of(self.type, "r"):
+        if self._kb.types.is_descendant_of(self.type, "r"):
             name = "at"
-        elif KnowledgeBase.default().types.is_descendant_of(self.type, ["c", "I"]):
+        elif self._kb.types.is_descendant_of(self.type, ["c", "I"]):
             name = "in"
-        elif KnowledgeBase.default().types.is_descendant_of(self.type, "s"):
+        elif self._kb.types.is_descendant_of(self.type, "s"):
             name = "on"
         else:
             raise ValueError("Unexpected type {}".format(self.type))
@@ -173,11 +180,11 @@ def add(self, *entities: List["WorldEntity"]) -> None:
             entity.parent = self
 
     def remove(self, *entities):
-        if KnowledgeBase.default().types.is_descendant_of(self.type, "r"):
+        if self._kb.types.is_descendant_of(self.type, "r"):
             name = "at"
-        elif KnowledgeBase.default().types.is_descendant_of(self.type, ["c", "I"]):
+        elif self._kb.types.is_descendant_of(self.type, ["c", "I"]):
             name = "in"
-        elif KnowledgeBase.default().types.is_descendant_of(self.type, "s"):
+        elif self._kb.types.is_descendant_of(self.type, "s"):
             name = "on"
         else:
             raise ValueError("Unexpected type {}".format(self.type))
@@ -283,7 +290,8 @@ class WorldPath:
 
     def __init__(self, src: WorldRoom, src_exit: WorldRoomExit,
                  dest: WorldRoom, dest_exit: WorldRoomExit,
-                 door: Optional[WorldEntity] = None) -> None:
+                 door: Optional[WorldEntity] = None,
+                 kb: Optional[KnowledgeBase] = None) -> None:
         """
         Args:
             src: The source room.
@@ -297,6 +305,7 @@ def __init__(self, src: WorldRoom, src_exit: WorldRoomExit,
         self.dest = dest
         self.dest_exit = dest_exit
         self.door = door
+        self._kb = kb or KnowledgeBase.default()
         self.src.exits[self.src_exit].dest = self.dest.exits[self.dest_exit]
         self.dest.exits[self.dest_exit].dest = self.src.exits[self.src_exit]
 
@@ -307,7 +316,7 @@ def door(self) -> Optional[WorldEntity]:
 
     @door.setter
     def door(self, door: WorldEntity) -> None:
-        if door is not None and not KnowledgeBase.default().types.is_descendant_of(door.type, "d"):
+        if door is not None and not self._kb.types.is_descendant_of(door.type, "d"):
             msg = "Expecting a WorldEntity of 'door' type."
             raise TypeError(msg)
 
@@ -348,7 +357,7 @@ class GameMaker:
         paths (List[WorldPath]): The connections between the rooms.
     """
 
-    def __init__(self) -> None:
+    def __init__(self, kb: Optional[KnowledgeBase] = None) -> None:
         """
         Creates an empty world, with a player and an empty inventory.
         """
@@ -357,7 +366,7 @@ def __init__(self) -> None:
         self.quests = []
         self.rooms = []
         self.paths = []
-        self._kb = KnowledgeBase.default()
+        self._kb = kb or KnowledgeBase.default()
         self._types_counts = self._kb.types.count(State(self._kb.logic))
         self.player = self.new(type='P')
         self.inventory = self.new(type='I')
@@ -442,15 +451,15 @@ def new(self, type: str, name: Optional[str] = None,
             * Otherwise, a `WorldEntity` is returned.
         """
         var_id = type
-        if not KnowledgeBase.default().types.is_constant(type):
+        if not self._kb.types.is_constant(type):
             var_id = get_new(type, self._types_counts)
 
         var = Variable(var_id, type)
         if type == "r":
             entity = WorldRoom(var, name, desc)
             self.rooms.append(entity)
         else:
-            entity = WorldEntity(var, name, desc)
+            entity = WorldEntity(var, name, desc, kb=self._kb)
 
         self._entities[var_id] = entity
         if entity.name:
@@ -546,7 +555,7 @@ def connect(self, exit1: WorldRoomExit, exit2: WorldRoomExit) -> WorldPath:
                              exit2.dest.src, exit2.dest.direction)
             raise ExitAlreadyUsedError(msg)
 
-        path = WorldPath(exit1.src, exit1.direction, exit2.src, exit2.direction)
+        path = WorldPath(exit1.src, exit1.direction, exit2.src, exit2.direction, kb=self._kb)
         self.paths.append(path)
         return path
 
@@ -658,6 +667,9 @@ def set_quest_from_commands(self, commands: List[str], ask_for_state: bool = Fal
             winning_facts = [user_query.query_for_important_facts(actions=recorder.actions,
                                                                   facts=recorder.last_game_state.state.facts,
                                                                   varinfos=self._working_game.infos)]
+        if len(commands) != len(actions):
+            unrecognized_commands = [c for c, a in zip(commands, recorder.actions) if a is None]
+            raise QuestError("Some of the actions were unrecognized: {}".format(unrecognized_commands))
 
         event = Event(actions=actions, conditions=winning_facts)
         self.quests = [Quest(win_events=[event])]
@@ -726,7 +738,7 @@ def validate(self) -> bool:
             msg = "Player position has not been specified. Use 'M.set_player(room)'."
             raise MissingPlayerError(msg)
 
-        failed_constraints = get_failing_constraints(self.state)
+        failed_constraints = get_failing_constraints(self.state, self._kb)
         if len(failed_constraints) > 0:
             raise FailedConstraintsError(failed_constraints)
 
@@ -747,7 +759,7 @@ def build(self, validate: bool = True) -> Game:
         if validate:
             self.validate()  # Validate the state of the world.
 
-        world = World.from_facts(self.facts)
+        world = World.from_facts(self.facts, kb=self._kb)
         game = Game(world, quests=self.quests)
 
         # Keep names and descriptions that were manually provided.

textworld/generator/tests/test_game.py

@@ -27,7 +27,7 @@
 from textworld.generator.game import ActionDependencyTree, ActionDependencyTreeElement
 from textworld.generator.inform7 import Inform7Game
 
-from textworld.logic import Proposition
+from textworld.logic import Proposition, GameLogic
 
 
 def _find_action(command: str, actions: Iterable[Action], inform7: Inform7Game) -> None:
@@ -65,6 +65,36 @@ def test_game_comparison():
     assert game1 != game3
 
 
+def test_reloading_game_with_custom_kb():
+    twl = KnowledgeBase.default().logic._document
+    twl += """
+        type customobj : o {
+            inform7 {
+                type {
+                    kind :: "custom-obj-like";
+                }
+            }
+        }
+    """
+
+    logic = GameLogic.parse(twl)
+    kb = KnowledgeBase(logic, "")
+
+    M = GameMaker(kb=kb)
+
+    room = M.new_room("room")
+    M.set_player(room)
+
+    custom_obj = M.new(type='customobj', name='customized object')
+    M.inventory.add(custom_obj)
+
+    commands = ["drop customized object"]
+    quest = M.set_quest_from_commands(commands)
+    assert quest.commands == commands
+    game = M.build()
+    assert game == Game.deserialize(game.serialize())
+
+
 def test_variable_infos(verbose=False):
     options = textworld.GameOptions()
     options.nb_rooms = 5