Deduplicate published jobs using STAC catalog

CHANGELOG.md

@@ -4,13 +4,26 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.8.0]
+### Added
+- An `itslive` module with functions to deduplicate already published pairs based on the STAC catalog.
+
+### Changed
+- `deduplicate_hyp3_pairs`, `submit_pairs_for_processing`, and other HyP3 specific functionality has been moved to the `hyp3` module from `main`. 
+
+### Fixed
+- Sentinel-1 burst or SLC image pairs are now deduplicated against already published ITS_LIVE pairs in the STAC catalog.See [#331](https://github.com/ASFHyP3/its-live-monitoring/issues/331) for more info.
+
+### Removed
+- `deduplicate_s3_pairs` and supporting functions in favor of `itslive.deduplicate_published_pairs`.
+
 ## [0.7.0]
-## Added
+### Added
 - `StacItemsEndpoint` and `StacExistsOk` cloudformation parameters to allow publishing STAC items directly to a STAC catalog. Accordingly:
   - `STAC_ITEMS_ENDPOINT` and `STAC_EXISTS_OK` environment variables are set for the monitoring lambda.
   - The cloud formation parameters are set in the build and deploy GitHub Actions workflow by the `STAC_ITEMS_ENDPOINT` and `STAC_EXISTS_OK` deploy environment variables.
 
-## Changed
+### Changed
 - Updated the `AUTORIFT_JOB_TEMPLATE` for [HyP3 v10.13.0+](https://github.com/ASFHyP3/hyp3/pull/3003) to allow posting/putting STAC items in the catalog instead of writing them to an alternate ingest location.
 
 ### Removed

its_live_monitoring/src/hyp3.py

@@ -0,0 +1,180 @@
+"""Functions for interacting with HyP3 ITS_LIVE."""
+
+import logging
+import os
+from copy import deepcopy
+from datetime import UTC, datetime
+
+import boto3
+import geopandas as gpd
+import hyp3_sdk as sdk
+import pandas as pd
+from boto3.dynamodb.conditions import Attr, Key
+
+
+log = logging.getLogger('its_live_monitoring')
+log.setLevel(os.environ.get('LOGGING_LEVEL', 'INFO'))
+
+dynamo = boto3.resource('dynamodb')
+
+# NOTE: Commented items will get set when submitting
+AUTORIFT_JOB_TEMPLATE = {
+    'job_parameters': {
+        # 'reference': list[str],
+        # 'secondary': list[str],
+        'parameter_file': '/vsicurl/https://its-live-data.s3.amazonaws.com/autorift_parameters/v001/autorift_landice_0120m.shp',
+        # 'publish_bucket': str | None,
+        'use_static_files': True,
+        # 'frame_id' = str | None,
+        # 'stac_items_endpoint': str | None,
+        # 'stac_exists_ok': bool,
+    },
+    'job_type': 'AUTORIFT',
+    # 'name': str | None,
+}
+
+
+def format_time(time: datetime) -> str:
+    """Format time to ISO with UTC timezone.
+
+    Args:
+        time: a datetime object to format
+
+    Returns:
+        datetime: the UTC time in ISO format
+    """
+    if time.tzinfo is None:
+        raise ValueError(f'missing tzinfo for datetime {time}')
+    utc_time = time.astimezone(UTC)
+    return utc_time.isoformat(timespec='seconds')
+
+
+def query_jobs_by_status_code(status_code: str, user: str, name: str, start: datetime) -> sdk.Batch:
+    """Query dynamodb for jobs by status_code, then filter by user, name, and date.
+
+    Args:
+        status_code: `status_code` of the desired jobs
+        user: the `user_id` that submitted the jobs
+        name: the name of the jobs
+        start: the earliest submission date of the jobs
+
+    Returns:
+        sdk.Batch: batch of jobs matching the filters
+    """
+    table = dynamo.Table(os.environ['JOBS_TABLE_NAME'])
+
+    key_expression = Key('status_code').eq(status_code)
+
+    filter_expression = Attr('user_id').eq(user) & Attr('name').eq(name) & Attr('request_time').gte(format_time(start))
+
+    params = {
+        'IndexName': 'status_code',
+        'KeyConditionExpression': key_expression,
+        'FilterExpression': filter_expression,
+        'ScanIndexForward': False,
+    }
+
+    jobs = []
+    while True:
+        response = table.query(**params)
+        jobs.extend(response['Items'])
+        if (next_key := response.get('LastEvaluatedKey')) is None:
+            break
+        params['ExclusiveStartKey'] = next_key
+
+    return sdk.Batch([sdk.Job.from_dict(job) for job in jobs])
+
+
+def get_reference_secondary_from_job(job: sdk.Job) -> tuple[tuple | str, tuple | str]:
+    """Get the reference and secondary scenes from an AUTORIFT HyP3 job."""
+    granules = job.job_parameters.get('granules')
+    if granules:
+        reference = granules[:1]
+        secondary = granules[1:]
+    else:
+        reference = job.job_parameters['reference']
+        secondary = job.job_parameters['secondary']
+
+    return tuple(reference), tuple(secondary)
+
+
+def deduplicate_hyp3_pairs(pairs: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
+    """Search HyP3 jobs since the reference scene's acquisition date and remove already submitted (in PENDING or RUNNING state) pairs.
+
+    Args:
+         pairs: A GeoDataFrame containing *at least*  these columns: `reference`, `reference_acquisition`, and
+          `secondary`.
+
+    Returns:
+         The pairs GeoDataFrame with any already submitted pairs removed.
+    """
+    earthdata_username = os.environ['EARTHDATA_USERNAME']
+    assert earthdata_username is not None
+
+    pending_jobs = query_jobs_by_status_code(
+        status_code='PENDING',
+        user=earthdata_username,
+        name=pairs.iloc[0].job_name,
+        start=pairs.iloc[0].reference_acquisition,
+    )
+    running_jobs = query_jobs_by_status_code(
+        status_code='RUNNING',
+        user=earthdata_username,
+        name=pairs.iloc[0].job_name,
+        start=pairs.iloc[0].reference_acquisition,
+    )
+    jobs = pending_jobs + running_jobs
+
+    df = pd.DataFrame([get_reference_secondary_from_job(job) for job in jobs], columns=['reference', 'secondary'])
+    df = df.set_index(['reference', 'secondary'])
+    pairs = pairs.set_index(['reference', 'secondary'])
+
+    duplicates = df.loc[df.index.isin(pairs.index)]
+    if len(duplicates) > 0:
+        pairs = pairs.drop(duplicates.index)
+
+    return pairs.reset_index()
+
+
+def _nullable_str(s: str) -> str | None:
+    s = s.replace('None', '').strip()
+    return s if s else None
+
+
+def _string_is_true(s: str) -> bool:
+    return s.lower() == 'true'
+
+
+def submit_pairs_for_processing(pairs: gpd.GeoDataFrame) -> sdk.Batch:  # noqa: D103
+    prepared_jobs = []
+    for reference, secondary, name in pairs[['reference', 'secondary', 'job_name']].itertuples(index=False):
+        prepared_job: dict = deepcopy(AUTORIFT_JOB_TEMPLATE)
+        prepared_job['name'] = name
+        prepared_job['job_parameters']['reference'] = reference
+        prepared_job['job_parameters']['secondary'] = secondary
+
+        if publish_bucket := os.environ.get('PUBLISH_BUCKET', ''):
+            prepared_job['job_parameters']['publish_bucket'] = _nullable_str(publish_bucket)
+
+        if stac_items_endpoints := os.environ.get('STAC_ITEMS_ENDPOINT', ''):
+            prepared_job['job_parameters']['stac_items_endpoint'] = _nullable_str(stac_items_endpoints)
+            prepared_job['job_parameters']['stac_exists_ok'] = _string_is_true(os.environ.get('STAC_EXISTS_OK', ''))
+
+        if name.startswith('OPERA'):
+            prepared_job['job_parameters']['frame_id'] = name.split('_')[1]
+
+        prepared_jobs.append(prepared_job)
+
+    log.debug(prepared_jobs)
+
+    hyp3 = sdk.HyP3(
+        os.environ.get('HYP3_API', 'https://hyp3-its-live-test.asf.alaska.edu'),
+        username=os.environ.get('EARTHDATA_USERNAME'),
+        password=os.environ.get('EARTHDATA_PASSWORD'),
+    )
+
+    jobs = sdk.Batch()
+    for batch in sdk.util.chunk(prepared_jobs):
+        jobs += hyp3.submit_prepared_jobs(batch)
+
+    return jobs

its_live_monitoring/src/itslive.py

@@ -0,0 +1,121 @@
+"""Functions for interacting with published ITS_LIVE products."""
+
+from datetime import datetime
+
+import geopandas as gpd
+import pystac
+import pystac_client
+
+from sentinel1 import get_safe_acquisition_times
+
+
+ITS_LIVE_CATALOG_API = 'https://stac.itslive.cloud/'
+ITS_LIVE_CATALOG = pystac_client.Client.open(ITS_LIVE_CATALOG_API)
+ITS_LIVE_COLLECTION_NAME = 'itslive-granules'
+ITS_LIVE_COLLECTION = ITS_LIVE_CATALOG.get_collection(ITS_LIVE_COLLECTION_NAME)
+
+
+def get_datetime(scene_name: str) -> datetime:
+    """Get the acquisition start time from a Landsat, Sentinel-1 (SLC or Burst), or Sentinel-2 scene name."""
+    if 'BURST' in scene_name:
+        return datetime.strptime(scene_name[14:29], '%Y%m%dT%H%M%S')
+    if scene_name.startswith('S1'):
+        return datetime.strptime(scene_name[17:32], '%Y%m%dT%H%M%S')
+    if scene_name.startswith('S2') and len(scene_name) > 25:  # ESA
+        return datetime.strptime(scene_name[11:26], '%Y%m%dT%H%M%S')
+    if scene_name.startswith('S2'):  # COG
+        return datetime.strptime(scene_name.split('_')[2], '%Y%m%d')
+    if scene_name.startswith('L'):
+        return datetime.strptime(scene_name[17:25], '%Y%m%d')
+
+    raise ValueError(f'Unsupported scene format: {scene_name}')
+
+
+def sort_earliest_first(reference: str, secondary: str) -> tuple[str, str]:
+    """Sort reference and secondary scene names according to the ITS_LIVE convention."""
+    ref_datetime = get_datetime(reference)
+    sec_datetime = get_datetime(secondary)
+
+    if ref_datetime > sec_datetime:
+        return secondary, reference
+
+    return reference, secondary
+
+
+def bursts_in_item(ref_datetime: datetime, sec_datetime: datetime, item: pystac.Item) -> bool:
+    """Determines if the reference and secondary burst pairs fall within an ITS_LIVE granules.
+
+    ITS_LIVE granules for Sentinel-1 report the synthetic burst2safe name for scene_1 (reference), scene_2 (secondary),
+    so we must check and see if the bursts' acquisition start times falls with the start,stop time reported in the name.
+    """
+    scene_1, scene_2 = sort_earliest_first(item.properties['scene_1_id'], item.properties['scene_2_id'])
+
+    scene_1_start, scene_1_stop = get_safe_acquisition_times(scene_1)
+    scene_2_start, scene_2_stop = get_safe_acquisition_times(scene_2)
+
+    # Bounds need to be inclusive because burst2safe just uses the first and last bursts datetimes
+    if scene_1_start <= ref_datetime <= scene_1_stop and scene_2_start <= sec_datetime <= scene_2_stop:
+        return True
+
+    return False
+
+
+def pair_exists(reference: str, secondary: str, name: str) -> bool:
+    """Determine if a velocity granule for a scene pair has already been published to the ITS_LIVE STAC catalog."""
+    reference, secondary = sort_earliest_first(reference, secondary)
+    ref_datetime = get_datetime(reference)
+    sec_datetime = get_datetime(secondary)
+
+    if reference.startswith('S1'):
+        frame = name.split('_')[1]
+        results = ITS_LIVE_CATALOG.search(
+            collections=[ITS_LIVE_COLLECTION_NAME],
+            datetime=[ref_datetime, sec_datetime],
+            filter={
+                'op': 'and',
+                'args': [
+                    {'op': 'like', 'args': [{'property': 'platform'}, 'S1%']},
+                    {'op': '=', 'args': [{'property': 'scene_1_frame'}, frame]},
+                ],
+            },
+        )
+        items = [item for page in results.pages() for item in page]
+        items = [item for item in items if bursts_in_item(ref_datetime, sec_datetime, item)]
+
+    else:
+        results = ITS_LIVE_CATALOG.search(
+            collections=[ITS_LIVE_COLLECTION_NAME],
+            datetime=[ref_datetime, sec_datetime],
+            query=[f'scene_1_id={reference}', f'scene_2_id={secondary}'],
+        )
+        items = [item for page in results.pages() for item in page]
+
+    # This will effectively work just like `if items: return True`
+    # but will put the reference scene message back into the DeadLetter queue so we can learn about and remove duplicates
+    if (n_items := len(items)) > 1:
+        raise ValueError(
+            f'{n_items} items for ({reference}, {secondary}) found in ITS_LIVE STAC collection: '
+            f'{ITS_LIVE_CATALOG_API}/collections/{ITS_LIVE_COLLECTION_NAME}'
+        )
+
+    if items:
+        return True
+    return False
+
+
+def deduplicate_published_pairs(pairs: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
+    """Ensures that pairs aren't submitted if they already have a product in S3.
+
+    Args:
+         pairs: A GeoDataFrame containing *at least*  these columns: `reference`, `reference_acquisition`, and
+          `secondary`.
+
+    Returns:
+         The pairs GeoDataFrame with any already submitted pairs removed.
+    """
+    drop_indexes = []
+    for idx, reference, secondary, name in pairs[['reference', 'secondary', 'job_name']].itertuples():
+        if pair_exists(reference, secondary, name):
+            drop_indexes.append(idx)
+
+    return pairs.drop(index=drop_indexes)