Codex Relay: zyra-project/zyra PR #226 — Add THREDDS catalog connector to the acquire stage

samples/pipelines/thredds_to_local.yaml

@@ -0,0 +1,41 @@
+# SPDX-License-Identifier: Apache-2.0
+name: THREDDS → Transform → Video → Local
+stages:
+  # 1) Acquire frames by syncing a THREDDS catalog to local.
+  #    Datasets are enumerated from catalog.xml and downloaded via the
+  #    HTTPServer (fileServer) service. Self-updating: existing non-empty
+  #    files are skipped on re-runs.
+  - stage: acquire
+    command: thredds
+    args:
+      catalog_url: https://gsl.noaa.gov/thredds/catalog/fv3-chem-0p25deg-grib2/catalog.xml
+      sync_dir: ./frames
+      pattern: "\\.grib2$"
+      since_period: "P1D"
+      # Follow nested catalogRef entries (optional)
+      recursive: false
+      max_depth: 3
+
+  # 2) Compute frames metadata (JSON)
+  - stage: transform
+    command: metadata
+    args:
+      frames_dir: ./frames
+      datetime_format: "%Y%m%d"
+      period_seconds: 3600
+      output: frames_meta.json
+
+  # 3) Compose frames into a video
+  - stage: visualize
+    command: compose-video
+    args:
+      frames: ./frames
+      output: video.mp4
+      fps: 24
+
+  # 4) Copy outputs to local filesystem (no network required)
+  - stage: export
+    command: local
+    args:
+      input: video.mp4
+      path: /tmp/video.mp4

src/zyra/api/models/domain_api.py

@@ -180,3 +180,8 @@ class AcquireFtpRun(DomainRunRequest):
 class AcquireApiRun(DomainRunRequest):
     tool: Literal["api"]
     args: da.AcquireApiArgs  # type: ignore[assignment]
+
+
+class AcquireThreddsRun(DomainRunRequest):
+    tool: Literal["thredds"]
+    args: da.AcquireThreddsArgs  # type: ignore[assignment]

src/zyra/api/routers/domain_acquire.py

@@ -25,6 +25,7 @@
     AcquireFtpRun,
     AcquireHttpRun,
     AcquireS3Run,
+    AcquireThreddsRun,
     DomainRunResponse,
 )
 from zyra.api.routers.cli import get_cli_matrix, run_cli_endpoint
@@ -37,7 +38,13 @@
 
 
 AcquireRequest = Annotated[
-    Union[AcquireHttpRun, AcquireS3Run, AcquireFtpRun, AcquireApiRun],
+    Union[
+        AcquireHttpRun,
+        AcquireS3Run,
+        AcquireFtpRun,
+        AcquireApiRun,
+        AcquireThreddsRun,
+    ],
     Body(discriminator="tool"),
 ]
 

src/zyra/api/schemas/domain_args.py

@@ -186,6 +186,30 @@ def _require_path_or_listing(self):  # type: ignore[override]
         return self
 
 
+class AcquireThreddsArgs(BaseModel):
+    catalog_url: str
+    output: str | None = None
+    # Listing/sync/batch
+    list_mode: bool | None = Field(default=None, alias="list")
+    sync_dir: str | None = None
+    output_dir: str | None = None
+    # Enumeration
+    recursive: bool | None = None
+    max_depth: int | None = None
+    pattern: str | None = None
+    since: str | None = None
+    since_period: str | None = None
+    until: str | None = None
+    date_format: str | None = None
+    header: list[str] | None = None
+    # Sync mode options (subset meaningful over HTTP)
+    overwrite_existing: bool | None = None
+    recheck_existing: bool | None = None
+    min_remote_size: str | int | None = None
+    prefer_remote: bool | None = None
+    skip_if_local_done: bool | None = None
+
+
 # New: acquire api (generic REST)
 class AcquireApiArgs(BaseModel):
     url: str | None = None
@@ -303,6 +327,8 @@ def normalize_and_validate(stage: str, tool: str, args: dict) -> dict:
         out = _normalize_credentials(out)
     elif stage == "acquire" and tool == "ftp":
         out = _normalize_credentials(out)
+    elif stage == "acquire" and tool == "thredds":
+        out = _normalize_headers(out)
     elif stage == "decimate" and tool == "post":
         out = _normalize_headers(out)
         out = _normalize_credentials(out)
@@ -651,11 +677,15 @@ def resolve_model(stage: str, tool: str) -> type[BaseModel] | None:
         return AcquireHttpArgs
     if key == ("acquire", "api"):
         return AcquireApiArgs
+    if key == ("acquire", "thredds"):
+        return AcquireThreddsArgs
     # Aliases for acquire
     if key == ("import", "http"):
         return AcquireHttpArgs
     if key == ("import", "api"):
         return AcquireApiArgs
+    if key == ("import", "thredds"):
+        return AcquireThreddsArgs
     if key == ("process", "convert-format"):
         return ProcessConvertFormatArgs
     if key == ("process", "decode-grib2"):