Write custom schemas to fiboa metadata for use in improve/merge/etc. #113 and minor fixes

Author: m-mohr

CHANGELOG.md

@@ -17,6 +17,12 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   - rename columns
 - Converter for Switzerland
 
+### Changed
+
+- `fiboa convert` writes custom schemas to collection metadata
+- `fiboa validate` uses custom schemas for validation
+- `fiboa merge` keeps custom schemas when needed
+
 ## [v0.8.0] - 2024-11-12
 
 ### Added

fiboa_cli/__init__.py

@@ -601,7 +601,7 @@ def merge(datasets, out, crs, include, exclude, extension, compression, geoparqu
     type=click.STRING,
     help='Coordinate Reference System (CRS) to use for the GeoParquet file.',
     show_default=True,
-    default=DEFAULT_CRS,
+    default=None,
 )
 @click.option(
     '--compression', '-pc',

fiboa_cli/improve.py

@@ -1,10 +1,8 @@
 import os
 
-from pyproj import CRS
-
 from .const import CORE_COLUMNS
 from .parquet import create_parquet
-from .util import load_parquet_data, load_parquet_schema, log, parse_metadata
+from .util import load_parquet_data, load_parquet_schema, log, parse_metadata, pick_schemas
 
 
 def improve(input, out = None, rename_columns = {}, add_sizes = False, fix_geometries = False, crs = None, compression = None, geoparquet1 = False):
@@ -64,6 +62,12 @@ def improve(input, out = None, rename_columns = {}, add_sizes = False, fix_geome
         gdf["area"] = gdf_m["area"].fillna(gdf_m.geometry.area * 0.0001)
         gdf["perimeter"] = gdf_m["perimeter"].fillna(gdf_m.geometry.length)
 
+    custom_schemas = collection.get("fiboa_custom_schemas", {})
+    custom_schemas = pick_schemas(custom_schemas, columns, rename_columns)
+    if len(custom_schemas) > 0:
+        collection["fiboa_custom_schemas"] = custom_schemas
+
+
     # Write the merged dataset to the output file
     create_parquet(gdf, columns, collection, out, {}, compression=compression, geoparquet1=geoparquet1)
     log(f"Wrote data to {out}", "success")

fiboa_cli/merge.py

@@ -4,7 +4,7 @@
 
 from .const import CORE_COLUMNS
 from .parquet import create_parquet
-from .util import load_parquet_data, load_parquet_schema, log, parse_metadata
+from .util import load_parquet_data, load_parquet_schema, log, merge_schemas, parse_metadata, pick_schemas
 from .version import fiboa_version
 
 DEFAULT_CRS = "EPSG:4326"
@@ -20,22 +20,27 @@ def merge(datasets, out, crs = DEFAULT_CRS, includes = [], excludes = [], extens
 
     # Load the datasets
     all_gdf = []
+    custom_schemas = {}
     for dataset in datasets:
         # Load the dataset
         schema = load_parquet_schema(dataset)
+        collection = parse_metadata(schema, b"fiboa")
         file_columns = list(set(columns) & set(schema.names))
         gdf = load_parquet_data(dataset, columns=file_columns)
 
         # Change the CRS if necessary
         gdf.to_crs(crs=crs, inplace=True)
 
         # Add collection column to each dataset
-        collection = parse_metadata(schema, b"fiboa")
         if collection is not None and "id" in collection:
             gdf["collection"] = collection["id"]
         else:
             gdf["collection"] = os.path.splitext(os.path.basename(dataset))[0]
 
+        # Merge custom schemas
+        custom_schema = collection.get("fiboa_custom_schemas", {})
+        custom_schemas = merge_schemas(custom_schemas, custom_schema)
+
         all_gdf.append(gdf)
 
     merged = pd.concat(all_gdf, ignore_index=True)
@@ -51,6 +56,11 @@ def merge(datasets, out, crs = DEFAULT_CRS, includes = [], excludes = [], extens
         "fiboa_extensions": extensions,
     }
 
+    # Add custom schemas
+    custom_schemas = pick_schemas(custom_schemas, columns)
+    if len(custom_schemas) > 0:
+        collection["fiboa_custom_schemas"] = custom_schemas
+
     # Write the merged dataset to the output file
     create_parquet(merged, columns, collection, out, {}, compression=compression, geoparquet1=geoparquet1)
     log(f"Merged data to {out}", "success")

fiboa_cli/parquet.py

@@ -15,6 +15,11 @@ def create_parquet(data, columns, collection, output_file, config, missing_schem
     fiboa_schema = load_fiboa_schema(config)
     schemas = merge_schemas(missing_schemas, fiboa_schema)
 
+    # Add the custom schemas to the collection for future use
+    if len(missing_schemas) > 0:
+        collection = collection.copy()
+        collection["fiboa_custom_schemas"] = missing_schemas
+
     # Load all extension schemas
     extensions = {}
     if "fiboa_extensions" in collection and isinstance(collection["fiboa_extensions"], list):

fiboa_cli/util.py

@@ -276,6 +276,22 @@ def merge_schemas(*schemas):
     return result
 
 
+def pick_schemas(schemas, properties, rename = {}):
+    """Pick and rename schemas for specific properties"""
+    result = {
+        "required": [],
+        "properties": {}
+    }
+    for prop in properties:
+        prop2 = rename[prop] if prop in rename else prop
+        if prop in schemas["required"]:
+            result["required"].append(prop2)
+        if prop in schemas["properties"]:
+            result["properties"][prop2] = schemas["properties"][prop]
+
+    return result
+
+
 def migrate_schema(schema):
     """Migrate schema to a new version"""
     return schema.copy()

fiboa_cli/validate.py

@@ -207,6 +207,10 @@ def validate_parquet(file, config):
     for ext in extensions.values():
         schemas = merge_schemas(schemas, ext)
 
+    # Add custom schemas
+    custom_schemas = collection.get("fiboa_custom_schemas", {})
+    schemas = merge_schemas(schemas, custom_schemas)
+
     # Check that all required fields are present
     for key in schemas.get("required", []):
         if key not in parquet_schema.names: