Viz - Dependent products postprocess, add states to attribute table, peak flow arrival optimization

Core/EC2/RDSBastion/scripts/viz/postgresql_setup.sh.tftpl

@@ -28,7 +28,7 @@ egisDB_fim_catchments="egisDB_fim_catchments_2023_0406.dump"
 vizDB_admin="vizDB_admin_2023_0424.dump"
 vizDB_archive="vizDB_archive_2023_0214.dump"
 vizDB_cache="vizDB_cache_2023_0424.dump"
-vizDB_derived="vizDB_derived_2023_0424.dump"
+vizDB_derived="vizDB_derived_2023_0426.dump"
 vizDB_ingest="vizDB_ingest_2023_0424.dump"
 vizDB_publish="vizDB_publish_2023_0224.dump"
 vizDB_external="vizDB_external_2023_0214.dump"

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/ana_inundation.sql

@@ -16,8 +16,9 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_conus.strm_order, 
-    derived.channels_conus.name
+	channels.strm_order, 
+    channels.name,
+	channels.state
 INTO publish.ana_inundation
 FROM ingest.ana_inundation as inun 
-left join derived.channels_conus ON derived.channels_conus.feature_id = inun.feature_id;
+left join derived.channels_conus as channels ON channels.feature_id = inun.feature_id;

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/ana_inundation_hi.sql

@@ -16,12 +16,13 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_hi.strm_order, 
-    derived.channels_hi.name
+	channels.strm_order, 
+    channels.name,
+	'HI' as state
 INTO publish.ana_inundation_hi
 FROM ingest.ana_inundation_hi as inun 
-left join derived.channels_hi ON derived.channels_hi.feature_id = inun.feature_id_str
+left join derived.channels_hi AS channels ON channels.feature_id = inun.feature_id_str
 --Add an empty row so that service monitor will pick up a reference and update time in the event of no fim features.
 UNION SELECT -9999, '-9999', 'NA', -9999, '-9999', -9999, -9999, -9999, -9999, 'NA', to_char('1900-01-01 00:00:00'::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC'), 
 to_char('1900-01-01 00:00:00'::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC'), '-9999', NULL,
-to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, -9999, NULL;
+to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, -9999, NULL, 'HI';

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/ana_inundation_prvi.sql

@@ -16,12 +16,13 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_prvi.strm_order, 
-    derived.channels_prvi.name
+	channels.strm_order, 
+    channels.name,
+	'PRVI' AS state
 INTO publish.ana_inundation_prvi
 FROM ingest.ana_inundation_prvi as inun 
-left join derived.channels_prvi ON derived.channels_prvi.feature_id = inun.feature_id
+left join derived.channels_prvi as channels ON channels.feature_id = inun.feature_id
 --Add an empty row so that service monitor will pick up a reference and update time in the event of no fim features.
 UNION SELECT -9999, '-9999', 'NA', -9999, '-9999', -9999, -9999, -9999, -9999, 'NA', to_char('1900-01-01 00:00:00'::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC'), 
 to_char('1900-01-01 00:00:00'::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC'), '-9999', NULL,
-to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, -9999, NULL;
+to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, -9999, NULL, 'PRVI';

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/ana_past_14day_max_inundation.sql

@@ -16,9 +16,10 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_conus.strm_order, 
-    derived.channels_conus.name,
+	channels.strm_order, 
+    channels.name,
+	channels.state,
 	'ana_past_14day' AS config
 INTO publish.ana_past_14day_max_inundation
 FROM ingest.ana_past_14day_max_inundation as inun 
-left join derived.channels_conus ON derived.channels_conus.feature_id = inun.feature_id;
+left join derived.channels_conus as channels ON channels.feature_id = inun.feature_id;

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/ana_past_7day_max_inundation.sql

@@ -16,9 +16,10 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_conus.strm_order, 
-    derived.channels_conus.name,
+	channels.strm_order, 
+    channels.name,
+	channels.state,
 	'ana_past_7day' AS config
 INTO publish.ana_past_7day_max_inundation
 FROM ingest.ana_past_7day_max_inundation as inun 
-left join derived.channels_conus ON derived.channels_conus.feature_id = inun.feature_id;
+left join derived.channels_conus as channels ON channels.feature_id = inun.feature_id;

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/mrf_max_inundation_10day.sql

@@ -15,9 +15,10 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_conus.strm_order, 
-    derived.channels_conus.name,
+	channels.strm_order, 
+    channels.name,
+	channels.state,
     'mrf_10day' AS config
 INTO publish.mrf_max_inundation_10day
 FROM ingest.mrf_max_inundation_10day as inun 
-left join derived.channels_conus ON derived.channels_conus.feature_id = inun.feature_id;
+left join derived.channels_conus as channels ON inun.feature_id = channels.feature_id;

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/mrf_max_inundation_3day.sql

@@ -15,9 +15,10 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_conus.strm_order, 
-    derived.channels_conus.name,
+	channels.strm_order, 
+    channels.name,
+	channels.state,
     'mrf_3day' AS config
 INTO publish.mrf_max_inundation_3day
 FROM ingest.mrf_max_inundation_3day as inun 
-left join derived.channels_conus ON derived.channels_conus.feature_id = inun.feature_id;
+left join derived.channels_conus as channels ON inun.feature_id = channels.feature_id;

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/mrf_max_inundation_5day.sql

@@ -15,9 +15,10 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_conus.strm_order, 
-    derived.channels_conus.name,
+	channels.strm_order, 
+    channels.name,
+	channels.state,
     'mrf_5day' AS config
 INTO publish.mrf_max_inundation_5day
 FROM ingest.mrf_max_inundation_5day as inun 
-left join derived.channels_conus ON derived.channels_conus.feature_id = inun.feature_id;
+left join derived.channels_conus as channels ON inun.feature_id = channels.feature_id;

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/rfc_5day_max_downstream_inundation.sql

@@ -23,11 +23,12 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_conus.strm_order, 
-    derived.channels_conus.name,
+	channels.strm_order, 
+    channels.name,
+	channels.state,
     agg_status.inherited_rfc_forecasts,
     agg_status.max_status
 INTO publish.rfc_5day_max_downstream_inundation
 FROM ingest.rfc_5day_max_downstream_inundation as inun 
 JOIN agg_status ON inun.feature_id = agg_status.feature_id
-left join derived.channels_conus ON derived.channels_conus.feature_id = inun.feature_id;
+left join derived.channels_conus as channels ON channels.feature_id = inun.feature_id;

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/srf_max_inundation.sql

@@ -15,8 +15,9 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_conus.strm_order, 
-    derived.channels_conus.name
+	channels.strm_order, 
+    channels.name,
+	channels.state
 INTO publish.srf_max_inundation
 FROM ingest.srf_max_inundation as inun 
-left join derived.channels_conus ON derived.channels_conus.feature_id = inun.feature_id;
+left join derived.channels_conus as channels ON channels.feature_id = inun.feature_id;

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/srf_max_inundation_hi.sql

@@ -15,11 +15,12 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_hi.strm_order, 
-    derived.channels_hi.name
+	channels.strm_order, 
+    channels.name,
+	'HI' AS state
 INTO publish.srf_max_inundation_hi
 FROM ingest.srf_max_inundation_hi as inun 
-left join derived.channels_hi ON derived.channels_hi.feature_id = inun.feature_id
+left join derived.channels_hi as channels ON channels.feature_id = inun.feature_id
 --Add an empty row so that service monitor will pick up a reference and update time in the event of no fim features.
 UNION SELECT -9999, '-9999', 'NA', -9999, '-9999', -9999, -9999, -9999, -9999, 'NA', to_char('1900-01-01 00:00:00'::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC'),
-'-9999', NULL, to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, -9999, NULL;
+'-9999', NULL, to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, -9999, NULL, 'HI';

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/fim_configs/srf_max_inundation_prvi.sql

@@ -15,11 +15,12 @@ SELECT
 	inun.huc8,
 	inun.geom,
 	to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, 
-	derived.channels_prvi.strm_order, 
-    derived.channels_prvi.name
+	channels.strm_order, 
+    channels.name,
+	'PRVI' AS state
 INTO publish.srf_max_inundation_prvi
 FROM ingest.srf_max_inundation_prvi as inun 
-left join derived.channels_prvi ON derived.channels_prvi.feature_id = inun.feature_id
+left join derived.channels_conus as channels ON channels.feature_id = inun.feature_id
 --Add an empty row so that service monitor will pick up a reference and update time in the event of no fim features.
 UNION SELECT -9999, '-9999', 'NA', -9999, '-9999', -9999, -9999, -9999, -9999, 'NA', to_char('1900-01-01 00:00:00'::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC'),
-'-9999', NULL, to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, -9999, NULL;
+'-9999', NULL, to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time, -9999, NULL, 'PRVI';

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/lambda_function.py

@@ -2,44 +2,51 @@
 import os
 from viz_classes import database
 
+class RequiredTableNotUpdated(Exception):
+    """ This is a custom exception to report back to the AWS Step Function that a required table (dependent_on_table) has not yet been updated with the current reference time. """
+
 def lambda_handler(event, context):
     step = event['step']
     folder = event['folder']
     reference_time = event['args']['reference_time']
     sql_replace = event['args']['sql_rename_dict']
     sql_replace.update({'1900-01-01 00:00:00': reference_time}) #setup a replace dictionary, starting with the reference time of the current pipeline.
 
+    # Don't run any SQL if it's a reference service
     if step in ["products", "fim_config"]:
         if event['args']['product']['configuration'] == "reference":
             return
 
+    # Admin tasks
     if folder == 'admin':
          run_admin_tasks(event, folder, step, sql_replace)
     else:
-        # TODO: Clean up this conditional logic to be more readable.
-        if step == 'summaries':
-            folder = os.path.join(folder, event['args']['product']['product'])
-            sql_file = event['args']['postprocess_summary']['sql_file']
-
-        elif step == "max_flows":
+        # Max Flow
+        if step == "max_flows":
             sql_file = event['args']['db_max_flow']['max_flows_sql_file']
-            
+        # FIM Config
         elif step == 'fim_config':
             if not event['args']['fim_config'].get('postprocess'):
                 return
-
             sql_file = event['args']['fim_config']['postprocess']['sql_file']
-            
-        else:
+        # Product
+        elif step == "products":
             folder = os.path.join(folder, event['args']['product']['configuration'])
             sql_file = event['args']['postprocess_sql']['sql_file']
+            if event.get('args').get('postprocess_sql').get('dependent_on_db_tables'):
+                dependent_on_tables = event['args']['postprocess_sql']['dependent_on_db_tables']
+                if len(dependent_on_tables) > 0:
+                    print("Dependent DB Table Specified - Checking if complete.")
+                    if check_required_tables(dependent_on_tables, reference_time, sql_replace) is False:
+                        print("Reference Time not yet present in required table. Pausing and Retrying.")
+                        raise RequiredTableNotUpdated
+        # Summary
+        elif step == 'summaries':
+            folder = os.path.join(folder, event['args']['product']['product'])
+            sql_file = event['args']['postprocess_summary']['sql_file'] 
 
-            
+        ### Run the Appropriate SQL File ###
         sql_path = f"{folder}/{sql_file}.sql"
-
-        if step == 'max_flows' and max_flows_already_processed(sql_path, reference_time, sql_replace):
-            return True
-
         run_sql(sql_path, sql_replace)
 
     return True
@@ -94,19 +101,19 @@ def run_sql(sql_path_or_str, sql_replace=None):
     print(f"Finished executing the SQL statement above.")
     return result
 
-def max_flows_already_processed(sql_path, reference_time, sql_replace):
-    sql = open(sql_path, 'r').read().lower()
-    for word, replacement in sql_replace.items():
-        sql = re.sub(word, replacement, sql, flags=re.IGNORECASE).replace('utc', 'UTC')
-    schema, table = re.search('into (\w+)\.(\w+)', sql).groups()
-    sql = f'SELECT reference_time FROM {schema}.{table} LIMIT 1;'
-    result = run_sql(sql)
-
-    if not result:
-        return False
-
-    if result[0] == reference_time:
-        print(f"NOTE: {sql_path} was already executed for reference time {reference_time}")
-        return True
-    else:
-        return False
+# Check if a specified reference time is present in a database table
+def check_required_tables(dependent_on_tables, reference_time, sql_replace):
+    for table in dependent_on_tables:
+        sql = f"SELECT reference_time FROM {table} LIMIT 1"
+        for word, replacement in sql_replace.items():
+            sql = re.sub(word, replacement, sql, flags=re.IGNORECASE).replace('utc', 'UTC')
+        try:
+            result = run_sql(sql)
+        except: # table doesn't exist
+            return False
+        if not result: # table is empty
+            return False
+        elif result[0].replace(" UTC", "") == reference_time: # table reference time matches current reference time
+            return True
+        else: # table reference time doesn't match current reference time
+            return False

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/analysis_assim/ana_high_flow_magnitude.sql

@@ -27,6 +27,7 @@ SELECT channels.feature_id,
 	channels.feature_id::TEXT AS feature_id_str,
 	channels.strm_order,
 	channels.name,
+	channels.state,
 	channels.huc6,
 	high_flow_mag.nwm_vers,
 	high_flow_mag.reference_time,

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/analysis_assim/ana_past_14day_max_high_flow_magnitude.sql

@@ -5,6 +5,7 @@ SELECT channels.feature_id,
 	channels.strm_order,
 	channels.name,
 	channels.huc6,
+	channels.state,
 	hfm_14day.nwm_vers,
 	hfm_14day.reference_time,
 	hfm_14day.reference_time AS valid_time,

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/analysis_assim/ana_past_7day_max_high_flow_magnitude.sql

@@ -5,6 +5,7 @@ SELECT channels.feature_id,
 	channels.strm_order,
 	channels.name,
 	channels.huc6,
+	channels.state,
 	hfm_7day.nwm_vers,
 	hfm_7day.reference_time,
 	hfm_7day.reference_time AS valid_time,

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/analysis_assim/ana_streamflow.sql

@@ -10,6 +10,7 @@ SELECT
     channels.strm_order::integer,
     channels.name,
     channels.huc6,
+    channels.state,
     to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time,
     channels.geom
 INTO publish.ana_streamflow

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/analysis_assim_alaska/ana_streamflow_ak.sql

@@ -12,6 +12,7 @@ SELECT
     channels.strm_order::integer,
     channels.name,
     channels.huc6,
+    'AK' AS state,
     channels.geom
 INTO publish.ana_streamflow_ak
 FROM cache.max_flows_ana_ak as ana

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/analysis_assim_hawaii/ana_high_flow_magnitude_hi.sql

@@ -33,6 +33,7 @@ SELECT channels.feature_id,
 	channels.strm_order,
 	channels.name,
 	channels.huc6,
+	'HI' as state,
 	high_flow_mag.nwm_vers,
 	high_flow_mag.reference_time,
 	high_flow_mag.reference_time AS valid_time,

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/analysis_assim_hawaii/ana_streamflow_hi.sql

@@ -10,6 +10,7 @@ SELECT
     channels.strm_order::integer,
     channels.name,
     channels.huc6,
+    'HI' as state,
     to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time,
     channels.geom
 INTO publish.ana_streamflow_hi

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/analysis_assim_puertorico/ana_high_flow_magnitude_prvi.sql

@@ -33,6 +33,7 @@ SELECT channels.feature_id,
 	channels.strm_order,
 	name,
 	huc6,
+	'PRVI' as state,
 	high_flow_mag.nwm_vers,
 	high_flow_mag.reference_time,
 	high_flow_mag.reference_time AS valid_time,

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/analysis_assim_puertorico/ana_streamflow_prvi.sql

@@ -10,6 +10,7 @@ SELECT
     channels.strm_order::integer,
     channels.name,
     channels.huc6,
+    'PRVI' AS state,
     to_char(now()::timestamp without time zone, 'YYYY-MM-DD HH24:MI:SS UTC') AS update_time,
     channels.geom
 INTO publish.ana_streamflow_prvi