#!/usr/bin/env python # -*- coding: utf-8 -*- import argparse, os, re, shutil, subprocess, sys def run(cmd: str): p = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True) return p.returncode, p.stdout def die(msg, code=1): print(msg); sys.exit(code) def argp(): ap = argparse.ArgumentParser(description="GDB→GDB with NTv2, preserving Feature Datasets. Default output driver: FileGDB.") ap.add_argument("SRC_GDB", help="Source .gdb (Drukref03)") ap.add_argument("DST_GDB", help="Destination .gdb (will be created)") ap.add_argument("GRID_GSB", help="NTv2 grid .gsb") ap.add_argument("DST_PRJ", help="Target CRS: EPSG code (e.g. EPSG:11341) or path to .prj file") ap.add_argument("--src-epsg", default="EPSG:5266", help="Source CRS (default EPSG:5266)") ap.add_argument("--strict", action="store_true", help="No ',null' fallback in +grids (fail outside grid)") ap.add_argument("--gt", default="65536", help="Transaction size for ogr2ogr (-gt). Default 65536") ap.add_argument("--no-dim2", action="store_true", help="Do not force -dim 2 (keep Z/M if present)") ap.add_argument("--driver", choices=["FileGDB","OpenFileGDB"], default="FileGDB", help="Output driver. Default FileGDB (requires Esri FileGDB API).") return ap.parse_args() def build_ct(grid_path: str, strict: bool) -> str: # Normalize path separators for PROJ compatibility grid_norm = grid_path.replace("\\", "/") # Use strict grid only or allow fallback (grid,null) grids = grid_norm if strict else f"{grid_norm},null" return ( "+proj=pipeline " # 1) TM DrukRef03 → Geographic DrukRef03 "+step +inv +proj=tmerc " "+lat_0=0 +lon_0=90 +k=1 " "+x_0=250000 +y_0=0 +ellps=GRS80 " # 2) Geographic DrukRef03 → Geographic DrukRef23 (NTv2) f"+step +proj=hgridshift +grids={grids} " # 3) Geographic DrukRef23 → LCC BNG2023 (final projection) "+step +proj=lcc " "+lat_1=27.65 +lat_2=26.90 +lat_0=27.50 " "+lon_0=90 +x_0=500000 +y_0=200000 " "+ellps=GRS80 +units=m +no_defs" ) def list_groups_layers(src_gdb:str): rc, out = run(f'ogrinfo -ro -q "{src_gdb}"') if rc != 0: die(out, 6) groups = {}; cur = "" rx = re.compile(r'^ Layer:\s*(.+?)\s*\((.+)\)\s*$') for ln in out.splitlines(): if ln.startswith("Group "): g = ln[6:].strip() if g.endswith(":"): g = g[:-1] cur = g; groups.setdefault(cur, []) elif ln.strip().startswith("Layer:"): m = rx.match(ln) if m: groups.setdefault(cur, []).append((m.group(1).strip(), m.group(2).strip())) if not groups: # fallback sin grupos names=[] for ln in out.splitlines(): s=ln.strip() if s.lower().startswith("layer:"): nm=s.split("Layer:",1)[1].strip() if " (" in nm: nm=nm.split(" (",1)[0].strip() names.append((nm,"unknown")) groups[""]=names return groups def nlt_of(gstr:str)->str: s=(gstr or "").lower() if "polygon" in s: return "MULTIPOLYGON" if "line" in s: return "MULTILINESTRING" if "multipoint" in s: return "MULTIPOINT" if "point" in s: return "POINT" return "" # fallback a PROMOTE_TO_MULTI def main(): a = argp() # Validate SRC_GDB and GRID_GSB (must be existing paths) for pth, msg, code in [ (a.SRC_GDB, "*** ERROR: Source not found: ", 2), (a.GRID_GSB, "*** ERROR: Grid not found: ", 3), ]: if not os.path.exists(pth): die(f'{msg}"{pth}"', code) # Validate DST_PRJ: accept EPSG code (e.g. EPSG:11341) OR a .prj file path if a.DST_PRJ.upper().startswith("EPSG:"): # It's an EPSG code — pass directly to ogr2ogr as -t_srs dst_crs = a.DST_PRJ print(f'[INFO] Target CRS (EPSG code): {dst_crs}') else: # It's a file path — must exist if not os.path.exists(a.DST_PRJ): die(f'*** ERROR: PRJ not found: "{a.DST_PRJ}"', 4) dst_crs = a.DST_PRJ print(f'[INFO] Target CRS (.prj file): {dst_crs}') # Drivers disponibles rc, fmts = run("ogrinfo --formats") if rc!=0: die(fmts, 5) if a.driver == "FileGDB": if "FileGDB" not in fmts: die("*** ERROR: 'FileGDB' driver not available. Install gdal-filegdb and Esri FileGDB API.", 5) else: if "OpenFileGDB" not in fmts: die("*** ERROR: 'OpenFileGDB' driver not available in this shell.", 5) print("=== 1) Inspect NTv2 grid ===") rc, gi = run(f'gdalinfo "{a.GRID_GSB}"') if rc != 0: die(gi, 5) print(gi.strip(), "\n") ct = build_ct(a.GRID_GSB, a.strict) print("Mode:", "STRICT (no fallback)" if a.strict else 'FALLBACK (",null" allowed)') print("CT :", ct, "\n") print("=== 2) Discover source datasets/layers ===") groups = list_groups_layers(a.SRC_GDB) print("\n=== 3) Prepare destination ===") if os.path.exists(a.DST_GDB): try: shutil.rmtree(a.DST_GDB) except Exception: die(f'*** ERROR: Destination locked (close apps using it): "{a.DST_GDB}"', 7) print(f'[INFO] Source : {a.SRC_GDB}') print(f'[INFO] Target : {a.DST_GDB}') print(f'[INFO] Driver : {a.driver}') print(f'[INFO] -s_srs : {a.src_epsg}') print(f'[INFO] -t_srs : {dst_crs}') print(f'[INFO] -ct : {ct}\n') # Crear destino vacío (driver elegido) run(f'ogr2ogr -f "{a.driver}" "{a.DST_GDB}" "{a.SRC_GDB}" -nln __init__ -sql "SELECT NULL as dummy"') run(f'ogrinfo -if {a.driver} "{a.DST_GDB}" -sql "DROP TABLE __init__"') dim_opt = "" if a.no_dim2 else "-dim 2" total=copiadas=fallos=0 first_ok=False for fds, items in groups.items(): fd_opt = f'-lco FEATURE_DATASET="{fds}"' if fds else "" for name, gtyp in items: total += 1 nlt = nlt_of(gtyp) nlt_opt = f"-nlt {nlt}" if nlt else "-nlt PROMOTE_TO_MULTI" sql = f'SELECT * FROM "{name}"' cmd = ( f'ogr2ogr -f "{a.driver}" ' f'{"-update " if first_ok else ""}"{a.DST_GDB}" "{a.SRC_GDB}" ' f'-nln "{name}" {fd_opt} -lco GEOMETRY_NAME=SHAPE ' f'-s_srs {a.src_epsg} -t_srs "{dst_crs}" -ct "{ct}" ' f'{nlt_opt} {dim_opt} -gt {a.gt} -sql "{sql}" -progress' ) print(f'[COPY] {(fds+"/") if fds else ""}{name} ({gtyp})') rc, out = run(cmd) if rc==0: copiadas += 1 first_ok = True else: print(" [ERROR]"); print(out.strip()); fallos += 1 print("\n=== 4) Destination summary ===") rc, summ = run(f'ogrinfo -if {a.driver} -ro -so "{a.DST_GDB}"') print(summ) print("== SUMMARY ==") print("Total layers :", total) print("Copied layers :", copiadas) print("Failed layers :", fallos) print("Output :", a.DST_GDB) sys.exit(0 if fallos==0 else 4) if __name__ == "__main__": main()