From 900813a3e77e6df0f74ef6fe873fec4738d7c97a Mon Sep 17 00:00:00 2001 From: camante Date: Mon, 7 Sep 2026 15:49:02 -0400 Subject: [PATCH 1/2] Add TNM elevation provider hierarchy support --- pyproject.toml | 1 + src/fetchez/cli/pipeline.py | 8 + src/fetchez/core.py | 26 +- src/fetchez/hooks/stream_init.py | 20 +- src/fetchez/modules/tnm.py | 109 +++- src/fetchez/modules/tnm_ned.py | 137 +++++ src/fetchez/modules/tnm_raster.py | 63 +++ src/fetchez/modules/tnm_wesm.py | 473 ++++++++++++++++++ src/fetchez/registry.py | 36 +- src/fetchez/utils.py | 15 +- tests/modules/test_tnm_ned.py | 160 ++++++ tests/modules/test_tnm_provider_foundation.py | 296 ++++++++++- tests/modules/test_tnm_raster.py | 45 ++ tests/modules/test_tnm_wesm.py | 268 ++++++++++ tests/test_http_file.py | 104 ++++ tests/test_recipe.py | 88 ++++ tests/test_stream_init.py | 61 +++ 17 files changed, 1866 insertions(+), 44 deletions(-) create mode 100644 src/fetchez/modules/tnm_ned.py create mode 100644 src/fetchez/modules/tnm_raster.py create mode 100644 src/fetchez/modules/tnm_wesm.py create mode 100644 tests/modules/test_tnm_ned.py create mode 100644 tests/modules/test_tnm_raster.py create mode 100644 tests/modules/test_tnm_wesm.py create mode 100644 tests/test_http_file.py create mode 100644 tests/test_stream_init.py diff --git a/pyproject.toml b/pyproject.toml index 5602cbb..b89fff9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ dependencies = [ "pyproj", "shapely", "pyogrio", + "rasterio>1.4.0", ] keywords = ["Geospatial"] diff --git a/src/fetchez/cli/pipeline.py b/src/fetchez/cli/pipeline.py index ddd8237..3058cbf 100644 --- a/src/fetchez/cli/pipeline.py +++ b/src/fetchez/cli/pipeline.py @@ -81,6 +81,14 @@ def get_command(self, ctx, name): if bundle_yml: help_text = bundle_yml.get("description", "") mod_args = [] + if bundle_yml.get("products"): + mod_args.append( + [ + "--products", + "Comma-separated products to include (default: all)", + "all", + ] + ) @click.command(name=name, help=help_text, hidden=True, cls=FetchezMainCommand) @click.option("--weight", type=float, default=1.0) diff --git a/src/fetchez/core.py b/src/fetchez/core.py index 0fb2ec7..8f5d15e 100644 --- a/src/fetchez/core.py +++ b/src/fetchez/core.py @@ -405,10 +405,11 @@ def __init__(self, url, session=None, callback=None): self.size = self._get_size() def _get_size(self): - resp = self.session.head(self.url) - if "Content-Length" not in resp.headers: - return 0 - return int(resp.headers["Content-Length"]) + with self.session.head(self.url, timeout=(10, 60)) as resp: + resp.raise_for_status() + if "Content-Length" not in resp.headers: + raise OSError("HTTP file has no Content-Length") + return int(resp.headers["Content-Length"]) def seek(self, offset, whence=io.SEEK_SET): if whence == io.SEEK_SET: @@ -436,10 +437,19 @@ def read(self, size=-1): # Fetch ONLY the specific bytes requested headers = {"Range": f"bytes={self.offset}-{end}"} - response = self.session.get(self.url, headers=headers, timeout=(10, 60)) - response.raise_for_status() - - data = response.content + with self.session.get( + self.url, headers=headers, timeout=(10, 60), stream=True + ) as response: + response.raise_for_status() + expected = f"bytes {self.offset}-{end}/{self.size}" + if ( + response.status_code != 206 + or response.headers.get("Content-Range") != expected + ): + raise OSError("HTTP server did not honor the requested byte range") + data = response.raw.read(end - self.offset + 2, decode_content=True) + if len(data) != end - self.offset + 1: + raise OSError("HTTP server returned an incomplete byte range") if self.callback: self.callback(len(data)) diff --git a/src/fetchez/hooks/stream_init.py b/src/fetchez/hooks/stream_init.py index feead1e..b070d3e 100644 --- a/src/fetchez/hooks/stream_init.py +++ b/src/fetchez/hooks/stream_init.py @@ -14,6 +14,10 @@ import math import logging +from pyproj import CRS +from pyproj.crs import CompoundCRS +from pyproj.exceptions import CRSError + from fetchez.spatial import Region from fetchez.hooks import FetchHook from fetchez.registry import ReaderRegistry, ProfileRegistry @@ -124,8 +128,20 @@ def run(self, entries): base_srs = reader.get_srs() or base_srs vert_srs = kwargs_copy.get("vert_srs") - if vert_srs and "+" not in base_srs: - base_srs = f"{base_srs}+{vert_srs}" + if vert_srs: + try: + horizontal = CRS.from_user_input(base_srs) + vertical = CRS.from_user_input(vert_srs) + except CRSError: + # Retain custom vertical references such as global:mss. + if "+" not in base_srs: + base_srs = f"{base_srs}+{vert_srs}" + else: + if len(horizontal.axis_info) == 2: + base_srs = CompoundCRS( + f"{horizontal.name} + {vertical.name}", + [horizontal, vertical], + ).to_wkt() logger.debug(f"[{self.name}] Using SRS: {base_srs}") diff --git a/src/fetchez/modules/tnm.py b/src/fetchez/modules/tnm.py index c5acc85..54714b9 100644 --- a/src/fetchez/modules/tnm.py +++ b/src/fetchez/modules/tnm.py @@ -20,6 +20,8 @@ from fetchez import utils from fetchez import spatial from fetchez import cli +from fetchez.modules import tnm_ned, tnm_raster +from fetchez.modules.tnm_wesm import WESM logger = logging.getLogger(__name__) @@ -56,12 +58,19 @@ "US Topo Historical", "Land Cover - Woodland", "3D Hydrography Program (3DHP)", + "Seamless 1-m DEM (S1M)", ] DATASET_ALIASES = { "1m": 2, "1_9as": 4, "1_3as": 3, "1_as": 1, + "2_as": 5, + "5m": 6, + "s1m": 29, +} +DATASET_PRODUCTS = { + DATASET_CODES[index]: alias for alias, index in DATASET_ALIASES.items() } @@ -76,6 +85,8 @@ q="Free text search query", date_start="Start date (YYYY-MM-DD)", date_end="End date (YYYY-MM-DD)", + strict_datasets="Fail instead of broadening a rejected dataset query", + source_coverage="Attach authoritative USGS source coverage", ) class TheNationalMap(FetchModule): name = "tnm" @@ -114,6 +125,8 @@ def __init__( date_start: Optional[str] = None, date_end: Optional[str] = None, dedupe: bool = True, + strict_datasets: bool = False, + source_coverage: bool = False, **kwargs, ): super().__init__(name="tnm", **kwargs) @@ -127,6 +140,12 @@ def __init__( self.dedupe = utils.str2bool(dedupe) if self.dedupe is None: self.dedupe = True + self.strict_datasets = utils.str2bool(strict_datasets) + if self.strict_datasets is None: + self.strict_datasets = False + self.source_coverage = utils.str2bool(source_coverage) + if self.source_coverage is None: + self.source_coverage = False def run(self): """Run the TNM fetching module.""" @@ -138,7 +157,7 @@ def run(self): bbox_str = f"{w},{s},{e},{n}" offset = 0 - total = 0 + expected_total: Optional[int] = None # Determine Datasets to query dataset_names = [] @@ -160,6 +179,9 @@ def run(self): if not dataset_names: dataset_names = ["National Elevation Dataset (NED) 1 arc-second"] + product = ( + DATASET_PRODUCTS.get(dataset_names[0]) if len(dataset_names) == 1 else None + ) best_tiles = {} all_tiles = [] @@ -188,10 +210,12 @@ def run(self): req = core.Fetch(TNM_API_PRODUCTS_URL).fetch_req(params=params) if ( - req + req is not None and "All dataset queries failed" in req.text and "datasets" in params ): + if self.strict_datasets or self.source_coverage: + raise RuntimeError("TNM API rejected the strict dataset query") logger.warning( "USGS rejected the strict dataset strings. Retrying with broad text search..." ) @@ -207,24 +231,44 @@ def run(self): req = core.Fetch(TNM_API_PRODUCTS_URL).fetch_req(params=params) if req is None or req.status_code != 200: - logger.error( - f"TNM API Failed: {req.status_code if req else 'No Response'}" - ) - break - - if req.text.strip().startswith("{errorMessage"): - logger.error(f"TNM API Error: {req.text}") - break + status = req.status_code if req is not None else "no response" + raise RuntimeError(f"TNM API request failed: {status}") try: - data = req.json() - total = data.get("total", 0) - items = data.get("items", []) + try: + data = req.json() + except Exception as exc: + raise RuntimeError("TNM API returned invalid JSON") from exc + if not isinstance(data, dict): + raise RuntimeError("TNM API returned an invalid response") + if data.get("errorMessage"): + raise RuntimeError(f"TNM API error: {data['errorMessage']}") + if "total" not in data or "items" not in data: + raise RuntimeError("TNM API response is missing total or items") + total = int(data["total"]) + items = data["items"] + if total < 0 or not isinstance(items, list): + raise RuntimeError("TNM API returned an invalid result page") + if expected_total is None: + expected_total = total + elif total != expected_total: + raise RuntimeError( + "TNM API result total changed during pagination: " + f"{expected_total} to {total}" + ) + if offset > total or len(items) > total - offset: + raise RuntimeError("TNM API returned an invalid result page") + if offset < total and not items: + raise RuntimeError( + f"TNM API pagination stopped after {offset} of {total} products" + ) for item in items: + if not isinstance(item, dict): + raise RuntimeError("TNM API returned an invalid product entry") url = item.get("downloadURL") - if not url: - continue + if not isinstance(url, str) or not url.strip(): + raise RuntimeError("TNM product is missing its download URL") filename = url.split("/")[-1] fmt = item.get("format", "Unknown") @@ -279,6 +323,7 @@ def run(self): "remote_size": item.get("sizeInBytes"), "title": item.get("title"), "tnm_project": project, + "tnm_product": product, "tnm_source_id": item.get("sourceId"), "tnm_publication_date": item.get("publicationDate"), "tnm_last_updated": item.get("lastUpdated"), @@ -297,15 +342,39 @@ def run(self): if date and date > existing_date: best_tiles[fn_bn] = item_data - except Exception as e: - logger.exception(f"Error parsing TNM JSON: {e}") - break + except RuntimeError: + raise + except Exception as exc: + raise RuntimeError("Error parsing TNM API response") from exc - offset += 100 + offset += len(items) if offset >= total: break - tiles = best_tiles.values() if self.dedupe else all_tiles + tiles = list(best_tiles.values()) if self.dedupe else all_tiles + if self.source_coverage and tiles: + unsupported = set(dataset_names).difference( + { + DATASET_CODES[2], + DATASET_CODES[4], + DATASET_CODES[6], + DATASET_CODES[29], + } + ) + if unsupported: + raise ValueError( + "TNM source coverage supports only S1M, 1 m, 5 m and 1/9 arc-second DEMs" + ) + if len(dataset_names) != 1: + raise ValueError("TNM source coverage requires one dataset per module") + if product in {"s1m", "5m"}: + tiles = tnm_raster.add_source_coverage(tiles, self.wgs_region) + elif product == "1_9as": + tiles = tnm_ned.add_source_coverage(tiles, self.wgs_region) + else: + tiles = WESM.add_source_coverage( + tiles, self.wgs_region, require_year=True + ) for tile_data in tiles: self.add_entry_to_results(**tile_data) diff --git a/src/fetchez/modules/tnm_ned.py b/src/fetchez/modules/tnm_ned.py new file mode 100644 index 0000000..4bfd5c0 --- /dev/null +++ b/src/fetchez/modules/tnm_ned.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +"""Read source footprints packaged with legacy USGS NED 1/9 arc-second DEMs.""" + +import hashlib +import time +import zipfile +from datetime import datetime, timezone +from pathlib import Path, PurePosixPath +from tempfile import TemporaryDirectory + +import requests +import shapely +from pyogrio.raw import read +from pyproj import CRS, Transformer +from shapely.ops import transform as shapely_transform + +from fetchez import core, spatial +from fetchez.modules.tnm_wesm import collection_year, project_name + + +def _read_footprint(url): + for attempt in range(3): + try: + digest = hashlib.sha256(url.encode()) + with TemporaryDirectory(prefix="fetchez-ned-") as directory: + with ( + requests.Session() as session, + core.HttpFile(url, session=session) as remote, + zipfile.ZipFile(remote) as archive, + ): + names = archive.namelist() + shapes = [name for name in names if name.lower().endswith(".shp")] + if len(shapes) != 1: + raise RuntimeError( + "NED archive must contain one source footprint" + ) + source = shapes[0] + stem = str(PurePosixPath(source).with_suffix("")) + members = {name.lower(): name for name in names} + for suffix in (".shp", ".shx", ".dbf", ".prj", ".cpg"): + member = members.get((stem + suffix).lower()) + if member is None: + if suffix == ".cpg": + continue + raise RuntimeError( + f"NED source footprint is missing {suffix}" + ) + payload = archive.read(member) + digest.update(member.encode()) + digest.update(payload) + (Path(directory) / ("source" + suffix)).write_bytes(payload) + return ( + read(str(Path(directory) / "source.shp"), return_fids=True), + source, + digest.hexdigest(), + ) + except Exception as exc: + if attempt == 2: + raise RuntimeError( + f"Unable to read NED source footprint from {url}" + ) from exc + time.sleep(2**attempt) + + +def add_source_coverage(entries, region): + """Attach packaged source polygons; omit only known non-intersections.""" + + roi = spatial.region_to_shapely(region) + selected = [] + digest = hashlib.sha256() + for entry in entries: + project = project_name(entry) + if not project: + raise RuntimeError( + "TNM source coverage requires a provider project identity" + ) + bounds = entry.get("bounds") + if not bounds or len(bounds) != 4 or any(value is None for value in bounds): + raise RuntimeError("TNM source coverage requires product bounds") + url = entry["url"] + (meta, fids, geometry_wkb, fields), member, checksum = _read_footprint(url) + digest.update(checksum.encode()) + if not meta.get("crs"): + raise RuntimeError("NED source footprint has no CRS") + if fids is None or geometry_wkb is None or len(fids) == 0: + raise RuntimeError("NED source footprint has no source features") + names = [str(name).lower() for name in meta["fields"]] + if len(geometry_wkb) != len(fids) or any( + len(field) != len(fids) for field in fields + ): + raise RuntimeError("NED source footprint has incomplete source features") + transformer = None + if CRS.from_user_input(meta["crs"]) != CRS.from_epsg(4326): + transformer = Transformer.from_crs(meta["crs"], "EPSG:4326", always_xy=True) + source = spatial.region_to_shapely(bounds).intersection(roi) + claims = [] + for offset, fid in enumerate(fids): + row = {name: fields[pos][offset] for pos, name in enumerate(names)} + identity = row.get("proj_name") or row.get("demname") + if not isinstance(identity, str) or not identity.strip(): + raise RuntimeError("NED source footprint has no source identity") + geometry = shapely.from_wkb(geometry_wkb[offset]) + if ( + geometry is None + or geometry.geom_type not in ("Polygon", "MultiPolygon") + or geometry.is_empty + or not geometry.is_valid + ): + raise RuntimeError("NED source footprint has invalid polygon geometry") + if transformer is not None: + geometry = shapely_transform(transformer.transform, geometry) + coverage = geometry.intersection(source) + if coverage.is_empty or coverage.area == 0: + continue + claims.append( + { + "geometry": shapely.to_wkt(coverage, rounding_precision=-1), + "fid": int(fid), + "project": str(row.get("proj_name") or project), + "source_dem": str(row["demname"]) if row.get("demname") else None, + "year": collection_year({"collect_start": row.get("s_date")}), + } + ) + if not claims: + continue + entry["tnm_project"] = project + entry["tnm_source_coverage"] = claims + entry["tnm_ned_source_url"] = f"{url}#{member}" + selected.append(entry) + + retrieved_at = datetime.now(timezone.utc).isoformat() + for entry in selected: + entry["tnm_ned_snapshot_sha256"] = digest.hexdigest() + entry["tnm_ned_snapshot_retrieved_at"] = retrieved_at + return selected diff --git a/src/fetchez/modules/tnm_raster.py b/src/fetchez/modules/tnm_raster.py new file mode 100644 index 0000000..a768ee4 --- /dev/null +++ b/src/fetchez/modules/tnm_raster.py @@ -0,0 +1,63 @@ +"""Read the published raster extent for projected TNM elevation products.""" + +import time + +import requests +import rasterio +import shapely +from pyproj import CRS, Transformer +from shapely.ops import transform + +from fetchez import core, spatial + + +def add_source_coverage(entries, region): + """Use raster edges, never valid-data holes, as the source footprint.""" + roi = spatial.region_to_shapely(region) + selected = [] + with requests.Session() as session: + for entry in entries: + for attempt in range(3): + try: + with core.HttpFile(entry["url"], session=session) as remote: + with rasterio.open(remote) as source: + if source.crs is None: + raise RuntimeError("TNM raster has no source CRS") + crs = CRS(source.crs) + if crs.is_compound: + crs = crs.sub_crs_list[0] + # Densify at pixel spacing before projecting the perimeter. + corners = [ + source.transform * point + for point in ( + (0, 0), + (source.width, 0), + (source.width, source.height), + (0, source.height), + ) + ] + footprint = shapely.Polygon(corners) + footprint = shapely.segmentize(footprint, min(source.res)) + footprint = transform( + Transformer.from_crs( + crs, 4326, always_xy=True + ).transform, + footprint, + ) + entry["tnm_raster_crs"] = source.crs.to_wkt() + break + except (requests.RequestException, OSError): + if attempt == 2: + raise + time.sleep(2**attempt) + coverage = footprint.intersection(roi) + if coverage.is_empty or coverage.area == 0: + continue + entry["tnm_source_coverage"] = [ + { + "geometry": shapely.to_wkt(coverage, rounding_precision=-1), + "source": entry["url"], + } + ] + selected.append(entry) + return selected diff --git a/src/fetchez/modules/tnm_wesm.py b/src/fetchez/modules/tnm_wesm.py new file mode 100644 index 0000000..1bd5795 --- /dev/null +++ b/src/fetchez/modules/tnm_wesm.py @@ -0,0 +1,473 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- + +"""Read authoritative USGS WESM source coverage for TNM products.""" + +import csv +import hashlib +import io +import logging +import os +import re +import time +from contextlib import contextmanager +from datetime import datetime, timezone + +import shapely +from pyogrio.raw import read +from pyproj import CRS, Transformer +from shapely.ops import transform as shapely_transform + +from fetchez import core, spatial + + +WESM_CSV_URL = ( + "https://prd-tnm.s3.amazonaws.com/StagedProducts/Elevation/metadata/WESM.csv" +) +WESM_GPKG_URL = ( + "https://prd-tnm.s3.amazonaws.com/StagedProducts/Elevation/metadata/WESM.gpkg" +) +WESM_GPKG_PATH = "/vsis3/prd-tnm/StagedProducts/Elevation/metadata/WESM.gpkg" +WESM_LAYER = "WESM" +WESM_FIELDS = ( + "project", + "project_id", + "workunit", + "workunit_id", + "collect_start", + "collect_end", + "sourcedem_link", +) +WESM_IDENTITY_FIELDS = ("workunit", "workunit_id", "project", "project_id") +WESM_FEATURE_CHUNK_SIZE = 100 +WESM_TIMEOUT = 60 +WESM_RETRIES = 3 + +logger = logging.getLogger(__name__) + + +def _value(value): + if value is None: + return None + value = str(value).strip() + return None if not value or value.lower() == "nan" else value + + +def _path_names(value): + if not isinstance(value, str): + return [] + + names = [] + for marker in ("/Projects/", "/metadata/"): + if marker not in value: + continue + for part in value.split(marker, 1)[1].split("/"): + part = part.split("?", 1)[0].strip() + if part and part not in names: + names.append(part) + return names + + +def project_name(entry): + """Return the TNM project identity exposed by the product itself.""" + + project = _value(entry.get("tnm_project")) + if project: + return project + + for field in ("dst_fn", "url"): + value = entry.get(field) + if not isinstance(value, str): + continue + name = value.split("?", 1)[0].rsplit("/", 1)[-1] + for suffix in (".zip", ".img", ".xml", ".txt", ".html"): + if name.lower().endswith(suffix): + name = name[: -len(suffix)] + break + match = re.match(r"^ned19_[^_]+_[^_]+_(.+)$", name, re.IGNORECASE) + if match: + return match.group(1) + return None + + +def entry_names(entry): + """Return provider-exposed names that may identify an entry in WESM.""" + + names = [] + project = project_name(entry) + if project: + names.append(project) + for field in ("tnm_vendor_meta_url", "tnm_meta_url", "url"): + for name in _path_names(entry.get(field)): + if name not in names: + names.append(name) + return names + + +def _row_names(row): + names = [] + for field in ("workunit", "project"): + value = _value(row.get(field)) + if value and value not in names: + names.append(value) + for name in _path_names(row.get("sourcedem_link")): + if name not in names: + names.append(name) + return names + + +def _normalized_name(value, drop_compass=False): + compass = { + "eastern": "e", + "northern": "n", + "southern": "s", + "western": "w", + } + tokens = re.sub(r"[^a-z0-9]+", " ", str(value).lower()).split() + tokens = [compass.get(token, token) for token in tokens] + if drop_compass: + tokens = [token for token in tokens if token not in {"e", "n", "s", "w"}] + return "".join(tokens) or None + + +def _row_project(row, aliases): + row_names = set(_row_names(row)) + matches = [ + project for project, names in aliases.items() if row_names.intersection(names) + ] + if not matches: + normalized = {_normalized_name(name) for name in row_names} + normalized.discard(None) + matches = [ + project + for project, names in aliases.items() + if normalized.intersection({_normalized_name(name) for name in names}) + ] + if not matches: + # TNM occasionally omits an otherwise authoritative cardinal qualifier + # from its project directory (for example CA_SanDiegoCo_2016 versus the + # WESM work unit CA_E_SanDiegoCo_2016). Accept that provider naming + # difference only when it identifies one TNM project unambiguously. + directionless = { + _normalized_name(name, drop_compass=True) for name in row_names + } + directionless.discard(None) + matches = [ + project + for project, names in aliases.items() + if directionless.intersection( + {_normalized_name(name, drop_compass=True) for name in names} + ) + ] + if len(matches) > 1: + raise RuntimeError( + "WESM work unit matches multiple TNM projects: " + + ", ".join(sorted(matches)) + ) + return matches[0] if matches else None + + +def collection_year(row): + for field in ("collect_end", "collect_start"): + value = _value(row.get(field)) + if value is None: + continue + try: + number = float(value) + except ValueError: + number = None + if number is not None and number > 10_000_000_000: + return datetime.fromtimestamp(number / 1000, timezone.utc).year + match = re.search(r"(? str: f"{key}={args[key]}" for key in [ "datatype", + "dataset", "datasets", "formats", "layer", @@ -729,7 +730,40 @@ def expand_modules( bundle_def = recipe_meta.get("config", {}) if bundle_def: - child_modules = bundle_def.get("modules", []) + child_modules = copy.deepcopy(bundle_def.get("modules", [])) + products = bundle_def.get("products") + selected = user_args.get("products") + if products: + unknown_args = set(user_args).difference({"products", "weight"}) + if unknown_args: + raise ValueError( + f"Unknown bundle argument(s): {', '.join(sorted(unknown_args))}. " + "Use products=s1m/1m/1_as in source strings." + ) + if products and selected and str(selected).lower() != "all": + selected_products = { + value.strip() + for value in str(selected).replace(",", "/").split("/") + if value.strip() + } + unknown = selected_products.difference(products) + if unknown: + raise ValueError( + f"Unknown product(s) for {target}: " + f"{', '.join(sorted(unknown))}" + ) + child_modules = [ + child + for child in child_modules + if child.get("args", {}).get("datasets") + in selected_products + ] + start_hook = bundle_def.get("product_start_hook") + if products and child_modules and start_hook: + for hook in child_modules[0].get("hooks", []): + if hook.get("name") == start_hook: + hook.setdefault("args", {})["start"] = True + break child_expanded = cls.expand_modules(child_modules, current_weight) for child_mod in child_expanded: diff --git a/src/fetchez/utils.py b/src/fetchez/utils.py index fbfcad4..b36daa5 100644 --- a/src/fetchez/utils.py +++ b/src/fetchez/utils.py @@ -693,14 +693,7 @@ def compile_sources(sources): compiled_modules = [] for src in sources: - if str(src) in BundleRegistry.get_registry().keys(): - partial_recipe = BundleRegistry.get_yaml(str(src)) - if "modules" in partial_recipe: - compiled_modules.extend(partial_recipe["modules"]) - logger.debug( - f"Imported {len(partial_recipe['modules'])} modules from {src}" - ) - elif str(src).lower().endswith((".yaml", ".yml")) and Path(src).exists(): + if str(src).lower().endswith((".yaml", ".yml")) and Path(src).exists(): try: with open(src, "r") as f: partial_recipe = yaml.safe_load(f) @@ -718,7 +711,11 @@ def compile_sources(sources): elif src == "-": continue # TODO: add stdin support else: - compiled_modules.append(parse_source_string(src)) + parsed = parse_source_string(src) + name = parsed.get("module") + if name in BundleRegistry.get_registry(): + parsed["bundle"] = parsed.pop("module") + compiled_modules.append(parsed) return compiled_modules diff --git a/tests/modules/test_tnm_ned.py b/tests/modules/test_tnm_ned.py new file mode 100644 index 0000000..11f7f2b --- /dev/null +++ b/tests/modules/test_tnm_ned.py @@ -0,0 +1,160 @@ +import io +import zipfile + +import numpy as np +import pytest +import shapely +from pyogrio.raw import write +from pyproj import Transformer +from shapely.ops import transform + +from fetchez import spatial +from fetchez.modules import tnm_ned + + +REGION = spatial.Region(0, 1, 0, 1, srs="EPSG:4326") + + +def _entry(): + return { + "url": ( + "https://prd-tnm.s3.amazonaws.com/StagedProducts/Elevation/19/IMG/" + "ned19_n45x00_w067x00_me_northcoast_2010.zip" + ), + "bounds": (0, 1, 0, 1), + } + + +def _archive(tmp_path, nested=False, omit=None): + path = tmp_path / "footprint.shp" + fields = ( + { + "proj_name": "AK_Juneau_2013", + "demname": None, + "s_date": 2013, + "resolution": 100, + } + if nested + else {"DEMNAME": "me_nelot1_dem", "S_DATE": 2010, "RESOLUTION": 19} + ) + write( + path, + np.array([shapely.to_wkb(shapely.box(0.25, 0.25, 0.75, 0.75))]), + [np.array([value]) for value in fields.values()], + list(fields), + driver="ESRI Shapefile", + geometry_type="Polygon", + crs="EPSG:4269", + ) + buffer = io.BytesIO() + prefix = "ned19_n58x25_w134x00/" if nested else "" + with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive: + for member in sorted(tmp_path.glob("footprint.*")): + if member.suffix != omit: + archive.write(member, prefix + member.name) + archive.writestr(prefix + "raster.img", b"raster must not be read") + return buffer.getvalue(), prefix + path.name + + +@pytest.mark.parametrize("nested", [False, True]) +def test_reads_actual_flat_and_nested_footprints_without_raster( + monkeypatch, tmp_path, nested +): + payload, source = _archive(tmp_path, nested) + monkeypatch.setattr(tnm_ned.core, "HttpFile", lambda *a, **k: io.BytesIO(payload)) + original = zipfile.ZipFile.read + members = [] + + def read_member(archive, name, *args, **kwargs): + members.append(name) + return original(archive, name, *args, **kwargs) + + monkeypatch.setattr(zipfile.ZipFile, "read", read_member) + selected = tnm_ned.add_source_coverage([_entry()], REGION) + claim = selected[0]["tnm_source_coverage"][0] + assert claim["year"] == (2013 if nested else 2010) + assert claim["source_dem"] == (None if nested else "me_nelot1_dem") + if nested: + assert claim["project"] == "AK_Juneau_2013" + assert shapely.from_wkt(claim["geometry"]).equals( + shapely.box(0.25, 0.25, 0.75, 0.75) + ) + assert selected[0]["tnm_ned_source_url"] == _entry()["url"] + "#" + source + assert len(selected[0]["tnm_ned_snapshot_sha256"]) == 64 + assert "tnm_wesm_snapshot_sha256" not in selected[0] + assert all(not member.endswith(".img") for member in members) + + +@pytest.mark.parametrize("omit", [".shp", ".shx", ".dbf", ".prj"]) +def test_missing_archive_metadata_fails_after_bounded_retries( + monkeypatch, tmp_path, omit +): + payload, _ = _archive(tmp_path, omit=omit) + calls = [] + + def remote(*args, **kwargs): + calls.append(1) + return io.BytesIO(payload) + + monkeypatch.setattr(tnm_ned.core, "HttpFile", remote) + monkeypatch.setattr(tnm_ned.time, "sleep", lambda _: None) + with pytest.raises(RuntimeError, match="Unable to read NED source footprint"): + tnm_ned.add_source_coverage([_entry()], REGION) + assert len(calls) == 3 + + +def _raw(monkeypatch, geometry=None, crs="EPSG:4326", identity="me_nelot1_dem"): + if geometry is None: + geometry = shapely.box(0.25, 0.25, 0.75, 0.75) + result = ( + {"crs": crs, "fields": ["DEMNAME", "S_DATE"]}, + [0], + [shapely.to_wkb(geometry)], + [[identity], [2010]], + ) + monkeypatch.setattr( + tnm_ned, "_read_footprint", lambda _: (result, "source.shp", "abc") + ) + + +def test_known_source_outside_roi_is_omitted(monkeypatch): + _raw(monkeypatch, shapely.box(2, 2, 3, 3)) + assert tnm_ned.add_source_coverage([_entry()], REGION) == [] + + +def test_missing_source_identity_fails_closed(monkeypatch): + _raw(monkeypatch, identity=None) + with pytest.raises(RuntimeError, match="no source identity"): + tnm_ned.add_source_coverage([_entry()], REGION) + + +def test_missing_crs_fails_closed(monkeypatch): + _raw(monkeypatch, crs=None) + with pytest.raises(RuntimeError, match="no CRS"): + tnm_ned.add_source_coverage([_entry()], REGION) + + +@pytest.mark.parametrize( + "geometry", + [ + shapely.Point(0.5, 0.5), + shapely.Polygon(), + shapely.Polygon([(0, 0), (1, 1), (1, 0), (0, 1)]), + ], +) +def test_invalid_polygon_raises(monkeypatch, geometry): + _raw(monkeypatch, geometry) + with pytest.raises(RuntimeError, match="invalid polygon geometry"): + tnm_ned.add_source_coverage([_entry()], REGION) + + +def test_projected_footprint_is_clipped_to_tile_and_roi(monkeypatch): + geometry = transform( + Transformer.from_crs(4326, 3857, always_xy=True).transform, + shapely.box(-1, -1, 2, 2), + ) + _raw(monkeypatch, geometry, crs="EPSG:3857") + region = spatial.Region(0.5, 1.5, 0.5, 1.5, srs="EPSG:4326") + selected = tnm_ned.add_source_coverage([_entry()], region) + geometry = shapely.from_wkt(selected[0]["tnm_source_coverage"][0]["geometry"]) + assert geometry.equals(shapely.box(0.5, 0.5, 1, 1)) diff --git a/tests/modules/test_tnm_provider_foundation.py b/tests/modules/test_tnm_provider_foundation.py index a13bfae..e240a2f 100644 --- a/tests/modules/test_tnm_provider_foundation.py +++ b/tests/modules/test_tnm_provider_foundation.py @@ -9,13 +9,15 @@ class FakeResponse: - status_code = 200 - text = "{}" - - def __init__(self, payload): + def __init__(self, payload, status_code=200, text=None, json_error=None): self.payload = payload + self.status_code = status_code + self.text = text if text is not None else "{}" + self.json_error = json_error def json(self): + if self.json_error is not None: + raise self.json_error return self.payload @@ -69,6 +71,9 @@ def _item(title, url, publication_date, source_id): ("1_9as", tnm.DATASET_CODES[4]), ("1_3as", tnm.DATASET_CODES[3]), ("1_as", tnm.DATASET_CODES[1]), + ("2_as", tnm.DATASET_CODES[5]), + ("5m", tnm.DATASET_CODES[6]), + ("s1m", tnm.DATASET_CODES[29]), ("2", tnm.DATASET_CODES[2]), ("8/2", f"{tnm.DATASET_CODES[8]},{tnm.DATASET_CODES[2]}"), ], @@ -141,6 +146,67 @@ def test_default_dedupe_keeps_existing_newest_product_behavior(): assert "/metadata/waf/" in mod.results[0]["tnm_vendor_meta_url"] +def test_single_dataset_alias_preserves_product_identity(): + FakeFetch.payload = { + "total": 1, + "items": [ + _item( + "USGS Seamless 1 Meter", + "https://example.test/StagedProducts/Elevation/S1M/test.tif", + "2026-01-01", + "s1m-source", + ) + ], + } + mod = tnm.TheNationalMap( + src_region=SAMPLE_REGION, + datasets="s1m", + use_cache=False, + ) + mod.run() + + assert mod.results[0]["tnm_product"] == "s1m" + + +@pytest.mark.parametrize("product", ["1m", "1_9as"]) +def test_source_coverage_uses_the_provider_for_the_selected_product( + monkeypatch, product +): + FakeFetch.payload = { + "total": 1, + "items": [ + _item( + "source tile", + "https://example.test/Projects/ME_Project/USGS_1M_tile.tif", + "2020-01-01", + "source", + ) + ], + } + called = [] + + def wesm(entries, region, require_year=False): + called.append(("wesm", require_year)) + return entries + + def ned(entries, region): + called.append(("ned", False)) + return entries + + monkeypatch.setattr(tnm.WESM, "add_source_coverage", wesm) + monkeypatch.setattr(tnm.tnm_ned, "add_source_coverage", ned) + mod = tnm.TheNationalMap( + src_region=SAMPLE_REGION, + datasets=product, + source_coverage=True, + use_cache=False, + ) + + mod.run() + + assert called == ([("wesm", True)] if product == "1m" else [("ned", False)]) + + def test_dedupe_false_retains_overlapping_products_without_name_collision(): FakeFetch.payload = { "total": 2, @@ -177,3 +243,225 @@ def test_dedupe_false_retains_overlapping_products_without_name_collision(): "CA_2024_Test", "CA_2025LosAngelesPostWildfire_C25", ] + + +@pytest.mark.parametrize( + ("response", "message"), + [ + (FakeResponse({}, status_code=500), "request failed: 500"), + ( + FakeResponse({"errorMessage": "Search timed out"}), + "TNM API error: Search timed out", + ), + ( + FakeResponse({}, json_error=ValueError("broken")), + "returned invalid JSON", + ), + (FakeResponse({}), "missing total or items"), + ], +) +def test_provider_failures_raise_and_are_not_cached( + tmp_path, monkeypatch, response, message +): + class FailedFetch: + def __init__(self, _url): + pass + + def fetch_req(self, params=None): + return response + + monkeypatch.setattr(tnm.core, "Fetch", FailedFetch) + mod = tnm.TheNationalMap( + src_region=SAMPLE_REGION, + datasets="1m", + outdir=tmp_path, + ) + + with pytest.raises(RuntimeError, match=message): + mod.run() + + assert not list(tmp_path.rglob("*.json")) + + +def test_successful_empty_result_is_cacheable(tmp_path): + mod = tnm.TheNationalMap( + src_region=SAMPLE_REGION, + datasets="1m", + outdir=tmp_path, + ) + + mod.run() + + cache_files = list(tmp_path.rglob("*.json")) + assert len(cache_files) == 1 + assert cache_files[0].read_text().strip() == "[]" + + +def test_incomplete_pagination_raises(monkeypatch): + class PagedFetch: + calls = 0 + + def __init__(self, _url): + pass + + def fetch_req(self, params=None): + self.__class__.calls += 1 + items = ( + [ + _item( + "first", + "https://example.test/Projects/CA_Test/USGS_1M_tile.tif", + "2025-01-01", + "first", + ) + ] + if self.calls == 1 + else [] + ) + return FakeResponse({"total": 2, "items": items}) + + monkeypatch.setattr(tnm.core, "Fetch", PagedFetch) + mod = tnm.TheNationalMap( + src_region=SAMPLE_REGION, + datasets="1m", + use_cache=False, + ) + + with pytest.raises(RuntimeError, match="stopped after 1 of 2 products"): + mod.run() + + +def test_pagination_total_change_raises(monkeypatch): + class PagedFetch: + calls = 0 + + def __init__(self, _url): + pass + + def fetch_req(self, params=None): + self.__class__.calls += 1 + total = 2 if self.calls == 1 else 1 + return FakeResponse( + { + "total": total, + "items": [ + _item( + "page item", + "https://example.test/Projects/CA_Test/USGS_1M_tile.tif", + "2025-01-01", + str(self.calls), + ) + ], + } + ) + + monkeypatch.setattr(tnm.core, "Fetch", PagedFetch) + mod = tnm.TheNationalMap( + src_region=SAMPLE_REGION, + datasets="1m", + use_cache=False, + ) + + with pytest.raises(RuntimeError, match="total changed during pagination"): + mod.run() + + +def test_source_coverage_does_not_broaden_rejected_dataset_query(monkeypatch): + class RejectedFetch: + calls = 0 + + def __init__(self, _url): + pass + + def fetch_req(self, params=None): + self.__class__.calls += 1 + return FakeResponse( + {"errorMessage": "All dataset queries failed"}, + text='{"errorMessage":"All dataset queries failed"}', + ) + + monkeypatch.setattr(tnm.core, "Fetch", RejectedFetch) + mod = tnm.TheNationalMap( + src_region=SAMPLE_REGION, + datasets="1m", + source_coverage=True, + use_cache=False, + ) + + with pytest.raises(RuntimeError, match="rejected the strict dataset query"): + mod.run() + assert RejectedFetch.calls == 1 + + +def test_strict_dataset_query_does_not_broaden_without_source_coverage(monkeypatch): + class RejectedFetch: + calls = 0 + + def __init__(self, _url): + pass + + def fetch_req(self, params=None): + self.__class__.calls += 1 + return FakeResponse( + {"errorMessage": "All dataset queries failed"}, + text='{"errorMessage":"All dataset queries failed"}', + ) + + monkeypatch.setattr(tnm.core, "Fetch", RejectedFetch) + mod = tnm.TheNationalMap( + src_region=SAMPLE_REGION, + datasets="s1m", + strict_datasets=True, + use_cache=False, + ) + + with pytest.raises(RuntimeError, match="rejected the strict dataset query"): + mod.run() + assert RejectedFetch.calls == 1 + + +def test_source_coverage_is_requested_from_wesm(monkeypatch): + FakeFetch.payload = { + "total": 1, + "items": [ + _item( + "USGS 1 Meter", + "https://example.test/Projects/CA_Test/USGS_1M_tile.tif", + "2025-01-01", + "source", + ) + ], + } + seen = {} + + def add_source_coverage(entries, region, require_year=False): + seen["entries"] = entries + seen["region"] = region + seen["require_year"] = require_year + entries[0]["tnm_source_coverage"] = [{"geometry": "POINT (0 0)"}] + return entries + + monkeypatch.setattr( + tnm.WESM, + "add_source_coverage", + staticmethod(add_source_coverage), + ) + mod = tnm.TheNationalMap( + src_region=SAMPLE_REGION, + datasets="1m", + source_coverage=True, + use_cache=False, + ) + mod.run() + + assert seen["require_year"] is True + assert seen["region"] == SAMPLE_REGION + assert len(seen["entries"]) == 1 + assert mod.results[0]["tnm_source_coverage"] == [{"geometry": "POINT (0 0)"}] + + +def test_missing_download_url_fails_closed(): + FakeFetch.payload = {"total": 1, "items": [{"sourceId": "missing-url"}]} + mod = tnm.TheNationalMap(src_region=SAMPLE_REGION, datasets="1m", use_cache=False) + with pytest.raises(RuntimeError, match="download URL"): + mod.run() diff --git a/tests/modules/test_tnm_raster.py b/tests/modules/test_tnm_raster.py new file mode 100644 index 0000000..0d3b65d --- /dev/null +++ b/tests/modules/test_tnm_raster.py @@ -0,0 +1,45 @@ +import numpy as np +import pytest +import rasterio +import shapely +from rasterio.transform import from_origin + +from fetchez.modules import tnm_raster +from fetchez.spatial import Region + + +@pytest.fixture +def raster(monkeypatch, tmp_path): + path = tmp_path / "s1m.tif" + with rasterio.open( + path, + "w", + driver="GTiff", + width=10, + height=10, + count=1, + dtype="float32", + crs="EPSG:6350", + transform=from_origin(2240000, 2790000, 1000, 1000), + nodata=-9999, + ) as dst: + # Interior NoData must not create fallback holes. + dst.write(np.full((10, 10), -9999, dtype="float32"), 1) + monkeypatch.setattr(tnm_raster.core, "HttpFile", lambda *a, **kw: open(path, "rb")) + return {"url": "https://example.test/s1m.tif"} + + +def test_catalog_bbox_corner_is_not_raster_coverage(raster): + assert ( + tnm_raster.add_source_coverage([raster], Region(-67, -66.99, 44.90, 44.93)) + == [] + ) + + +def test_raster_extent_preserves_interior_nodata(raster): + selected = tnm_raster.add_source_coverage( + [raster], Region(-67.08, -67.07, 44.95, 44.96) + ) + assert len(selected) == 1 + coverage = shapely.from_wkt(selected[0]["tnm_source_coverage"][0]["geometry"]) + assert coverage.equals(shapely.box(-67.08, 44.95, -67.07, 44.96)) diff --git a/tests/modules/test_tnm_wesm.py b/tests/modules/test_tnm_wesm.py new file mode 100644 index 0000000..c48d41d --- /dev/null +++ b/tests/modules/test_tnm_wesm.py @@ -0,0 +1,268 @@ +import os + +import numpy as np +import pytest +import shapely +from pyproj import Transformer + +from fetchez import spatial +from fetchez.modules import tnm_wesm + + +REGION = spatial.Region(0, 1, 0, 1, srs="EPSG:4326") + + +def _entry(project="Test_Project"): + return { + "url": f"https://example.test/Projects/{project}/tile.tif", + "dst_fn": "tile.tif", + "bounds": (0, 1, 0, 1), + "tnm_project": project, + } + + +def _index_row(project="Test_Project"): + return { + "fid": 1, + "project": project, + "project_id": "10", + "workunit": project, + "workunit_id": "20", + "collect_start": "2020-01-01", + "collect_end": "2020-12-31", + "sourcedem_link": "", + "_tnm_project": project, + } + + +@pytest.fixture(autouse=True) +def reset_wesm(): + tnm_wesm.WESM.reset() + + +def test_legacy_one_ninth_filename_supplies_project_identity(): + entry = {"dst_fn": "ned19_n34x00_w118x00_CA_Orange_County_2016.zip"} + + assert tnm_wesm.project_name(entry) == "CA_Orange_County_2016" + + +def test_wesm_identity_accepts_omitted_cardinal_qualifier(): + aliases = {"CA_SanDiegoCo_2016": {"CA_SanDiegoCo_2016"}} + row = { + **_index_row("CA_Eastern_San_Diego_Co_Lidar_2016_B16"), + "workunit": "CA_E_SanDiegoCo_2016", + "sourcedem_link": ( + "https://prd-tnm.s3.amazonaws.com/index.html?prefix=" + "StagedProducts/Elevation/OPR/Projects/" + "CA_Eastern_San_Diego_Co_Lidar_2016_B16/CA_E_SanDiegoCo_2016" + ), + } + + assert tnm_wesm._row_project(row, aliases) == "CA_SanDiegoCo_2016" + + +def test_directionless_wesm_identity_remains_fail_closed_when_ambiguous(): + aliases = { + "CA_E_SanDiegoCo_2016": {"CA_E_SanDiegoCo_2016"}, + "CA_W_SanDiegoCo_2016": {"CA_W_SanDiegoCo_2016"}, + } + row = {**_index_row("CA_SanDiegoCo_2016"), "workunit": "CA_SanDiegoCo_2016"} + + with pytest.raises(RuntimeError, match="matches multiple TNM projects"): + tnm_wesm._row_project(row, aliases) + + +def test_wesm_index_snapshot_is_loaded_once(monkeypatch): + row = _index_row() + header = ",".join(tnm_wesm.WESM_FIELDS) + values = ",".join(str(row[field]) for field in tnm_wesm.WESM_FIELDS) + + class Response: + status_code = 200 + content = f"{header}\n{values}\n".encode() + + class Fetch: + calls = 0 + + def __init__(self, _url): + pass + + def fetch_req(self, **kwargs): + self.__class__.calls += 1 + return Response() + + monkeypatch.setattr(tnm_wesm.core, "Fetch", Fetch) + + first = tnm_wesm.WESM.index() + second = tnm_wesm.WESM.index() + + assert first is second + assert first[0]["fid"] == 1 + assert Fetch.calls == 1 + assert len(tnm_wesm.WESM.snapshot_sha256) == 64 + + +def test_wesm_index_request_failure_raises(monkeypatch): + class Response: + status_code = 500 + content = b"" + + class Fetch: + def __init__(self, _url): + pass + + def fetch_req(self, **kwargs): + return Response() + + monkeypatch.setattr(tnm_wesm.core, "Fetch", Fetch) + + with pytest.raises(RuntimeError, match="WESM CSV request failed: 500"): + tnm_wesm.WESM.index() + + +def test_missing_wesm_identity_fails_closed(monkeypatch): + monkeypatch.setattr( + tnm_wesm.WESM, + "matching_rows", + classmethod(lambda cls, aliases: []), + ) + + with pytest.raises(RuntimeError, match="no matching work-unit identity"): + tnm_wesm.WESM.add_source_coverage([_entry()], REGION, require_year=True) + + +def test_matching_identity_outside_query_is_a_valid_nonintersection(monkeypatch): + row = _index_row() + monkeypatch.setattr( + tnm_wesm.WESM, + "matching_rows", + classmethod(lambda cls, aliases: [row]), + ) + monkeypatch.setattr( + tnm_wesm.WESM, + "features", + classmethod( + lambda cls, fids=None, bbox=None: [ + {**row, "geometry": shapely.box(2, 2, 3, 3)} + ] + ), + ) + + assert tnm_wesm.WESM.add_source_coverage([_entry()], REGION) == [] + + +def test_matching_wesm_claim_is_attached_as_top_level_audit_data(monkeypatch): + row = _index_row() + tnm_wesm.WESM.snapshot_sha256 = "abc123" + tnm_wesm.WESM.snapshot_retrieved_at = "2026-09-06T00:00:00+00:00" + monkeypatch.setattr( + tnm_wesm.WESM, + "matching_rows", + classmethod(lambda cls, aliases: [row]), + ) + monkeypatch.setattr( + tnm_wesm.WESM, + "features", + classmethod( + lambda cls, fids=None, bbox=None: [ + {**row, "geometry": shapely.box(0.25, 0.25, 0.75, 0.75)} + ] + ), + ) + + selected = tnm_wesm.WESM.add_source_coverage([_entry()], REGION, require_year=True) + + assert len(selected) == 1 + assert selected[0]["tnm_source_coverage"][0]["year"] == 2020 + assert selected[0]["tnm_wesm_snapshot_sha256"] == "abc123" + assert "tnm_source_coverage" not in selected[0].get("metadata", {}) + + +def test_required_collection_year_fails_closed(monkeypatch): + row = {**_index_row(), "collect_start": "", "collect_end": ""} + monkeypatch.setattr( + tnm_wesm.WESM, + "matching_rows", + classmethod(lambda cls, aliases: [row]), + ) + monkeypatch.setattr( + tnm_wesm.WESM, + "features", + classmethod( + lambda cls, fids=None, bbox=None: [ + {**row, "geometry": shapely.box(0, 0, 1, 1)} + ] + ), + ) + + with pytest.raises(RuntimeError, match="no collection year"): + tnm_wesm.WESM.add_source_coverage([_entry()], REGION, require_year=True) + + +def test_feature_read_validates_csv_geopackage_identity(monkeypatch): + row = _index_row() + tnm_wesm.WESM._index = [row] + meta = {"fields": list(tnm_wesm.WESM_FIELDS), "crs": "EPSG:4326"} + fields = [ + np.array(["different" if field == "project" else row[field]]) + for field in tnm_wesm.WESM_FIELDS + ] + monkeypatch.setattr( + tnm_wesm.WESM, + "_read", + classmethod( + lambda cls, **kwargs: ( + meta, + np.array([1]), + np.array([shapely.to_wkb(shapely.box(0, 0, 1, 1))]), + fields, + ) + ), + ) + + with pytest.raises(RuntimeError, match="changed between the CSV index"): + tnm_wesm.WESM.features([1]) + + +@pytest.mark.parametrize("crs", ["EPSG:4326", "EPSG:3857"]) +def test_feature_read_transforms_source_geometry_to_wgs84(monkeypatch, crs): + row = _index_row() + tnm_wesm.WESM._index = [row] + expected = shapely.box(-67.0, 45.0, -66.9, 45.1) + transformer = Transformer.from_crs("EPSG:4326", crs, always_xy=True) + geometry = shapely.Polygon( + [transformer.transform(x, y) for x, y in expected.exterior.coords] + ) + meta = {"fields": list(tnm_wesm.WESM_FIELDS), "crs": crs} + fields = [np.array([row[field]]) for field in tnm_wesm.WESM_FIELDS] + monkeypatch.setattr( + tnm_wesm.WESM, + "_read", + classmethod( + lambda cls, **kwargs: ( + meta, + np.array([1]), + np.array([shapely.to_wkb(geometry)]), + fields, + ) + ), + ) + + selected = tnm_wesm.WESM.features([1]) + + assert len(selected) == 1 + assert selected[0]["fid"] == 1 + assert selected[0]["project"] == row["project"] + assert selected[0]["geometry"].equals_exact(expected, tolerance=1e-9) + + +def test_wesm_gdal_environment_is_restored(monkeypatch): + monkeypatch.setenv("AWS_NO_SIGN_REQUEST", "prior") + monkeypatch.delenv("GDAL_DISABLE_READDIR_ON_OPEN", raising=False) + + with tnm_wesm.WESM._gdal_env(): + assert os.environ["AWS_NO_SIGN_REQUEST"] == "YES" + assert os.environ["GDAL_DISABLE_READDIR_ON_OPEN"] == "EMPTY_DIR" + + assert os.environ["AWS_NO_SIGN_REQUEST"] == "prior" + assert "GDAL_DISABLE_READDIR_ON_OPEN" not in os.environ diff --git a/tests/test_http_file.py b/tests/test_http_file.py new file mode 100644 index 0000000..aedae94 --- /dev/null +++ b/tests/test_http_file.py @@ -0,0 +1,104 @@ +import io + +import pytest +import requests + +from fetchez.core import HttpFile + + +class Response: + def __init__(self, status, headers, data=b""): + self.status_code = status + self.headers = headers + self.data = io.BytesIO(data) + self.raw = self + self.reads = 0 + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.HTTPError(str(self.status_code)) + + def read(self, size, decode_content=False): + self.reads += 1 + return self.data.read(size) + + +class Session: + def __init__(self): + self.payload = b"0123456789" + self.head_response = Response(200, {"Content-Length": "10"}) + self.get_response = None + self.ranges = [] + + def head(self, url, **kwargs): + return self.head_response + + def get(self, url, headers, **kwargs): + assert kwargs["stream"] is True + start, end = map(int, headers["Range"].removeprefix("bytes=").split("-")) + self.ranges.append((start, end)) + return self.get_response or Response( + 206, + {"Content-Range": f"bytes {start}-{end}/10"}, + self.payload[start : end + 1], + ) + + +def test_random_access_and_eof_only_read_requested_ranges(): + session = Session() + sizes = [] + with HttpFile("https://example.test/archive.zip", session, sizes.append) as remote: + assert remote.read(3) == b"012" + remote.seek(-2, io.SEEK_END) + assert remote.read() == b"89" + assert remote.read() == b"" + remote.seek(2) + assert remote.read(0) == b"" + assert remote.read(2) == b"23" + assert remote.tell() == 4 + assert session.ranges == [(0, 2), (8, 9), (2, 3)] + assert sizes == [3, 2, 2] + + +@pytest.mark.parametrize( + ("response", "error"), + [(Response(500, {}), requests.HTTPError), (Response(200, {}), OSError)], +) +def test_failed_or_missing_file_size_raises(response, error): + session = Session() + session.head_response = response + with pytest.raises(error): + HttpFile("https://example.test/archive.zip", session) + + +@pytest.mark.parametrize( + "response", + [ + Response(200, {}, b"0123456789"), + Response(206, {"Content-Range": "bytes 0-2/11"}, b"012"), + Response(206, {"Content-Range": "bytes 1-3/10"}, b"123"), + ], +) +def test_ignored_or_inconsistent_range_is_rejected_before_body_read(response): + session = Session() + session.get_response = response + with HttpFile("https://example.test/archive.zip", session) as remote: + with pytest.raises(OSError, match="did not honor"): + remote.read(3) + assert response.reads == 0 + + +@pytest.mark.parametrize("body", [b"01", b"0123"]) +def test_truncated_or_oversized_range_is_rejected(body): + session = Session() + session.get_response = Response(206, {"Content-Range": "bytes 0-2/10"}, body) + with HttpFile("https://example.test/archive.zip", session) as remote: + with pytest.raises(OSError, match="incomplete byte range"): + remote.read(3) + assert remote.tell() == 0 diff --git a/tests/test_recipe.py b/tests/test_recipe.py index 76d497c..df0afe2 100644 --- a/tests/test_recipe.py +++ b/tests/test_recipe.py @@ -5,6 +5,8 @@ import pytest from unittest.mock import patch from fetchez.recipe import Recipe +from fetchez.registry import BundleRegistry +from fetchez.utils import compile_sources def test_recipe_initialization(): @@ -46,6 +48,22 @@ def test_get_module_signature(): assert sig_1 != sig_3 +def test_module_signature_distinguishes_single_dataset_filters(): + orange_county = { + "module": "ncei_thredds", + "args": {"dataset": "orange_county_13_navd88_2015"}, + } + santa_monica = { + "module": "ncei_thredds", + "args": {"dataset": "santa_monica_13_navd88_2010"}, + } + + orange_signature = BundleRegistry.get_module_signature(orange_county) + santa_monica_signature = BundleRegistry.get_module_signature(santa_monica) + + assert orange_signature != santa_monica_signature + + @patch("fetchez.registry.BundleRegistry.get_yaml") def test_expand_modules_recursive_and_deduplucate(mock_get_bundle): """Test that bundles expand recursively and parent definitions override child arguments.""" @@ -76,6 +94,55 @@ def test_expand_modules_recursive_and_deduplucate(mock_get_bundle): assert final_mod["hooks"][0]["name"] == "unzip" +@patch("fetchez.registry.BundleRegistry.get_yaml") +def test_bundle_products_select_children_in_bundle_order(mock_get_bundle): + mock_get_bundle.return_value = { + "products": ["fine", "medium", "coarse"], + "modules": [ + {"module": "tnm", "args": {"datasets": "fine"}}, + {"module": "tnm", "args": {"datasets": "medium"}}, + {"module": "tnm", "args": {"datasets": "coarse"}}, + ], + } + + expanded = BundleRegistry.expand_modules( + [{"bundle": "test", "args": {"products": "coarse,fine"}}] + ) + + assert [module["args"]["datasets"] for module in expanded] == [ + "fine", + "coarse", + ] + + +@patch("fetchez.registry.BundleRegistry.get_yaml") +def test_bundle_products_reject_unknown_product(mock_get_bundle): + mock_get_bundle.return_value = { + "products": ["fine"], + "modules": [{"module": "tnm", "args": {"datasets": "fine"}}], + } + + with pytest.raises(ValueError, match="Unknown product.*missing"): + BundleRegistry.expand_modules( + [{"bundle": "test", "args": {"products": "missing"}}] + ) + + +@patch("fetchez.registry.BundleRegistry.get_registry") +def test_compile_sources_preserves_bundle_arguments(mock_registry): + mock_registry.return_value = {"test-bundle": {}} + + compiled = compile_sources(["test-bundle:products=fine/coarse"]) + + assert compiled == [ + { + "bundle": "test-bundle", + "args": {"products": "fine/coarse"}, + "hooks": [], + } + ] + + def test_to_cli_translation(): """Verify the Recipe configuration correctly translates into a Fetchez CLI string.""" @@ -127,3 +194,24 @@ def test_to_json_translation(): def test_recipe_utils(): resolved_path = Recipe({})._resolve_path("./test") assert Path(resolved_path) == Path.cwd() / "test" + + +@patch("fetchez.registry.BundleRegistry.get_registry") +@patch("fetchez.registry.BundleRegistry.get_yaml") +def test_product_bundle_rejects_truncated_comma_source(get_yaml, get_registry): + get_registry.return_value = {"test-products": {}} + get_yaml.return_value = { + "products": ["s1m", "1m", "1_as"], + "modules": [ + {"module": "tnm", "args": {"datasets": product}} + for product in ["s1m", "1m", "1_as"] + ], + } + with pytest.raises(ValueError, match="Use products="): + Recipe({})._expand_modules( + compile_sources(["test-products:products=s1m,1m,1_as"]) + ) + modules = Recipe({})._expand_modules( + compile_sources(["test-products:products=s1m/1m/1_as"]) + ) + assert [module["args"]["datasets"] for module in modules] == ["s1m", "1m", "1_as"] diff --git a/tests/test_stream_init.py b/tests/test_stream_init.py new file mode 100644 index 0000000..4ff5c62 --- /dev/null +++ b/tests/test_stream_init.py @@ -0,0 +1,61 @@ +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from pyproj import CRS +from pyproj.crs import CompoundCRS + +from fetchez.hooks.stream_init import DataStream +from fetchez.spatial import Region + + +@pytest.mark.parametrize( + "source", + [ + "EPSG:4269", + CRS(4269).to_wkt("WKT1_GDAL"), + CRS(6350).to_wkt(), + CRS.from_proj4("+proj=longlat +datum=NAD83").to_wkt(), + ], +) +def test_stream_adds_vertical_reference(monkeypatch, source): + entry = initialize(monkeypatch, source) + crs = CRS(entry["src_srs"]) + assert crs.is_compound + assert crs.sub_crs_list[0].equals(CRS(source)) + assert crs.sub_crs_list[1].equals(CRS(5703)) + + +@pytest.mark.parametrize( + "source", + [ + CRS(4979).to_wkt(), + CompoundCRS("Albers + height", [CRS(6350), CRS(5703)]).to_wkt(), + ], +) +def test_stream_preserves_existing_vertical_reference(monkeypatch, source): + assert initialize(monkeypatch, source)["src_srs"] == source + + +def test_stream_preserves_custom_vertical_reference(monkeypatch): + assert ( + initialize(monkeypatch, "EPSG:4326", "global:mss")["src_srs"] + == "EPSG:4326+global:mss" + ) + + +def initialize(monkeypatch, source, vertical="EPSG:5703"): + from fetchez.hooks import stream_init + + reader = SimpleNamespace( + name="test", get_srs=lambda: source, yield_chunks=lambda: iter([1]) + ) + monkeypatch.setattr(stream_init.ReaderRegistry, "load_all", Mock()) + monkeypatch.setattr(stream_init.ProfileRegistry, "load_all", Mock()) + monkeypatch.setattr( + stream_init.ReaderRegistry, "get_reader", lambda *args, **kwargs: reader + ) + entry = {"dst_fn": "dem.tif"} + module = SimpleNamespace(region=Region(-67.001, -67, 44.9, 44.901)) + DataStream(vert_srs=vertical).run([(module, entry)]) + return entry From 867d7b778b15b5e80bd1f91a1bfc45804f378706 Mon Sep 17 00:00:00 2001 From: camante Date: Mon, 7 Sep 2026 17:01:35 -0400 Subject: [PATCH 2/2] Handle Rasterio in mypy checks --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index b89fff9..5a41039 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -129,6 +129,7 @@ module = [ "filelock", "pyogrio.*", "pyogrio.raw", + "rasterio", "pyproj", "sentinelsat", "shapefile",