def locate_graph_header()

in scripts/parser.py [0:0]

• 19 lines of code
• 6 McCabe index (conditional complexity)

def locate_graph_header(fetcher, probe_size=1*1024*1024):
    """
    Download the first `probe_size` bytes, scan for ModelProto.graph (field 7, wire 2).
    Returns (before_bytes, graph_payload_offset, graph_length).
    """
    data = fetcher.fetch(0, probe_size-1)
    stream = io.BytesIO(data)
    before = bytearray()

    while True:
        key_pos = stream.tell()
        try:
            key, key_len = read_varint(stream)
        except EOFError:
            raise RuntimeError("Couldn't find graph header in initial bytes")
        field_num, wire_type = key >> 3, key & 0x7

        if field_num == 7 and wire_type == 2:
            length, length_len = read_varint(stream)
            header_len = key_len + length_len
            payload_off = key_pos + header_len
            return data[:key_pos], payload_off, length

        # skip and accumulate everything before
        skipped = skip_field(stream, wire_type)
        stream.seek(key_pos + key_len + skipped)
        before.extend(data[key_pos:key_pos + key_len + skipped])