Source Engine .dem parser (header)

demo.py

# Copyright 2019 Cameron Brown

# Licensed under the Apache License, Version 2.0 (the 'License');
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an 'AS IS' BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""A Source Engine .DEM file format is comprised of two parts,
a header and a stream of 'frames'/events.

Source: https://developer.valvesoftware.com/wiki/DEM_Format

Types:
    Integer     4 bytes
    Float       4 bytes
    String      260 bytes

Header:
    Type        Field               Value
    ===========================================================================
    String      Header              8 characters, should be 'HL2DEMO'+NULL
    Int         Demo Protocol       Demo protocol version
    Int         Network Protocol    Network protocol version number
    String      Server name         260 characters long
    String      Client name         260 characters long
    String      Map name            260 characters long
    String      Game directory      260 characters long
    Float       Playback time       The length of the demo, in seconds
    Int         Ticks               Number of ticks in the demo
    Int         Frames              Number of frames in the demo (possibly)
    Int         Sign on length	    Length of signon data (Init if first frame)

Each frame begins with 0 or more of these commands. These are described in
Source SDK code files.

Frame (Network Protocols 7 and 8):
    Type                Value               Notes
    ===========================================================================
    dem_signon          1                   Ignore.
    dem_packet          2                   Ignore.
    dem_synctick        3                   Ignore.
    dem_consolecmd      4                   Read a standard data packet.
    dem_usercmd         5                   Read a standard data packet.
    dem_datatables      6                   Read a standard data packet.
    dem_stop            7                   A signal that the demo is over.
    dem_lastcommand     dem_stop

Frame (Network Protocols 14 and 15):
    Type                Value               Notes
    ===========================================================================
    dem_stringtables    8                   Read a standard data packet.
    dem_lastcommand     dem_stringtables

Frame (Network Protocols 36 and Higher):
    Type                Value               Notes
    ===========================================================================
    dem_customdata      8                   n/a
    dem_stringtables    9                   Read a standard data packet.
    dem_lastcommand     dem_stringtables
"""

import struct

from absl import app
from absl import flags
from absl import logging


FLAGS = flags.FLAGS


flags.DEFINE_string('file', None, 'Demo file to parse.')


# First header field. Every demo must begin with this.
HEADER = 'HL2DEMO\x00'

# Default number of bytes for various data types.
TYPE_INTEGER_LEN = 4
TYPE_FLOAT_LEN = 4
TYPE_STRING_LEN = 260


def __read_float(byte_arr, num_bytes=TYPE_FLOAT_LEN):
    """Reads a float from bytearray.
    Args:
        byte_arr (bytearray) Bytes to read float from.
        num_bytes (integer) Number of bytes to read. Source Engine
            has a default length of 4 bytes for float.
    Returns:
        buffer (string) The float that's been read.
        byte_arr (bytearray) The input bytearray with the read bytes
            trimmed off.
    """
    buffer = struct.unpack('f', byte_arr[0:num_bytes])[0]
    return buffer, byte_arr[num_bytes:]


def __read_int(byte_arr, num_bytes=TYPE_INTEGER_LEN):
    """Reads a integer from bytearray.
    Args:
        byte_arr (bytearray) Bytes to read integer from.
        num_bytes (integer) Number of bytes to read. Source Engine
            has a default length of 4 bytes for integer.
    Returns:
        buffer (string) The integer that's been read.
        byte_arr (bytearray) The input bytearray with the read bytes
            trimmed off.
    """
    integer = int.from_bytes(
        byte_arr[0:num_bytes], byteorder='little', signed=False)
    return integer, byte_arr[num_bytes:]


def __read_string(byte_arr, num_bytes=TYPE_STRING_LEN, strip=True):
    """Reads a string from bytearray.
    Args:
        byte_arr (bytearray) Bytes to read string from.
        num_bytes (integer) Number of bytes to read. Source Engine
            has a default length of 260 bytes, so that's what we're
            going with here.
        strip (boolean) Strip the null bytes.
    Returns:
        buffer (string) The string that's been read.
        byte_arr (bytearray) The input bytearray with the read bytes
            trimmed off.
    """
    buffer = str(byte_arr[0:num_bytes], 'utf-8')
    if strip:
        buffer = buffer.replace('\x00', '')
    return buffer, byte_arr[num_bytes:]


def parse_header(byte_arr):
    """Parse the demo's header.
    Args:
        byte_arr (bytearray) The byte array we're reading from.
    Returns:
        byte_arr (bytearray) Trimmed byte array without header.
    """
    header = {}

    header['header'], byte_arr = __read_string(byte_arr, 8, False)
    # Check that the header field matches with the file format.
    if header['header'] != HEADER:
        raise Exception("Bad file format!")

    header['demo_protocol'], byte_arr = __read_int(byte_arr)
    header['network_protocol'], byte_arr = __read_int(byte_arr)
    header['server_name'], byte_arr = __read_string(byte_arr)
    header['client_name'], byte_arr = __read_string(byte_arr)
    header['map_name'], byte_arr = __read_string(byte_arr)
    header['game_directory'], byte_arr = __read_string(byte_arr)
    header['playback_time'], byte_arr = __read_float(byte_arr)
    header['total_ticks'], byte_arr = __read_int(byte_arr)
    header['total_frames'], byte_arr = __read_int(byte_arr)
    header['sign_on_length'], byte_arr = __read_int(byte_arr)
    header['tickrate'] = header['total_ticks'] / header['playback_time']

    return header, byte_arr


def parse_frame(byte_arr):
    """Parse a demo's frame.
    Args:
        byte_arr (bytearray) The byte array we're reading from.
    Returns:
        byte_arr (bytearray) Trimmed byte array without header.
    """
    return None, byte_arr[4:]


def parse_packet(byte_arr):
    """Parse a packet.
    Args:
        byte_arr (bytearray) The byte array we're reading from.
    Returns:
        byte_arr (bytearray) Trimmed byte array without header.
    """
    length, byte_arr = __read_int(byte_arr)  # Length of the data packet.
    print(length)
    return byte_arr[:length], byte_arr[length:]


class Demo:
    """Represents a CS:GO demo."""

    @staticmethod
    def from_bytes(byte_arr):
        """Create a Demo object from a raw file.
        Args:
            raw_file (bytearray) The raw contents of a demo.
        Returns:
            demo (Demo) The parsed demo object.
        """
        if not isinstance(byte_arr, bytearray):
            raise Exception('File must be a bytearray type.')
        demo = Demo()
        header, byte_arr = parse_header(byte_arr)
        demo.header = header

        packet, byte_arr = parse_packet(byte_arr)
        # print(packet)

        # while len(byte_arr) > 0:
        #     frame, byte_arr = parse_frame(byte_arr)
        #     demo.frames.append(frame)
        #     print(len(byte_arr))
        return demo

    def __init__(self):
        self.header = {}
        self.frames = []


def main(argv):
    del argv  # Unused.

    logging.info('Loading file %s.', FLAGS.file)
    if not FLAGS.file:
        print('You must provide a file!')
        return

    with open(FLAGS.file, 'rb') as f:
        output = bytearray(f.read())

    logging.info('Parsing demo')
    demo = Demo.from_bytes(output)
    logging.info('Done!')


if __name__ == '__main__':
    app.run(main)