Simplify fiboa publish for catalog-driven publication

CHANGELOG.md

@@ -6,16 +6,36 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
-
+- 
+- Add DuckDB BaseConverter for efficiently transforming large datasets
+- Fix `use_variant_as_determination`: the determination:datetime column was dropped again because it was not listed in `columns` (affected DK, HR)
+- Declare the beautifulsoup4 dependency that the ES-PV and ES-VC converters import
+- EE: name the cached WFS responses (ee_gsaa_<year>.gml)
+- ES-MD: the archive no longer nests RECINTO.shp in a folder
+- SK: KODKD is the (non-unique, sometimes empty) LPIS block code, keep it as block_id and use the row index as id
+- ES-CM: always read the year-named SIGPAC service (the unnamed one moved on to 2025 with a different id field)
+- REST converters: do not keep an error response as a cached page
+- ES-CB: determination date from the variant year (was an empty string, which broke the STAC temporal extent)
+- ES-CAT: the 2024 download is a shapefile package, not a GeoPackage; 34 crop names new in 2024 added to the mapping
+- CZ: find the shapefile in nested archive folders (2026)
+- DE-BB: read the shapefile as cp1252 (its .cpg wrongly says UTF-8)
+- NL: new PDOK download location (rvo/gewaspercelen/atom), add the 2026 concept edition
+- DE-TH: note the INSPIRE download service
+- Drop rows without a crop:code (required by the crop extension) with a warning instead of failing the conversion (BE-VLG 2023, ES-CN had one such row each); more than 1% missing is an error
+- Europe-LAND converters: use crop_name as crop:code when the file's crop_code column is empty (LT 2024)
+- BE-VLG: derive determination:datetime from the variant year instead of a constant date
+- `fiboa publish` no longer uploads to S3 or generates README/LICENSE files. It creates GeoParquet, PMTiles and a STAC Collection with relative links, `file:size`/`file:checksum` and a web-map-links v1.3.0 `pmtiles` link. Publishing is done by catalogs such as the [harmonized field data catalog](https://github.com/fieldsoftheworld/harmonized-field-data-catalog).
 - Add Italy Tuscany (IT-1) basd on EuroCrops v2
 - Suuport multiple years for CZ
 - Multiple years for DE_sh
 - Multiple year support for HR
 - Introduce FiboaBaseConverter.use_variant_as_determination for setting proper determination_date
 - Update years for DK (2025, 2026)
+- Converter for Spain (whole), based on the FEGA 2025+ data
 - Update fr-converter to support 2021/2022 files
 - Converter for Bavaria, Germany LPIS field blocks (de_by_block)
 - Converter for Hesse, Germany LPIS reference parcels
+- PerFileBaseConverter: per-file migration and merging the result, decreasing memory requirements for large data sets
 - Update vecorel-cli to v0.2.16:
   - Converter output is sorted by Hilbert distance
   - Commands exit with a non-zero exit code when they report a failure
@@ -32,7 +52,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - Fix the column additions of the determination fields in the AI4SF converter
 - Add HCAT to datasets where possible
 - Updated years & variants for at_crop, be_vlg, es_an, es_cl, es_pv, ie, pt, se
-- Extend create_stac, include include fiboa data
+- Extend create_stac, include fiboa data
 - Publish command; skip hidden files, generate better texts
 - Fix to vecorel: converter.license and provider should be string
 - Added a Dockerfile to simplify working with fiboa

README.md

@@ -62,7 +62,7 @@ fiboa CLI supports various commands to work with the files:
     - [Improve a fiboa Parquet file](#improve-a-fiboa-parquet-file)
     - [Update an extension template with new names](#update-an-extension-template-with-new-names)
     - [Converter for existing datasets](#converter-for-existing-datasets)
-    - [Publish datasets to source coop or your own s3 repository](#publish-datasets-to-source-coop-or-your-own-s3-repository)
+    - [Publish datasets](#publish-datasets)
   - [Development](#development)
     - [Implement a converter](#implement-a-converter)
   - [Run in Docker](#run-in-docker)
@@ -193,46 +193,43 @@ Use any of the IDs from the list to convert an existing dataset to fiboa:
 
 See [Implement a converter](#implement-a-converter) for details about how to
 
-### Publish datasets to source coop or your own s3 repository
+### Publish datasets
 
 `fiboa publish <dataset> -o <target>`
 
-The publish converts and publishes a fiboa dataset to source coop or your own s3 repository. The target directory 
-will be filled with the following files:
+Converts and validates a fiboa dataset and prepares everything that is needed to publish it
+in a (STAC-based) catalog. The target directory will be filled with the following files:
 
 ```
 <target>/
-  <dataset>.parquet
-  <dataset>.pmtiles      # requires working ogr2ogr and tippecanoe
-  stac/collection.json
-  README.md              # generated if --generate-meta/-gm flag is present
-  LICENSE.txt            # generated if --generate-meta/-gm flag is present
+  <dataset>[-<variant>].parquet
+  <dataset>[-<variant>].pmtiles   # requires working ogr2ogr and tippecanoe
+  collection.json                 # STAC Collection with relative links to the files above
 ```
 
-This directory is synchronized to the s3 repository (default source.coop/fiboa/data).
+The STAC Collection carries `file:size` and `file:checksum` for the files and a
+`pmtiles` link (web-map-links extension). Existing files in the target directory are reused,
+delete them to regenerate. Uploading to a bucket and catalog-specific metadata
+(README, styles, thumbnails, ...) are the job of the catalog that publishes the data, e.g. the
+[harmonized field data catalog](https://github.com/fieldsoftheworld/harmonized-field-data-catalog).
 
-**Requirements**: Requires the [aws CLI](https://docs.aws.amazon.com/cli/latest/userguide/getting-started-install.html) to be installed, 
-and `AWS_ACCESS_KEY_ID` with `AWS_SECRET_ACCESS_KEY` environment variables. Also, for generating the pmtiles file, 
-it requires [ogr2ogr](https://gdal.org/programs/ogr2ogr.html) and [tippecanoe](https://github.com/mapbox/tippecanoe).
+The command runs:
 
-The command executes the following steps:
-
-- `fiboa convert` to generate a fiboa parquet dataset. All convert parameters are passed to the converter.
-- `fiboa validate` to validate the fiboa dataset
-- creates a <dataset>.pmtiles from the parquet file. Uses ogr2ogr and tippecanoe
-- `fiboa create-stac-collection` to create a STAC collection
-- `fiboa publish` to publish the fiboa dataset to a source coop or your own s3 repository
+- `fiboa convert` to create a `<dataset>[-<variant>].parquet` file
+- `fiboa validate` to validate the GeoParquet file
+- `ogr2ogr | tippecanoe` to create the PMTiles file
+- `fiboa create-stac-collection` to create the STAC Collection
 
 Examples:
 
-- `fiboa publish at_crop -o data/at_crop`
-- `fiboa publish -c /tmp/cache -gm br_conab -o data/br_conab`
+- `fiboa publish at -o data/at`
+- `fiboa publish -c /tmp/cache nl --variant 2025 -o data/nl/2025`
 
 Relevant parameters:
 
-- `--generate-meta/-gm` Generatse the README.md and LICENSE.txt files if absent, based on data-survey and converter properties.
-- `--data-url` The URL to the data repository, used when generating the README
-- `--s3-upload-path` The `aws s3 sync` target. Defaults to `s3://source.coop/fiboa/data` . Uploading requires the `aws` CLI, and `AWS_ACCESS_KEY_ID` with `AWS_SECRET_ACCESS_KEY` environment variables.
+- `--variant` Choose the variant (e.g. year) of a dataset, defaults to the first variant.
+- `--no-pmtiles` Skip PMTiles generation.
+- `--tippecanoe-opts` Options passed to tippecanoe, defaults to `-zg --drop-densest-as-needed --extend-zooms-if-still-dropping`.
 
 Check `fiboa publish --help` for more details.
 

build/lib/fiboa_cli/__init__.py

@@ -0,0 +1,5 @@
+from vecorel_cli.registry import Registry
+
+from .registry import FiboaRegistry
+
+Registry.instance = FiboaRegistry()

build/lib/fiboa_cli/cli/setup.py

@@ -0,0 +1,3 @@
+from vecorel_cli.cli.setup import setup_cli
+
+run = setup_cli()

build/lib/fiboa_cli/conversion/convert_gml.py

@@ -0,0 +1,34 @@
+import os
+
+import geopandas
+from loguru import logger
+
+
+def gml_assure_columns(data, path, uri, layer, **kwargs):
+    # if GDAL opens a GML file, it generates a gfs file in which it tries to 'guess' a
+    # mapping from the GML XML file to features. This is not always correct.
+    # Call this function to add additional attributes from the GML
+    # We modify the GFS file, a more elegant solution is preferred
+    # See https://gdal.org/en/latest/drivers/vector/gml.html#schema for more info
+
+    if next(iter(kwargs)) not in data.columns:
+        logger.info("Patching generated GFS file")
+        assert path.endswith(".gml"), "Expected a gml file"
+        gfs_file = path[:-4] + ".gfs"
+        assert os.path.exists(gfs_file), "Expected a local, generated GFS file by OGR-GML driver"
+        # Fix GFS template file
+        with open(gfs_file, mode="r") as file:
+            gfs_xml = file.read()
+
+        for property in kwargs:
+            assert f"<Name>{property}</Name>" not in gfs_xml, "Expected unpatched gfs file"
+
+        lines = gfs_xml.splitlines()
+        for property, elements in kwargs.items():
+            element_str = "\n".join(f"<{k}>{v}</{k}>" for k, v in elements.items())
+            lines.insert(-2, f"<PropertyDefn><Name>{property}</Name>{element_str}</PropertyDefn>")
+        with open(gfs_file, mode="w") as file:
+            file.write("\n".join(lines))
+
+        data = geopandas.read_file(path, layer=layer)
+    return data

build/lib/fiboa_cli/conversion/converter_rest.py

@@ -0,0 +1,100 @@
+import os
+from urllib.parse import urlencode
+
+import geopandas as gpd
+import requests
+from vecorel_cli.vecorel.util import get_fs, stream_file
+
+
+class EsriRESTConverterMixin:
+    cache_folder = None
+    rest_base_url = None
+    rest_params = {}
+    rest_attribute = "OBJECTID"  # orderable, filterable, indexed
+
+    def rest_layer_filter(self, layers):
+        return next(iter(layers))
+
+    def get_urls(self):
+        assert self.rest_base_url, (
+            "Either define {c}.rest_base_url or override {c}.get_urls()".format(
+                c=self.__class__.__name__
+            )
+        )
+        return {"REST": self.rest_base_url}
+
+    def download_files(self, uris, cache_folder=None):
+        # Read-data will just stream alle pages of rest-service
+        if next(iter(uris), "").startswith("REST"):
+            self.cache_folder = cache_folder
+            return list(uris.values())
+
+        # This happens when input_file param is used
+        return super().download_files(uris, cache_folder)
+
+    def get_data(self, paths, **kwargs):
+        if isinstance(paths[0], tuple):
+            # (path, uri) pairs from the base downloader: input_file param was used
+            yield from super().get_data(paths, **kwargs)
+            return
+
+        base_url = paths[0]  # loop over paths to support more than 1 source
+        source_fs = get_fs(base_url)
+        cache_fs, cache_folder = self.get_cache(self.cache_folder)
+
+        service_metadata = requests.get(base_url, {"f": "pjson"}).json()
+        layer = self.rest_layer_filter(service_metadata["layers"])
+        page_size = service_metadata["maxRecordCount"]
+        layer_url = f"{base_url}/{layer['id']}/query"
+        get_dict = self.rest_params | {
+            "outFields": "*",
+            "returnGeometry": "true",
+            "f": "geojson",
+            "sortBy": self.rest_attribute,
+            "resultRecordCount": page_size,
+        }
+        gdfs = []
+        last_id = -1
+        while True:
+            get_dict["where"] = f"{self.rest_attribute}>{last_id}"
+            url = f"{layer_url}?{urlencode(get_dict)}"
+            if cache_fs is not None:
+                cache_file = os.path.join(
+                    cache_folder, f"{self.id}_{layer['id']}_{last_id}.geojson"
+                )
+                if not cache_fs.exists(cache_file):
+                    try:
+                        with cache_fs.open(cache_file, mode="wb") as file:
+                            stream_file(source_fs, url, file)
+                    except Exception:
+                        # A download that broke off must not survive as a cached page
+                        if cache_fs.exists(cache_file):
+                            cache_fs.rm(cache_file)
+                        raise
+                url = cache_file
+
+            try:
+                data = gpd.read_file(url)
+            except Exception as e:
+                # An error response from the server must not survive as a cached page
+                if cache_fs is not None and cache_fs.exists(url):
+                    cache_fs.rm(url)
+                raise RuntimeError(f"Could not read page {len(gdfs)} of {layer_url}: {e}") from e
+            print(
+                f"Read {len(data)} features, page {len(gdfs)} from [{data.iloc[0, 0]} ... {data.iloc[-1, 0]}]"
+            )
+            # joined layers return the field as <table>.<name>
+            id_column = next(
+                (
+                    c
+                    for c in data.columns
+                    if c == self.rest_attribute or c.endswith("." + self.rest_attribute)
+                ),
+                self.rest_attribute,
+            )
+            last_id = data[id_column].values[-1]
+
+            yield data, base_url, base_url, layer["id"]
+
+            if not len(data) >= page_size:
+                break