LGTM Devlog 26: Python Graphlib DAGs for Quest Stages
Tags: python, devjournal, gamedev
Published: 2021-01-26 09:54:46

Some further cleanup was done, None values have been switched out for [sentinel objects](https://python-patterns.guide/python/sentinel-object/), and the dependency injection started last time has been expanded to also deal with the return values, making the `main` function more trim, and the quests and tests have been adjusted to work better (there was an unfortunate mis-use of class variables previously). The code can be found at commit [1718538](https://github.com/meseta/lgtm/tree/1718538b03717c81f3cce4873baa1f4cf0daee58)

Here's some highlights:

# Quest stages
I'm defining a Quest Stage object, whose job is to contain the details that our game needs to interact with the player. This starts with our base `Stage` ABC. At the moment it just contains this:

```py
class Stage(ABC):
    @property
    @abstractmethod
    def children(cls) -> List[str]:
        """ List of children nodes of this stage """
        return NotImplemented

    quest: Union[Quest, NoQuestType] = NoQuest

    def __repr__(self):
        return f"{self.__class__.__name__}(quest={repr(self.quest)})"
```

A __repr__, `children` property which subclasses need to fill in (indicating the next quest(s) in the graph), and a `quest` instance property, which will be populated by the quest that spans the stage, which provides this class a parent reference to access things like quest data, or the user object.

From this `Stage` baseclass, I will implement subclasses for each of the quest stage types. Right now, I only have a debug stage (which doesn't do anything, it's just there for testing), and the definition (but not implementation for) the `CreateIssueStage` which later will contain implementation for starting a post to GitHub Issues

```py
class DebugStage(Stage):
    """ For debugging purposes """


class CreateIssueStage(Stage):
    """ This stage posts a new issue to a user's fork """
```

There will be future subclasses that do things like check conditions (to allow branching), check issue replies, and so on. I'll figure out how to do these things later.

Therefore a quest with stages can be defined like this:

```py
class DebugQuest(Quest):
    class QuestDataModel(QuestBaseModel):
        a: int = 1

    version = VersionInfo.parse("1.0.0")
    difficulty = Difficulty.RESERVED
    description = "This is a quest to facilitate testing/debugging"

    class Start(DebugStage):
        children = ["First"]

    class First(DebugStage):
        children = ["Second"]

    class Second(DebugStage):
        children = []
```

Each quest stage subclasses one of the existing stages, and provide metadata and other values needed by that class (they are not instances)

# Collecting stages
During subclassing, the following code runs inside the `Quest` object:

```py
class Quest(ABC):
    ... 
    @property
    @abstractmethod
    def stages(cls) -> Dict[str, Type[Stage]]:
        """ The initial default data to start quests with """
        return NotImplemented

    def __init_subclass__(cls):
        """ Subclasses instantiate by copying default data """
        # build class list
        cls.stages = {}
        for name, class_var in vars(cls).items():
            if isclass(class_var) and issubclass(class_var, Stage):
                cls.stages[name] = class_var
```

This __init__subclass method is called when `Quest` is subclassed! And when it does so, the code will collect all the properties of the class, find the ones that are subclasses of `Stage` and stick them into the class variable `stages`, so that this dictionary now contains a name: class pair of all the stages.  Note: this _still_ doesn't happen during instantiation, so at this point it's still only Classes we're dealing with.

# DAGs for quest stages
Each quest stage has a "children" property that defines one or more next stages. This means quests are not just a linear sequence of events, but can be possibly branching ones with things going on simultaneously. This might be overkill for a git learning game, but I can re-use this system for the future.

There's a specify type of graph in graph theory, called a [Directed Acyclic Graph](https://en.wikipedia.org/wiki/Directed_acyclic_graph), that makes a lot of these quest graphs easier to handle. They're characterised by being directional (which makes sense, since you should only be able to get from one quest stage to another in one direction, quest stages shouldn't send you back one stage); and do not contain cycles. While a quest could by cyclic, I think there's a chance that players can become stuck in the cycle, or the cycle becomes very obvious, with repeating events. So limiting quests to being acyclic has some benefits, and in exchange we have a much easier-to-deal with quest stage layout.

I previously talked a bit about DAGs/DiGraphs in my [Advent of Code 2020 Day 10 post](https://dev.to/meseta/advent-of-code-day-10-using-more-python-vectorized-calculations-in-numpy-2phe), where I used the versatile python [NetworkX](https://networkx.org/) package to compute the challenge solution very easily.

However, this time, I'm using Python's new [graphlib](https://docs.python.org/3/library/graphlib.html) which is new in Python 3.9!

What graphlib and it's `TopologicalSorter()` allows us to do two very important things:
1. It allows us to detect disallowed Cycles and loops in the quest. If someone accidentally makes a quest stage loop back on itself, then graphlib will complain (and tests will fail)
1. It allows us to easily find what the next available quest stage to do is, so we don't have to do any manual graph traversal to find the stages that are ready to run.

The code looks like this, and lives in the Quest ABC:

```py
class Quest(ABC):
    ...

    def load_stages(self) -> None:
        """ loads the stages """

        # load graph
        self.graph = TopologicalSorter()
        for stage_name, StageClass in self.stages.items():
            for child_name in cast(List[str], StageClass.children):
                if child_name not in self.stages:
                    raise QuestDefinitionError(
                        f"Quest {self.__class.__name__} does not have stage named '{child_name}'"
                    )
                self.graph.add(child_name, stage_name)

        try:
            self.graph.prepare()
        except CycleError as err:
            raise QuestDefinitionError(
                f"Quest {self.__class__.__name__} prepare failed! {err}"
            ) from err
```

The class `TopologicalSorter()` is provided by Python's built-in `graphlib`. We instantiate that, and load all the stages from our pre-build `stages` dictionary, using the stage's name as the node. We're unlikely to have a collision, because they're defined as class properties. This also does some checks for references to non-existent child stages.

Once the TopologicalSorter is loaded, running `self.graph.prepare()` will perform checks for cycles, and raise exceptions if they exist.

With this function, we can check that all quests are loadable and don't have any cycles in them. We can also do additional checks for things like having one start stages, and for having un-finishable quests, or disconnected stages. These are all classic graph theory problems that we could write later should we want to add checks!

# Tests
As per usual, all this new features have a full set of tests, including creation of bad quests, for example a couple of bad quests are defined in the tests files:

```py
class BadStageCycle(Quest):
    version = VersionInfo.parse("1.0.0")
    difficulty = Difficulty.RESERVED
    description = "Bad quest for testing, it has malformed stages"

    class Start(DebugStage):
        children = ["Loop"]

    class Loop(DebugStage):
        """ This should form a cycle, and get flagged by test """

        children = ["Start"]


class BadStageNotExist(Quest):
    version = VersionInfo.parse("1.0.0")
    difficulty = Difficulty.RESERVED
    description = "Bad quest for testing, it has malformed stages"

    class Start(DebugStage):
        """ This references a stage that doesn't exist, and get flagged """

        children = ["Loop"]

def test_fail_stage():
    """ Test bad quests that fail due to stage problems """

    quest = BadStageCycle()
    with pytest.raises(QuestDefinitionError):
        quest.load_stages()

    quest = BadStageNotExist()
    with pytest.raises(QuestDefinitionError):
        quest.load_stages()

```

And several other tests, which ensure we have good test coverage.

___
These are some major architectural changes, but things are nice and clean. The next step will be to add a way to actually execute the quest stages in the order that they're loaded.