diff --git a/CHANGELOG.md b/CHANGELOG.md index df0774e7..1d67a81b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - ES-MD: find RECINTO.shp wherever the archive puts it - DE-NDS: give the collection an id (the row index), which it was published without - DE-BB: ref_ident holds the FLIK (field block reference), not a farmer, and the shapefile is cp1252 +- Converters drop rows that cannot validate — no id, no crop:code, no geometry — up to 1% of a file, and fail rather than drop beyond that +- Every schema a conversion needs is fetched upfront with retries, so a transient outage cannot kill a long run at the last step - Update vecorel-cli to v0.2.17: - GeoJSON is read as UTF-8 as the format mandates, instead of the platform locale (cp1252 on Windows mangled umlauts) - GeoJSON files with a byte order mark no longer fail to read diff --git a/fiboa_cli/conversion/fiboa_converter.py b/fiboa_cli/conversion/fiboa_converter.py index c5ac7227..337dc67f 100644 --- a/fiboa_cli/conversion/fiboa_converter.py +++ b/fiboa_cli/conversion/fiboa_converter.py @@ -10,14 +10,109 @@ class FiboaBaseConverter(BaseConverter): area_is_in_ha = True area_calculate_missing = False use_variant_as_determination = False + # rows that cannot validate are dropped up to this share of the file, above + # which the conversion fails: a handful of bad rows in a source is normal, + # a broken mapping is not. Raise it for a source that is genuinely that + # patchy, and the message says how many rows it would have dropped. + max_dropped_share = 0.01 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.extensions.add(get_fiboa_uri()) + if self.use_variant_as_determination: + # The column is added in post_migrate; list it so it survives the + # "remove unlisted columns" step of the base converter. + self.columns = {**self.columns, "determination:datetime": "determination:datetime"} + + def convert(self, *args, **kwargs): + self._prewarm_schemas() + return super().convert(*args, **kwargs) + + def _prewarm_schemas(self): + """Fetch every schema this conversion will need before doing any real work. + + The schema hosts (vecorel.org, fiboa.org) fail intermittently, and + without this a blip after a long source download killed the conversion + at its very last step. load_file caches per process, so a successful + pre-warm makes the write network-free — and tells the converter what + every row must carry (see _required_properties).""" + import time + + from vecorel_cli.vecorel.util import load_file + from vecorel_cli.vecorel.version import vecorel_version + + uris = set(self.extensions) + uris.add(get_fiboa_uri()) + uris.add(f"https://vecorel.org/specification/v{vecorel_version}/schema.yaml") + # Nothing has been downloaded or converted yet, so a schema that cannot + # be fetched should say so now rather than after a wait: one retry for a + # dropped connection, then out. + attempts = 2 + for uri in sorted(uris): + for attempt in range(attempts): + try: + load_file(uri) + break + except Exception as e: + if attempt == attempts - 1: + raise RuntimeError( + f"Cannot load schema {uri} after {attempts} attempts: {e}" + ) from e + self.warning(f"Schema fetch failed ({uri}), retrying: {str(e)[:100]}") + time.sleep(2) + + def _required_properties(self) -> set[str]: + """What the schemas this conversion declares require of every row. + + A converter should not have to list them: the core schema requires id + and geometry, the crop extension crop:code and crop:code_list, and a + converter that declares an extension takes on its rules with it. The + schemas are already in memory, fetched by _prewarm_schemas. + """ + try: + schema = self.create_collection(self.id).merge_schemas() + except Exception as e: + self.warning(f"Cannot resolve the declared schemas ({e}); requiring an id only") + return {"id"} + return set(schema.get("required", [])) def post_migrate(self, gdf): gdf = super().post_migrate(gdf) + # post_migrate runs before columns are renamed, so look up the source column + for key in sorted(self._required_properties()): + for src, dst in self.columns.items(): + targets = dst if isinstance(dst, (list, tuple)) else [dst] + if key in targets and src in gdf.columns: + nulls = gdf[src].isna() + if nulls.any(): + share = nulls.mean() + if share > self.max_dropped_share: + raise ValueError( + f"{int(nulls.sum())} of {len(gdf)} rows ({share:.1%}) have no " + f"{key} ({src}); fix the converter instead of dropping them" + ) + self.warning( + f"Dropping {int(nulls.sum())} rows without a value for {key} ({src})" + ) + gdf = gdf[~nulls] + + # A null or empty geometry cannot be validated, tiled or Hilbert-sorted, + # and fails the run at its very last step, so it falls under the same + # bounded rule as the required properties. + if gdf.active_geometry_name is not None: + geom = gdf.geometry + blank = geom.isna() | geom.is_empty + if blank.any(): + share = blank.mean() + if share > self.max_dropped_share: + raise ValueError( + f"{int(blank.sum())} of {len(gdf)} rows ({share:.1%}) have an empty or " + f"missing geometry; fix the converter instead of dropping them" + ) + self.warning(f"Dropping {int(blank.sum())} rows with an empty or missing geometry") + gdf = gdf[~blank] + gdf_area_key = next((k for k, v in self.columns.items() if v == AREA_KEY), None) if self.area_calculate_missing: # If CRS is not in meters, reproject to an equal-area projection for area calculation diff --git a/fiboa_cli/datasets/commons/euro_land.py b/fiboa_cli/datasets/commons/euro_land.py index f33bee51..5294c22e 100644 --- a/fiboa_cli/datasets/commons/euro_land.py +++ b/fiboa_cli/datasets/commons/euro_land.py @@ -47,3 +47,11 @@ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) provider = "Europe-LAND HE Project " self.provider = (f"{self.provider}, {provider}") if self.provider else provider + + def migrate(self, gdf): + # Some Europe-LAND files (e.g. LT 2024) ship an empty crop_code column next to + # a populated crop_name; the name is then the best available crop code. + if "crop_code" in gdf.columns and gdf["crop_code"].isna().all(): + self.warning("crop_code is empty, using crop_name as crop:code") + gdf["crop_code"] = gdf["crop_name"] + return super().migrate(gdf)