Add run_fcre_aed_forecast entry function for FaaSr-deployable GLM-AED forecast

R/convert_vera4cast_inflow.R

@@ -1,4 +1,4 @@
-convert_vera4cast_inflow <- function(reference_date, model_id, save_path){
+convert_vera4cast_inflow <- function(reference_date, model_id, save_path, config = NULL){
 
 variables <- c("TP_ugL_sample", "NH4_ugL_sample","NO3NO2_ugL_sample",
                "SRP_ugL_sample","DOC_mgL_sample","DRSI_mgL_sample",
@@ -9,9 +9,14 @@ forecast_df <- NULL
 
 for(i in 1:length(variables)){
 
-  s3 <- arrow::s3_bucket(bucket = glue::glue("bio230121-bucket01/vera4cast/forecasts/archive-parquet/project_id=vera4cast/duration=P1D/variable={variables[i]}/model_id={model_id}/reference_date={reference_date}"),
-                         endpoint_override = "https://amnh1.osn.mghpcc.org",
-                         anonymous = TRUE)
+  if(!is.null(config)){
+    faasr_prefix <- glue::glue("vera4cast/forecasts/archive-parquet/project_id=vera4cast/duration=P1D/variable={variables[i]}/model_id={model_id}/reference_date={reference_date}")
+    s3 <- FLAREr::flare_arrow_s3_bucket(server_name = "vera4cast_forecasts", faasr_prefix = faasr_prefix, config = config)
+  } else {
+    s3 <- arrow::s3_bucket(bucket = glue::glue("bio230121-bucket01/vera4cast/forecasts/archive-parquet/project_id=vera4cast/duration=P1D/variable={variables[i]}/model_id={model_id}/reference_date={reference_date}"),
+                           endpoint_override = "https://amnh1.osn.mghpcc.org",
+                           anonymous = TRUE)
+  }
 
   ## test to see if inflow forecast exists ##
   tryCatch({
@@ -28,11 +33,6 @@ for(i in 1:length(variables)){
 
   })
 
-  # df <- arrow::open_dataset(s3) |>
-  #   dplyr::filter(site_id == "tubr") |>
-  #   dplyr::collect() |>
-  #   dplyr::mutate(variable = variables[i])
-
   forecast_df <- dplyr::bind_rows(forecast_df, df)
 
 }

R/generate_forecast_score_arrow.R

@@ -12,10 +12,17 @@ generate_forecast_score_arrow <- function(targets_df,
                                           bucket = NULL,
                                           endpoint = NULL,
                                           local_directory = NULL,
-                                          variable_types = "state"){
+                                          variable_types = "state",
+                                          config = NULL){
 
 
-  if(use_s3){
+  if(!is.null(config)){
+    vars <- arrow_env_vars()
+    output_directory <- FLAREr::flare_arrow_s3_bucket(server_name = "scores",
+                                                      faasr_prefix = "flare/scores/parquet",
+                                                      config = config)
+    on.exit(unset_arrow_vars(vars))
+  }else if(use_s3){
     if(is.null(bucket) | is.null(endpoint)){
       stop("scoring function needs bucket and endpoint if use_s3=TRUE")
     }

configuration/glm_aed_flare_rs/configure_flare_glm_aed.yml

@@ -20,6 +20,14 @@ s3:
   scores:
     endpoint: amnh1.osn.mghpcc.org
     bucket: bio230121-bucket01/flare/scores/parquet
+  vera4cast_targets:
+    endpoint: amnh1.osn.mghpcc.org
+    bucket: bio230121-bucket01/vera4cast/targets
+    anonymous: true
+  vera4cast_forecasts:
+    endpoint: amnh1.osn.mghpcc.org
+    bucket: bio230121-bucket01/vera4cast/forecasts/archive-parquet
+    anonymous: true
 location:
    site_id: fcre
    name: Falling Creek Reservoir
@@ -28,25 +36,30 @@ location:
 da_setup:
     da_method: enkf
     par_fit_method: perturb
-    ensemble_size:  221
+    ensemble_size:  31
     localization_distance: .na #distance in meters were covariances in the model error are used
     no_negative_states: TRUE
     assimilate_first_step: FALSE
     use_obs_constraint: TRUE
-    obs_filename: fcre-targets-rs.csv
+    inflation_factor: 1.0
+    add_random_noise: 2
+    obs_filename: fcre-targets-insitu.csv
 model_settings:
    ncore: 4
    model_name: glm_aed
    base_GLM_nml: glm3.nml
    base_AED_nml: aed2.nml
    base_AED_phyto_pars_nml: aed_phyto_pars.csv
+   base_AED_phyto_pars_nml_file: aed2.nml
    base_AED_zoop_pars_nml: aed2_zoop_pars.nml
    max_model_layers: 100
    modeled_depths: [0.00,1.00,1.60,2.00,3.00,4.00,5.00,6.00,7.00,8.00,9.00]
    par_config_file: parameter_calibration_config_aed.csv
    obs_config_file: observations_config_aed.csv
    states_config_file: states_config_aed.csv
    depth_model_sd_config_file: depth_model_sd_aed.csv
+   non_vertical_noise_config_file: non_vertical_noise_config.csv
+   depth_sd: 0.0
 default_init:
    lake_depth: 9.4  #not a modeled state
    temp: [25.667, 24.9101, 23.067, 21.8815, 19.6658, 16.5739, 12.9292, 12.8456, 12.8127, 12.8079, 12.778]
@@ -87,6 +100,7 @@ output_settings:
    diagnostics_names:  [extc]
    generate_plots: FALSE
    evaluate_past: FALSE
+   restart_save_timesteps: "all"
    diagnostics_daily:
      names: ['CAR_ch4_atm','CAR_atm_co2_flux', 'temp', 'OXY_oxy', 'temp', 'temp']
      save_names: ['ch4_flux_mean', 'co2_flux_mean', 'temp_1.6m_mean', 'oxy_mean', 'temp_1.0m_mean', 'temp_8.0m_mean']

configuration/glm_aed_flare_rs/configure_run.yml

@@ -1,9 +1,10 @@
 restart_file: .na
-start_datetime: 2024-10-01 00:00:00
+start_datetime: 2026-04-30 00:00:00
 end_datetime: .na
-forecast_start_datetime: 2024-12-20 00:00:00
-forecast_horizon: 34
-sim_name: glm_aed_flare_rs
+forecast_start_datetime: 2026-05-04 00:00:00
+forecast_horizon: 1
+sim_name: glm_aed_flare_rs_demo_v4
 configure_flare: configure_flare_glm_aed.yml
 configure_obs: observation_processing.yml
 use_s3: TRUE
+use_faasr: TRUE

configuration/glm_aed_flare_rs/glm3.nml

@@ -57,6 +57,8 @@
    out_fn = "output"
    nsave = 1
    csv_lake_fname = 'lake'
+   restart_fname = 'glm_restart.nc'
+   restart_nsave = 1
 /
 &init_profiles
    num_heights = 28
@@ -73,6 +75,8 @@
    avg_surf_temp = 6
    restart_variables = 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
    restart_mixer_count = 0
+   init_restart_fname = 'glm_restart.nc'
+   init_restart_from_file = 0
 /
 &meteorology
    met_sw = .true.

configuration/glm_aed_flare_rs/non_vertical_noise_config.csv

@@ -0,0 +1,4 @@
+model_variable,model_source,model_depth_m,process_noise_sd
+lake_depth,state,NA,0.01
+extc,diagnostic,1.0,NA
+CAR_ch4_atm,diagnostic_daily,NA,0.0

configuration/glm_aed_flare_rs/observations_config_aed.csv

@@ -1,7 +1,6 @@
-state_names_obs,obs_units,obs_sd,target_variable, multi_depth
-temp,degC,1,Temp_C_mean,1
-OXY_oxy,mmol m-3,15,DO_mgL_mean,1
-OGM_doc_total,mmol m-3,18.7,fDOM_QSU_mean,1
-PHY_TCHLA,ugL,2,Chla_ugL_mean,1
-secchi, unitless,0.25, Secchi_m_sample,0
-
+state_names_obs,obs_units,obs_sd,target_variable,multi_depth,model_source,model_variable,model_depth_m
+temp,degC,1,Temp_C_mean,1,NA,NA,NA
+OXY_oxy,mmol m-3,15,DO_mgL_mean,1,NA,NA,NA
+OGM_doc_total,mmol m-3,18.7,fDOM_QSU_mean,1,NA,NA,NA
+PHY_TCHLA,ugL,2,Chla_ugL_mean,1,NA,NA,NA
+secchi,unitless,0.25,Secchi_m_sample,0,diagnostic,extc_coeff,1.0

workflows/glm_aed_flare_rs/generate_inflow_forecast.R

@@ -1,14 +1,27 @@
 library(tidyverse)
 
-lake_directory <- here::here()
-config_set_name <- "glm_aed_flare_v3"
-configure_run_file <- "configure_run.yml"
-config <- FLAREr::set_up_simulation(configure_run_file,lake_directory, config_set_name = config_set_name)
+if (!exists("lake_directory", inherits = FALSE))     lake_directory <- here::here()
+if (!exists("config_set_name", inherits = FALSE))    config_set_name <- "glm_aed_flare_v3"
+if (!exists("configure_run_file", inherits = FALSE)) configure_run_file <- "configure_run.yml"
+if (!exists("config", inherits = FALSE)) {
+  config <- FLAREr::set_up_simulation(configure_run_file, lake_directory, config_set_name = config_set_name)
+}
 
 print('read VERA targets...')
 
-targets_vera <- readr::read_csv("https://amnh1.osn.mghpcc.org/bio230121-bucket01/vera4cast/targets/project_id=vera4cast/duration=P1D/daily-inflow-targets.csv.gz",
-                                show_col_types = FALSE)
+if(!is.null(config$s3$vera4cast_targets)){
+  targets_local <- file.path(tempdir(), "daily-inflow-targets.csv.gz")
+  FLAREr::flare_get_file(local_file = "daily-inflow-targets.csv.gz",
+                         remote_file = "daily-inflow-targets.csv.gz",
+                         server_name = "vera4cast_targets",
+                         local_folder = tempdir(),
+                         remote_folder = "vera4cast/targets/project_id=vera4cast/duration=P1D",
+                         config = config)
+  targets_vera <- readr::read_csv(targets_local, show_col_types = FALSE)
+} else {
+  targets_vera <- readr::read_csv("https://amnh1.osn.mghpcc.org/bio230121-bucket01/vera4cast/targets/project_id=vera4cast/duration=P1D/daily-inflow-targets.csv.gz",
+                                  show_col_types = FALSE)
+}
 
 inflow_hist_dates <- tibble(datetime = seq(min(targets_vera$datetime), max(targets_vera$datetime), by = "1 day"))