pbf2json

Convert a Protobuf file to JSON using its corresponding .proto definitions for easy inspection.

#!/usr/bin/env -S uv run -q --script 
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "protobuf",
#     "typer",
# ]
# ///
# SPDX-FileCopyrightText: 2025 kurt.town
# SPDX-License-Identifier: MIT

from pathlib import Path
from types import ModuleType
import importlib.util
import sys
import os
import json
from google.protobuf.json_format import MessageToDict
import tempfile
import shutil
import subprocess
import typer

def import_file_as_module(file_path: str) -> ModuleType:
    module_name = os.path.splitext(os.path.basename(file_path))[0]

    # Add the containing directory to sys.path
    module_dir = os.path.dirname(file_path)
    sys.path.insert(0, module_dir)

    # Load the module
    spec = importlib.util.spec_from_file_location(module_name, file_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module

def init_class(module: ModuleType, class_name: str) -> None:
    cls = getattr(module, class_name, None)
    if cls is None:
        raise ImportError(f"Class {class_name} not found in module {module.__name__}")
    return cls()

def pbf2json(proto_file: Path, pbf_file: Path, message_name: str) -> str:
    """
    Convert a Protocol Buffer file to JSON using its corresponding .proto file.
    """
    pbf_file = pbf_file.resolve()
    proto_file = proto_file.resolve()
    tmpdir = Path(tempfile.mkdtemp())
    os.system(f"protoc -I=. --python_out={tmpdir} {proto_file}")
    try:
        subprocess.run(["protoc", f"-I={proto_file.parent}", "--python_out", str(tmpdir), str(proto_file)], check=True)
    except subprocess.CalledProcessError as e:
        print(f"Error running protoc: {e}", file=sys.stderr)
        shutil.rmtree(tmpdir, ignore_errors=True)
        sys.exit(1)

    python_file = tmpdir / (os.path.splitext(proto_file.name)[0] + "_pb2.py")
    
    module = import_file_as_module(python_file)
    message = init_class(module, message_name)
    with open(pbf_file, "rb") as fd:
        message.ParseFromString(fd.read())
        dict_obj = MessageToDict(message)
        print(json.dumps(dict_obj, indent=2))
    
    shutil.rmtree(tmpdir, ignore_errors=True)


if __name__ == "__main__":
    typer.run(pbf2json)