Create binary layover/shadow masks from CSLC static layers

src/disp_s1/_masking.py

@@ -1,8 +1,15 @@
+import logging
+from pathlib import Path
+from typing import Sequence
+
 import numpy as np
-from dolphin._types import PathOrStr
-from dolphin.io import load_gdal, write_arr
+from dolphin._types import Filename, PathOrStr
+from dolphin.io import format_nc_filename, load_gdal, write_arr
+from opera_utils import group_by_burst
 from scipy import ndimage
 
+logger = logging.getLogger(__name__)
+
 
 def create_mask_from_distance(
     water_distance_file: PathOrStr,
@@ -109,3 +116,53 @@ def convert_distance_to_binary(
         binary_mask.filled(0), structure=np.ones((3, 3)), border_value=1
     )
     return closed_mask
+
+
+def create_layover_shadow_masks(
+    cslc_static_files: Sequence[Filename],
+    output_dir: Filename,
+) -> list[Path]:
+    """Create binary masks from the layover shadow CSLC static files.
+
+    In the outputs, 0 indicates a bad masked pixel, 1 is a good pixel.
+
+    Parameters
+    ----------
+    cslc_static_files : Sequence[Filename]
+        List of CSLC static layer files to process
+    output_dir : Filename
+        Directory where output masks will be saved
+
+    Returns
+    -------
+    list[Path]
+        List of paths to the created binary layover shadow mask files
+
+    """
+    output_path = Path(output_dir)
+    output_path.mkdir(exist_ok=True)
+
+    output_files = []
+
+    for burst_id, files in group_by_burst(cslc_static_files).items():
+        if len(files) > 1:
+            logger.warning(f"Found multiple static files for {burst_id}: {files}")
+        f = files[0]
+        input_name = format_nc_filename(f, ds_name="/data/layover_shadow_mask")
+        out_file = output_path / f"layover_shadow_{burst_id}.tif"
+
+        logger.info(f"Extracting layover shadow mask from {f} to {out_file}")
+        layover_data = load_gdal(input_name)
+        # we'll ignore the nodata region to be conservative
+        layover_data[layover_data == 127] = 0
+        not_layover_pixels = layover_data == 0
+        write_arr(
+            arr=not_layover_pixels,
+            output_name=out_file,
+            like_filename=input_name,
+            nodata=127,
+        )
+
+        output_files.append(out_file)
+
+    return output_files

src/disp_s1/main.py

@@ -20,7 +20,7 @@
 from opera_utils import get_dates, group_by_date
 
 from disp_s1 import __version__, product
-from disp_s1._masking import create_mask_from_distance
+from disp_s1._masking import create_layover_shadow_masks, create_mask_from_distance
 from disp_s1.pge_runconfig import RunConfig
 
 from ._reference import ReferencePoint, read_reference_point
@@ -60,6 +60,13 @@ def run(
         )
         cfg.mask_file = water_binary_mask
 
+    if len(cfg.correction_options.geometry_files) > 0:
+        layover_binary_mask_files = create_layover_shadow_masks(
+            cslc_static_files=cfg.correction_options.geometry_files,
+            output_dir=cfg.work_directory / "layover_shadow_masks",
+        )
+        cfg.layover_shadow_mask_files = layover_binary_mask_files
+
     # Run dolphin's displacement workflow
     out_paths = run_displacement(cfg=cfg, debug=debug)