cs_extract.py

import os
import csv
import json
import argparse
import sys
import xml.etree.ElementTree as ET

# ======================
# CONSTANTS
# ======================
NS = {'cs': 'http://www.osgeo.org/mapguide/coordinatesystem'}

# ======================
# FUNCTION DEFINITIONS
# ======================

def process_xml_file(xml_path):
    """
    Parse the given XML file and extract coordinate system details.
    Returns a list of records, one per coordinate system found.
    """
    records = []
    try:
        tree = ET.parse(xml_path)
        root = tree.getroot()

        cs_types = ['ProjectedCoordinateSystem', 'GeographicCoordinateSystem']

        for cs_type in cs_types:
            for cs_elem in root.findall(f".//cs:{cs_type}", NS):
                record = {}
                record["CS_Category"] = cs_type

                # Human_Readable_Name
                name_elem = cs_elem.find("cs:Name", NS)
                record["Human_Readable_Name"] = name_elem.text.strip() if name_elem is not None and name_elem.text else ""

                # Description
                desc_elem = cs_elem.find("cs:Description", NS)
                record["Description"] = desc_elem.text.strip() if desc_elem is not None and desc_elem.text else ""

                # Authority (using the <Authority> element)
                auth_elem = cs_elem.find("cs:Authority", NS)
                record["Authority"] = auth_elem.text.strip() if auth_elem is not None and auth_elem.text else ""

                # EPSG_Code: try to extract a code starting with "EPSG:" from the Authority text.
                epsg = ""
                if record["Authority"]:
                    parts = record["Authority"].split()
                    for part in parts:
                        if part.startswith("EPSG:"):
                            epsg = part
                            break
                record["EPSG_Code"] = epsg

                # Units: taken from the uom attribute of the <Axis> element.
                axis_elem = cs_elem.find("cs:Axis", NS)
                record["Units"] = axis_elem.attrib.get("uom", "") if axis_elem is not None else ""

                # DatumId: from the <DatumId> element.
                datum_elem = cs_elem.find("cs:DatumId", NS)
                record["DatumId"] = datum_elem.text.strip() if datum_elem is not None and datum_elem.text else ""

                # AdditionalInformation: extract all ParameterItem key-value pairs.
                add_info = {}
                for param in cs_elem.findall("cs:AdditionalInformation/cs:ParameterItem", NS):
                    key_elem = param.find("cs:Key", NS)
                    # Sometimes the value might be in <IntegerValue> or <Value>
                    val_elem = param.find("cs:IntegerValue", NS)
                    if val_elem is None:
                        val_elem = param.find("cs:Value", NS)
                    if key_elem is not None and key_elem.text:
                        add_info[key_elem.text.strip()] = val_elem.text.strip() if (val_elem is not None and val_elem.text) else ""
                record["AdditionalInformation"] = add_info

                records.append(record)
    except Exception as e:
        print(f"Error processing {xml_path}: {e}")

    return records

def main():
    parser = argparse.ArgumentParser(
        description='Extract coordinate system metadata from CSLibrary XML files into CSV and JSON.',
        epilog='Example: python cs_extract.py /path/to/CSLibrary --csv --json'
    )
    parser.add_argument('folder', help='Path to folder containing CSLibrary XML files')
    parser.add_argument('--output-dir', default=None,
                        help='Directory for output files (default: same as input folder)')
    parser.add_argument('--csv', action='store_true', dest='write_csv',
                        help='Write CSV output')
    parser.add_argument('--json', action='store_true', dest='write_json',
                        help='Write JSON output')
    args = parser.parse_args()

    # Default to both formats if neither flag is specified
    if not args.write_csv and not args.write_json:
        args.write_csv = True
        args.write_json = True

    # Validate input folder
    if not os.path.isdir(args.folder):
        print(f"Error: folder not found: {args.folder}")
        sys.exit(1)

    output_dir = args.output_dir if args.output_dir else args.folder
    if not os.path.isdir(output_dir):
        print(f"Error: output directory not found: {output_dir}")
        sys.exit(1)

    # Collect records from all XML files
    all_records = []
    for filename in os.listdir(args.folder):
        if filename.lower().endswith(".xml"):
            xml_path = os.path.join(args.folder, filename)
            print(f"Processing {xml_path}...")
            all_records.extend(process_xml_file(xml_path))

    # Write CSV
    if args.write_csv:
        csv_path = os.path.join(output_dir, "coordinate_systems_extended.csv")
        with open(csv_path, mode="w", newline="", encoding="utf-8") as csvfile:
            fieldnames = ["CS_Category", "Human_Readable_Name", "Description", "Authority", "EPSG_Code", "Units", "DatumId", "AdditionalInformation"]
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            for record in all_records:
                row = dict(record)
                row["AdditionalInformation"] = json.dumps(row["AdditionalInformation"])
                writer.writerow(row)
        print(f"CSV output saved to: {csv_path}")

    # Write JSON
    if args.write_json:
        json_path = os.path.join(output_dir, "coordinate_systems_extended.json")
        with open(json_path, mode="w", encoding="utf-8") as jsonfile:
            json.dump(all_records, jsonfile, indent=2)
        print(f"JSON output saved to: {json_path}")

    print("Extraction complete.")

if __name__ == "__main__":
    main()