Extract RPFDES NITF segment information

rpfdes.py

import argparse
import sys
import struct
import codecs
import json

import logging
import sys

logger = logging.getLogger("rpfddes")
formatter = logging.Formatter("[%(levelname)s] %(message)s")
handler = logging.StreamHandler(stream=sys.stderr)
handler.setFormatter(formatter)
logger.addHandler(handler)

def readInt(data, offset, size):
    """Read an integer from a byte array at offset of a given size"""
    byt = data[offset:offset + size]
    output = int.from_bytes(byt, byteorder='big')
    return output

def readAscii(data, offset, size):
    """Read an ASCII string from a byte array at offset of a given size"""
    byt = data[offset:offset + size]
    s = byt.decode('utf-8').strip()
    return s

def readOffsets(data, offset):
    """Read the offset records starting at the provided byte offset for the coverage sequences"""

    # First 2 bytes are the number of offset records we have
    size = 2
    count = readInt(data, offset, size)
    logger.info(f'There were {count} offset records')
    offset += size

    offsets = []
    for i in range(count):

        size = 2
        attribute_id = readInt(data, offset, size)
        offset += size

        size = 1
        parameter_id = readInt(data, offset, size)
        offset += size

        size = 1
        covseqnum = readInt(data, offset, size)
        offset += size

        size = 4
        recordoffset = readInt(data, offset, size)
        offset += size

        record = {'ATTRIBUTEID':attribute_id,
                  'PARAMETERID':parameter_id,
                  'COVSEQNUM':covseqnum,
                  'RECORDOFFSET': recordoffset}
        offsets.append(record)
        logger.info(f'Reading {record} offset record')

    return offsets


def readAttributes(data, offsets):
    """Using the offsets, go read the attributes"""

    attributes = {}
    for record in offsets:
        coverage = record['COVSEQNUM']
        if coverage not in attributes.keys():
            attributes[coverage] = {}


        # We need to slide over 10 bytes which is the ATTRIBUTE_SECTION_SUBHEADER_LENGTH
        # https://github.com/codice/imaging-nitf/blob/master/core/src/main/java/org/codice/imaging/nitf/core/image/impl/RasterProductFormatAttributeParser.java#L33C30-L33C64
        offset = record['RECORDOFFSET'] + 10

        # The size of the attribute is fixed and depends on its definition, which
        # comes from an ancient PDF – MIL-PRF-89041A w/ AMENDMENT 1
        if record['PARAMETERID'] == 1:
            if record['ATTRIBUTEID'] == 1: # SIGNIFICANT_DATE_ATTR_ID
                size = 8
                value = readAscii(data, offset, size)
                attributes[coverage]['SIGNIFICANT_DATE'] = value
            elif record['ATTRIBUTEID'] == 2: # CURRENCY_DATE_ATTR_ID
                size = 8
                value = readAscii(data, offset, size)
                attributes[coverage]['CURRENCY_DATE'] = value
            elif record['ATTRIBUTEID'] == 3: # PRODUCTION_DATE_ATTR_ID
                size = 8
                value = readAscii(data, offset, size)
                attributes[coverage]['PRODUCTION_DATE'] = value
            elif record['ATTRIBUTEID'] == 8: # VERTICAL_ABSOLUTE_ACCURACY_ATTR_ID
                size = 4
                value = readInt(data, offset, size)
                attributes[coverage]['VERTICAL_ABSOLUTE_ACCURACY'] = value
            elif record['ATTRIBUTEID'] == 9: # HORIZONTAL_ABSOLUTE_ACCURACY_ATTR_ID
                size = 4
                value = readInt(data, offset, size)
                attributes[coverage]['HORIZONTAL_ABSOLUTE_ACCURACY'] = value
            elif record['ATTRIBUTEID'] == 10: # VERTICAL_RELATIVE_ACCURACY_ATTR_ID
                size = 4
                value = readInt(data, offset, size)
                attributes[coverage]['VERTICAL_RELATIVE_ACCURACY'] = value
            elif record['ATTRIBUTEID'] == 11: # HORIZONTAL_RELATIVE_ACCURACY_ATTR_ID
                size = 4
                value = readInt(data, offset, size)
                attributes[coverage]['HORIZONTAL_RELATIVE_ACCURACY'] = value
            elif record['ATTRIBUTEID'] == 22: # DATA_LEVEL_ATTR_ID
                size = 12
                value = readAscii(data, offset, size)
                attributes[coverage]['SOURCE'] = value
        elif record['PARAMETERID'] == 2:
            if record['ATTRIBUTEID'] == 9: # HORIZONTAL_ACCURACY_UOM
                size = 2
                value = readInt(data, offset, size)
                attributes[coverage]['HORIZONTAL_ACCURACY_UOM'] = value
            if record['ATTRIBUTEID'] == 11: # VERTICAL_ACCURACY_UOM
                size = 2
                value = readInt(data, offset, size)
                attributes[coverage]['VERTICAL_ACCURACY_UOM'] = value
            if record['ATTRIBUTEID'] == 22: # GSD
                size = 4
                value = readInt(data, offset, size)
                attributes[coverage]['GSD'] = value
        # attributes.append(attribute)
        logger.info(f'Extracted {attributes} attribute')

    return attributes

def extractAttributes(data, offset):
    offsets = readOffsets(data, offset)
    attributes = readAttributes(data, offsets)
    return attributes



def doIt(args):
    if sys.stdin.isatty():
        raise Exception('Could not read any data from stdin!')

    args.data = sys.stdin.read()
    if args.escaped:
        args.data, length = codecs.escape_decode(args.data)

    logger.info(f'Read {len(args.data)} bytes from stdin')
    attributes = extractAttributes(args.data, args.offset)
    sys.stdout.write(json.dumps(attributes))

    return True

def get_parser(args):

    parser = argparse.ArgumentParser(description='Extract RPFDES NITF TRE segment binary data to JSON')
    parser.add_argument('--debug',
                        action='store_true',
                        help='print debug messages to stderr')
    parser.add_argument('--offset',
                        type=int,default = 0,
                        help='Offset to begin reading data from input bytes/string')
    parser.add_argument('--escaped',
                        action='store_true',default = False,
                        help='Incoming data is unicode escaped string instead of binary')

    args = parser.parse_args(args)
    return args

def main():
    args = get_parser(sys.argv[1:])
    if args.debug:
        logger.setLevel(logging.DEBUG)
        handler.setLevel(logging.DEBUG)

    doIt(args)

if __name__ == "__main__":
    sys.exit(main())


class Test:

    example = b"""\x00\x19\x00\x03\x00\x00\x00\x00\x00\x08\x00\x02\x01\x00\x00\x00\x00\xc8\x00\x01\x01\x01\x00\x00\x00\xd0\x00\x03\x01\x01\x00\x00\x00\xd8\x00\x16\x01\x01\x00\x00\x00\xe0\x00\x16\x02\x01\x00\x00\x00\xec\x00\t\x01\x01\x00\x00\x00\xf0\x00\t\x02\x01\x00\x00\x00\xf4\x00\x0b\x01\x01\x00\x00\x00\xf6\x00\x0b\x02\x01\x00\x00\x00\xfa\x00\x01\x01\x02\x00\x00\x00\xfc\x00\x03\x01\x02\x00\x00\x01\x04\x00\x16\x01\x02\x00\x00\x01\x0c\x00\x16\x02\x02\x00\x00\x01\x18\x00\t\x01\x02\x00\x00\x01\x1c\x00\t\x02\x02\x00\x00\x01 \x00\x0b\x01\x02\x00\x00\x01"\x00\x0b\x02\x02\x00\x00\x01&\x00\x01\x01\x03\x00\x00\x01(\x00\x03\x01\x03\x00\x00\x010\x00\x16\x01\x03\x00\x00\x018\x00\x16\x02\x03\x00\x00\x01D\x00\t\x01\x03\x00\x00\x01H\x00\t\x02\x03\x00\x00\x01L\x00\x0b\x01\x03\x00\x00\x01N\x00\x0b\x02\x03\x00\x00\x01R202104082021040820070616NTM         \x00\x00\x00\x01\x00\x00\x00\x07\x00\x01\x00\x00\x00\x00\x00\x012021040820030328NTM         \x00\x00\x00\x01\x00\x00\x00\x07\x00\x01\x00\x00\x00\x00\x00\x012021040820181006WORLDVIEW2  \x00\x00\x00\x01\x00\x00\x00\x07\x00\x01\x00\x00\x00\x00\x00\x01\x00\x00\x00\x08\x00\x02\x00\x10\x00\x1a@"\xa22O\xc1\xdb\xed\xc0S\xca\x15\xc1_\x06\xc7@"\xa2\x8fKU\x86\xf1\xc0S\xca\x10\xb4B\xe3\x88@"\xa2X\xcd+\xe0\xb7\xc0S\xca\x00\x97\x84\xc0\xf1@"\xa4\xdd\xa1\x069*\xc0S\xc8\x9e\tY\xba\x1b@"\xab\\j\xe67\xa4\xc0S\xc9\xf3^qN\xa5@"\xab\x82K\x96\xc7L\xc0S\xcbV\x92\xd6\xa4\xf8@"\xa5VN.\xaff\xc0S\xcb\xd0\xd8\t]!@"\xa8\x06\x89(\x85\x03\xc0S\xcc\xd7\xbd\x9e$d@"\xab\xad\x88\xf6"0\xc0S\xcc\xec\x0f\xce_\xa7@"\xacg4\xfe\xf7\xa6\xc0S\xd3\xb9;\xbcC\x10@"\xa5n\xa6`\x8d9\xc0S\xd4E\xfdg\x13\n@"\xa4@}D9\x93\xc0S\xd3\x0b\x818B\xe7@"\xa30\x8e\x97\x92A\xc0S\xd2\xea\x18H\xfe>@"\xa2\xf0\xf9\x85e\x0e\xc0S\xd2\xb7\xd9\xe2\x17\xe0@"\xa3\xba\xd0\xe2\xc7\xc6\xc0S\xd2\x80a\x0f\xa4l@"\xa1\xf6b\x1a\x98\xd8\xc0S\xd0\xa9~\x1b\xf3\xa2@"\xa33\xcb\xb4\xf0\x80\xc0S\xd0\x80\x99\x133\x89@"\xa1\x8e,\x844P\xc0S\xd0=\x08f\x04\x89@"\xa11i\xfb\xb0\xf1\xc0S\xcf\xdc}K+\xe5@"\xa23\x11[4\x8e\xc0S\xcf\xa4\xed\xd6R\xc6@"\xa0\xcc~\xa6\x8a\xda\xc0S\xcfst1\x0f>@"\xa06\xa4H\'\\\xc0S\xce\xd7}\x1a\x12\x80@"\xa8\x06\t)e\x16\xc0S\xce\xca\x9fg\xb7)@"\xa8\xf1X\x02F]\xc0S\xce.\x1e0E\xa8@"\xa8\xe9?8\x8f%\xc0S\xcd\xf0\xfa\xa8v\x93@"\x9f\xa2\x0c\xdc\x12\xbf\xc0S\xcb~\x98\xf5\xf6^\x00\x0b@"\xc3B\x93\x8e\xacN\xc0S\xc6x\xf1\x93\x18\xb4@"\xbd\xe0$\xd1\xc4\x18\xc0S\xc7T"\xf4\xb3\xb5@"\xae\xed\xc4\xdc\x8c\x89\xc0S\xc8\xf0\x82\x80\xbbU@"\xa4\xdd\xa1\x069*\xc0S\xc8\x9e\tY\xba\x1b@"\xa4\xd5\xe9\xfa`2\xc0S\xc6K)\xf2\x88\xd2@"\xa4\x97\xf0\x96\x9d\xd0\xc0S\xc6:7h\x83\xf2@"\xa4\xd5\x9a\xbd\xb0h\xc0S\xc63LP\xc6\xae@"\xa4\xcd?\xb2\x8a\x92\xc0S\xc3\xaf\x07\xc80\xa9@"\xa5zO>J\xfe\xc0S\xc0\xbf\x02\xc8"\xb0@"\xa5\xd7[\xe5\x10\xbc\xc0S\xbf\xc8\x15\xb0\xaf\xd0@"\xc3\ro\xd6\xb3\xbb\xc0S\xbf\xcbjR!\xcf\x00\x18@$\x08x>\xcc\xc8\xff\xc0S\xcf\x08Js\xc4\xb8@#*\x01\x05\x18\xa6\'\xc0S\xcf!\xd2\x94\xdc\x83@#$Q\xad\\\x7f\x9d\xc0S\xd0\x95Rk\xaa\xcc@#\x1b\xab~\xfae\x0b\xc0S\xd0;\x89-\x99\x93@#\x19\xe1\xbd\xa4\xce}\xc0S\xd0\x9a\x81a\x87\xca@#\x18\xbbu\xe93*\xc0S\xd0\x90\xce\xedg\x01@#\x18\xb8\xef\xd8\xbc\xba\xc0S\xd0\x9b\x1bRS\xfe@#\x0b\x1cY\x84\x1c\xbe\xc0S\xd0\xa2*\xa4&\xea@"\xf01@(\x9c\x88\xc0S\xd0\xaf5\xa5t\xfc@"\xe02\xe5?\xa0\x9e\xc0S\xd0\xf9#jU\xff@"\xc8\xad#\x0e\x16\x14\xc0S\xd1\x13*[\xdd\xe0@"\xc0\'\xa7\xc3\x12\xfc\xc0S\xd0\xfcL9\xdb\xad@"\xb4M\x9c@\xac\xc2\xc0S\xd0I\x9f\x9e\x9e\x86@"\xb4\xdc\xf9\xd4=\xbb\xc0S\xcc\xbch)\xfew@"\xab\\j\xe67\xa4\xc0S\xc9\xf3^qN\xa5@"\xa4\xdd\xa1\x069*\xc0S\xc8\x9e\tY\xba\x1b@"\xae\xed\xc4\xdc\x8c\x89\xc0S\xc8\xf0\x82\x80\xbbU@"\xbd\xe0$\xd1\xc4\x18\xc0S\xc7T"\xf4\xb3\xb5@"\xe7\xd4\xa4S8U\xc0S\xc7\x87\xba?w@@"\xfb\x1aL\'\x860\xc0S\xc8/\xd8\xa2\xcc\x06@"\xfcW\xc9>\x19\xef\xc0S\xc7\xdb;\xf2\xed\xe3@#\x02.t-yo\xc0S\xc8+\xaee\x8e\xf1@#\x1e\x07\x83\x8e6s\xc0S\xc8\\\xa8\xb4\xa07@$\t\xa9\x1c\x01\x00\x1a\xc0S\xc8\xe2\x7f\xe5\xed\xe7"""

    def test_extraction(self):
        attributes = extractAttributes(self.example, 0)
        assert len(attributes) == 4
        assert attributes[1]['SIGNIFICANT_DATE'] == "20210408"
        assert attributes[1]['PRODUCTION_DATE'] == "20070616"
        assert attributes[1]['SOURCE'] == "NTM"
        assert attributes[3]['PRODUCTION_DATE'] == "20181006"
        assert attributes[3]['SOURCE'] == "WORLDVIEW2"