Product bundles

packages/gnss-product-management/src/gnss_product_management/client/gnss_client.py

@@ -281,6 +281,9 @@ def resolve_dependencies(
         date: datetime.datetime,
         *,
         sink_id: str,
+        force_download: bool = False,
+        centers: list[str] | None = None,
+        bundle_centers: list[str] | None = None,
     ) -> tuple[DependencyResolution, AnyPath | None]:
         """Resolve all dependencies in a spec for the given date.
 
@@ -297,6 +300,8 @@ def resolve_dependencies(
             date: Target date (timezone-aware datetime, midnight UTC).
             sink_id: Local resource alias for storing resolved files
                 (e.g. ``"local"``).
+            centers: Optional remote resource IDs to search.  When omitted,
+                all configured centers are eligible.
 
         Returns:
             A ``(DependencyResolution, lockfile_path)`` tuple.
@@ -312,4 +317,11 @@ def resolve_dependencies(
             workspace=self._workspace,
             transport=self._transport,
         )
-        return pipeline.run(dep_spec, date, sink_id=sink_id)
+        return pipeline.run(
+            dep_spec,
+            date,
+            sink_id=sink_id,
+            force_download=force_download,
+            centers=centers,
+            bundle_centers=bundle_centers,
+        )

packages/gnss-product-management/src/gnss_product_management/client/product_query.py

@@ -24,6 +24,12 @@
 logger = logging.getLogger(__name__)
 
 
+def _remote_uri(protocol: str, hostname: str, directory: str, filename: str) -> str:
+    """Build a remote URI without duplicating schemes or path separators."""
+    base = hostname if "://" in hostname else f"{protocol}://{hostname}"
+    return f"{base.rstrip('/')}/{directory.strip('/')}/{filename}"
+
+
 class ProductQuery:
     """Fluent builder for constructing and executing a GNSS product search.
 
@@ -342,6 +348,13 @@ def search(self) -> list[FoundResource]:
                 continue
             seen[key] = True
             params = {p.name: p.value for p in rq.product.parameters if p.value is not None}
+            # Search targets may retain wildcard regexes for unconstrained
+            # filename axes (for example TTT/PPP on Wuhan ORBIT).  Replace
+            # those patterns with metadata parsed from the actual match so
+            # callers can group candidates by their real product family.
+            classified = self._search_planner._product_registry.classify(filename)
+            if classified and classified.get("product") == rq.product.name:
+                params.update(classified.get("parameters", {}))
             protocol = (rq.server.protocol or "").upper()
             is_local = protocol in ("FILE", "LOCAL")
             if is_local:
@@ -352,8 +365,11 @@ def search(self) -> list[FoundResource]:
                 )
             else:
                 proto = (rq.server.protocol or "ftp").lower()
-                uri = (
-                    f"{proto}://{hostname}/{rq.directory.value or rq.directory.pattern}/{filename}"  # type: ignore[union-attr]
+                uri = _remote_uri(
+                    proto,
+                    hostname,
+                    rq.directory.value or rq.directory.pattern,  # type: ignore[union-attr]
+                    filename,
                 )
             r = FoundResource(
                 product=rq.product.name,

packages/gnss-product-management/src/gnss_product_management/factories/connection_pool.py

@@ -2,11 +2,16 @@
 
 import logging
 import os
+import random
+import re
+import shutil
 import threading
+import time
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Optional
-from urllib.parse import urlparse
+from urllib.parse import quote, urlparse
+from urllib.request import Request, urlopen
 
 import fsspec
 import fsspec.utils
@@ -23,16 +28,24 @@ class ConnectionPool:
         max_connections: Maximum number of concurrent connections.
     """
 
-    def __init__(self, hostname: str, max_connections: int = 4):
+    def __init__(
+        self,
+        hostname: str,
+        max_connections: int = 4,
+        listing_url: str | None = None,
+    ):
         """Initialise a connection pool for *hostname*.
 
         Args:
             hostname: Server address or local path.
             max_connections: Maximum number of concurrent connections.
+            listing_url: Optional URL template for a separate HTML directory
+                listing. ``{directory}`` is replaced with the quoted path.
         """
         self.hostname = hostname
         self.protocol = fsspec.utils.get_protocol(hostname) or "file"
         self.max_connections = max_connections
+        self.listing_url = listing_url
         self._pool: list[fsspec.AbstractFileSystem] = []
         self._semaphore: threading.Semaphore | None = None
         self._pool_lock = threading.Lock()
@@ -193,15 +206,22 @@ def __init__(self, max_connections: int = 4):
         self._listing_cache: dict[str, list[str]] = {}
         self._listing_cache_lock = threading.Lock()
 
-    def add_connection(self, hostname: str):
+    def add_connection(self, hostname: str, *, listing_url: str | None = None):
         """Ensure a connection pool exists for *hostname*.
 
         Args:
             hostname: Server address to pool.
+            listing_url: Optional separate HTML directory-listing URL template.
         """
         with self._factory_lock:
             if hostname not in self._pools:
-                self._pools[hostname] = ConnectionPool(hostname, self.max_connections)
+                self._pools[hostname] = ConnectionPool(
+                    hostname,
+                    self.max_connections,
+                    listing_url=listing_url,
+                )
+            elif self._pools[hostname].listing_url != listing_url:
+                raise ValueError(f"Conflicting listing configuration for: {hostname}")
 
     @contextmanager
     def get_connection(self, hostname: str):
@@ -249,6 +269,14 @@ def list_directory(self, hostname: str, directory: str) -> list[str]:
         full_path = pool.full_path(directory)
 
         def _ls(conn: "fsspec.AbstractFileSystem") -> list[str]:
+            if pool.listing_url:
+                quoted_directory = quote(directory.strip("/"), safe="/")
+                listing_url = pool.listing_url.format(directory=quoted_directory)
+                html = conn.cat_file(listing_url)
+                if isinstance(html, bytes):
+                    html = html.decode("utf-8", errors="replace")
+                download_prefix = re.escape(f"{hostname.rstrip('/')}/{directory.strip('/')}/")
+                return sorted(set(re.findall(rf'href="{download_prefix}([^"/]+)"', html)))
             raw = conn.ls(full_path, detail=False)
             return [Path(p).name for p in raw]
 
@@ -335,6 +363,30 @@ def download_file(self, hostname: str, remote_path: str, target_dir: str) -> Pat
         filename = Path(remote_path).name
         local_path = Path(target_dir) / filename
 
+        # GithubFileSystem's Contents API path can return only the first 4 MiB
+        # of larger blobs.  Use GitHub's immutable raw-content endpoint for
+        # the actual transfer while retaining GithubFileSystem for listings.
+        if pool.protocol == "github":
+            match = re.fullmatch(r"github://([^:]+):([^@]+)@(.+)", hostname)
+            if match:
+                owner, repository, revision = match.groups()
+                raw_url = (
+                    f"https://raw.githubusercontent.com/{owner}/{repository}/"
+                    f"{revision}/{remote_path.lstrip('/')}"
+                )
+                try:
+                    request = Request(raw_url, headers={"Accept-Encoding": "identity"})
+                    with (
+                        urlopen(request, timeout=60) as source,
+                        local_path.open("wb") as destination,
+                    ):
+                        shutil.copyfileobj(source, destination, length=1024 * 1024)
+                    return local_path if local_path.stat().st_size > 0 else None
+                except Exception as exc:
+                    logger.warning("Raw GitHub download failed for %s: %s", raw_url, exc)
+                    local_path.unlink(missing_ok=True)
+                    return None
+
         def _get(conn: "fsspec.AbstractFileSystem") -> Path | None:
             conn.get(full_path, str(local_path))
             if local_path.exists() and local_path.stat().st_size > 0:
@@ -348,6 +400,8 @@ def _get(conn: "fsspec.AbstractFileSystem") -> Path | None:
                 return _get(conn)
             except (BrokenPipeError, ConnectionError, EOFError, OSError) as e:
                 logger.debug("Stale connection for %s, reconnecting: %s", hostname, e)
+                if pool.protocol != "file":
+                    time.sleep(random.uniform(1.0, 3.0))
                 fresh = pool.replace_connection(conn)
                 if fresh is None:
                     return None

packages/gnss-product-management/src/gnss_product_management/factories/pipelines/download.py

@@ -61,6 +61,7 @@ def run(
         date: datetime.datetime,
         *,
         sink_id: str = "local_config",
+        force: bool = False,
     ) -> Path | None | list[Path | None]:
         """Download found resources to the workspace.
 
@@ -79,7 +80,7 @@ def run(
 
         paths: list[Path | None] = []
         for r in resources:
-            path = self._download_one(r, date, sink_id)
+            path = self._download_one(r, date, sink_id, force=force)
             paths.append(path)
 
         if single:
@@ -91,6 +92,8 @@ def _download_one(
         resource: FoundResource,
         date: datetime.datetime,
         sink_id: str,
+        *,
+        force: bool = False,
     ) -> Path | None:
         """Download a single resource and write its sidecar lockfile.
 
@@ -102,7 +105,7 @@ def _download_one(
         Returns:
             Path to the resolved file, or ``None`` on failure.
         """
-        if resource.is_local:
+        if resource.is_local and not force:
             local_path = resource.path
             if local_path and local_path.exists():
                 logger.debug("Already local: %s", local_path)
@@ -121,9 +124,10 @@ def _download_one(
             local_resource_id=sink_id,
             local_factory=self._planner._workspace,
             date=date,
+            force=force,
         )
         if path is not None:
-            if get_lock_product(path) is None:
+            if force or get_lock_product(path) is None:
                 lock = build_lock_product(sink=path, url=resource.uri, name=resource.product)
                 write_lock_product(lock)
             logger.info("Downloaded %s → %s", resource.product, path)