feat: Fmask as standalone pipeline

src/hls-workflows/src/hls_workflows/cli.py

@@ -254,5 +254,43 @@ def lasrc(
     )
 
 
+@cli.command("fmask")
+@click.option(
+    "--granule",
+    envvar="GRANULE",
+    required=True,
+    help="Sentinel-2 or Landsat granule ID, auto-detected (env: GRANULE).",
+)
+@click.option(
+    "--local-granule",
+    envvar="LOCAL_GRANULE",
+    default=None,
+    type=click.Path(path_type=Path),
+    help="Pre-downloaded granule (.zip for S2, dir for Landsat) (env: LOCAL_GRANULE).",
+)
+@_cleanup_option
+def fmask(
+    granule: str,
+    local_granule: Path | None,
+    cleanup_working_dir: bool,
+) -> None:
+    """Run the standalone Fmask pipeline (Download -> Fmask -> Upload).
+
+    Debug pipeline for testing Fmask performance in isolation. Uploads the
+    Fmask cloud mask to DEBUG_BUCKET. Does not skip granules with an invalid
+    solar zenith angle.
+    """
+    from hls_workflows.fmask import construct_pipeline
+
+    _run(
+        construct_pipeline(
+            granule_id=granule,
+            local_granule=local_granule,
+            cleanup_working_dir=cleanup_working_dir,
+        ),
+        "fmask-workflow",
+    )
+
+
 if __name__ == "__main__":
     cli()

src/hls-workflows/src/hls_workflows/fmask/__init__.py

@@ -0,0 +1,11 @@
+"""Standalone ``fmask`` pipeline for Fmask performance testing.
+
+A debug-oriented pipeline that runs only the Fmask step (Download -> Fmask ->
+Upload) so changes to Fmask can be measured in isolation from the rest of the
+production workflow. Unlike production, it does not skip granules with an
+invalid solar zenith angle.
+"""
+
+from .workflow import construct_pipeline
+
+__all__ = ["construct_pipeline"]

src/hls-workflows/src/hls_workflows/fmask/landsat.py

@@ -0,0 +1,65 @@
+"""Landsat tasks specific to the standalone ``fmask`` pipeline.
+
+Kept out of the production ``landsat_ac`` package so the main L30-AC pipeline
+stays clean.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+import boto3
+
+from hls_workflows.base import AssetBundle, Task
+from hls_workflows.fmask import products
+from hls_workflows.landsat_ac.assets import (
+    CONFIG,
+    FMASK_BIN,
+    UPLOAD_COMPLETE,
+    EnvConfig,
+)
+
+if TYPE_CHECKING:
+    from mypy_boto3_s3 import S3Client
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class UploadFmaskDebug(Task):
+    """Upload the Fmask cloud mask to the debug bucket.
+
+    Uploads only the ``*_UPL.tif`` mask, flattened into a single
+    ``<prefix>/<fmask version>/<granule_id>/`` directory (see
+    ``products.debug_upload_base``). Depends on the Fmask ENVI binary purely to
+    order after Fmask.
+
+    No-ops (with a warning) when ``DEBUG_BUCKET`` is unset.
+    """
+
+    prefix: str
+
+    requires = (CONFIG, FMASK_BIN)
+    provides = (UPLOAD_COMPLETE,)
+
+    def run(self, inputs: AssetBundle) -> AssetBundle:
+        config: EnvConfig = inputs[CONFIG]
+
+        if not config.debug_bucket:
+            logger.warning("DEBUG_BUCKET not set; skipping Fmask debug upload")
+            return {UPLOAD_COMPLETE: True}
+
+        s3: S3Client = boto3.client("s3")
+        base = products.debug_upload_base(
+            config.debug_bucket, self.prefix, config.granule
+        )
+        logger.info(f"Uploading Fmask debug files to {base}")
+
+        for f in config.granule_dir.rglob("*"):
+            if f.is_file() and products.is_fmask_product(f.name):
+                dest = base / f.name
+                s3.upload_file(str(f), dest.bucket, dest.key)
+
+        return {UPLOAD_COMPLETE: True}

src/hls-workflows/src/hls_workflows/fmask/products.py

@@ -0,0 +1,36 @@
+"""Helpers for the standalone ``fmask`` pipeline's outputs.
+
+Kept free of imports from the rest of the ``fmask`` package so both
+``fmask.sentinel`` and ``fmask.landsat`` can import it without an import cycle
+through ``fmask.workflow``.
+"""
+
+from __future__ import annotations
+
+from importlib.metadata import version
+
+from hls_workflows.common import S3Path
+
+# Both the Landsat and Sentinel-2 ``RunFmaskV5`` tasks run the "UPL" model, and
+# Fmask names its mask "<scene>_UPL.tif". The ENVI ``fmask.bin`` is a plain
+# gdal_translate copy of it, so only the TIF is uploaded.
+FMASK_PRODUCT_SUFFIX = "_UPL.tif"
+
+
+def is_fmask_product(name: str) -> bool:
+    """True if ``name`` is the Fmask cloud mask GeoTIFF."""
+    return name.endswith(FMASK_PRODUCT_SUFFIX)
+
+
+def fmask_version() -> str:
+    """Installed ``fmask`` Python package version (PEP 440 normalized)."""
+    return version("fmask")
+
+
+def debug_upload_base(bucket: str, prefix: str, granule_id: str) -> S3Path:
+    """Debug-bucket location for one granule's Fmask outputs.
+
+    Keyed by the installed Fmask version so runs of different Fmask builds do
+    not overwrite each other: ``<prefix>/<fmask version>/<granule_id>/``.
+    """
+    return S3Path(bucket, f"{prefix}/{fmask_version()}/{granule_id}")

src/hls-workflows/src/hls_workflows/fmask/sentinel.py

@@ -0,0 +1,78 @@
+"""Sentinel-2 tasks specific to the standalone ``fmask`` pipeline.
+
+Kept out of the production ``sentinel`` package so the main S30 pipeline stays
+clean.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+import boto3
+
+from hls_workflows.base import AssetBundle, MappedTask
+from hls_workflows.fmask import products
+from hls_workflows.sentinel.assets import (
+    CONFIG,
+    UPLOAD_COMPLETE,
+    EnvConfig,
+    fmask_bin_asset,
+    granule_dir_asset,
+)
+from hls_workflows.sentinel.mapped_tasks import ApplyQualityMask
+
+if TYPE_CHECKING:
+    from mypy_boto3_s3 import S3Client
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True, kw_only=True)
+class ApplyQualityMaskNoSolar(ApplyQualityMask):
+    """ApplyQualityMask without the solar zenith check dependency.
+
+    The standalone Fmask pipeline runs Fmask regardless of the solar zenith
+    angle, so it never runs ``CheckSolarZenith``.
+    """
+
+    requires_factory = lambda gid: (granule_dir_asset(gid),)
+
+
+@dataclass(frozen=True, kw_only=True)
+class UploadFmaskDebug(MappedTask):
+    """Upload the Fmask cloud mask to the debug bucket.
+
+    Uploads only the ``*_UPL.tif`` mask, flattened into a single
+    ``<prefix>/<fmask version>/<granule_id>/`` directory (see
+    ``products.debug_upload_base``). Depends on the Fmask ENVI binary purely to
+    order after Fmask.
+
+    No-ops (with a warning) when ``DEBUG_BUCKET`` is unset.
+    """
+
+    prefix: str
+
+    requires_factory = lambda gid: (CONFIG, fmask_bin_asset(gid))
+    provides = (UPLOAD_COMPLETE,)
+
+    def run(self, bundle: AssetBundle) -> AssetBundle:
+        config: EnvConfig = bundle[CONFIG]
+
+        if not config.debug_bucket:
+            logger.warning("DEBUG_BUCKET not set; skipping Fmask debug upload")
+            return {UPLOAD_COMPLETE: True}
+
+        s3: S3Client = boto3.client("s3")
+        base = products.debug_upload_base(
+            config.debug_bucket, self.prefix, self.granule_id
+        )
+        logger.info(f"Uploading Fmask debug files to {base}")
+
+        for f in (config.working_dir / self.granule_id).rglob("*"):
+            if f.is_file() and products.is_fmask_product(f.name):
+                dest = base / f.name
+                s3.upload_file(str(f), dest.bucket, dest.key)
+
+        return {UPLOAD_COMPLETE: True}

src/hls-workflows/src/hls_workflows/fmask/workflow.py

@@ -0,0 +1,131 @@
+"""Construct the standalone ``fmask`` pipeline.
+
+One unified ``construct_pipeline`` detects whether the granule is Sentinel-2 or
+Landsat and assembles the minimal chain needed to reach Fmask, reusing the
+existing per-mission tasks:
+
+- Sentinel-2: ``Download -> GetInnerDir -> ApplyMask -> Fmask -> Upload``. The
+  ESA quality mask is applied first because it changes the radiances Fmask
+  reads in production.
+- Landsat: ``Download -> Fmask -> Upload``.
+
+Neither path runs the solar zenith check, so Fmask runs on granules that the
+production pipelines would skip.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+
+from hls_workflows.granules import LandsatGranule, Sentinel2Granule, resolve_granule
+from hls_workflows.landsat_ac import tasks as l_tasks
+from hls_workflows.metrics import MetricsCollector, MetricSink
+from hls_workflows.pipeline import Pipeline, PipelineBuilder
+from hls_workflows.sentinel import mapped_tasks as s2_mapped
+from hls_workflows.sentinel import tasks as s2_tasks
+
+from . import landsat as l_fmask
+from . import sentinel as s2_fmask
+
+logger = logging.getLogger(__name__)
+
+UPLOAD_PREFIX = "fmask"
+
+
+def construct_pipeline(
+    granule_id: str,
+    *,
+    working_dir: Path | None = None,
+    local_granule: Path | None = None,
+    metric_sink: MetricSink | None = None,
+    cleanup_working_dir: bool = True,
+) -> Pipeline:
+    """Construct the standalone Fmask pipeline for one granule.
+
+    Parameters
+    ----------
+    granule_id
+        Sentinel-2 or Landsat granule ID (auto-detected).
+    working_dir
+        Override local processing directory.
+    local_granule
+        Pre-downloaded granule path (a .zip for Sentinel-2, a directory for
+        Landsat). If omitted, the granule is downloaded from S3.
+    metric_sink
+        Optional metrics sink.
+    cleanup_working_dir
+        Remove the working directory when the pipeline finishes.
+    """
+    match resolve_granule(granule_id):
+        case Sentinel2Granule():
+            builder = _build_sentinel(granule_id, working_dir, local_granule)
+        case LandsatGranule():
+            builder = _build_landsat(granule_id, working_dir, local_granule)
+        case other:
+            raise ValueError(
+                f"Expected a Sentinel-2 or Landsat granule ID, got {type(other).__name__}"
+            )
+
+    return builder.build(
+        metrics=MetricsCollector(
+            pipeline_dims={"workflow": "fmask-workflow"},
+            pipeline_props={"input_granule_id": granule_id},
+            sink=metric_sink,
+        ),
+        cleanup_working_dir=cleanup_working_dir,
+    )
+
+
+def _build_sentinel(
+    granule_id: str,
+    working_dir: Path | None,
+    local_granule: Path | None,
+) -> PipelineBuilder:
+    """Sentinel-2 ``fmask`` pipeline builder (single granule, no twin consolidation)."""
+    # The Sentinel EnvSource reads GRANULE_LIST from the environment.
+    os.environ["GRANULE_LIST"] = granule_id
+
+    download: s2_mapped.LocalSentinelGranule | s2_mapped.DownloadSentinelGranule
+    if local_granule:
+        download = s2_mapped.LocalSentinelGranule.map(granule_id)(
+            "LocalGranule", local_granule_zip=local_granule
+        )
+    else:
+        download = s2_mapped.DownloadSentinelGranule.map(granule_id)("Download")
+
+    return (
+        PipelineBuilder()
+        .add(s2_tasks.EnvSource("EnvConfig", working_dir=working_dir))
+        .add(download)
+        .add(s2_mapped.GetGranuleDir.map(granule_id)("GetInnerDir"))
+        .add(s2_fmask.ApplyQualityMaskNoSolar.map(granule_id)(name="ApplyMask"))
+        .add(s2_mapped.RunFmaskV5.map(granule_id)("Fmask"))
+        .add(s2_fmask.UploadFmaskDebug.map(granule_id)("Upload", prefix=UPLOAD_PREFIX))
+    )
+
+
+def _build_landsat(
+    granule_id: str,
+    working_dir: Path | None,
+    local_granule: Path | None,
+) -> PipelineBuilder:
+    """Landsat ``fmask`` pipeline builder."""
+    download: l_tasks.LocalGranule | l_tasks.DownloadGranule
+    if local_granule:
+        download = l_tasks.LocalGranule("LocalGranule", local_granule_dir=local_granule)
+    else:
+        download = l_tasks.DownloadGranule("DownloadGranule")
+
+    return (
+        PipelineBuilder()
+        .add(
+            l_tasks.EnvSource(
+                "EnvConfig", granule_id=granule_id, working_dir=working_dir
+            )
+        )
+        .add(download)
+        .add(l_tasks.RunFmaskV5("Fmask"))
+        .add(l_fmask.UploadFmaskDebug("Upload", prefix=UPLOAD_PREFIX))
+    )