feat(DIST-S1): DIST-S1 historical state-config

Author: chrisjrd

conf/sds/files/datasets.json

@@ -511,6 +511,14 @@
           "s3://{{ DATASET_S3_ENDPOINT }}:80/{{ DATASET_BUCKET }}/browse/inputs/GSLC_NI/{id}"
         ]
       }
+    },
+    {
+      "ipath": "hysds::data/DIST_S1-state-config",
+      "level": "STATE-CONFIG",
+      "type": "DIST_S1-STATE-CONFIG",
+      "match_pattern": "/(?P<id>DIST_S1_state-config_.+)$",
+      "alt_match_pattern": null,
+      "extractor": null
     }
   ]
 }

conf/sds/rules/user_rules.json

@@ -111,6 +111,48 @@
       "username": "hysdsops",
       "workflow": "hysds-io-SCIFLO_L3_DSWx_HLS:__TAG__",
       "job_spec": "job-SCIFLO_L3_DSWx_HLS:__TAG__"
+    },
+    {
+      "enabled": true,
+      "job_type": "hysds-io-rtc_for_dist_query_dist_on_pub:__TAG__",
+      "kwargs": "{}",
+      "passthru_query": false,
+      "priority": 0,
+      "query_all": false,
+      "query_string": "{\"bool\": {\"must\": [{\"term\": {\"dataset_type.keyword\": \"L3_DIST_S1\"}}], \"must_not\": [{\"term\": {\"metadata.restaged\": \"true\"}}]}}",
+      "queue": "factotum-job_worker-small",
+      "rule_name": "trigger-rtc_for_dist_query_dist_on_pub",
+      "username": "hysdsops",
+      "workflow": "hysds-io-rtc_for_dist_query_dist_on_pub:__TAG__",
+      "job_spec": "job-rtc_for_dist_query_dist_on_pub:__TAG__"
+    },
+    {
+      "enabled": true,
+      "job_type": "hysds-io-rtc_for_dist_query_sc_on_first:__TAG__",
+      "kwargs": "{}",
+      "passthru_query": false,
+      "priority": 0,
+      "query_all": false,
+      "query_string": "{\"bool\": {\"must\": [{\"term\": {\"dataset_type.keyword\": \"DIST_S1-STATE-CONFIG\"}}, {\"term\": {\"metadata.first\": true}}]}}",
+      "queue": "factotum-job_worker-small",
+      "rule_name": "trigger-rtc_for_dist_query_sc_on_first",
+      "username": "hysdsops",
+      "workflow": "hysds-io-rtc_for_dist_query_sc_on_first:__TAG__",
+      "job_spec": "job-rtc_for_dist_query_sc_on_first:__TAG__"
+    },
+    {
+      "enabled": true,
+      "job_type": "hysds-io-rtc_for_dist_query_sc_on_complete:__TAG__",
+      "kwargs": "{}",
+      "passthru_query": false,
+      "priority": 0,
+      "query_all": false,
+      "query_string": "{\"bool\": {\"must\": [{\"term\": {\"dataset_type.keyword\": \"DIST_S1-STATE-CONFIG\"}}, {\"term\": {\"metadata.is_complete\": true}}]}}",
+      "queue": "factotum-job_worker-small",
+      "rule_name": "trigger-rtc_for_dist_query_sc_on_complete",
+      "username": "hysdsops",
+      "workflow": "hysds-io-rtc_for_dist_query_sc_on_complete:__TAG__",
+      "job_spec": "job-rtc_for_dist_query_sc_on_complete:__TAG__"
     }
   ],
   "mozart": []

data_subscriber/cslc_utils.py

@@ -215,17 +215,21 @@ def process_frame_geo_json(file):
     return frame_geo_map
 
 def parse_r2_product_file_name(native_id, product_type):
+    match_product_id = _datasets_json_match(product_type, native_id)
+    burst_id = match_product_id.group("burst_id")  # e.g. T074-157286-IW3 (for RTC and CSLC)
+    acquisition_dts = match_product_id.group("acquisition_ts")  # e.g. 20210705T183117Z
+    return burst_id, acquisition_dts
 
+# TODO chrisjrd: move to dataset_util.py or similar
+def _datasets_json_match(product_type, native_id):
     dataset_json = datasets_json_util.DatasetsJson()
     cslc_granule_regex = dataset_json.get(product_type)["match_pattern"]
     match_product_id = re.match(cslc_granule_regex, native_id)
 
     if not match_product_id:
         raise ValueError(f"{product_type} native ID {native_id} could not be parsed with regex from datasets.json")
+    return match_product_id
 
-    burst_id = match_product_id.group("burst_id")  # e.g. T074-157286-IW3 (for RTC and CSLC)
-    acquisition_dts = match_product_id.group("acquisition_ts")  # e.g. 20210705T183117Z
-    return burst_id, acquisition_dts
 
 def parse_cslc_file_name(native_id):
     return parse_r2_product_file_name(native_id, "L2_CSLC_S1")

data_subscriber/daac_data_subscriber.py

@@ -55,7 +55,7 @@ def run(argv: list[str]):
 
     es_conn = supply_es_conn(args)
 
-    logger.debug(f"daac_data_subscriber.py invoked with {args=}")
+    logger.info(f"daac_data_subscriber.py invoked with {args=}")
 
     job_id = supply_job_id()
     logger.debug(f"Using {job_id=}")

data_subscriber/dist_s1_utils.py

@@ -87,21 +87,21 @@ def process_dist_burst_db(file):
     logger.info(f"Processing {df.shape[0]} rows in the DIST-S1 burst database file...")
 
     # Create a dictionary of tile ids and the products that are associated with them
-    for index, row in df.iterrows():
+    for row in df.itertuples():
         #print(row['mgrs_tile_id'], row['acq_group_id_within_mgrs_tile'])
-        tile_id = row['mgrs_tile_id']
-        unique_acquisition = row['acq_group_id_within_mgrs_tile']
+        tile_id = row.mgrs_tile_id
+        unique_acquisition = row.acq_group_id_within_mgrs_tile
         product_id = tile_id + "_" + str(unique_acquisition)
         if product_id not in dist_products[tile_id]:
             dist_products[tile_id].add(product_id)
 
-        jpl_burst_id = row['jpl_burst_id']
+        jpl_burst_id = row.jpl_burst_id
         bursts_to_products[jpl_burst_id].add(product_id)
         product_to_bursts[product_id].add(jpl_burst_id)
 
         if jpl_burst_id in all_burst_ids:
             rtc_bursts_reused += 1
-        all_burst_ids.add(row['jpl_burst_id'])
+        all_burst_ids.add(row.jpl_burst_id)
 
     print(f"Total of {len(all_burst_ids)} unique RTC bursts in this database file.")
     print(f"RTC Bursts were reused {rtc_bursts_reused} times in this database file.")