How to encode traffic stuff for valhalla

valhalla_traffic_tile_encoder.py

from ctypes import *
import struct
from enum import Enum

"""
Converts valhalla's tile header & edge speeds into byte objects to be compressed.
"""

HEADER_FORMAT = '<2Q4L'  # 2 uint_64, 4 uint_32 from the traffic header

# Encode the tile speeds into this bit fielded structure
class SpeedStruct(LittleEndianStructure):
    _pack_ = 1
    _fields_ = [('overall_encoded_speed', c_uint64, 7),  # in 2 kmph resolution (up to 255)!
                ('encoded_speed1', c_uint64, 7),
                ('encoded_speed2', c_uint64, 7),
                ('encoded_speed3', c_uint64, 7),
                ('breakpoint1', c_uint64, 8),  # position = length * (breakpoint1 / 255)
                ('breakpoint2', c_uint64, 8),
                ('congestion1', c_uint64, 6),  # Stores 0 (unknown), or 1->63 (no congestion->max congestion)
                ('congestion2', c_uint64, 6),
                ('congestion3', c_uint64, 6),
                ('has_incidents', c_uint64, 1),  # Are there incidents on this edge in the corresponding incident tile
                ('spare', c_uint64, 1)]

def struct_to_bytes(struc: SpeedStruct) -> bytes:
    """
    Converts a struct to raw bytes, there seems to be no API in Structure.

    :param struc: The structure to be converted.
    :returns: the raw bytes of the structure to put into tiles
    """
    buffer = create_string_buffer(sizeof(struc))
    memmove(buffer, addressof(struc), sizeof(struc))
    return buffer.raw

# initialize with the proper values
# be careful: there's no "protection" for overflowing: if a 1-bit field
# gets assigned a number > 1 it will _silently_ only take the most significant bit
speed_struct = SpeedStruct(58, 58, 58, 58, 58, 58, 58, 58, 0, 0)

speed_bytes = struct_to_bytes(speed_struct)
header_bytes = struct.pack(HEADER_FORMAT, 12441, 112040199501, 44104, 3, 0, 0)

# then lz4 compress it and push it to redis