Viz EC2 Setup Access and Raster Processing Fixes

Core/EC2/viz/scripts/viz_ec2_setup.ps1

@@ -253,9 +253,15 @@ LogWrite "-->TRANFERRING PRISTINE DATA"
 $s3_pristine = "s3://" + $DEPLOYMENT_DATA_BUCKET + "/" + $DEPLOY_FILES_PREFIX + "pristine_data/"
 aws s3 cp $s3_pristine $PRISTINE_ROOT --recursive
 
+Install-Module -Name Invoke-CommandAs -force
+$ec2host = hostname
+$strScriptUser = "$ec2host\$PIPELINE_USER"
+$PSS = ConvertTo-SecureString $PIPELINE_USER_ACCOUNT_PASSWORD -AsPlainText -Force
+$cred = new-object system.management.automation.PSCredential $strScriptUser,$PSS
+
 LogWrite "CREATING CONNECTION FILES FOR $FIM_DATA_BUCKET"
-Set-Location -Path $AWS_SERVICE_REPO
-& "C:\Program Files\ArcGIS\Pro\bin\Python\envs\viz\python.exe" "aws_loosa\deploy\create_s3_connection_files.py"
+$python_file = "$AWS_SERVICE_REPO\aws_loosa\deploy\create_s3_connection_files.py"
+Invoke-CommandAs -ScriptBlock { param($python_file) & "C:\Program Files\ArcGIS\Pro\bin\Python\envs\viz\python.exe" $python_file } -ArgumentList $python_file -AsUser $cred
 
 LogWrite "UPDATING PYTHON PERMISSIONS FOR $PIPELINE_USER"
 $ACL = Get-ACL -Path "C:\Program Files\ArcGIS\Pro\bin\Python"
@@ -270,15 +276,8 @@ $ACL.SetAccessRule($AccessRule)
 $ACL | Set-Acl -Path "D:\"
 
 LogWrite "ADDING $PUBLISHED_ROOT TO $EGIS_HOST"
-Set-Location -Path $AWS_SERVICE_REPO
-& "C:\Program Files\ArcGIS\Pro\bin\Python\envs\viz\python.exe" "aws_loosa\deploy\update_data_stores_and_sd_files.py"
-
-LogWrite "DELETING PUBLISHED FLAGS IF THEY EXIST"
-$EXISTING_PUBLISHED_FLAGS = aws s3 ls $FLAGS_ROOT
-if ($EXISTING_PUBLISHED_FLAGS) {
-    LogWrite "DELETING PUBLISHED FLAGS"
-    aws s3 rm $FLAGS_ROOT --recursive
-}
+$python_file = "$AWS_SERVICE_REPO\aws_loosa\deploy\update_data_stores_and_sd_files.py"
+Invoke-CommandAs -ScriptBlock { param($python_file) & "C:\Program Files\ArcGIS\Pro\bin\Python\envs\viz\python.exe" $python_file } -ArgumentList $python_file -AsUser $cred
 
 Set-Location HKCU:\Software\ESRI\ArcGISPro
 Remove-Item -Recurse -Force -Confirm:$false Licensing

Core/LAMBDA/layers/viz_lambda_shared_funcs/python/viz_classes.py

@@ -243,10 +243,10 @@ def check_required_tables_updated(self, sql_path_or_str, sql_replace={}, referen
                 sql = f'''
                     SELECT EXISTS (
                         SELECT FROM 
-                            pg_tables
+                            information_schema.tables
                         WHERE 
-                            schemaname = '{schemaname}' AND 
-                            tablename  = '{tablename}'
+                            table_schema = '{schemaname}' AND 
+                            table_name  = '{tablename}'
                     );
                 '''
                 cur.execute(sql)

Core/LAMBDA/viz_functions/image_based/viz_raster_processing/lambda_function.py

@@ -14,8 +14,9 @@ def lambda_handler(event, context):
     file_window = event['product']['raster_input_files']['file_window']
     product_file = event['product']['raster_input_files']['product_file']
     input_bucket = event['product']['raster_input_files']['bucket']
+
     output_bucket = event['product']['raster_outputs']['output_bucket']
-    output_workspace = event['product']['raster_outputs']['output_raster_workspaces'][product_name]
+    output_workspace = event['product']['raster_outputs']['output_raster_workspaces'][0][product_name]
     reference_time = event['reference_time']
     reference_date = datetime.strptime(reference_time, "%Y-%m-%d %H:%M:%S")
 

Core/LAMBDA/viz_functions/image_based/viz_schism_fim_processing/lambda_function.py

@@ -55,7 +55,8 @@ def lambda_handler(event, context):
     max_elevs_file = event['args']['fim_config']['max_file']
 
     output_bucket = event['args']['product']['raster_outputs']['output_bucket']
-    output_workspace = event['args']['product']['raster_outputs']['output_raster_workspaces'][fim_config]
+    output_workspaces = event['args']['product']['raster_outputs']['output_raster_workspaces']
+    output_workspace = next(list(workspace.values())[0] for workspace in output_workspaces if list(workspace.keys())[0] == fim_config)
 
     schism_fim_s3_uri = f's3://{max_elevs_file_bucket}/{max_elevs_file}'
 

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/medium_range_mem1/mrf_gfs_10day_peak_flow_arrival_time.sql

@@ -30,7 +30,7 @@ JOIN derived.channels_conus AS channels ON forecasts.feature_id = channels.featu
 JOIN derived.recurrence_flows_conus AS rf ON forecasts.feature_id = rf.feature_id
 
 -- Join in high water arrival time for return time (the yaml config file ensures that arrival time finishes first for this, but we'll join on reference_time as well to ensure)
-JOIN publish.mrf_high_water_arrival_time AS arrival_time ON forecasts.feature_id = arrival_time.feature_id and forecasts.reference_time = arrival_time.reference_time
+JOIN publish.mrf_gfs_10day_high_water_arrival_time AS arrival_time ON forecasts.feature_id = arrival_time.feature_id and forecasts.reference_time = arrival_time.reference_time
 
 WHERE round((forecasts.streamflow*35.315)::numeric, 2) >= rf.high_water_threshold
 GROUP BY forecasts.feature_id, forecasts.reference_time, forecasts.nwm_vers, forecasts.streamflow, channels.name, channels.strm_order, channels.huc6, channels.state, rf.high_water_threshold, max_flows.maxflow_10day_cfs, arrival_time.t_normal, channels.geom;

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/short_range/srf_18hr_peak_flow_arrival_time.sql

@@ -30,7 +30,7 @@ JOIN derived.channels_conus AS channels ON forecasts.feature_id = channels.featu
 JOIN derived.recurrence_flows_conus AS rf ON forecasts.feature_id = rf.feature_id
 
 -- Join in high water arrival time for return time (the yaml config file ensures that arrival time finishes first for this, but we'll join on reference_time as well to ensure)
-JOIN publish.srf_high_water_arrival_time AS arrival_time ON forecasts.feature_id = arrival_time.feature_id and forecasts.reference_time = arrival_time.reference_time
+JOIN publish.srf_18hr_high_water_arrival_time AS arrival_time ON forecasts.feature_id = arrival_time.feature_id and forecasts.reference_time = arrival_time.reference_time
 
 WHERE round((forecasts.streamflow*35.315)::numeric, 2) >= rf.high_water_threshold
 GROUP BY forecasts.feature_id, forecasts.reference_time, forecasts.nwm_vers, channels.name, channels.strm_order, channels.huc6, channels.state, rf.high_water_threshold, arrival_time.t_normal, max_flows.maxflow_18hour_cms, channels.geom

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/short_range_hawaii/srf_48hr_peak_flow_arrival_time_hi.sql

@@ -30,7 +30,7 @@ JOIN derived.channels_hi as channels ON forecasts.feature_id = channels.feature_
 JOIN derived.recurrence_flows_hi as rf ON forecasts.feature_id = rf.feature_id
 
 -- Join in high water arrival time for return time (the yaml config file ensures that arrival time finishes first for this, but we'll join on reference_time as well to ensure)
-JOIN publish.srf_high_water_arrival_time_hi as arrival_time ON forecasts.feature_id = arrival_time.feature_id and forecasts.reference_time = arrival_time.reference_time
+JOIN publish.srf_48hr_high_water_arrival_time_hi as arrival_time ON forecasts.feature_id = arrival_time.feature_id and forecasts.reference_time = arrival_time.reference_time
 
 WHERE (rf.high_water_threshold > 0 OR rf.high_water_threshold = '-9999') AND forecasts.streamflow * 35.315::double precision >= rf.high_water_threshold
 GROUP BY forecasts.feature_id, forecasts.reference_time, forecasts.nwm_vers, channels.name, channels.strm_order, channels.huc6, rf.high_water_threshold, arrival_time.t_normal, max_flows.maxflow_48hour_cms, channels.geom

Core/LAMBDA/viz_functions/viz_db_postprocess_sql/products/short_range_puertorico/srf_48hr_peak_flow_arrival_time_prvi.sql

@@ -30,7 +30,7 @@ JOIN derived.channels_prvi as channels ON forecasts.feature_id = channels.featur
 JOIN derived.recurrence_flows_prvi as rf ON forecasts.feature_id = rf.feature_id
 
 -- Join in high water arrival time for return time (the yaml config file ensures that arrival time finishes first for this, but we'll join on reference_time as well to ensure)
-JOIN publish.srf_high_water_arrival_time_prvi as arrival_time ON forecasts.feature_id = arrival_time.feature_id and forecasts.reference_time = arrival_time.reference_time
+JOIN publish.srf_48hr_high_water_arrival_time_prvi as arrival_time ON forecasts.feature_id = arrival_time.feature_id and forecasts.reference_time = arrival_time.reference_time
 
 WHERE round((forecasts.streamflow*35.315)::numeric, 2) >= rf.high_water_threshold
 GROUP BY forecasts.feature_id, forecasts.reference_time, forecasts.nwm_vers, channels.name, channels.strm_order, channels.huc6, rf.high_water_threshold, arrival_time.t_normal, max_flows.maxflow_48hour_cms, channels.geom

Core/LAMBDA/viz_functions/viz_initialize_pipeline/lambda_function.py

@@ -472,10 +472,11 @@ def get_product_metadata(self, specific_products=None, run_only=True):
             raster_output_bucket = os.environ['RASTER_OUTPUT_BUCKET']
             raster_output_prefix = os.environ['RASTER_OUTPUT_PREFIX']
             product_metadata['raster_outputs'] = {}
-            product_metadata['raster_outputs']['output_raster_workspaces'] = {}
+            product_metadata['raster_outputs']['output_bucket'] = ""
+            product_metadata['raster_outputs']['output_raster_workspaces'] = []
             if product_metadata['product_type'] == "raster":
                 product_metadata['raster_outputs']['output_bucket'] = raster_output_bucket
-                product_metadata['raster_outputs']['output_raster_workspaces'][product_name] = f"{raster_output_prefix}/{product_name}/{pipeline_run_date}/{pipeline_run_hour}/workspace"
+                product_metadata['raster_outputs']['output_raster_workspaces'].append({product_name: f"{raster_output_prefix}/{product_name}/{pipeline_run_date}/{pipeline_run_hour}/workspace"})
 
             if not product_metadata.get("fim_configs"):
                 product_metadata['fim_configs'] = []
@@ -492,7 +493,7 @@ def get_product_metadata(self, specific_products=None, run_only=True):
                     if fim_config['fim_type'] == "coastal":
                         if not product_metadata['raster_outputs'].get('output_bucket'):
                             product_metadata['raster_outputs']['output_bucket'] = raster_output_bucket
-                        product_metadata['raster_outputs']['output_raster_workspaces'][fim_config_name] = f"{raster_output_prefix}/{product_name}/{fim_config_name}/{pipeline_run_date}/{pipeline_run_hour}/workspace"
+                        product_metadata['raster_outputs']['output_raster_workspaces'].append({fim_config_name: f"{raster_output_prefix}/{product_name}/{fim_config_name}/{pipeline_run_date}/{pipeline_run_hour}/workspace"})
 
             if not product_metadata.get("postprocess_sql"):
                 product_metadata['postprocess_sql'] = []

Core/LAMBDA/viz_functions/viz_initialize_pipeline/product_configs/short_range_coastal/srf_18hr_max_coastal_inundation.yml

@@ -16,7 +16,7 @@ fim_configs:
   - name: srf_18hr_max_coastal_inundation_atlgulf
     target_table: ingest.srf_18hr_max_coastal_inundation_atlgulf
     fim_type: coastal
-    sql_file: coastal_pacific
+    sql_file: coastal_atlgulf
     preprocess:
         file_format: common/data/model/com/nwm/para/nwm.{{datetime:%Y%m%d}}/short_range_coastal_atlgulf/nwm.t{{datetime:%H}}z.short_range_coastal.total_water.f{{range:1,19,1,%03d}}.atlgulf.nc
         file_step: None