JezK

Edit File: build_jp2_from_xml.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ******************************************************************************
#  $Id$
#
#  Project:  GDAL
#  Purpose:  Build a JPEG2000 file from the XML structure dumped by dump_jp2.py
#            Mostly useful to build non-conformant files
#  Author:   Even Rouault, <even dot rouault at spatialys dot com>
#
# ******************************************************************************
#  Copyright (c) 2015, European Union (European Environment Agency)
#
#  Permission is hereby granted, free of charge, to any person obtaining a
#  copy of this software and associated documentation files (the "Software"),
#  to deal in the Software without restriction, including without limitation
#  the rights to use, copy, modify, merge, publish, distribute, sublicense,
#  and/or sell copies of the Software, and to permit persons to whom the
#  Software is furnished to do so, subject to the following conditions:
#
#  The above copyright notice and this permission notice shall be included
#  in all copies or substantial portions of the Software.
#
#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
#  OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
#  THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
#  FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
#  DEALINGS IN THE SOFTWARE.
# ******************************************************************************

import os
import struct
import sys

from osgeo import gdal

XML_TYPE_IDX = 0
XML_VALUE_IDX = 1
XML_FIRST_CHILD_IDX = 2

def Usage():
    print("Usage: build_jp2_from_xml in.xml out.jp2")
    return 2

def find_xml_node(ar, element_name, immediate_child=False, only_attributes=False):
    # type = ar[XML_TYPE_IDX]
    value = ar[XML_VALUE_IDX]
    if not immediate_child and value == element_name:
        return ar
    for child_idx in range(XML_FIRST_CHILD_IDX, len(ar)):
        child = ar[child_idx]
        if only_attributes and child[XML_TYPE_IDX] != gdal.CXT_Attribute:
            continue
        if immediate_child:
            if child[XML_VALUE_IDX] == element_name:
                return child
        else:
            found = find_xml_node(child, element_name)
            if found is not None:
                return found
    return None

def get_attribute_val(ar, attr_name):
    node = find_xml_node(ar, attr_name, True)
    if node is None or node[XML_TYPE_IDX] != gdal.CXT_Attribute:
        return None
    if (
        len(ar) > XML_FIRST_CHILD_IDX
        and node[XML_FIRST_CHILD_IDX][XML_TYPE_IDX] == gdal.CXT_Text
    ):
        return node[XML_FIRST_CHILD_IDX][XML_VALUE_IDX]
    return None

def get_node_content(node):
    if node is None:
        return None
    for child_idx in range(XML_FIRST_CHILD_IDX, len(node)):
        child = node[child_idx]
        if child[XML_TYPE_IDX] == gdal.CXT_Text:
            return child[XML_VALUE_IDX]
    return None

def hex_letter_to_number(ch):
    val = 0
    if ch >= "0" and ch <= "9":
        val = ord(ch) - ord("0")
    elif ch >= "a" and ch <= "f":
        val = (ord(ch) - ord("a")) + 10
    elif ch >= "A" and ch <= "F":
        val = (ord(ch) - ord("A")) + 10
    return val

def write_hexstring_as_binary(hex_binary_content, out_f):
    for i in range(int(len(hex_binary_content) / 2)):
        val = hex_letter_to_number(
            hex_binary_content[2 * i]
        ) * 16 + hex_letter_to_number(hex_binary_content[2 * i + 1])
        out_f.write(chr(val).encode("latin1"))

def parse_field(xml_tree, out_f, src_jp2file):
    # pylint: disable=unused-argument
    if not (
        xml_tree[XML_TYPE_IDX] == gdal.CXT_Element
        and xml_tree[XML_VALUE_IDX] == "Field"
    ):
        print("Not a Field element")
        return False
    field_name = get_attribute_val(xml_tree, "name")
    if field_name is None:
        print("Cannot find Field.name attribute")
        # return False
    field_type = get_attribute_val(xml_tree, "type")
    if field_type is None:
        print("Cannot find Field.type attribute")
        return False
    val = get_node_content(xml_tree)
    if val is None:
        print("Cannot find Field content")
        return False
    if field_type == "uint8":
        out_f.write(struct.pack(">B" * 1, int(val)))
    elif field_type == "uint16":
        out_f.write(struct.pack(">H" * 1, int(val)))
    elif field_type == "uint32":
        out_f.write(struct.pack(">I" * 1, int(val)))
    elif field_type == "string":
        field_size = get_attribute_val(xml_tree, "size")
        if field_size is not None:
            assert len(val) == int(field_size)
        out_f.write(val.encode("latin1"))
    elif field_type == "hexint":
        field_size = get_attribute_val(xml_tree, "size")
        if field_size is not None:
            assert len(val) == 2 + 2 * int(field_size)
        write_hexstring_as_binary(val[2:], out_f)
    else:
        print("Unhandled type %s" % field_type)
        return False
    return True

marker_map = {
    "SOC": 0xFF4F,
    "EOC": 0xFFD9,
    "SOT": 0xFF90,
    "SIZ": 0xFF51,
    "COD": 0xFF52,
    "COC": 0xFF53,
    "TLM": 0xFF55,
    "PLM": 0xFF57,
    "PLT": 0xFF58,
    "QCD": 0xFF5C,
    "QCC": 0xFF5D,
    "RGN": 0xFF5E,
    "POC": 0xFF5F,
    "PPM": 0xFF60,
    "PPT": 0xFF61,
    "CRG": 0xFF63,
    "COM": 0xFF64,
}

def parse_jpc_marker(xml_tree, out_f, src_jp2file):
    if not (
        xml_tree[XML_TYPE_IDX] == gdal.CXT_Element
        and xml_tree[XML_VALUE_IDX] == "Marker"
    ):
        print("Not a Marker element")
        return False
    marker_name = get_attribute_val(xml_tree, "name")
    if marker_name is None:
        print("Cannot find Marker.name attribute")
        return False
    if find_xml_node(xml_tree, "Field", immediate_child=True):
        if marker_name not in marker_map:
            print("Cannot find marker signature for %s" % marker_name)
            return False
        marker_signature = marker_map[marker_name]
        out_f.write(struct.pack(">H" * 1, marker_signature))
        pos = out_f.tell()
        out_f.write(struct.pack(">H" * 1, 0))

for child_idx in range(XML_FIRST_CHILD_IDX, len(xml_tree)):
            child = xml_tree[child_idx]
            if child[XML_TYPE_IDX] != gdal.CXT_Element:
                continue
            if not parse_field(child, out_f, src_jp2file):
                return False

new_pos = out_f.tell()
        out_f.seek(pos, 0)
        out_f.write(struct.pack(">H" * 1, new_pos - pos))
        out_f.seek(new_pos, 0)
    else:
        offset = get_attribute_val(xml_tree, "offset")
        if offset is None:
            if marker_name == "SOC" or marker_name == "EOC":
                marker_signature = marker_map[marker_name]
                out_f.write(struct.pack(">H" * 1, marker_signature))
                return True

print("Cannot find Marker.offset attribute")
            return False
        offset = int(offset)

length = get_attribute_val(xml_tree, "length")
        if length is None:
            print("Cannot find Marker.length attribute")
            return False
        length = int(length)

src_jp2file.seek(offset, 0)
        data = src_jp2file.read(length)

out_f.write(data)

return True

def parse_jp2codestream(inpath, xml_tree, out_f, src_jp2file=None):

if src_jp2file is None:
        src_jp2filename = get_attribute_val(xml_tree, "filename")
        if src_jp2filename is None:
            print("Cannot find JP2KCodeStream.filename attribute")
            return False
        if os.path.exists(src_jp2filename):
            src_jp2file = open(src_jp2filename, "rb")
        else:
            src_jp2file = open(os.path.join(inpath, src_jp2filename), "rb")

for child_idx in range(XML_FIRST_CHILD_IDX, len(xml_tree)):
        child = xml_tree[child_idx]
        if child[XML_TYPE_IDX] != gdal.CXT_Element:
            continue
        if not parse_jpc_marker(child, out_f, src_jp2file):
            return False
    return True

def parse_jp2_box(xml_tree, out_f, src_jp2file):
    if not (
        xml_tree[XML_TYPE_IDX] == gdal.CXT_Element
        and xml_tree[XML_VALUE_IDX] == "JP2Box"
    ):
        print("Not a JP2Box element")
        return False
    jp2box_name = get_attribute_val(xml_tree, "name")
    if jp2box_name is None:
        print("Cannot find JP2Box.name attribute")
        return False
    if len(jp2box_name) != 4:
        print("Invalid JP2Box.name : %s" % jp2box_name)
        return False
    hex_binary_content = get_node_content(
        find_xml_node(xml_tree, "BinaryContent", immediate_child=True)
    )
    decoded_content = find_xml_node(xml_tree, "DecodedContent", immediate_child=True)
    decoded_geotiff = find_xml_node(xml_tree, "DecodedGeoTIFF", immediate_child=True)
    text_content = get_node_content(
        find_xml_node(xml_tree, "TextContent", immediate_child=True)
    )
    xml_content = find_xml_node(xml_tree, "XMLContent", immediate_child=True)
    jp2box = find_xml_node(xml_tree, "JP2Box", immediate_child=True)
    jp2codestream = find_xml_node(xml_tree, "JP2KCodeStream", immediate_child=True)

if hex_binary_content:
        if decoded_content or decoded_geotiff or text_content or xml_content or jp2box:
            print(
                "BinaryContent found, and one of DecodedContent/DecodedGeoTIFF/TextContent/XMLContent/JP2Box. The latter will be ignored"
            )
        if jp2box_name == "uuid":
            uuid = get_node_content(
                find_xml_node(xml_tree, "UUID", immediate_child=True)
            )
            if uuid is None:
                print("Cannot find JP2Box.UUID element")
                return False
        else:
            uuid = ""
        out_f.write(
            struct.pack(
                ">I" * 1, 8 + int(len(hex_binary_content) / 2) + int(len(uuid) / 2)
            )
        )
        out_f.write(jp2box_name.encode("ascii"))
        write_hexstring_as_binary(uuid, out_f)
        write_hexstring_as_binary(hex_binary_content, out_f)

elif decoded_content:
        if decoded_geotiff or text_content or xml_content or jp2box:
            print(
                "DecodedContent found, and one of DecodedGeoTIFF/TextContent/XMLContent/JP2Box. The latter will be ignored"
            )
        pos = out_f.tell()
        out_f.write(struct.pack(">I" * 1, 0))
        out_f.write(jp2box_name.encode("ascii"))
        for child_idx in range(XML_FIRST_CHILD_IDX, len(decoded_content)):
            child = decoded_content[child_idx]
            if (
                child[XML_TYPE_IDX] == gdal.CXT_Element
                and child[XML_VALUE_IDX] == "Field"
            ):
                if not parse_field(child, out_f, src_jp2file):
                    return False
        new_pos = out_f.tell()
        out_f.seek(pos, 0)
        out_f.write(struct.pack(">I" * 1, new_pos - pos))
        out_f.seek(new_pos, 0)

elif text_content:
        if decoded_geotiff or xml_content or jp2box:
            print(
                "TextContent found, and one of DecodedGeoTIFF/XMLContent/JP2Box. The latter will be ignored"
            )
        out_f.write(struct.pack(">I" * 1, 8 + len(text_content)))
        out_f.write(jp2box_name.encode("ascii"))
        out_f.write(text_content.encode("latin1"))

elif xml_content:
        if decoded_geotiff or jp2box:
            print(
                "XMLContent found, and one of DecodedGeoTIFF/JP2Box. The latter will be ignored"
            )
        serialized_xml_content = gdal.SerializeXMLTree(xml_content[XML_FIRST_CHILD_IDX])
        out_f.write(struct.pack(">I" * 1, 8 + len(serialized_xml_content)))
        out_f.write(jp2box_name.encode("ascii"))
        out_f.write(serialized_xml_content.encode("latin1"))

elif jp2box:
        if decoded_geotiff:
            print("JP2Box found, and one of DecodedGeoTIFF. The latter will be ignored")
        pos = out_f.tell()
        out_f.write(struct.pack(">I" * 1, 0))
        out_f.write(jp2box_name.encode("ascii"))
        for child_idx in range(XML_FIRST_CHILD_IDX, len(xml_tree)):
            child = xml_tree[child_idx]
            if (
                child[XML_TYPE_IDX] == gdal.CXT_Element
                and child[XML_VALUE_IDX] == "JP2Box"
            ):
                if not parse_jp2_box(child, out_f, src_jp2file):
                    return False
        new_pos = out_f.tell()
        out_f.seek(pos, 0)
        out_f.write(struct.pack(">I" * 1, new_pos - pos))
        out_f.seek(new_pos, 0)

elif decoded_geotiff:
        serialized_xml_content = gdal.SerializeXMLTree(
            decoded_geotiff[XML_FIRST_CHILD_IDX]
        )

vrt_ds = gdal.Open(serialized_xml_content)
        if vrt_ds is None:
            print("Cannot decode VRTDataset. Outputting empty content")
            binary_content = ""
        else:
            tmpfilename = "/vsimem/build_jp2_from_xml_tmp.tif"
            gdal.GetDriverByName("GTiff").CreateCopy(tmpfilename, vrt_ds)
            tif_f = gdal.VSIFOpenL(tmpfilename, "rb")
            binary_content = gdal.VSIFReadL(1, 10000, tif_f)
            gdal.VSIFCloseL(tif_f)
            gdal.Unlink(tmpfilename)

uuid = get_node_content(find_xml_node(xml_tree, "UUID", immediate_child=True))
        if uuid is None:
            uuid = "B14BF8BD083D4B43A5AE8CD7D5A6CE03"

out_f.write(struct.pack(">I" * 1, 8 + len(binary_content) + int(len(uuid) / 2)))
        out_f.write(jp2box_name.encode("ascii"))
        write_hexstring_as_binary(uuid, out_f)
        out_f.write(binary_content)

elif jp2codestream:
        pos = out_f.tell()
        out_f.write(struct.pack(">I" * 1, 0))
        out_f.write(jp2box_name.encode("ascii"))
        if not parse_jp2codestream(None, jp2codestream, out_f, src_jp2file):
            return False
        new_pos = out_f.tell()
        out_f.seek(pos, 0)
        out_f.write(struct.pack(">I" * 1, new_pos - pos))
        out_f.seek(new_pos, 0)

else:
        data_offset = get_attribute_val(xml_tree, "data_offset")
        if data_offset is None:
            print("Cannot find JP2Box.data_offset attribute")
            return False
        data_offset = int(data_offset)

data_length = get_attribute_val(xml_tree, "data_length")
        if data_length is None:
            print("Cannot find JP2Box.data_length attribute")
            return False
        data_length = int(data_length)

src_jp2file.seek(data_offset, 0)
        data = src_jp2file.read(data_length)

out_f.write(struct.pack(">I" * 1, 8 + data_length))
        out_f.write(jp2box_name.encode("ascii"))
        out_f.write(data)

return True

def parse_jp2file(inpath, xml_tree, out_f):
    src_jp2filename = get_attribute_val(xml_tree, "filename")
    if src_jp2filename is None:
        print("Cannot find JP2File.filename attribute")
        return False
    if os.path.exists(src_jp2filename):
        src_jp2file = open(src_jp2filename, "rb")
    else:
        src_jp2file = open(os.path.join(inpath, src_jp2filename), "rb")
    for child_idx in range(XML_FIRST_CHILD_IDX, len(xml_tree)):
        child = xml_tree[child_idx]
        if child[XML_TYPE_IDX] != gdal.CXT_Element:
            continue
        if not parse_jp2_box(child, out_f, src_jp2file):
            return False
    return True

# Wrapper class for GDAL VSI*L API with class Python file interface

class VSILFile(object):
    def __init__(self, filename, access):
        self.f = gdal.VSIFOpenL(filename, access)

def write(self, data):
        gdal.VSIFWriteL(data, 1, len(data), self.f)

def tell(self):
        return gdal.VSIFTellL(self.f)

def seek(self, pos, ref):
        gdal.VSIFSeekL(self.f, pos, ref)

def close(self):
        gdal.VSIFCloseL(self.f)
        self.f = None

def build_file(inname, outname):

inpath = os.path.dirname(inname)
    xml_tree = gdal.ParseXMLString(open(inname).read())
    if xml_tree is None:
        print("Cannot parse %s" % inname)
        return False

# out_f = open(outname, 'wb+')
    out_f = VSILFile(outname, "wb+")
    if (
        xml_tree[XML_TYPE_IDX] == gdal.CXT_Element
        and xml_tree[XML_VALUE_IDX] == "JP2File"
    ):
        ret = parse_jp2file(inpath, xml_tree, out_f)
    elif (
        xml_tree[XML_TYPE_IDX] == gdal.CXT_Element
        and xml_tree[XML_VALUE_IDX] == "JP2KCodeStream"
    ):
        ret = parse_jp2codestream(inpath, xml_tree, out_f)
    else:
        print("Unexpected node: %s" % xml_tree[XML_VALUE_IDX])
        ret = False
    out_f.close()

return ret

def main(argv=sys.argv):
    i = 1
    inname = None
    outname = None
    while i < len(argv):
        if sys.argv[i][0] == "-":
            return Usage()
        elif inname is None:
            inname = sys.argv[i]
        elif outname is None:
            outname = sys.argv[i]
        else:
            return Usage()

i = i + 1

if inname is None or outname is None:
        return Usage()

if build_file(inname, outname):
        return 0
    return 1

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

JezK

Edit File: build_jp2_from_xml.py

Events