add partial testing and refactor merge_bursts

src/hyp3_isce2/merge_tops_bursts.py

@@ -572,23 +572,40 @@ def get_merged_orbit(products: Iterable[Sentinel1]) -> Orbit:
     return orb
 
 
-def merge_bursts(range_looks: int, azimuth_looks: int, mergedir: str = 'merged') -> None:
-    """Merge burst products into a multi-swath product, and multilook
+def get_frames_and_indexes(burst_ifg_dir: str | Path) -> Tuple:
+    """Get the frames and burst indexes from a directory of burst interferograms.
+
+    Args:
+        burst_ifg_dir: The directory containing the burst interferograms
+
+    Returns:
+        A tuple of the frames and burst indexes
+    """
+    frames = []
+    burst_index = []
+
+    swath_list = get_swath_list(burst_ifg_dir)
+    for swath in swath_list:
+        ifg = load_product(os.path.join(burst_ifg_dir, 'IW{0}_multilooked.xml'.format(swath)))
+        min_burst = ifg.bursts[0].burstNumber - 1
+        max_burst = ifg.bursts[-1].burstNumber
+        frames.append(ifg)
+        burst_index.append([int(swath), min_burst, max_burst])
+
+    return frames, burst_index
+
+
+def merge_bursts(range_looks: int, azimuth_looks: int, merge_dir: str = 'merged') -> None:
+    """Merge burst products into a multi-swath product, and multilook.
+    Note: Can't test this function without polluting home directory with ISCE2 files since
+            mergeBursts2 sets up pathing assuming it is in the working directory.
 
     Args:
         azimuth_looks: The number of azimuth looks
         range_looks: The number of range looks
         mergedir: The directory to write the merged product to
     """
-    frames = []
-    burstIndex = []
-    swathList = get_swath_list(BURST_IFG_DIR)
-    for swath in swathList:
-        ifg = load_product(os.path.join(BURST_IFG_DIR, 'IW{0}_multilooked.xml'.format(swath)))
-        minBurst = ifg.bursts[0].burstNumber - 1
-        maxBurst = ifg.bursts[-1].burstNumber
-        frames.append(ifg)
-        burstIndex.append([int(swath), minBurst, maxBurst])
+    frames, burstIndex = get_frames_and_indexes(BURST_IFG_DIR)
 
     box = mergeBox(frames)
     file_types = {
@@ -601,9 +618,8 @@ def merge_bursts(range_looks: int, azimuth_looks: int, mergedir: str = 'merged')
     for file_type in file_types:
         directory, file_pattern, name = file_types[file_type]
         burst_paths = os.path.join(directory, 'IW%d', file_pattern)
-        out_path = os.path.join(mergedir, name)
-        merged_path = out_path
-        mergeBursts2(frames, burst_paths, burstIndex, box, merged_path, virtual=True, validOnly=True)
+        out_path = os.path.join(merge_dir, name)
+        mergeBursts2(frames, burst_paths, burstIndex, box, out_path, virtual=True, validOnly=True)
         with TemporaryDirectory() as tmpdir:
             out_tmp_path = str(Path(tmpdir) / Path(out_path).name)
             gdal.Translate(out_tmp_path, out_path + '.vrt', format='ENVI', creationOptions=['INTERLEAVE=BIL'])
@@ -996,6 +1012,23 @@ def check_burst_group_validity(products) -> None:
             raise ValueError(f'Products from swaths {swath1} and {swath2} do not overlap')
 
 
+def get_product_multilook(product_dir: Path) -> Tuple:
+    """Get the multilook values for a set of ASF burst products.
+    You should have already checked that all products have the same multilook,
+    so you can just use the first product's values.
+
+    Args:
+        product_dir: The path to the directory containing the UNZIPPED ASF burst product directories
+
+    Returns:
+        The number of azimuth looks and range looks
+    """
+    product_path = list(product_dir.glob('S1_??????_IW?_*'))[0]
+    metadata_path = product_path / f'{product_path.name}.txt'
+    meta = read_product_metadata(metadata_path)
+    return int(meta['Rangelooks']), int(meta['Azimuthlooks'])
+
+
 def prepare_products(directory: Path) -> None:
     """Set up a directory for ISCE2-based burst merging using a set of ASF burst products.
     This includes:
@@ -1030,23 +1063,6 @@ def prepare_products(directory: Path) -> None:
     download_dem_for_multiple_bursts(swath_objs)
 
 
-def get_product_multilook(product_dir: Path) -> Tuple:
-    """Get the multilook values for a set of ASF burst products.
-    You should have already checked that all products have the same multilook,
-    so you can just use the first product's values.
-
-    Args:
-        product_dir: The path to the directory containing the UNZIPPED ASF burst product directories
-
-    Returns:
-        The number of azimuth looks and range looks
-    """
-    product_path = list(product_dir.glob('S1_??????_IW?_*'))[0]
-    metadata_path = product_path / f'{product_path.name}.txt'
-    meta = read_product_metadata(metadata_path)
-    return int(meta['Rangelooks']), int(meta['Azimuthlooks'])
-
-
 def run_isce2_workflow(
     range_looks: int, azimuth_looks: int, mergedir='merged', filter_strength=0.5, apply_water_mask=False
 ) -> None:
@@ -1060,7 +1076,7 @@ def run_isce2_workflow(
         apply_water_mask: Whether or not to apply a water body mask to the coherence file before unwrapping
     """
     Path(mergedir).mkdir(exist_ok=True)
-    merge_bursts(range_looks, azimuth_looks, mergedir=mergedir)
+    merge_bursts(range_looks, azimuth_looks, merge_dir=mergedir)
     goldstein_werner_filter(filter_strength=filter_strength, mergedir=mergedir)
     if apply_water_mask:
         log.info('Water masking requested, downloading water mask')

tests/conftest.py

@@ -1,5 +1,6 @@
 import os
 import shutil
+from datetime import datetime
 from pathlib import Path
 
 import lxml.etree as ET
@@ -44,3 +45,111 @@ def annotation_manifest_dirs(tmp_path, test_data_dir):
     ET.ElementTree(burst_metadata.annotation).write(annotation_dir / burst_metadata.annotation_name, **et_args)
     ET.ElementTree(burst_metadata.manifest).write(manifest_dir / f'{burst_metadata.safe_name}.xml', **et_args)
     return annotation_dir, manifest_dir
+
+
+@pytest.fixture
+def burst_product(test_merge_dir):
+    product_path = list(test_merge_dir.glob('*'))[0]
+    product = merge.BurstProduct(
+        granule='bar',
+        reference_date=datetime(2020, 6, 4, 2, 23, 15),
+        secondary_date=datetime(2020, 6, 16, 2, 23, 16),
+        burst_id=0,
+        swath='IW2',
+        polarization='VV',
+        burst_number=1,
+        product_path=product_path,
+        n_lines=377,
+        n_samples=1272,
+        range_looks=20,
+        azimuth_looks=4,
+        first_valid_line=8,
+        n_valid_lines=363,
+        first_valid_sample=9,
+        n_valid_samples=1220,
+        az_time_interval=0.008222225199999992,
+        rg_pixel_size=46.59124229430646,
+        start_utc=datetime(2020, 6, 4, 2, 22, 54, 655908),
+        stop_utc=datetime(2020, 6, 4, 2, 23, 18, 795712),
+        relative_orbit=1,
+    )
+    return product
+
+
+@pytest.fixture
+def burst_product1(test_merge_dir):
+    product_path1 = list(test_merge_dir.glob('*'))[0]
+    product1 = merge.BurstProduct(
+        granule='S1A_IW_SLC__1SDV_20200604T022251_20200604T022318_032861_03CE65_7C85',
+        reference_date=datetime(2020, 6, 4, 2, 23, 12),
+        secondary_date=datetime(2020, 6, 16, 2, 23, 13),
+        burst_id=136231,
+        swath='IW2',
+        polarization='VV',
+        burst_number=7,
+        product_path=product_path1,
+        n_lines=377,
+        n_samples=1272,
+        range_looks=20,
+        azimuth_looks=4,
+        first_valid_line=8,
+        n_valid_lines=363,
+        first_valid_sample=9,
+        n_valid_samples=1220,
+        az_time_interval=0.008222225199999992,
+        rg_pixel_size=46.59124229430646,
+        start_utc=datetime(2020, 6, 4, 2, 23, 13, 963847),
+        stop_utc=datetime(2020, 6, 4, 2, 23, 16, 30988),
+        relative_orbit=64,
+    )
+    return product1
+
+
+@pytest.fixture
+def burst_product2(test_merge_dir):
+    product_path2 = list(test_merge_dir.glob('*'))[0]
+
+    product2 = merge.BurstProduct(
+        granule='S1A_IW_SLC__1SDV_20200604T022251_20200604T022318_032861_03CE65_7C85',
+        reference_date=datetime(2020, 6, 4, 2, 23, 15),
+        secondary_date=datetime(2020, 6, 16, 2, 23, 16),
+        burst_id=136232,
+        swath='IW2',
+        polarization='VV',
+        burst_number=8,
+        product_path=product_path2,
+        n_lines=377,
+        n_samples=1272,
+        range_looks=20,
+        azimuth_looks=4,
+        first_valid_line=8,
+        n_valid_lines=363,
+        first_valid_sample=9,
+        n_valid_samples=1220,
+        az_time_interval=0.008222225199999992,
+        rg_pixel_size=46.59124229430646,
+        start_utc=datetime(2020, 6, 4, 2, 23, 16, 722124),
+        stop_utc=datetime(2020, 6, 4, 2, 23, 18, 795712),
+        relative_orbit=64,
+    )
+    return product2
+
+
+@pytest.fixture
+def burst_products(burst_product1, burst_product2):
+    return [burst_product1, burst_product2]
+
+
+@pytest.fixture
+def isce2_merge_setup(annotation_manifest_dirs, burst_products):
+    base_dir = annotation_manifest_dirs[0].parent
+    s1_obj = merge.create_burst_cropped_s1_obj(2, burst_products, 'VV', base_dir=base_dir)
+    for product, burst_obj in zip(burst_products, s1_obj.product.bursts):
+        product.isce2_burst_number = burst_obj.burstNumber
+
+    save_dir = str(base_dir / 'fine_interferogram')
+    multilooked_swath_obj = merge.modify_for_multilook(burst_products, s1_obj, save_dir)
+    multilooked_swath_obj.write_xml()
+
+    merge.spoof_isce2_setup(burst_products, s1_obj, base_dir=base_dir)
+    return base_dir

tests/test_merge_tops_bursts.py

@@ -37,35 +37,6 @@ def mock_asf_search_results(
     return results
 
 
-@pytest.fixture
-def burst_product(test_merge_dir):
-    product_path = list(test_merge_dir.glob('*'))[0]
-    product = merge.BurstProduct(
-        granule='bar',
-        reference_date=datetime(2020, 6, 4, 2, 23, 15),
-        secondary_date=datetime(2020, 6, 16, 2, 23, 16),
-        burst_id=0,
-        swath='IW2',
-        polarization='VV',
-        burst_number=1,
-        product_path=product_path,
-        n_lines=377,
-        n_samples=1272,
-        range_looks=20,
-        azimuth_looks=4,
-        first_valid_line=8,
-        n_valid_lines=363,
-        first_valid_sample=9,
-        n_valid_samples=1220,
-        az_time_interval=0.008222225199999992,
-        rg_pixel_size=46.59124229430646,
-        start_utc=datetime(2020, 6, 4, 2, 22, 54, 655908),
-        stop_utc=datetime(2020, 6, 4, 2, 23, 18, 795712),
-        relative_orbit=1,
-    )
-    return product
-
-
 def test_to_burst_params(burst_product):
     assert burst_product.to_burst_params() == burst_utils.BurstParams('bar', 'IW2', 'VV', 1)
 
@@ -272,3 +243,14 @@ def test_get_merged_orbit(test_data_dir):
     assert len(merged_orbit.stateVectors) == 17  # This number will change if the test data changes
 
 
+def test_get_frames_and_indexes(isce2_merge_setup):
+    frames, burst_index = merge.get_frames_and_indexes(isce2_merge_setup / 'fine_interferogram')
+    assert len(frames) == 1
+    assert isinstance(frames[0], isceobj.Sensor.TOPS.TOPSSwathSLCProduct.TOPSSwathSLCProduct)
+    assert burst_index[0] == [2, 0, 2]
+
+# FIX: test_merge_bursts doesn't work due to pathing issue.
+# def test_merge_bursts(isce2_merge_setup):
+#     import os
+#     os.chdir(isce2_merge_setup)
+#     merge.merge_bursts(20, 4)