We still need game content to test the facade, and it's even better if we can load and save that content from a file!

This post is part of the OpenGL 2D Facade series

Objective

The expected result is our current level editor able to load and save all the current game state:

Serialization

Design

To load and save data, I propose to use serialization. It is a common approach that reads or writes content sequentially. It is a fast approach, and the only drawback is that we can not read or write a specific item. In our context, we always want to load or save a whole level, so it is not an issue.

We also have to choose the encoding or how we translate data into bytes or characters. Many people like text-based encodings like JSON or XML, but it is relatively slow and leads to large files. In our case, I think that it is better to consider binary encoding to get quick I/O and small files. The only drawback is that binary files are not human-readable.

Here are the classes I propose to create:

The IO class is the main class, and it can read or write primary content like strings or numbers. It holds a binary stream in the stream attribute we use to read or write byte arrays.

For specific objects, like the state of our game, we can create implementations of the ObjectSerializer class. In the read() and write() methods, we explain how to serialize them. They can use primary types (like numbers or strings) and any data type with an ObjectSerializer implementation registered with the registerObjectSerializer() method.

Note: In the diagram above, we only see a few cases. The complete IO class contains most integers (from 8 to 64 bits, signed or not), floats, usual structures, Numpy arrays, etc.

Integer serialization

Let's see an example with the serialization of an unsigned 8-bit integer:

def writeUInt8(self, i: int):
    assert 0 <= i <= 255
    self.writeBlockHeader(self.BLOCK_UINT8)
    data = i.to_bytes(1, byteorder='little', signed=False)
    self.__stream.write(data)

Line 2 checks that the input number fits the available space; otherwise, we could lose data without knowing it.

Line 3 writes some data that tells that the following content is an unsigned 8-bit integer. This step is not mandatory and increases the size of the file. However, it robustifies the file and helps in the detection of corrupted data. It is worth the cost, especially if we save most of our data in large structures like Numpy arrays.

self.BLOCK_UINT8 is a unique value to identify unsigned 8-bit integers. Its actual value does matter; what matters is to define it and never change it.

Line 4 encodes the number into a Python bytes array.

Line 5 writes the bytes array to the stream.

The writeBlockHeader() method writes a single byte:

def writeBlockHeader(self, code: int):
    assert 0 <= code <= 255
    data = code.to_bytes(1, byteorder='little', signed=False)
    self.__stream.write(data)

Integer deserialization

The decoding of an integer is as follow:

def readUInt8(self) -> int:
    assert self.readBlockHeader() == self.BLOCK_UINT8
    data = self.__stream.read(1)
    return int.from_bytes(data, byteorder='little', signed=False)

Line 2 reads the type code in the block header and checks it is as expected.

Line 3 reads a bytes array with a single byte.

Line 4 converts the bytes array into an integer.

The readBlockHeader() also reads a single byte:

def readBlockHeader(self) -> int:
    data = self.__stream.read(1)
    return int.from_bytes(data, byteorder='little', signed=False)

Objects serialization

For the serialization of objects, we need to explain how they can be read and written. Let's take an example with a simple class with two attributes:

class User:
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

The serialization of instances of this class can be as follow:

class UserSerializer(ObjectSerializer):
    def write(self, io: IO, user: User):
        io.writeString(user.name)
        io.writeUInt8(user.age)

    def read(self, io: IO) -> User:
        name = io.readString()
        age = io.readUInt8()
        return User(name, age)

The implementation is straightforward:

• We write the name and then the age for the serialization;
• We read the name and then the age for the deserialization.

Note that we could also add a header to tell that the content is a user.

To use this serializer (read or write), we first need to register it:

io = IO()
io.registerObjectSerializer("User", UserSerializer())

The implementation of the registerObjectSerializer() is straightforward:

def registerObjectSerializer(self, t: str, serializer: ObjectSerializer):
    self.__types[t] = serializer

We can write instances of the User class with the writeObject() method:

user = User("Peter", 23)
io.writeObject(user)

Here is the implementation of this method:

def writeObject(self, o: Any):
    t = type(o).__name__
    if t not in self.__types:
        raise ValueError("Type {} is not registered".format(t))
    self.writeBlockHeader(self.BLOCK_OBJECT)
    self.writeString(t)
    serializer = self.__types[t]
    serializer.write(self, o)

Line 2 gets the name of the class. Note that it is the relative name; we can have several classes with the same name in various locations.

Lines 3-4 check that we registered the type.

Line 5 writes the block header.

Line 6 writes a string with the name of the class.

Line 7 gets the object serializer.

Line 8 serializes the object.

The deserialization of the object is thanks to the readObject() method:

user = io.readObject()

NB: don't forget to move the stream pointer to the stream's start if you want to test this!

The implementation of the readObject() method is as follow:

def readObject(self) -> Any:
    assert self.readBlockHeader() == self.BLOCK_OBJECT
    t = self.readString()
    if t not in self.__types:
        raise ValueError("Type {} is not registered".format(t))
    serializer = self.__types[t]
    return serializer.read(self)

Other data types

Please have a look at the complete code below to see all cases.

Game state serialization

Using the IO class, the serialization of all our game content is easy:

class StateSerializer(ObjectSerializer):

    def write(self, io: IO, state: State):
        io.writeString("state")
        io.writeStringObjectDict(state.regions)

    def read(self, io: IO) -> State:
        assert io.readString() == "state"
        state = State()
        state.regions = io.readStringObjectDict()
        return state

class RegionSerializer(ObjectSerializer):

    def write(self, io: IO, region: Region):
        io.writeString("region")
        io.writeNumpyArray(region.cells)

    def read(self, io: IO) -> Region:
        assert io.readString() == "region"
        region = Region()
        region.cells = io.readNumpyArray()
        return region

Handling versions

Our game data will evolves with time, so we should handle these changes as soon as possible. The approach in this section consists of creating sets of object serializers for each version:

The diagram above shows the state and region serializers in a v1 package. The idea is to create a new package for each version (v2, v3, ...) with all the serializers. As we see below, we will only need to add the new serializers since we can reuse the previous ones if they have not changed.

For each version, we create a static method in the MainIO class that register all the serializers. For instance, for the first version:

def createV1(stream: Union[None, io.BytesIO] = None) -> MainIO:
    from .v1.StateSerializer import StateSerializer
    from .v1.RegionSerializer import RegionSerializer

    mainIO = MainIO(stream)
    mainIO.registerObjectSerializer("State", StateSerializer())
    mainIO.registerObjectSerializer("Region", RegionSerializer())

    return mainIO

Then, we can save our game state in the following way:

def save(fileName: str, state: State):
    mainIO = MainIO.createV1()
    mainIO.createStream()
    mainIO.writeFileHeader("level", 1)
    mainIO.writeObject(state)
    mainIO.saveFile(fileName)

Line 2 creates a MainIO with the latest version (here version 1).

Line 3 asks MainIO to create a new empty stream in memory.

Line 4 writes the main file header: a string to identify the file's content (here a game level) and an integer for the version (here version 1).

Line 5 writes the game state.

Line 6 saves all the stream content into a file.

The loading of the file have to deal with the versions:

def load(fileName: str) -> State:
    mainIO = IO()
    mainIO.loadFile(fileName)
    version = mainIO.readFileHeader("level")
    if version == 1:
        mainIO = MainIO.createV1(mainIO.stream)
        state = mainIO.readObject()
    else:
        raise RuntimeError("Unsupported version {}".format(version))
    return state

Line 2 creates a new instance of IO.

Line 3 loads the content of a file into the stream of mainIO.

Line 4 reads the file header: it checks that the file is a "level" file and returns its version.

If the version is the first one, then line 5 creates the corresponding MainIO. It uses the previous stream, so we don't have to decode the file header again. Line 6 reads the game state.

For other versions, we can add elif statements and build corresponding deserializers.

Final progam

Download code and assets

In the next post, I'll show how to animate water.