Add gpkg generator

geopackage_validator/cli.py

@@ -416,6 +416,77 @@ def geopackage_validator_command_generate_table_definitions(
         sys.exit(1)
 
 
+
+@cli.command(
+    name="generate-gpkg",
+    help=(
+        "Generate an empty geopackage based on a geopackage validator table definition. Provide the table definitions "
+        "with the --table-definitions-path parameter. The generated geopackage will be valid except for the fact that it"
+        "will be empty.\n\n"
+        "When the filepath is preceded with '/vsi' the gdal virtual file system method will be used to access the file on "
+        "S3 and will not be directly downloaded. See https://gdal.org/user/virtual_file_systems.html for further "
+        "explanation. For convenience the gdal vsi environment parameters and optional parameters are provided with "
+        "an S3_ instead of an AWS_ prefix. The AWS_ environment parameters will also work.\n\n"
+        "Examples:\n\n"
+        "viscurl:\n\n"
+        "geopackage-validator generate-definitions --gpkg-path /vsicurl/http://minio-url.nl/bucketname/key/to/public.gpkg\n\n"
+        "vsis3:\n\n"
+        "geopackage-validator generate-definitions --gpkg-path /vsis3/bucketname/key/to/public.gpkg --s3-signing-region eu-central-1 --s3-secret-key secret --s3-access-key acces-key --s3-secure=false --s3-virtual-hosting false --s3-endpoint-no-protocol minio-url.nl\n\n"
+        "S3_SECRET_KEY=secret S3_ACCESS_KEY=acces-key S3_SIGNING_REGION=eu-central-1 S3_SECURE=false S3_VIRTUAL_HOSTING=false S3_ENDPOINT_NO_PROTOCOL=minio-url.nl geopackage-validator generate-definitions --gpkg-path /vsis3/bucketname/key/to/public.gpkg\n\n"
+        "AWS_SECRET_ACCESS_KEY=secret AWS_ACCESS_KEY_ID=acces-key AWS_DEFAULT_REGION=eu-central-1 AWS_HTTPS=NO AWS_VIRTUAL_HOSTING=FALSE AWS_S3_ENDPOINT=minio-url.nl geopackage-validator generate-definitions --gpkg-path /vsis3/bucketname/key/to/public.gpkg"
+    ),
+)
+@click.option(
+    "--gpkg-path",
+    envvar="GPKG_PATH",
+    required=False,
+    default=None,
+    show_envvar=True,
+    help="Path pointing to the geopackage.gpkg file",
+    type=click.types.Path(
+        file_okay=False,
+        dir_okay=False,
+        readable=False,
+        writable=False,
+        resolve_path=False,
+        allow_dash=False,
+    ),
+)
+@click.option(
+    "--validations-path",
+    show_envvar=True,
+    required=False,
+    default=None,
+    envvar="VALIDATIONS_FILE",
+    help=(
+        "Path pointing to the set of validations to run. If validations-path and validations are not given, validate "
+        "runs all validations"
+    ),
+    type=click.types.Path(
+        exists=True,
+        file_okay=True,
+        dir_okay=False,
+        readable=True,
+        writable=False,
+        allow_dash=False,
+    ),
+)
+@click_log.simple_verbosity_option(logger)
+def geopackage_validator_command_generate_gpkg(
+    gpkg_path,
+    validations_path,
+):
+    gpkg_path_not_exists = gpkg_path is None
+    if gpkg_path_not_exists:
+        logger.error("Give a valid --gpkg-path or (/vsi)s3 location")
+        sys.exit(1)
+    try:
+        generate.generate_empty_geopackage(gpkg_path, validations_path)
+    except Exception:
+        logger.exception("Error while generating table definitions")
+        sys.exit(1)
+
+
 @cli.command(
     name="show-validations",
     help="Show all the possible validations that can be executed in the validate command.",

geopackage_validator/generate.py

@@ -14,6 +14,34 @@
 TableDefinition = Dict[str, Union[int, Dict[str, ColumnDefinition]]]
 
 
+OGR_GEOMETRY_TYPES = {
+    "POINT": ogr.wkbPoint,
+    "LINESTRING": ogr.wkbLineString,
+    "POLYGON": ogr.wkbPolygon,
+    "MULTIPOINT": ogr.wkbMultiPoint,
+    "MULTILINESTRING": ogr.wkbMultiLineString,
+    "MULTIPOLYGON": ogr.wkbMultiPolygon,
+}
+
+
+OGR_FIELD_TYPES = dict(**OGR_GEOMETRY_TYPES, **{
+    "DATE": ogr.OFTDate,
+    "DATETIME": ogr.OFTDateTime,
+    "TIME": ogr.OFTTime,
+    "INTEGER": ogr.OFTInteger,
+    "INTEGER64": ogr.OFTInteger64,
+    "REAL": ogr.OFTReal ,
+    "STRING": ogr.OFTString ,
+    "BINARY": ogr.OFTBinary,
+    "INTEGERLIST": ogr.OFTIntegerList,
+    "INTEGER64LIST": ogr.OFTInteger64List,
+    "REALLIST": ogr.OFTRealList,
+    "STRINGLIST": ogr.OFTStringList,
+    "WIDESTRING": ogr.OFTWideString,
+    "WIDESTRINGLIST": ogr.OFTWideStringList,
+})
+
+
 def columns_definition(table, geometry_column) -> ColumnDefinition:
     layer_definition = table.GetLayerDefn()
 
@@ -83,10 +111,52 @@ def generate_table_definitions(dataset: DataSource) -> TableDefinition:
     return result
 
 
+def generate_geopackage_from_table_definition(dataset: DataSource, table_definition: TableDefinition):
+    version = table_definition["geopackage_validator_version"]
+    projection = table_definition["projection"]
+    tables = table_definition["tables"]
+
+    srs = osr.SpatialReference()
+    srs.ImportFromEPSG(projection)
+
+    for table in tables:
+        columns = {c["name"]: c["type"] for c in table["columns"]}
+
+        try:
+            geometry_type = OGR_GEOMETRY_TYPES[columns[table["geometry_column"]]]
+        except KeyError:
+            raise ValueError(f"Unknown geometry type for table {table['name']}")
+
+        layer = dataset.CreateLayer(table["name"], srs=srs, geom_type=geometry_type)
+        try:
+            fields = [
+                ogr.FieldDefn(column["name"], OGR_FIELD_TYPES[column["type"]])
+                for column in table["columns"]
+                if column["name"] != table["geometry_column"]
+            ]
+        except KeyError:
+            raise ValueError(f"Unknown field type for table {table['name']}")
+
+        layer.CreateFields(fields)
+
+
 def generate_definitions_for_path(gpkg_path: str) -> TableDefinition:
     """Starts the geopackage validation."""
     utils.check_gdal_version()
 
     dataset = utils.open_dataset(gpkg_path)
 
     return generate_table_definitions(dataset)
+
+
+def generate_empty_geopackage(gpkg_path: str, table_definition_path: str):
+    utils.check_gdal_version()
+
+    dataset = utils.create_dataset(gpkg_path)
+    table_definition = load_table_definitions(table_definition_path)
+
+    return generate_geopackage_from_table_definition(dataset, table_definition)
+
+
+def load_table_definitions(table_definitions_path) -> TableDefinition:
+    return utils.load_config(table_definitions_path)

geopackage_validator/utils.py

@@ -69,6 +69,35 @@ def silence_gdal():
     return dataset
 
 
+def create_dataset(filename=None, error_handler=None):
+    if error_handler is not None:
+        gdal.UseExceptions()
+        gdal.PushErrorHandler(error_handler)
+
+    @contextmanager
+    def silence_gdal():
+        if error_handler is None:
+            warnings.warn("cannot silence gdal without error handler")
+            return
+        gdal.PopErrorHandler()
+        yield
+        gdal.PushErrorHandler(error_handler)
+
+    driver = ogr.GetDriverByName("GPKG")
+
+    dataset = None
+    try:
+        dataset = driver.CreateDataset(filename)
+    except Exception as e:
+        error_handler(gdal.CE_Failure, 0, e.args[0])
+
+    if dataset is not None:
+        dataset.silence_gdal = silence_gdal
+
+    return dataset
+
+
+
 def check_gdal_version():
     """This method checks if GDAL has the right version and exits with an error otherwise."""
     version_num = int(gdal.VersionInfo("VERSION_NUM"))

geopackage_validator/validate.py

@@ -5,7 +5,7 @@
 
 from osgeo import gdal
 
-from geopackage_validator.generate import TableDefinition
+from geopackage_validator.generate import TableDefinition, load_table_definitions
 from geopackage_validator import validations as validation
 from geopackage_validator.validations.validator import (
     Validator,
@@ -226,7 +226,3 @@ def get_validator_classes():
         if issubclass(getattr(validation, validator), Validator)
     ]
     return sorted(validator_classes, key=lambda v: (v.level, v.code))
-
-
-def load_table_definitions(table_definitions_path) -> TableDefinition:
-    return utils.load_config(table_definitions_path)

tests/validations/test_table_definitions_check.py

@@ -1,5 +1,4 @@
-from geopackage_validator.generate import generate_definitions_for_path
-from geopackage_validator.validate import load_table_definitions
+from geopackage_validator.generate import generate_definitions_for_path, load_table_definitions
 from geopackage_validator.validations.table_definitions_check import (
     TableDefinitionValidator,
     TableDefinitionValidatorV0,