Adding a New Source¶

from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field
from gathering.base import Gatherer
from gathering.grid import Grid, parse_grid
from gathering.integrity import verify_file_integrity
from gathering.logging import logger


class MyNetworkParameters(BaseModel):
    station: str = Field(description="Station identifier code")
    product: list[str] = Field(default_factory=lambda: ["default_prod"], description="Products to fetch")
    start_date: str | None = Field(default=None, description="Start date YYYY-MM-DD")
    end_date: str | None = Field(default=None, description="End date YYYY-MM-DD")
    grid: Grid | dict[str, Any] | None = Field(default=None, description="Spatial Region of Interest")
    max_workers: int = Field(default=1, ge=1, description="Concurrent download threads")


class MyNetworkGatherer(Gatherer):
    NAME = "my_network"
    ALIASES = ["mynet", "my_source"]
    PARAMS_MODEL = MyNetworkParameters

    def __init__(self, *, caching_sub_location: Path | str = Path("my_network"), **kwargs: Any):
        super().__init__(caching_sub_location=caching_sub_location, **kwargs)

    def _validate_parameters(self, **kwargs: Any) -> dict[str, Any]:
        """Validate input arguments against the Pydantic model."""
        kwargs.pop("overwrite_cache", None)
        return MyNetworkParameters(**kwargs).model_dump()

    def _set_caching_name(self, **kwargs: Any) -> None:
        """Define the local caching directory or filename."""
        self.caching_name = f"{kwargs['station']}_{kwargs.get('start_date', 'all')}"

    def _call_source_fetch(self, **kwargs: Any) -> list[str]:
        """Search, authenticate, download, and verify files. Return file paths."""
        params = MyNetworkParameters(**kwargs)
        # Download logic here (see sections below for Grid and Integrity details)...
        return []

from gathering.grid import Grid

class MySatelliteParameters(BaseModel):
    grid: Grid | dict[str, Any] | list[float] | None = Field(
        default=None,
        description="Spatial ROI defined as a Grid instance, site dictionary, or [min_lon, min_lat, max_lon, max_lat].",
    )
    bounding_box: list[float] | None = Field(
        default=None,
        description="Bounding box [min_lon, min_lat, max_lon, max_lat].",
    )

from gathering.grid import parse_grid, get_aeronet_site_coords

def _call_source_fetch(self, **kwargs: Any) -> list[str]:
    params = MySatelliteParameters(**kwargs)

    # 1. Parse Grid object (resolves AERONET sites, dicts, or bbox lists)
    roi_grid: Grid | None = parse_grid(params.grid) if params.grid else None

    # 2. Convert to STAC / OGC Bounding Box: [min_lon, min_lat, max_lon, max_lat]
    if roi_grid:
        stac_bbox = roi_grid.to_stac_bbox()
        # e.g. [-3.705, 37.064, -3.505, 37.264]

    # 3. Convert to NASA CMR Bounding Box: (min_lon, min_lat, max_lon, max_lat)
    if roi_grid:
        cmr_bbox = roi_grid.to_cmr_bounding_box()
        # Pass directly to cmr_base or search query: search_cmr_granules(..., bounding_box=cmr_bbox)

    # 4. Convert to WKT Polygon (for OData / Copernicus APIs)
    if roi_grid:
        wkt_polygon = roi_grid.to_wkt_polygon()
        # e.g. "POLYGON((-3.705 37.064, -3.505 37.064, ...))"

    # 5. Direct AERONET Site Coordinate Lookup (if needed)
    lat, lon, elev = get_aeronet_site_coords("Granada")
    # Returns (37.164, -3.605, 680.0)

import requests
from gathering.integrity import verify_file_integrity
from gathering.logging import logger

def _download_and_verify(self, url: str, dest_path: Path, expected_md5: str | None = None) -> Path | None:
    dest_path.parent.mkdir(parents=True, exist_ok=True)

    # 1. Check existing cache
    if dest_path.exists() and not self.overwrite_cache:
        valid, reason = verify_file_integrity(dest_path, expected_checksum=expected_md5)
        if valid:
            logger.debug(f"Cache hit: {dest_path.name}")
            return dest_path
        logger.warning(f"Purging corrupted cached file ({reason}): {dest_path.name}")
        dest_path.unlink(missing_ok=True)

    # 2. Download to temporary file
    temp_path = dest_path.with_suffix(dest_path.suffix + ".tmp")
    try:
        with requests.get(url, stream=True, timeout=60) as resp:
            resp.raise_for_status()
            with open(temp_path, "wb") as f:
                for chunk in resp.iter_content(chunk_size=65536):
                    if chunk:
                        f.write(chunk)

        # 3. Verify downloaded file integrity (magic bytes, size, HTML errors, checksum)
        is_valid, error_msg = verify_file_integrity(
            temp_path,
            min_size_bytes=256,
            check_archive=True,
            expected_checksum=expected_md5,
            checksum_algo="md5",
        )

        if not is_valid:
            logger.error(f"Download verification failed for {dest_path.name}: {error_msg}")
            temp_path.unlink(missing_ok=True)
            return None

        # 4. Atomic move to final cache location
        temp_path.replace(dest_path)
        return dest_path

    except Exception as exc:
        logger.error(f"Download failed for {dest_path.name}: {exc}")
        if temp_path.exists():
            temp_path.unlink(missing_ok=True)
        return None

from concurrent.futures import ThreadPoolExecutor, as_completed

def _call_source_fetch(self, **kwargs: Any) -> list[str]:
    params = MyNetworkParameters(**kwargs)
    items_to_download = self._search_remote_items(params)

    if params.max_workers <= 1 or len(items_to_download) == 1:
        # Sequential execution
        return [self._download_item(it, params) for it in items_to_download]

    # Parallel execution
    workers = min(params.max_workers, len(items_to_download))
    downloaded: list[str] = []

    with ThreadPoolExecutor(max_workers=workers) as executor:
        futures = {executor.submit(self._download_item, item, params): item for item in items_to_download}
        for future in as_completed(futures):
            res = future.result()
            if res:
                downloaded.append(str(res))

    return downloaded