artlynk-sei-msp-displayport-parser.py

import subprocess
import re
import struct
import sys
from pathlib import Path

# Artlynk SEI MSP_DISPLAYPORT parser
#
# SEI Message Data structure
# Header:
# 1 byte number of commands
# 4 byte timestamp
# 4 byte total length without the ff seperator

# Command, "number of commands" times:
# 1 byte command length, following this byte, 0 indexed, 0a -> 11 glyphs
# 1 byte command, b6 -> MSP_DISPLAYPORT
# 1 byte sub-command, 03 -> MSP_DP_WRITE_STRING
# 1 byte pos x
# 1 byte pos y
# 1 byte glyph table index
# command length - 5 glyphs, glyph index from font file

class OsdParser:
    GRID_WIDTH = 53
    GRID_HEIGHT = 20
    BYTES_PER_GLYPH = 2
    TIMESTAMP_BYTES = 4
    HEADER_BYTES = 40
    FRAME_BYTES = TIMESTAMP_BYTES + (GRID_WIDTH * GRID_HEIGHT * BYTES_PER_GLYPH)

    def __init__(self, ffmpeg_path="ffmpeg"):
        self.ffmpeg_path = ffmpeg_path

    def get_data(self, video_path):
        """Extracts PTS and User Data using a multiline-aware regex."""
        cmd = [
            self.ffmpeg_path,
            '-i', str(video_path),
            '-vf', 'showinfo',
            '-f', 'null', '-'
        ]
        # We need to capture stderr where showinfo prints
        process = subprocess.Popen(cmd, stderr=subprocess.PIPE, stdout=subprocess.DEVNULL, text=True)
        stderr = process.communicate()[1]
        
        # This regex looks for pts_time and then finds the 'User Data=' string 
        # which may contain hex and spaces/newlines
        pattern = r"pts_time:([\d\.]+).*?User Data=([0-9a-fA-F\s]+)"
        return re.findall(pattern, stderr, re.DOTALL)

    def parse_msp_payload(self, hex_string):
        """Cleans hex and removes structural 'ff' padding."""
        # Clean hex string from all whitespace
        clean_hex = "".join(hex_string.split())
        raw_bytes = bytes.fromhex(clean_hex)
        
        # Structural removal: Remove every 3rd byte (the 'ff' padding)
        data = bytearray()
        for i, b in enumerate(raw_bytes):
            if (i + 1) % 3 != 0:
                data.append(b)
        
        if len(data) < 9:
            return None

        num_commands = data[0]
        # We ignore the internal timestamp data[1:5] now because we use PTS
        
        offset = 9
        active_glyphs = []

        for _ in range(num_commands):
            if offset >= len(data): break
            
            # The length byte indicates how many bytes follow it in this command
            cmd_payload_len = data[offset]
            
            # Ensure we have enough data for the header [b6, 03, x, y, page]
            if offset + 6 <= len(data):
                row = data[offset + 3]
                col = data[offset + 4]
                attribute = data[offset + 5]
                # page_offset = (font_page & 0x03) * 0x100
                
                # The number of glyphs is simply the remaining bytes.
                num_glyphs = cmd_payload_len - 4
                
                offset += 6
                for i in range(num_glyphs):
                    glyph_byte = data[offset]
                    character = ((attribute & 0x03) << 8) | glyph_byte
                    active_glyphs.append((row, col + i, character))
                    offset += 1

        return active_glyphs

    def convert(self, video_path, output_osd_path):
        entries = self.get_data(video_path)
        if not entries:
            print("No OSD data found in SEI messages.")
            return

        with open(output_osd_path, "wb") as f:
            # Write standard 40-byte .osd header
            header = bytearray(self.HEADER_BYTES)
            header[0:4] = b"BTFL"
            f.write(header)

            for pts_str, hex_line in entries:
                glyphs = self.parse_msp_payload(hex_line)
                if glyphs is None: continue
                
                # Timestamp in ms from PTS
                ts_ms = int(float(pts_str) * 1000)
                
                # Initialize empty frame buffer
                frame_buf = bytearray(self.FRAME_BYTES)
                struct.pack_into("<I", frame_buf, 0, ts_ms)
                
                for row, col, idx in glyphs:
                    # Direct coordinate check: only print if it fits on the 53x20 grid
                    if 0 <= col < self.GRID_WIDTH and 0 <= row < self.GRID_HEIGHT:
                        # Position = Header(4 bytes) + (row * width + col) * 2 bytes per glyph
                        pos = self.TIMESTAMP_BYTES + ((row * self.GRID_WIDTH) + col) * self.BYTES_PER_GLYPH
                        struct.pack_into("<H", frame_buf, pos, idx)

                f.write(frame_buf)

        print(f"Extraction complete: {output_osd_path}")

if __name__ == "__main__":
    if len(sys.argv) > 2:
        parser = OsdParser()
        parser.convert(sys.argv[1], sys.argv[2])
    else:
        print("Usage: python3 artlynk-sei-msp-displayport-parser.py <video.mp4> <output.osd>")

Hi!

On what license is your gist? Can i use it in my project?