fix: empty gauge_ids bug, add CorruptedDataError, re-extract 7 countries, update README

- Fix get_watersheds(country, []) returning all rows instead of empty
  (falsy empty list in _store._read_geo filter condition)
- Add CorruptedDataError for ArrowInvalid parquet files
- Re-extract australia, chile, germany, lithuania, norway, poland,
  south_africa through fixed hydra-shed with GDAL geometry repair
- Add make_valid() safety net to convert_to_parquet.py
- Fix south_africa RegionSpec key in convert_to_parquet.py
- Update README gauge counts from live R2 (60,141 total across 16 countries)
- 4 new unit tests (80 total)

Author: CooperBigFoot

README.md

@@ -60,7 +60,7 @@ wr.available_countries()
 
 # List gauge IDs for a country
 wr.available_gauges("portugal")
-# -> ['02G-02H', '02O-01H', ..., '16J-01H']  (73 gauges)
+# -> ['02G-02H', '02O-01H', ..., '16J-01H']  (~710 gauges)
 ```
 
 ### Single Watershed
@@ -78,7 +78,7 @@ watershed, rivers = result
 
 ```python
 # All watersheds for a country
-gdf = wr.get_watersheds("portugal")  # -> GeoDataFrame (73 rows)
+gdf = wr.get_watersheds("portugal")  # -> GeoDataFrame (~710 rows)
 
 # Subset by gauge IDs
 gdf = wr.get_watersheds("portugal", ["04K/04A", "05G/01A"])
@@ -117,24 +117,24 @@ from watershed_retrieve import (
 
 | Country | Gauges | Status |
 |---------|--------|--------|
-| Australia | ~5,200 | Available |
-| Brazil | ~4,300 | Available |
-| Canada | ~5,800 | Available |
-| Chile | ~500 | Available |
-| Czech Republic | ~400 | Available |
-| France | ~5,400 | Available |
-| Germany | ~500 | Available |
-| Japan | ~1,200 | Available |
-| Lithuania | ~70 | Available |
-| Norway | ~600 | Available |
-| Poland | ~600 | Available |
-| Portugal | 73 | Available |
-| Slovenia | ~200 | Available |
-| South Africa | ~900 | Available |
-| Spain | ~600 | Available |
-| UK (EA) | ~1,500 | Pending — MERIT-Hydro coverage gap |
-| UK (NRFA) | ~1,500 | Pending — MERIT-Hydro coverage gap |
-| USA | ~32,000 | Available |
+| Australia | ~6,200 | Available |
+| Brazil | ~4,600 | Available |
+| Canada | ~7,600 | Available |
+| Chile | ~540 | Available |
+| Czech Republic | ~830 | Available |
+| France | ~5,300 | Available |
+| Germany | ~190 | Available |
+| Japan | ~820 | Available |
+| Lithuania | ~100 | Available |
+| Norway | ~4,500 | Available |
+| Poland | ~1,300 | Available |
+| Portugal | ~710 | Available |
+| Slovenia | ~710 | Available |
+| South Africa | ~1,290 | Available |
+| Spain | ~1,500 | Available |
+| UK (EA) | — | Pending — MERIT-Hydro coverage gap |
+| UK (NRFA) | — | Pending — MERIT-Hydro coverage gap |
+| USA | ~23,900 | Available |
 
 ## Development
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "watershed-retrieve"
-version = "1.0.3"
+version = "1.0.4"
 description = "Pre-delineated MERIT-Hydro watershed boundaries for ~60,000 gauging stations across 16 countries"
 readme = "README.md"
 authors = [{ name = "Nicolas Lazaro", email = "nlazaro@ethz.ch" }]
@@ -88,7 +88,7 @@ markers = [
 # Bump My Version
 # -----------------------
 [tool.bumpversion]
-current_version = "1.0.3"
+current_version = "1.0.4"
 commit = false
 tag = false
 allow_dirty = true

scripts/convert_to_parquet.py

@@ -37,7 +37,7 @@ class RegionSpec(NamedTuple):
     RegionSpec("poland", "poland"),
     RegionSpec("portugal", "portugal"),
     RegionSpec("slovenia", "slovenia"),
-    RegionSpec("south", "south_africa"),
+    RegionSpec("south_africa", "south_africa"),
     RegionSpec("spain", "spain"),
     RegionSpec("uk_ea", "uk_ea"),
     RegionSpec("uk_nrfa", "uk_nrfa"),
@@ -82,6 +82,11 @@ def convert(input_dir: Path, output_dir: Path) -> None:
             t0 = time.time()
             print(f"Reading {region.key}.gpkg layer={layer} ...", flush=True)
             gdf = gpd.read_file(path, layer=layer)
+            invalid_mask = ~gdf.geometry.is_valid
+            n_invalid = invalid_mask.sum()
+            if n_invalid > 0:
+                print(f"  Repairing {n_invalid} invalid geometries ...", flush=True)
+                gdf.geometry = gdf.make_valid()
             nrows = len(gdf)
 
             if region.key != region.output_name and "country" in gdf.columns:

src/watershed_retrieve/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "1.0.3"
+__version__ = "1.0.4"
 
 from ._api import (
     available_countries,
@@ -10,6 +10,7 @@
     get_watersheds_with_rivers,
 )
 from ._errors import (
+    CorruptedDataError,
     CountryNotFoundError,
     DataNotFoundError,
     DataUnavailableError,
@@ -29,6 +30,7 @@
     "get_watersheds",
     "get_watersheds_with_rivers",
     "Backend",
+    "CorruptedDataError",
     "CountryNotFoundError",
     "DataNotFoundError",
     "DataUnavailableError",

src/watershed_retrieve/_errors.py

@@ -23,3 +23,7 @@ class DataUnavailableError(WatershedRetrieveError):
 
 class R2ConnectionError(WatershedRetrieveError):
     pass
+
+
+class CorruptedDataError(WatershedRetrieveError):
+    pass

src/watershed_retrieve/_store.py

@@ -7,8 +7,9 @@
 import geopandas as gpd
 import pandas as pd
 from pyarrow.fs import FSSpecHandler, PyFileSystem
+from pyarrow.lib import ArrowInvalid
 
-from ._errors import DataNotFoundError, R2ConnectionError
+from ._errors import CorruptedDataError, DataNotFoundError, R2ConnectionError
 from ._registry import CountryInfo
 from ._types import CompositeGaugeId
 
@@ -51,9 +52,19 @@ def _read_geo(
         path: Path,
         gauge_ids: list[CompositeGaugeId] | None,
     ) -> gpd.GeoDataFrame:
+        if gauge_ids is not None and len(gauge_ids) == 0:
+            return gpd.GeoDataFrame()
         self._ensure_exists(path)
-        filters: list[tuple[str, str, list[str]]] | None = [("gauge_id", "in", gauge_ids)] if gauge_ids else None
-        return gpd.read_parquet(path, filters=filters)
+        filters: list[tuple[str, str, list[str]]] | None = (
+            [("gauge_id", "in", gauge_ids)] if gauge_ids is not None else None
+        )
+        try:
+            gdf = gpd.read_parquet(path, filters=filters)
+        except ArrowInvalid as exc:
+            raise CorruptedDataError(
+                f"Data file appears corrupted: {path}. Try re-downloading or report this issue. Detail: {exc}"
+            ) from exc
+        return gdf
 
     def read_watersheds(
         self,
@@ -93,11 +104,19 @@ def _remote_path(self, country: CountryInfo, suffix: str) -> str:
         return f"{_R2_BASE_URL}/{country.file_stem}_{suffix}.parquet"
 
     def _read_geo(self, url: str, gauge_ids: list[CompositeGaugeId] | None) -> gpd.GeoDataFrame:
-        filters: list[tuple[str, str, list[str]]] | None = [("gauge_id", "in", gauge_ids)] if gauge_ids else None
+        if gauge_ids is not None and len(gauge_ids) == 0:
+            return gpd.GeoDataFrame()
+        filters: list[tuple[str, str, list[str]]] | None = (
+            [("gauge_id", "in", gauge_ids)] if gauge_ids is not None else None
+        )
         try:
             return gpd.read_parquet(url, filesystem=self._fs, filters=filters)
         except FileNotFoundError:
             raise DataNotFoundError(f"Remote data file not found: {url}") from None
+        except ArrowInvalid as exc:
+            raise CorruptedDataError(
+                f"Remote data file appears corrupted: {url}. This is a known issue — please report it. Detail: {exc}"
+            ) from exc
         except (OSError, ConnectionError) as exc:
             raise R2ConnectionError(f"Failed to fetch data from R2: {exc}") from exc
 

tests/test_api_unit.py

@@ -94,3 +94,8 @@ def test_missing_gauge_in_list_raises(self, fake_store: FakeWatershedStore) -> N
     def test_returns_geodataframe(self, fake_store: FakeWatershedStore) -> None:
         api_mod._store = fake_store
         assert isinstance(wr.get_watersheds("portugal"), gpd.GeoDataFrame)
+
+    def test_empty_gauge_ids_returns_empty(self, fake_store: FakeWatershedStore) -> None:
+        api_mod._store = fake_store
+        result = wr.get_watersheds("portugal", [])
+        assert len(result) == 0

tests/test_local_store.py

@@ -3,6 +3,7 @@
 import geopandas as gpd
 import pytest
 
+from watershed_retrieve import CorruptedDataError
 from watershed_retrieve._errors import DataNotFoundError
 from watershed_retrieve._registry import CountryInfo
 from watershed_retrieve._store import LocalParquetStore
@@ -38,6 +39,11 @@ def test_missing_file_raises(self, tmp_path: Path) -> None:
         with pytest.raises(DataNotFoundError, match="Data file not found"):
             store.read_watersheds(PORTUGAL)
 
+    def test_empty_gauge_ids_returns_empty(self, synthetic_parquet_dir: Path) -> None:
+        store = LocalParquetStore(synthetic_parquet_dir)
+        result = store.read_watersheds(PORTUGAL, [])
+        assert len(result) == 0
+
 
 class TestLocalParquetStoreReadRivers:
     def test_reads_all_rows(self, synthetic_parquet_dir: Path) -> None:
@@ -55,6 +61,11 @@ def test_filters_by_gauge_ids(self, synthetic_parquet_dir: Path) -> None:
         gdf = store.read_rivers(PORTUGAL, [CompositeGaugeId("portugal_G001")])
         assert len(gdf) == 1
 
+    def test_empty_gauge_ids_returns_empty(self, synthetic_parquet_dir: Path) -> None:
+        store = LocalParquetStore(synthetic_parquet_dir)
+        result = store.read_rivers(PORTUGAL, [])
+        assert len(result) == 0
+
 
 class TestLocalParquetStoreReadGaugeIds:
     def test_returns_all_gauge_ids(self, synthetic_parquet_dir: Path) -> None:
@@ -71,3 +82,12 @@ def test_missing_file_raises(self, tmp_path: Path) -> None:
         store = LocalParquetStore(tmp_path)
         with pytest.raises(DataNotFoundError):
             store.read_gauge_ids(PORTUGAL)
+
+
+class TestCorruptedParquet:
+    def test_corrupted_file_raises_corrupted_data_error(self, tmp_path: Path) -> None:
+        store = LocalParquetStore(tmp_path)
+        corrupt_path = tmp_path / "portugal_watersheds.parquet"
+        corrupt_path.write_bytes(b"not a parquet file")
+        with pytest.raises(CorruptedDataError, match="corrupted"):
+            store.read_watersheds(PORTUGAL)