DNG OpcodeList parser (DNG 1.7.1)

parse_dng_opcodelist.py

#!/usr/bin/env python3
#
# parse_dng_opcodelist.py
#
# Parse a DNG OpcodeList binary blob (big-endian) and print decoded opcodes.
# DNG 1.7.1.0 Specification: https://helpx.adobe.com/camera-raw/digital-negative.html
# This script can be used after extracting an OpcodeList from a DNG file using
# ExifTool, for example:
#
#   exiftool -b -OpcodeList2 image.dng > OpcodeList2.bin
#
# Usage:
#   python parse_dng_opcodelist.py OpcodeList2.bin
#
# Features:
#   - Prints statistics of opcode types at the beginning
#   - Each opcode is printed in full detail
#   - Long arrays are truncated (first 20 and last 20 entries shown)
#   - Output is formatted on a single line for readability
#

import struct
import io
import sys
from collections import Counter

OPCODE_NAMES = {
    1: "WarpRectilinear",
    2: "WarpFisheye",
    3: "FixVignetteRadial",
    4: "FixBadPixelsConstant",
    5: "FixBadPixelsList",
    6: "TrimBounds",
    7: "MapTable",
    8: "MapPolynomial",
    9: "GainMap",
    10: "DeltaPerRow",
    11: "DeltaPerColumn",
    12: "ScalePerRow",
    13: "ScalePerColumn",
    14: "WarpRectilinear2",
}

# ---------- binary helpers ----------
def read(fmt, f):
    size = struct.calcsize(fmt)
    b = f.read(size)
    if len(b) != size:
        raise EOFError("Unexpected EOF while reading %s" % fmt)
    return struct.unpack(fmt, b)

def read_u32(f): return read(">I", f)[0]
def read_i32(f): return read(">i", f)[0]
def read_u16(f): return read(">H", f)[0]
def read_i16(f): return read(">h", f)[0]
def read_double(f): return read(">d", f)[0]
def read_float(f): return read(">f", f)[0]
def read_bytes(f, n):
    b = f.read(n)
    if len(b) != n:
        raise EOFError("Unexpected EOF while reading bytes")
    return b

def version_from_u32(v):
    b = struct.pack(">I", v)
    return ".".join(str(x) for x in b)

# ---------- pretty-printer with truncation ----------
def pretty(obj, max_items=40, edge=20):
    if isinstance(obj, dict):
        items = [f"{repr(k)}: {pretty(v, max_items, edge)}" for k, v in obj.items()]
        return "{ " + ", ".join(items) + " }"
    elif isinstance(obj, list):
        n = len(obj)
        if n > max_items:
            shown = obj[:edge] + ["..."] + obj[-edge:]
            parts = [pretty(x, max_items, edge) for x in shown]
        else:
            parts = [pretty(x, max_items, edge) for x in obj]
        return "[ " + ", ".join(parts) + " ]"
    elif isinstance(obj, tuple):
        return "(" + ", ".join(pretty(x, max_items, edge) for x in obj) + ")"
    else:
        return repr(obj)

# ---------- opcode parsers ----------
def parse_warp_rectilinear(f):
    N = read_i32(f)
    sets = []
    for _ in range(N):
        kr = [read_double(f) for _ in range(4)]
        kt = [read_double(f) for _ in range(2)]
        cx, cy = read_double(f), read_double(f)
        sets.append({"kr": kr, "kt": kt, "cx": cx, "cy": cy})
    return {"N": N, "coeff_sets": sets}

def parse_warp_fisheye(f):
    N = read_i32(f)
    sets = []
    for _ in range(N):
        kr = [read_double(f) for _ in range(4)]
        cx, cy = read_double(f), read_double(f)
        sets.append({"kr": kr, "cx": cx, "cy": cy})
    return {"N": N, "coeff_sets": sets}

def parse_fix_vignette_radial(f):
    k = [read_double(f) for _ in range(5)]
    cx, cy = read_double(f), read_double(f)
    return {"k": k, "cx": cx, "cy": cy}

def parse_fix_bad_pixels_constant(f):
    return {"Constant": read_i32(f), "BayerPhase": read_i32(f)}

def parse_fix_bad_pixels_list(f):
    BayerPhase = read_i32(f)
    BadPointCount = read_i32(f)
    BadRectCount  = read_i32(f)
    points = [(read_i32(f), read_i32(f)) for _ in range(BadPointCount)]
    rects = [(read_i32(f), read_i32(f), read_i32(f), read_i32(f)) for _ in range(BadRectCount)]
    return {"BayerPhase": BayerPhase, "BadPointCount": BadPointCount,
            "BadRectCount": BadRectCount, "BadPoints": points, "BadRects": rects}

def parse_trim_bounds(f):
    return {"Top": read_i32(f), "Left": read_i32(f),
            "Bottom": read_i32(f), "Right": read_i32(f)}

def parse_map_table(f):
    Top, Left, Bottom, Right = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    Plane, Planes, RowPitch, ColPitch = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    TableSize = read_i32(f)
    table = [read_u16(f) for _ in range(TableSize)]
    return {"Top":Top,"Left":Left,"Bottom":Bottom,"Right":Right,
            "Plane":Plane,"Planes":Planes,"RowPitch":RowPitch,"ColPitch":ColPitch,
            "TableSize":TableSize,"Table":table}

def parse_map_polynomial(f):
    Top, Left, Bottom, Right = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    Plane, Planes, RowPitch, ColPitch = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    Degree = read_i32(f)
    coeffs = [read_double(f) for _ in range(Degree+1)]
    return {"Top":Top,"Left":Left,"Bottom":Bottom,"Right":Right,
            "Plane":Plane,"Planes":Planes,"RowPitch":RowPitch,"ColPitch":ColPitch,
            "Degree":Degree,"Coefficients":coeffs}

def parse_gain_map(f):
    Top, Left, Bottom, Right = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    Plane, Planes, RowPitch, ColPitch = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    MapPointsV, MapPointsH = read_i32(f), read_i32(f)
    MapSpacingV, MapSpacingH = read_double(f), read_double(f)
    MapOriginV, MapOriginH = read_double(f), read_double(f)
    MapPlanes = read_i32(f)
    gains = [[[read_float(f) for _ in range(MapPlanes)] for _ in range(MapPointsH)] for _ in range(MapPointsV)]
    return {"Top":Top,"Left":Left,"Bottom":Bottom,"Right":Right,
            "Plane":Plane,"Planes":Planes,"RowPitch":RowPitch,"ColPitch":ColPitch,
            "MapPointsV":MapPointsV,"MapPointsH":MapPointsH,
            "MapSpacingV":MapSpacingV,"MapSpacingH":MapSpacingH,
            "MapOriginV":MapOriginV,"MapOriginH":MapOriginH,
            "MapPlanes":MapPlanes,"MapGains":gains}

def parse_delta_per_row(f):
    Top, Left, Bottom, Right = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    Plane, Planes, RowPitch, ColPitch = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    Count = read_i32(f)
    deltas = [read_float(f) for _ in range(Count)]
    return {"Top":Top,"Left":Left,"Bottom":Bottom,"Right":Right,
            "Plane":Plane,"Planes":Planes,"RowPitch":RowPitch,"ColPitch":ColPitch,
            "Count":Count,"Deltas":deltas}

def parse_delta_per_column(f): return parse_delta_per_row(f)

def parse_scale_per_row(f):
    Top, Left, Bottom, Right = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    Plane, Planes, RowPitch, ColPitch = read_i32(f), read_i32(f), read_i32(f), read_i32(f)
    Count = read_i32(f)
    scales = [read_float(f) for _ in range(Count)]
    return {"Top":Top,"Left":Left,"Bottom":Bottom,"Right":Right,
            "Plane":Plane,"Planes":Planes,"RowPitch":RowPitch,"ColPitch":ColPitch,
            "Count":Count,"Scales":scales}

def parse_scale_per_column(f): return parse_scale_per_row(f)

def parse_warp_rectilinear2(f):
    N = read_i32(f)
    sets = []
    for _ in range(N):
        kr = [read_double(f) for _ in range(15)]
        kt = [read_double(f), read_double(f)]
        min_valid_radius, max_valid_radius = read_double(f), read_double(f)
        cx, cy = read_double(f), read_double(f)
        reciprocalRadial = read_i32(f)
        sets.append({"kr": kr, "kt": kt, "min_valid_radius": min_valid_radius,
                     "max_valid_radius": max_valid_radius, "cx": cx, "cy": cy,
                     "reciprocalRadial": reciprocalRadial})
    return {"N": N, "coeff_sets": sets}

PARSERS = {
    1: parse_warp_rectilinear,
    2: parse_warp_fisheye,
    3: parse_fix_vignette_radial,
    4: parse_fix_bad_pixels_constant,
    5: parse_fix_bad_pixels_list,
    6: parse_trim_bounds,
    7: parse_map_table,
    8: parse_map_polynomial,
    9: parse_gain_map,
    10: parse_delta_per_row,
    11: parse_delta_per_column,
    12: parse_scale_per_row,
    13: parse_scale_per_column,
    14: parse_warp_rectilinear2,
}

# ---------- main ----------
def parse_opcodelist_file(path):
    data = open(path, "rb").read()
    f = io.BytesIO(data)

    try:
        total_count = read_u32(f)
    except EOFError:
        print("File too short or empty.")
        return

    # Pre-read headers for stats
    headers = []
    for idx in range(total_count):
        try:
            opcode_id = read_u32(f)
            version_u32 = read_u32(f)
            flags = read_u32(f)
            param_bytes = read_u32(f)
        except EOFError:
            break
        headers.append((opcode_id,))
        f.seek(param_bytes, io.SEEK_CUR)

    counts = Counter(OPCODE_NAMES.get(op, "Unknown") for (op,) in headers)
    print(f"OpcodeList: total opcodes = {total_count}")
    for name, cnt in counts.items():
        print(f"Opcode [{name}] count = {cnt}")

    # Reset and parse fully
    f.seek(4)
    for idx in range(total_count):
        try:
            opcode_id = read_u32(f)
            version_u32 = read_u32(f)
            flags = read_u32(f)
            param_bytes = read_u32(f)
        except EOFError:
            break

        version_str = version_from_u32(version_u32)
        name = OPCODE_NAMES.get(opcode_id, "Unknown")

        print(f"\nOpcode #{idx}: ID={opcode_id} ({name}), version={version_str}, flags=0x{flags:08X}, param_bytes={param_bytes}")

        try:
            param_blob = read_bytes(f, param_bytes)
        except EOFError:
            print("Unexpected EOF while reading parameter block.")
            break

        param_f = io.BytesIO(param_blob)
        parser = PARSERS.get(opcode_id)
        if parser is None:
            snippet = param_blob[:64].hex()
            print(f"  Unknown opcode ID: raw parameter blob (first 64 bytes hex): {snippet}...")
            continue

        try:
            parsed = parser(param_f)
            consumed = param_f.tell()
            if consumed != param_bytes:
                rem_bytes = param_f.read()
                if rem_bytes.strip(b"\x00") != b"":
                    print(f"  Warning: parser consumed {consumed} bytes but param_bytes={param_bytes}")
            print(" ", pretty(parsed))
        except Exception as e:
            print("  Exception while parsing:", repr(e))

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python parse_dng_opcodelist.py <OpcodeList.bin>")
        sys.exit(1)
    parse_opcodelist_file(sys.argv[1])