Merge branch 'main' into dependabot/pip/pytest-gte-9.1.1

Author: ehavazli

next_pass.py

@@ -134,6 +134,11 @@ def create_parser() -> argparse.ArgumentParser:
             "in format YYYY-MM-DDTHH:MM"
         ),
     )
+    parser.add_argument(
+        "--include-hls",
+        action="store_true",
+        help="Fetch and append corresponding source HLS scenes for OPERA HLS products.",
+    )
     return parser
 
 
@@ -256,6 +261,7 @@ def run_next_pass(
     compute_tide: bool = False,
     products: List[str] | str | None = None,
     satellites: List[str] | str | None = None,
+    include_hls: bool = False,
 ):
     """
     Programmatic entry point for next_pass.
@@ -286,6 +292,8 @@ def run_next_pass(
         cli_args.append("-c")
     if compute_tide:
         cli_args.append("-t")
+    if include_hls:
+        cli_args.append("--include-hls")
 
     if date:
         cli_args += ["-d", date]
@@ -389,7 +397,10 @@ def main(cli_args: Any = None):
             timestamp_dir,
         )
         export_opera_products(
-            results_opera, timestamp_dir, compute_cloudiness=args.cloudiness
+            results_opera,
+            timestamp_dir,
+            compute_cloudiness=args.cloudiness,
+            include_hls=getattr(args, "include_hls", False),
         )
         make_opera_granule_map(results_opera, args.bbox, timestamp_dir)
 

requirements.txt

@@ -12,7 +12,7 @@ yagmail
 openpyxl>=3.1.5
 timezonefinder
 pre-commit>=4.6.1
-black>=24.4.0
+black>=26.5.1
 isort>=8.0.1
 flake8>=7.0.0
 pytest>=9.1.1

tests/test_hls.py

@@ -0,0 +1,36 @@
+import earthaccess
+import pytest
+
+from utils.opera_products import fetch_hls_granule_links
+
+
+@pytest.mark.integration
+def test_live_cmr_fallback_search():
+    """
+    Hit the live CMR API to ensure earthaccess.search_data still accepts
+    our specific combination of arguments without throwing an error.
+    """
+    try:
+        results = earthaccess.search_data(
+            short_name=["HLSS30", "HLSL30"],
+            temporal=("2023-01-01T00:00:00", "2023-01-01T23:59:59"),
+            bounding_box=(-120.0, 30.0, -119.0, 31.0),
+        )
+        assert isinstance(results, list)
+    except Exception as e:
+        pytest.fail(f"Live CMR search failed! The API contract may have changed: {e}")
+
+
+@pytest.mark.integration
+def test_live_hls_granule_fetch():
+    """
+    Hit the live CMR API to ensure our fetch_hls_granule_links function
+    successfully queries and extracts data links for a known collection.
+    """
+    known_granule_prefix = "HLS.S30.T11SLT.2023001"
+
+    try:
+        links = fetch_hls_granule_links(known_granule_prefix)
+        assert isinstance(links, list)
+    except Exception as e:
+        pytest.fail(f"Live CMR granule fetch failed: {e}")

tests/test_opera_products.py

@@ -3,6 +3,7 @@
 import json
 import zipfile
 from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -121,3 +122,135 @@ def test_export_opera_products_writes_workbook_and_skips_cloudiness_when_disable
         assert payload[0][0] == "Dataset"
         assert payload[1][1] == "granule-1"
         assert payload[1][5] == "https://example.com/file_B01_WTR.tif"
+
+
+@patch("utils.opera_products.earthaccess.search_data")
+def test_fetch_hls_granule_links(mock_search):
+    # Setup mock successful response
+    mock_result = MagicMock()
+    mock_result.data_links.return_value = ["https://example.com/B04.tif"]
+    mock_search.return_value = [mock_result]
+
+    # Test 1: Successful fetch
+    links = opera_products.fetch_hls_granule_links("HLS.S30.T11SLT.20230101.v2.0")
+    assert links == ["https://example.com/B04.tif"]
+    mock_search.assert_called_with(
+        short_name="HLSS30", granule_name="HLS.S30.T11SLT.20230101.v2.0"
+    )
+
+    # Test 2: Error case (API failure)
+    mock_search.side_effect = Exception("API Timeout")
+    error_links = opera_products.fetch_hls_granule_links("HLS.S30.T11SLT.20230101.v2.0")
+    assert error_links is None
+
+
+@patch("utils.opera_products.fetch_hls_granule_links")
+def test_export_hls_band_mapping_and_input_granules(mock_fetch_links, tmp_path):
+    # Simulate an S30 and L30 response with their specific band names
+    mock_fetch_links.side_effect = [
+        [
+            "https://fake/B02.tif",
+            "https://fake/B03.tif",
+            "https://fake/B04.tif",
+            "https://fake/B8A.tif",
+            "https://fake/Fmask.tif",
+        ],  # S30 response
+        [
+            "https://fake/B02.tif",
+            "https://fake/B03.tif",
+            "https://fake/B04.tif",
+            "https://fake/B05.tif",
+            "https://fake/Fmask.tif",
+        ],  # L30 response
+    ]
+
+    mock_results = {
+        "OPERA_L3_DSWX-HLS_V1": {
+            "results": [
+                {
+                    "umm": {
+                        "GranuleUR": "OPERA_S30",
+                        "InputGranules": ["HLS.S30.T11SLT"],
+                    }
+                },
+                {
+                    "umm": {
+                        "GranuleUR": "OPERA_L30",
+                        "InputGranules": ["HLS.L30.T11SLT"],
+                    }
+                },
+            ],
+            "gdf": None,  # Skipping geometry for simplicity
+        }
+    }
+
+    # Run the export function
+    opera_products.export_opera_products(
+        mock_results, tmp_path, compute_cloudiness=False, include_hls=True
+    )
+
+    # Read the generated Excel file to verify mappings
+    from openpyxl import load_workbook
+
+    wb = load_workbook(tmp_path / "opera_products_metadata.xlsx")
+    ws = wb.active
+
+    # Row 2 is S30 (B8A for NIR)
+    assert ws.cell(row=2, column=18).value == "https://fake/B04.tif"  # Red
+    assert ws.cell(row=2, column=21).value == "https://fake/B8A.tif"  # NIR
+    assert ws.cell(row=2, column=22).value == "https://fake/Fmask.tif"
+
+    # Row 3 is L30 (B05 for NIR)
+    assert ws.cell(row=3, column=21).value == "https://fake/B05.tif"  # NIR for Landsat
+
+
+@patch("utils.opera_products.earthaccess.search_data")
+def test_export_hls_fallback_cmr_search(mock_cmr_search, tmp_path):
+    # Setup mock geometry and mock CMR response
+    mock_geom = MagicMock()
+    mock_geom.bounds = (-120, 30, -119, 31)
+    mock_geom.wkt = "POLYGON((-120 30, -119 30, -119 31, -120 31, -120 30))"
+
+    mock_hls_result = MagicMock()
+    mock_hls_result.get.return_value = {"GranuleUR": "HLS.S30.T11SLT.123"}
+    mock_hls_result.data_links.return_value = ["https://fallback/B04.tif"]
+    mock_cmr_search.return_value = [mock_hls_result]
+
+    # Provide results missing "InputGranules", forcing the fallback logic
+    mock_results = {
+        "OPERA_L3_DIST-ALERT-HLS_V1": {
+            "results": [
+                {
+                    "umm": {
+                        "GranuleUR": "OPERA_L3_DIST-ALERT-HLS_T11SLT_20230101",
+                        "TemporalExtent": {
+                            "RangeDateTime": {
+                                "BeginningDateTime": "2023-01-01T00:00:00Z"
+                            }
+                        },
+                    }
+                }
+            ],
+            "gdf": FakeFrame([{"geometry": mock_geom}]),
+        }
+    }
+
+    opera_products.export_opera_products(
+        mock_results, tmp_path, compute_cloudiness=False, include_hls=True
+    )
+
+    # Verify fallback search triggered with bounds and date
+    mock_cmr_search.assert_called_once()
+    kwargs = mock_cmr_search.call_args.kwargs
+    assert kwargs["bounding_box"] == (-120, 30, -119, 31)
+    assert kwargs["temporal"] == ("2023-01-01T00:00:00", "2023-01-01T23:59:59")
+
+    # Verify the fallback successfully mapped the data
+    from openpyxl import load_workbook
+
+    wb = load_workbook(tmp_path / "opera_products_metadata.xlsx")
+    ws = wb.active
+    assert (
+        ws.cell(row=2, column=17).value == "HLS.S30.T11SLT.123"
+    )  # Extracted Fallback ID
+    assert ws.cell(row=2, column=18).value == "https://fallback/B04.tif"