CWL CommandLineTool for stage-in

Goal

Use a stage-in CWL workflow to stage a Landsat-9 acquisition

Lab

This step has a dedicated lab at practice-labs/4-CWL-Workflows/2.1-stage-in.ipynb.

Step 1 - Create a stage-in CWL Workflow

Below a stage-in.cwl CWL (Common Workflow Language) document for a command-line tool that executes a Python script.

The document invokes the installed stage-in --reference <item> command in localhost/stage-in:latest. The reference remains a string, so existing job documents continue to work.

cwlVersion: v1.2
$namespaces:
  s: https://schema.org/
# Training-example authors and creation date mirror codemeta.json.
s:name: Stage-in STAC assets
s:description: Download the assets of a STAC item for local processing.
s:dateCreated: '2022-09-01'
s:license:
  s:name: Creative Commons Attribution-ShareAlike 4.0 International
  s:url: https://creativecommons.org/licenses/by-sa/4.0/
  s:identifier: CC-BY-SA-4.0
s:softwareHelp:
  s:name: Mastering Earth Observation Application Packaging with CWL
  s:url: https://eoap.github.io/mastering-app-package
s:publisher:
  s:name: EO Application Packaging
  s:identifier: https://github.com/eoap
s:author:
  - s:givenName: Jane
    s:familyName: Doe
    s:email: jane.doe@acme.earth
    s:affiliation:
      s:name: ACME
  - s:givenName: John
    s:familyName: Doe
    s:email: john.doe@acme.earth
    s:affiliation:
      s:name: ACME
s:softwareRequirements:
  - container runtime
  - cwl runner
s:softwareVersion: 2.0.2

class: CommandLineTool
id: main
inputs:
  reference:
    type: string
    label: STAC item reference
    doc: URI of the STAC item whose assets are downloaded.
    inputBinding:
      prefix: --reference
  http_timeout:
    type: int
    default: 300
    label: HTTP transfer timeout
    doc: Timeout in seconds for each HTTP transfer.
    inputBinding:
      prefix: --http-timeout
  asset_keys:
    type:
      type: array
      items: string
      inputBinding:
        prefix: --asset
    default: []
    label: Asset keys
    doc: Download only these STAC asset keys; an empty list downloads all assets.
outputs:
  staged:
    type: Directory
    outputBinding:
      glob: .
    label: Staged acquisition
    doc: Directory containing the staged STAC catalog, item, and downloaded assets.
baseCommand: stage-in
requirements:
  DockerRequirement:
    dockerPull: localhost/stage-in:latest
  InlineJavascriptRequirement: {}
  NetworkAccess:
    networkAccess: true
  EnvVarRequirement:
    envDef:
      PATH: /app/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin

The Python package reads a STAC Item, downloads its assets with stac-asset, and writes a self-contained catalog.json with relative asset links. It keeps the working directory unchanged and validates that the item ID is a directory name. HTTP downloads default to a 300-second timeout and three attempts. Set STAGE_IN_HTTP_TIMEOUT and STAGE_IN_HTTP_MAX_ATTEMPTS to configure transfers; --http-timeout overrides the timeout environment variable. Failed asset downloads fail the command before publishing a catalog. Progress and errors are logged to stderr.

"""Download a STAC item and package its assets in a local catalog."""

import asyncio
import os
from pathlib import Path

import click
import pystac
import stac_asset
from loguru import logger
from stac_asset.errors import AssetOverwriteError, DownloadError


def _positive_env_int(name: str, default: int) -> int:
    """Read a positive integer setting, reporting invalid values clearly."""
    raw = os.environ.get(name, str(default))
    try:
        value = int(raw)
    except ValueError as exc:
        raise ValueError(f"{name} must be a positive integer, got {raw!r}") from exc
    if value <= 0:
        raise ValueError(f"{name} must be a positive integer, got {raw!r}")
    return value


async def stage_item(reference: str, *, http_timeout: float | None = None, asset_keys: tuple[str, ...] = ()) -> pystac.Catalog:
    """Stage an item's assets under its ID and write catalog.json."""
    # Timeout is in seconds; attempts includes the initial download attempt.
    timeout = http_timeout if http_timeout is not None else _positive_env_int("STAGE_IN_HTTP_TIMEOUT", 300)
    attempts = _positive_env_int("STAGE_IN_HTTP_MAX_ATTEMPTS", 3)
    config = stac_asset.Config(
        http_client_timeout=timeout,
        http_max_attempts=attempts,
    )
    logger.info("Stage-in HTTP timeout: {} seconds; max attempts: {}", timeout, attempts)
    logger.info("Reading input STAC item")
    item = pystac.read_file(reference)
    if not isinstance(item, pystac.Item):
        raise ValueError("Reference must point to a STAC item")
    if asset_keys:
        missing = set(asset_keys) - item.assets.keys()
        if missing: raise ValueError(f"Unknown STAC asset keys: {sorted(missing)}")
        item.assets = {key: item.assets[key] for key in asset_keys}
    directory = Path(item.id)
    if directory.name != item.id or item.id in {"", ".", ".."}:
        raise ValueError("STAC item id must be a single directory name")
    directory.mkdir(exist_ok=True)
    logger.info("Downloading assets for {} into {}", item.id, directory.resolve())
    item = await stac_asset.download_item(
        item=item, directory=directory.resolve(), config=config
    )
    catalog = pystac.Catalog(
        id="catalog",
        description=f"catalog with staged {item.id}",
        title=f"catalog with staged {item.id}",
    )
    catalog.add_item(item)
    catalog.normalize_hrefs("./")
    catalog.save(catalog_type=pystac.CatalogType.SELF_CONTAINED)
    logger.success("Wrote catalog.json for {}", item.id)
    return catalog


def execute(*, reference: str, http_timeout: float | None = None, asset_keys: tuple[str, ...] = ()) -> None:
    """Synchronous callback for the stage-in console command."""
    try:
        asyncio.run(stage_item(reference, http_timeout=http_timeout, asset_keys=asset_keys))
    except (TimeoutError, ValueError, OSError, pystac.STACError, DownloadError, AssetOverwriteError) as exc:
        logger.error("Stage-in failed: {}", exc)
        raise click.ClickException(str(exc)) from exc

Step 2 - Create a container for the stage-in

The container image is built with:

export WORKSPACE=/workspace/mastering-app-package

podman \
    build \
    --format docker \
    -t localhost/stage:latest \
    ${WORKSPACE}/water-bodies/command-line-tools/stage

# Current checkout uses separate installed staging commands.
podman build --format docker -t localhost/stage-in:latest ${WORKSPACE}/water-bodies/command-line-tools/stage-in
podman build --format docker -t localhost/stage-out:latest ${WORKSPACE}/water-bodies/command-line-tools/stage-out

Step 3 - Stage the Landsat-9 acquisition

Now the Landsat-9 acquisition "https://planetarycomputer.microsoft.com/api/stac/v1/collections/landsat-c2-l2/items/LC09_L2SP_042033_20231015_02_T1" is staged with:

ls9_ref="https://planetarycomputer.microsoft.com/api/stac/v1/collections/landsat-c2-l2/items/LC09_L2SP_042033_20231015_02_T1"

cwltool \
    --podman \
    --outdir ${WORKSPACE}/runs \
    ${WORKSPACE}/cwl-cli/stage-in.cwl \
    --reference \
    ${ls9_ref} > staged.json

Step 4 - Check the path to the stage Landsat-9 acquisition

The result is redirected to a file named staged.json as we use jq to get the path of the staged product:

terminal

cat staged.json | jq -r .staged.path

This returns a path like /workspace/mastering-app-package/runs/921x91vw