diff options
Diffstat (limited to 'geodata-import')
| -rwxr-xr-x | geodata-import | 794 |
1 files changed, 794 insertions, 0 deletions
diff --git a/geodata-import b/geodata-import new file mode 100755 index 0000000..2f0f5b4 --- /dev/null +++ b/geodata-import @@ -0,0 +1,794 @@ +#!/usr/bin/python3 + +#---------------------------------------------------------------------- +# Backend utilities for the Klimatanalys Norr project (extract/import layers) +# Copyright © 2024-2025 Guilhem Moulin <info@guilhem.se> +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <https://www.gnu.org/licenses/>. +#---------------------------------------------------------------------- + +# pylint: disable=invalid-name, missing-module-docstring, fixme + +from os import O_RDONLY, O_WRONLY, O_CREAT, O_TRUNC, O_CLOEXEC, O_PATH, O_DIRECTORY +import os +from stat import S_ISREG +import sys +from fcntl import flock, LOCK_EX, LOCK_SH +import logging +import argparse +import re +from datetime import datetime, timedelta, timezone, UTC +from math import modf +from pathlib import Path +from time import monotonic as time_monotonic +from typing import Any, Optional, NoReturn +import traceback + +from osgeo import gdal, ogr, osr +from osgeo.gdalconst import ( + CE_None as GDAL_CE_None, + DCAP_DEFAULT_FIELDS as GDAL_DCAP_DEFAULT_FIELDS, + DCAP_NOTNULL_FIELDS as GDAL_DCAP_NOTNULL_FIELDS, + DCAP_UNIQUE_FIELDS as GDAL_DCAP_UNIQUE_FIELDS, +) +from osgeo import gdalconst + +import common +from common import ( + BadConfiguration, + parse_config_dl, + escape_identifier, + escape_literal_str, + getSourcePathLockFileName +) +from common_gdal import ( + gdalGetMetadataItem, + getSRS, + getExtent, + parseGeomType, + parseFieldType, + parseSubFieldType, + parseTimeZone +) +from import_source import ( + openOutputDS, + createOutputLayer, + validateOutputLayer, + importSources, + ImportStatus +) +from export_mvt import exportMVT +from export_raster import processRaster + +def setFieldIf(cond : bool, + attrName : str, + val : Any, + data : dict[str, Any], + fldName : str, + drvName : str, + log = logging.warning) -> None: + """Conditionally set a field""" + if cond: + data[attrName] = val + else: + if isinstance(val, str): + val2 = '"' + val + '"' + else: + val2 = str(val) + log('Ignoring %s=%s on field "%s" (not supported by %s driver)', + attrName, val2, fldName, drvName) + +# pylint: disable-next=too-many-branches +def validate_schema(layers : dict[str, Any], + drvo : Optional[gdal.Driver] = None, + lco_defaults : Optional[dict[str, str]] = None) -> None: + """Validate layer creation options and schema. The schema is + modified in place with the parsed result. + (We need the driver of the output dataset to determine capability on + constraints.)""" + + # list of capability flags supported by the CreateField() API + drvoFieldDefnFlags = drvo.GetMetadataItem(gdalconst.DMD_CREATION_FIELD_DEFN_FLAGS) + drvoFieldDefnFlags = drvoFieldDefnFlags.split(' ') if drvoFieldDefnFlags is not None else [] + drvoSupportsFieldComment = 'Comment' in drvoFieldDefnFlags + + # cache driver capabilities + drvoSupportsFieldWidthPrecision = 'WidthPrecision' in drvoFieldDefnFlags + drvoSupportsFieldNullable = ('Nullable' in drvoFieldDefnFlags and + gdalGetMetadataItem(drvo, GDAL_DCAP_NOTNULL_FIELDS)) + drvoSupportsFieldUnique = ('Unique' in drvoFieldDefnFlags and + gdalGetMetadataItem(drvo, GDAL_DCAP_UNIQUE_FIELDS)) + drvoSupportsFieldDefault = ('Default' in drvoFieldDefnFlags and + gdalGetMetadataItem(drvo, GDAL_DCAP_DEFAULT_FIELDS)) + drvoSupportsFieldAlternativeName = 'AlternativeName' in drvoFieldDefnFlags + + for layername, layerdef in layers.items(): + create = layerdef.get('create', None) + if create is None or len(create) < 1: + logging.warning('Layer "%s" has no creation schema', layername) + continue + + # prepend global layer creation options (dataset:create-layer-options) + # and build the option=value list + lco = create.get('options', None) + if lco_defaults is not None or lco is not None: + options = [] + if lco_defaults is not None: + options += [ k + '=' + str(v) for k, v in lco_defaults.items() ] + if lco is not None: + options += [ k + '=' + str(v) for k, v in lco.items() ] + create['options'] = options + + # parse geometry type + create['geometry-type'] = parseGeomType(create.get('geometry-type', None)) + + fields = create.get('fields', None) + if fields is None: + create['fields'] = [] + else: + fields_set = set() + for idx, fld_def in enumerate(fields): + fld_name = fld_def.get('name', None) + if fld_name is None or fld_name == '': + raise BadConfiguration(f'Field #{idx} has no name') + if fld_name in fields_set: + raise BadConfiguration(f'Duplicate field "{fld_name}"') + fields_set.add(fld_name) + + fld_def2 = { 'Name': fld_name } + for k, v in fld_def.items(): + k2 = k.lower() + if k2 == 'name': + pass + elif k2 in ('alternativename', 'alias'): + setFieldIf(drvoSupportsFieldAlternativeName, + 'AlternativeName', v, fld_def2, fld_name, drvo.ShortName, + log=logging.debug) + elif k2 == 'comment': + setFieldIf(drvoSupportsFieldComment, + 'Comment', v, fld_def2, fld_name, drvo.ShortName, + log=logging.debug) + + elif k2 == 'type': + fld_def2['Type'] = parseFieldType(v) + elif k2 == 'subtype': + fld_def2['SubType'] = parseSubFieldType(v) + elif k2 == 'tz': + fld_def2['TZFlag'] = parseTimeZone(v) + elif k2 == 'width' and v is not None and isinstance(v, int): + setFieldIf(drvoSupportsFieldWidthPrecision, + 'Width', v, fld_def2, fld_name, drvo.ShortName) + elif k2 == 'precision' and v is not None and isinstance(v, int): + setFieldIf(drvoSupportsFieldWidthPrecision, + 'Precision', v, fld_def2, fld_name, drvo.ShortName) + + # constraints + elif k2 == 'default': + setFieldIf(drvoSupportsFieldDefault, + 'Default', v, fld_def2, fld_name, drvo.ShortName) + elif k2 == 'nullable' and v is not None and isinstance(v, bool): + setFieldIf(drvoSupportsFieldNullable, + 'Nullable', v, fld_def2, fld_name, drvo.ShortName) + elif k2 == 'unique' and v is not None and isinstance(v, bool): + setFieldIf(drvoSupportsFieldUnique, + 'Unique', v, fld_def2, fld_name, drvo.ShortName) + else: + raise BadConfiguration(f'Field "{fld_name}" has unknown key "{k}"') + + fields[idx] = fld_def2 + +def setOutputFieldMap(defn : ogr.FeatureDefn, sources : dict[str, Any]): + """Setup output field mapping, modifying the sources dictionary in place.""" + fieldMap = {} + n = defn.GetFieldCount() + for i in range(n): + fld = defn.GetFieldDefn(i) + fldName = fld.GetName() + fieldMap[fldName] = i + + for source in sources: + source_import = source['import'] + + fieldMap2 = source_import.get('field-map', None) + if fieldMap2 is None: + fieldMap2 = fieldMap + else: + if isinstance(fieldMap2, list): + # convert list to identity dictionary + fieldMap2 = { fld: fld for fld in fieldMap2 } + + for ifld, ofld in fieldMap2.items(): + i = fieldMap.get(ofld, None) + if i is None: + raise RuntimeError(f'Ouput layer has no field named "{ofld}"') + fieldMap2[ifld] = i + source_import['field-map'] = fieldMap2 + + # validate field value mapping + valueMap = source_import.get('value-map', None) + if valueMap is not None: + for fldName, rules in valueMap.items(): + if rules is None: + continue + if not isinstance(rules, list): + rules = [rules] + for idx, rule in enumerate(rules): + if rule is None or not isinstance(rule, dict): + raise RuntimeError(f'Field "{fldName}" has invalid rule #{idx}: {rule}') + if 'type' not in rule: + ruleType = rule['type'] = 'literal' + else: + ruleType = rule['type'] + if ('replace' not in rule or 'with' not in rule or len(rule) != 3 or + ruleType is None or ruleType not in ('literal', 'regex')): + raise RuntimeError(f'Field "{fldName}" has invalid rule #{idx}: {rule}') + if ruleType == 'regex': + rule['replace'] = re.compile(rule['replace']) + rules[idx] = ( rule['replace'], rule['with'] ) + +def processOutputLayer(ds : gdal.Dataset, + layername : str, + layerdef : dict[str,Any], + srs : Optional[osr.SpatialReference] = None, + cachedir : Path|None = None, + extent : ogr.Geometry|None = None, + dsTransaction : bool = False, + lyrcache : ogr.Layer|None = None, + force : bool = False) -> ImportStatus: + """Process an output layer.""" + + logging.info('Processing output layer "%s"', layername) + lyr = ds.GetLayerByName(layername) + if lyr is None: + raise RuntimeError(f'Failed to create output layer "{layername}"??') + if not lyr.TestCapability(ogr.OLCSequentialWrite): + raise RuntimeError(f'Output layer "{layername}" has no working ' + 'CreateFeature() method') + + sources = layerdef['sources'] + if not (lyrcache is None or force or + areSourceFilesNewer(layername, sources=sources, + lyrcache=lyrcache, + cachedir=cachedir)): + logging.info('Output layer "%s" is up to date, skipping', layername) + return ImportStatus.IMPORT_NOCHANGE + + validateOutputLayer(lyr, srs=srs, options=layerdef['create']) + + description = layerdef.get('description', None) + if (description is not None and + lyr.SetMetadataItem('DESCRIPTION', description) != GDAL_CE_None): + logging.warning('Could not set description metadata') + + # setup output field mapping in the sources dictionary + setOutputFieldMap(lyr.GetLayerDefn(), sources) + + return importSources(lyr=lyr, sources=sources, + cachedir=cachedir, extent=extent, + dsoTransaction=dsTransaction, + lyrcache=lyrcache, + force=force, + cluster_geometry=layerdef.get('cluster-geometry', False)) + +def validate_sources(layers : dict[str, Any]) -> None: + """Mangle and validate layer sources and import definitions""" + toremove = set() + for layername, layerdefs in layers.items(): + sources = layerdefs.get('sources', None) + if sources is None or len(sources) < 1: + logging.warning('Output layer "%s" has no definition, skipping', layername) + toremove.add(layername) + continue + + for idx, layerdef in enumerate(sources): + importdef = layerdef.get('import', None) + if importdef is None: + raise BadConfiguration(f'Source #{idx} of output layer "{layername}" ' + 'has no import definition') + + sourcedef = layerdef.get('source', None) + unar = None if sourcedef is None else sourcedef.get('unar', None) + src = None if sourcedef is None else sourcedef.get('path', None) + + ds_srcpath = importdef.get('path', None) + if src is None and unar is None and ds_srcpath is not None: + # fallback to importe:path if there is no unarchiving recipe + src = ds_srcpath + if unar is not None and ds_srcpath is None: + raise BadConfiguration(f'Source #{idx} of output layer "{layername}" ' + 'has no import source path') + if src is None: + raise BadConfiguration(f'Source #{idx} of output layer "{layername}" ' + 'has no source path') + layerdef['source'] = { 'path': src, 'unar': unar } + + for layername in toremove: + layers.pop(layername) + +def validateLayerCacheField(defn : ogr.FeatureDefn, idx : int, + name : str, + typ : int, + subtyp : int = ogr.OFSTNone, + width : int = 0, + unique : Optional[bool] = None, + nullable : Optional[bool] = None) -> bool: + """Validate field #idx from the layer cache table.""" + n = defn.GetFieldCount() + if idx >= n: + return False + defn = defn.GetFieldDefn(idx) + + b = True + name2 = defn.GetName() + if name2 != name: + logging.warning('Layer cache\'s field #%d has name "%s" != "%s"', idx, name2, name) + b = False + + if nullable is not None and defn.IsNullable() != nullable: + # non-fatal + logging.warning('Layer cache\'s field #%d ("%s") %s nullable', + idx, name2, 'is' if defn.IsNullable() else 'isn\'t') + + if unique is not None and defn.IsUnique() != unique: + # non-fatal + logging.warning('Layer cache\'s field #%d ("%s") %s unique', + idx, name2, 'is' if defn.IsUnique() else 'isn\'t') + + typ2 = defn.GetType() + if typ2 != typ: + logging.warning('Layer cache\'s field #%d ("%s") has type %s != %s', idx, name2, + ogr.GetFieldTypeName(typ2), ogr.GetFieldTypeName(typ)) + b = False + + subtyp2 = defn.GetSubType() + if subtyp2 != subtyp: + logging.warning('Layer cache\'s field #%d ("%s") has subtype %s != %s', idx, name2, + ogr.GetFieldSubTypeName(subtyp2), ogr.GetFieldSubTypeName(subtyp)) + b = False + + width2 = defn.GetWidth() + if width2 != 0 and (width == 0 or width2 < width): + # non-fatal + logging.warning('Layer cache\'s field #%d ("%s") is too small (width %d < %d)', + idx, name2, width2, width) + return b + +def validateCacheLayer(ds : gdal.Dataset, name : str) -> bool: + """Validate layer cache table.""" + drvName = ds.GetDriver().ShortName + if drvName != 'PostgreSQL': # we need hash_record_extended(), sha256() and ST_AsEWKB() + logging.warning('Unsupported cache layer for output driver %s', drvName) + return False + lyr = ds.GetLayerByName(name) + if lyr is None: + logging.warning('Table "%s" does not exist', name) + return False + + if not (lyr.TestCapability(ogr.OLCRandomWrite) and lyr.TestCapability(ogr.OLCUpdateFeature)): + logging.warning('Layer "%s" does not support OLCUpdateFeature capability, ' + 'ignoring cache', name) + return False + + defn = lyr.GetLayerDefn() + fields = [ + { 'name': 'layername', 'typ': ogr.OFTString, + 'nullable': False, 'unique': True, 'width': 255 }, + { 'name': 'last_updated', 'typ': ogr.OFTDateTime, + 'nullable': False }, + { 'name': 'fingerprint', 'typ': ogr.OFTBinary, + 'nullable': False, 'width': 32 }, + ] + m = len(fields) + n = defn.GetFieldCount() + if n < m: + # this is fatal, and `all(bs)` is False so we return False below + logging.warning('Layer cache "%s" has %d < %d fields', name, n, m) + elif n != m: + logging.warning('Layer cache "%s" has %d != %d fields', name, n, m) + bs = [ validateLayerCacheField(defn, i, **fld) for i,fld in enumerate(fields) ] + if not all(bs): + return False + + n = defn.GetGeomFieldCount() + if n > 0: + geomFieldNames = [ escape_identifier(defn.GetGeomFieldDefn(i).GetName()) + for i in range(n) ] + logging.warning('Layer cache "%s" has %d > 0 geometry field(s): %s', + name, n, ', '.join(geomFieldNames)) + + style = lyr.GetStyleTable() + if style is not None: + logging.warning('Layer cache "%s" has a style table "%s"', + name, style.GetLastStyleName()) + return True + +def areSourceFilesNewer(layername : str, + sources : dict[str,Any], + lyrcache : ogr.Layer, + cachedir : Optional[Path] = None) -> bool: + """Return a boolean indicating whether the layer cache is up to date with + respect to the source files found on disk. That is, the last modification + and last changed time of each source file needs to be equal or lower than + the `last_updated` value found in the layer cache.""" + + source_paths = set() + for source in sources: + # the same source_path can be used multiple times, stat(2) only once + source_path = source['source']['path'] + source_paths.add(source_path) + if len(source_paths) == 0: + return False + + t = None + mtimes_ns = {} + for source_path in source_paths: + path = source_path if cachedir is None else str(cachedir.joinpath(source_path)) + try: + st = os.stat(path) + if not S_ISREG(st.st_mode): + raise FileNotFoundError + mtimes_ns[source_path] = st.st_mtime_ns + # take max(mtime, ctime): if we lock source paths any update after + # aquiring the lock will yield a value larger than time.time_ns() + t2 = max(st.st_mtime_ns, st.st_ctime_ns) + if t is None or t < t2: + t = t2 + except (OSError, ValueError): + #logging.warning('Could not stat(%s)', path) + return True + assert t is not None + + attributeFilter = 'layername = ' + escape_literal_str(layername) + logging.debug('SetAttributeFilter("%s", "%s")', lyrcache.GetName(), attributeFilter) + lyrcache.SetAttributeFilter(attributeFilter) + + feature = lyrcache.GetNextFeature() + if feature is None: + # not in cache + return True + + if not feature.IsFieldSetAndNotNull(1): + ret = True + else: + # https://gdal.org/en/stable/api/python/vector_api.html#osgeo.ogr.Feature.GetFieldAsDateTime + # [ year, month, day, hour, minute, second, timezone flag ] + dt = feature.GetFieldAsDateTime(1) + if dt[6] == ogr.TZFLAG_UNKNOWN: + logging.warning('Datetime specified with unknown timezone in layer cache\'s ' + 'field #%d "%s", assuming local time', 1, + feature.GetDefnRef().GetFieldDefn(1).GetName()) + tz = None + elif dt[6] == ogr.TZFLAG_LOCALTIME: + tz = None + elif dt[6] == ogr.TZFLAG_UTC: + tz = UTC + else: + tz = timezone(offset=timedelta(seconds=(dt[6] - ogr.TZFLAG_UTC) * 900)) + ms, s = modf(dt[5]) + dt = datetime( + year=dt[0], # including century + month=dt[1], # 01 ≤ year ≤ 12 + day=dt[2], # 01 ≤ day ≤ 31 + hour=dt[3], # 00 ≤ hour ≤ 23 + minute=dt[4], # 00 ≤ minute ≤ 59 + second=int(s), # 00 ≤ second ≤ 59 + microsecond=round(ms*1000000), + tzinfo=tz + ) + fpr = feature.GetFieldAsBinary(2) if feature.IsFieldSetAndNotNull(2) else None + logging.debug('Found entry in layer cache for "%s", last_updated=%s, fingerprint=%s', + layername, + dt.isoformat(timespec='microseconds'), + fpr.hex() if fpr is not None else 'NULL') + ret = int(dt.timestamp() * 1000000.) * 1000 < t + + if lyrcache.GetNextFeature() is not None: + raise RuntimeError(f'Duplicate key {layername}') + + if not ret: + for source_path, mtime_ns in sorted(mtimes_ns.items()): + # XXX datetime.fromtimestamp() doesn't support nanosecond input + # https://github.com/python/cpython/issues/59648 + mtime = (mtime_ns // 1000) / 1000000. + dt = datetime.fromtimestamp(mtime) + logging.info('Source file %s is unchanged (last modification time %s)', + source_path, dt.astimezone().isoformat(timespec='seconds')) + return ret + +def getLastMTimes(layerdefs : dict[str,Any], basedir : Optional[Path] = None) -> dict[str,int]: + """Return a directing mapping source paths to their last modification time + (as a timestamp in milliseconds).""" + ret = {} + for layerdef in layerdefs: + for source in layerdef['sources']: + source_path = source['source']['path'] + if source_path in ret: + continue + path = source_path if basedir is None else str(basedir.joinpath(source_path)) + try: + st = os.stat(path) + if not S_ISREG(st.st_mode): + raise FileNotFoundError + ret[source_path] = st.st_mtime_ns // 1000000 + except (OSError, ValueError): + #logging.warning('Could not stat(%s)', path) + pass + return ret + +def lockSourcePaths(layerdefs : dict[str,Any], lockdir: str) -> dict[str,int]: + """Place shared locks on each source path and return their respective file + descriptors. We could do that one layerdef at a time (one output layer at a + time) to reduce the time during which the sources prevented from being + updated/downloaded, but their is some value in having consistency across the + whole import process.""" + umask = os.umask(0o002) + lockdir_fd = os.open(lockdir, O_RDONLY|O_CLOEXEC|O_PATH|O_DIRECTORY) + try: + ret = {} + for layerdef in layerdefs: + for source in layerdef['sources']: + source_path = source['source']['path'] + if source_path in ret: + continue + lockfile = getSourcePathLockFileName(source_path) + lock_fd = os.open(lockfile, O_WRONLY|O_CREAT|O_CLOEXEC, mode=0o664, + dir_fd=lockdir_fd) + logging.debug('Acquiring shared lock for %s: flock("%s", LOCK_SH)', + source_path, lockfile) + flock(lock_fd, LOCK_SH) + ret[source_path] = lock_fd + return ret + finally: + try: + os.close(lockdir_fd) + except (OSError, ValueError): + logging.exception('Could not close lockdir') + os.umask(umask) + +def releaseSourcePathLocks(lock_fds : dict[str,int]) -> None: + """Release shared locks on the source paths. Closed FDs are removed from + the dictionary in place.""" + toremove = set() + for path, lock_fd in lock_fds.items(): + try: + os.close(lock_fd) + except (OSError, ValueError): + logging.exception('Could not release lock for %s', path) + else: + logging.debug('Released lock for %s', path) + toremove.add(path) + for path in toremove: + lock_fds.pop(path) + +# pylint: disable-next=missing-function-docstring, too-many-branches, too-many-statements +def main() -> NoReturn: + common.init_logger(app=os.path.basename(__file__), level=logging.INFO) + + parser = argparse.ArgumentParser(description='Extract and import GIS layers.') + parser.add_argument('--cachedir', default=None, + help=f'cache directory (default: {os.curdir})') + parser.add_argument('--debug', action='count', default=0, + help=argparse.SUPPRESS) + parser.add_argument('--lockfile', default=None, + help='obtain an exclusive lock before processing') + parser.add_argument('--lockdir-sources', default=None, + help='optional directory for lock files to source paths') + parser.add_argument('--mvtdir', default=None, + help='optional directory for Mapbox Vector Tiles (MVT)') + parser.add_argument('--mvt-compress', default=False, action='store_true', + help='whether to compress Mapbox Vector Tiles (MVT) files') + parser.add_argument('--rasterdir', default=None, + help='optional directory for raster files') + parser.add_argument('--metadata-compress', default=False, action='store_true', + help='whether to compress metadata.json files') + parser.add_argument('--force', default=False, action='store_true', + help='import even if no new changes were detected') + parser.add_argument('groupname', nargs='*', help='group layer name(s) to process') + args = parser.parse_args() + + if args.debug > 0: # pylint: disable=duplicate-code + logging.getLogger().setLevel(logging.DEBUG) + if args.debug > 1: + gdal.ConfigurePythonLogging(enable_debug=True) + + config = common.parse_config(groupnames=None if args.groupname == [] else args.groupname) + + # validate configuration + if 'dataset' not in config: + raise BadConfiguration('Configuration does not specify output dataset') + + layers = config.get('layers', {}) + validate_sources(layers) + + # set global GDAL/OGR configuration options + for pszKey, pszValue in config.get('GDALconfig', {}).items(): + logging.debug('gdal.SetConfigOption(%s, %s)', pszKey, pszValue) + gdal.SetConfigOption(pszKey, pszValue) + + # get configured Spatial Reference System and extent + srs = getSRS(config.get('SRS', None)) + extent = getExtent(config.get('extent', None), srs=srs) + + if args.lockfile is not None: + # obtain an exclusive lock and don't release it until exiting the program + lock_fd = os.open(args.lockfile, O_WRONLY|O_CREAT|O_TRUNC|O_CLOEXEC, mode=0o644) + logging.debug('flock("%s", LOCK_EX)', args.lockfile) + flock(lock_fd, LOCK_EX) + + if args.mvtdir is not None: + args.mvtdir = Path(args.mvtdir) + if args.mvtdir == Path(): # make sure it's not curdir as we don't want to exchange it + raise RuntimeError('Invalid value for --mvtdir') + args.mvtdir.parent.mkdir(parents=True, exist_ok=True) + + if args.cachedir is not None: + args.cachedir = Path(args.cachedir) + if args.lockdir_sources is None: + sourcePathLocks = None + else: + sourcePathLocks = lockSourcePaths(layerdefs=layers.values(), + lockdir=args.lockdir_sources) + + # special handling for raster layers + if any(l.get('type') == 'raster' for l in layers.values()): + if not all(l.get('type') == 'raster' for l in layers.values()): + raise NotImplementedError('Mix of raster and vector layers is not supported') + if args.rasterdir is None: + raise RuntimeError('Missing required value for --rasterdir') + if len(layers) != 1: + raise RuntimeError('Raster layers need to be processed one at a time') + args.rasterdir = Path(args.rasterdir) + if args.rasterdir == Path(): # make sure it's not curdir as we don't want to exchange it + raise RuntimeError('Invalid value for --rasterdir') + args.rasterdir.parent.mkdir(parents=True, exist_ok=True) + last_modified = getLastMTimes(layerdefs=layers.values(), basedir=args.cachedir) + rv = 0 + for layername, layerdef in layers.items(): + try: + processRaster(layername, layerdef, + sources=parse_config_dl(config.get('downloads', [])), + license_info=config.get('license-info', {}), + last_modified=last_modified, + dst=args.rasterdir, + cachedir=args.cachedir, + extent=extent, + compress_metadata=args.metadata_compress) + except Exception: # pylint: disable=broad-exception-caught + rv = 1 + traceback.print_exc() + sys.exit(rv) + + # open output dataset (possibly create it first) + dso = openOutputDS(config['dataset']) + + validate_schema(layers, + drvo=dso.GetDriver(), + lco_defaults=config['dataset'].get('create-layer-options', None)) + + # create all output layers before starting the transaction + for layername, layerdef in layers.items(): + lyr = dso.GetLayerByName(layername) + if lyr is not None: + # TODO dso.DeleteLayer(layername) if --overwrite and + # dso.TestCapability(ogr.ODsCDeleteLayer) + # (Sets OVERWRITE=YES for PostgreSQL and GPKG.) + continue + if not dso.TestCapability(ogr.ODsCCreateLayer): + raise RuntimeError(f'Output driver {dso.GetDriver().ShortName} does not ' + 'support layer creation') + createOutputLayer(dso, layername, srs=srs, options=layerdef.get('create', None)) + + if (dso.TestCapability(ogr.ODsCTransactions) and + # we need SAVEPOINT support + dso.GetDriver().ShortName in ('PostgreSQL', 'SQLite', 'GPKG')): + logging.debug('Starting transaction') + dsoTransaction = dso.StartTransaction() == ogr.OGRERR_NONE + else: + logging.warning('Output driver %s does not support dataset transactions or SQL SAVEPOINTs', + dso.GetDriver().ShortName) + dsoTransaction = False + + # validate layer cache + lyr_cache = config['dataset'].get('layercache', None) + if lyr_cache is None: + pass + elif validateCacheLayer(dso, lyr_cache): + lyr_cache = dso.GetLayerByName(lyr_cache) + else: + if not args.force: + logging.warning('Ignoring invalid layer cache "%s" (implying --force)', lyr_cache) + args.force = True + lyr_cache = None + + rv = 0 + try: + r = {} + n = 0 + start = time_monotonic() + for layername, layerdef in layers.items(): + r[layername] = r0 = processOutputLayer(dso, layername, layerdef, + srs=srs, + cachedir=args.cachedir, + extent=extent, + dsTransaction=dsoTransaction, + lyrcache=lyr_cache, + force=args.force) + n += 1 + logging.info('Import result status for layer "%s": %s', layername, str(r0)) + if r0 == ImportStatus.IMPORT_ERROR: + rv = 1 + if dsoTransaction: + dsoTransaction = False + logging.debug('Rolling back transaction') + # no need to catch the exception here + if dso.CommitTransaction() != ogr.OGRERR_NONE: + logging.error('Could not rollback transaction') + break + elapsed = time_monotonic() - start + logging.info('Processed %d destination layers in %s', n, common.format_time(elapsed)) + + # get mtimes before releasing the source locks + last_modified = getLastMTimes(layerdefs=layers.values(), basedir=args.cachedir) + + if sourcePathLocks is not None: + releaseSourcePathLocks(sourcePathLocks) + + export_layers = { l:d for l,d in layers.items() if d.get('publish', None) is not None } + if args.mvtdir is None or any(r0 == ImportStatus.IMPORT_ERROR for r0 in r.values()): + pass + elif len(export_layers) == 0: + logging.warning('--mvtdir option used but no layer has a publication definition') + elif (all(r0 == ImportStatus.IMPORT_NOCHANGE for l,r0 in r.items() if l in export_layers) + and args.mvtdir.is_dir()): + logging.info('Skipping MVT export for group %s (no changes)', + ', '.join(args.groupname) if args.groupname is not None else '*') + else: + exportMVT(dso, + layers=export_layers, + sources=parse_config_dl(config.get('downloads', [])), + license_info=config.get('license-info', {}), + last_modified=last_modified, + dst=args.mvtdir, + default_options=config.get('vector-tiles', None), + compress=args.mvt_compress, + compress_metadata=args.metadata_compress) + + if dsoTransaction: + dsoTransaction = False + logging.debug('Committing transaction') + if dso.CommitTransaction() != ogr.OGRERR_NONE: + logging.error('Could not commit transaction') + rv = 1 + + except Exception: # pylint: disable=broad-exception-caught + if dsoTransaction: + logging.exception('Exception occured within transaction, rolling back') + try: + if dso.RollbackTransaction() != ogr.OGRERR_NONE: + logging.error('Could not rollback transaction') + except Exception: # pylint: disable=broad-exception-caught + logging.exception('Could not rollback transaction') + else: + traceback.print_exc() + sys.exit(1) + + finally: + lyr_cache = None + dso = None + extent = None + srs = None + sys.exit(rv) + +gdal.UseExceptions() +main() |