#!/usr/bin/env perl 
#
# This script looks for the latest cycle of a model and
# creates the gif files for the MAG website.
#
# Arguments:
#    
#     -m model           - (required) valid models defined in %Config::areaHash
#     -e exception_mode  - currently none 
#     -l logging_level   - overrides $Config::default_logging_level
#           0 - debug
#           1 - info
#           2 - warn
#           3 - fatal
#      -x and -y: specify the date (-x) and cycle (y) to process
#      -a area
#      -p product
#      -s simulation mode, don't actually execute command list
#      -d development mode - no POE, execute commands sequentially.
#  
# Required: $FIXmag/MAG.xml - defines everything required for each model:
#       areas
#       parameters 
#       input filename patterns
#       forecast hour ranges and timesteps
#       max time to make each parameter (gif)
#       max number of concurrent commands to run in poe
#       scripts to make the parameters 
#
#   magv3-xml-library.pl - functions that provide the interface to the XML file
#
#
#   ${Config::capture_output_to_log_script} - (run_and_log_script.sh) - 
#           Each command line in a poe script is preceeded by this.  
#           It handles routing the output to the individual gempak log files
#           and to the collective one for the cycle as described below.
#
#   Logs, temp dir, status dir:
#
#       Optional output log is written to $HOMEmag/log/MAG_processor_<model>.log.<date>
#          (if $Config::make_log_file != 0)
#
#       A temp directory is created: Config::$tmpdir/MAG_processor_<jobid> 
#       It is deleted if this script ends successfully. Under this directory:
#         -subdir for each poe execution called job_<num>
#         -log for each command called <model>_<cycle>_<date>_<fhr>_<param>
#
#      $Config::status_dir 
#              - one file for every model/cycle to keep track of what forecast
#                hours have been processed
#              - pid file to detect if this script is already running for this model
#                type (they are hidden, only because that code was lifted from DBNnet...)
#      
#
#  Paula Freeman - 3/01/2013  - new version based on old MAG_processor running at Bob.
#  Paula Freeman - 3/19/2013 - execute list of commands by grouping into
#                              jobs of $Config::max_threads commands
#                              each and execute the job with mpirun.lsf.
#                              With this change, this script now must run 
#                              in a batch queue. 
#  Paula Freeman - 3/27/2013 - Add '-d' switch, for development.  This will execute
#                              commands in a poe job in the forground (not in parallel)
#                              to allow executing the script at the command line
#                              for development and debugging.
#  Lin Gan       - 3/29/2013 - Add logging for each stage of execution.
#                              Add time elapsed info for each job executed
#                              and entire run.
#                              Fix bug in check_for_exceptions -- don't  use
#                              shift when cycling through forecast_hr_list
# Paula Freeman - 3/29/2013  - Where possible, move log messages in main program
#                              before and after call to a subroutine to the subroutine 
#                              itself to keep main program more readable.
# Paula Freeman - 4/3/2012   - Fix bug: in check_for_exceptions, if an area exception is found, 
#                              forecast_hr_list was repopulated with new set of hours, and left that way.
#                              Fix by setting a variable when the list is altered, and if 
#                              no exception is found, but the list was altered, then repopulate
#                              with all hours in the range being processed.
#                            - change "submitted" in variables to "processed" to be more 
#                              descriptive of the current processig.
#                            - Clean up: replace "determine_fhr_range" with get_model_first_last_hour, 
#                              and remove unused global variables "last_fhr_this_range" and "step".  
#                              Now local variables used in load_list.
# Paula Freeman - 4/9/2013   - Fix error checking in process_command_list. An error when opening
#                              the pipe indicates an error forking, not an error in execution.
#                              That is reported when the file pipe is closed.
#                            - Make log file optional with configuration file variable $make_log_file.
#                              When running in batch, all output is captured in a log file.
# Paula Freeman - 4/19/2013    Make log optional. Having a separate log in production is useless, 
#                              and one more thing to clean up.
# Paula Freeman - 6/07/2013  - nam-sim-radar exception no longer used, it has been defined as a separate
#                              model in the MAG.xml file.  Leaving the code in for now.
# Paula Freeman - 8/27/2013  - Add ecFlow event if files are found and processed, and create a "go" file for
#                              the rsynch job to tell it what cycle to synch.  Only create if this
#                              is not running in development mode.
# Paula Freeman - 9/05/2013  - Process all cycles in @latest_cycles instead of just the latest.
#                              It will contain 2 cycles, if two are found.
# Paula Freeman - 4/20/2015  - Use env variable PDY for log file date, if defined.
# Paula Freeman - 5/07/2015  - Redmine 5984: Change info passed to sync job: remove "go" file, 
#                              create "transfer file", the list of filenames for rsync transfer.
# Paula Freeman - 4/27/2015  - Redmine 5978: In poe job, set MP_PULSE to set a max number of seconds for job 
#                              torun. Set to 1200 (20 min).  
# D.Xu/K.Wolou  - 5/07/2015  - Redmine 5935: moved INFO to DEBUG to reduce log, also added "sub logmsg_gempak" to
#                              log GEMPAK running messages.
# Paula Freeman - 6/9/2015   - Require $COMIN environment variable, add as argument
#                              to get_filepathpats function.
# Paula Freeman - 6/15/2015  - Move transfer files to "transfer" subdirectory of status directory
# Paula Freeman - 6/15/2015  - Set latest_cycles array to 3 cycles for hrrr or hrrr-subh
# Paula Freeman - 6/16/2015  - Create check_output_dirs to check for and create the status
#                              dir and transfer dir. Move creation of status dir from 
#                              set_highest_fhr_processed to check_output_dirs
# Paula Freeman - 11/04/2015 - User Config::transfer_dir instead of Config::status_dir/transfer
#                              Remove random number from transfer file name.
# Paula Freeman - 11/25/2015 - Redmine 12930: Make number of cycles to process configurable.
#                              Read in from MAG.xml. Add a command line option "c" to allow it too be
#                              overridden if command line value is specified.
#                              pass Config::status_dir and Config::transfer_dir into check_output_dirs
#                              instead of accessing globally.
# Deyong Xu     - 12/22/2015 - Add system time to filename of transfer file to make it unique. 
# Paula Freeman - 1/27/2016  - Add subdir_levels, which define whether the GIF
#                              directorys will have subdirectories:
#                                 none           - no subdirectories
#                                 areas          - areas level subdirectories
#                                 areas/parameters - two levels: areas/parameters
# Paula Freeman - 2/09/2016  - Redmine 14708: Add field for areas - valid-cycle, which defines
#                              cycles valid for that area, and overrides
#                              cycles defined for model. add function  check_valid_cycle
#                              Add $date and $cycle to make_command_list argument list
#                              They were being access globally, but without main:: ?
# Paula Freeman - 2/23/2016  - Redmine 14708: Add cycle-specific filepathpats to enable
#                              combining HRW-NMMB-[PR|AK] into HRW-NMMB (same for HRW-ARW).
#                              Each area is only valid for 2 cycles. Fetch cycle-specific filepats
#                              from MAG.xml with get_cycle_filepathpats, and in get_highest_fhr_found
# Paula Freeman - 3/16/2016    Redmine 5981: Add ability ignore cycles that match file
#                              pattern if not a valid cycle. Call get_cycles_by_model_name in 
#                              main routine. In get_file_list, call check_valid_cycle
#                              before adding files to the @all_filenames array and $datecycles 
#                              hash.                                
# Paula Freeman - 3/21/2016  - Add sync to Boulder (call it 'B'). If env variable SYNC_TO_B
#                              is "YES", then copy transfer files to Config::$transfer_B_dir
# Paula Freeman - 4/14/2016 - Rename check_output_dirs to check_and_create_output_dirs.
# Paula Freeman - 5/24/2016 - Control making transfer files with environment
#                             variables:
#                                if SYNC_TO_A == YES, make in Config::transfer_A_dir 
#                                if SYNC_TO_B == YES, make in Config::transfer_B_dir
#
# Paula Freeman - 6/14/2016 - Changes required to merge wcoss and web MAG.xml files:
#                           * In get_area_exception function, ignore optional
#                             timestep field returned by get_area_exception_by_name
#                             (after second comma). TODO: Use the timestep field!
#                           * Change how we determine if forecast hour minutes are
#                             defined and if we make the minutes or just the hour, as
#                             follows:
#                             Fetch the forecast mins from MAG.xml to see if they are
#                             defined for this model.  If they exist, then fetch     
#                             make-fmins tag, if "yes", make only the minutes. 
#                             If "no", make only the hour. 
# Paula Freeman - 7/7/2016  - In get_area_exception function, hardcode gfs exceptions
#                             for precip production, ignore any timestep field after
#                             second comma.  TODO: parse timestep and use it. 
# Paula Freeman - 7/7/2016  - Hard-code exception to only make GFS precip_.* products
#                             every 3 hours.
# Kit Menlove   - 7/16/2016 - Requeue send2web jobs in ecFlow when processing begins and
#                             added option to use the Command File Processor utility
# Paula Freeman - 7/27/2016 - Redmine 20319: Recognize and use optional timestep field 
#                             in exceptions tag:
#                             - In get_area_exception function, recognize the optional 
#                               timestep field and return it.
#                             - In check_for_exceptions function, return timestep.
#                             - In make_command_list, if timestep defined, skip fhrs that
#                               do not match.
#                             - Remove the two hardcoded GFS exceptions: precip and 
#                               dom_precip_type. No longer needed now that timestep 
#                               field is recognized.
#                             - Add options to make command line testing easier:
#                                * -x and -y: specify the date (-x) and cycle (y) to process
#                                * -s simulation mode, don't actually execute command list
#                             - Rename variables/functions to clarify their meaning:
#                                  (function) check_for_exceptions -> get_fhrs_to_process
#                                  (function) get_model_info -> get_model_fhr_ranges
#                                  @model_info -> @model_fhr_ranges
#                                  $first_fhr, $last_fhr -> $MODEL_FIRST_FHR, $MODEL_LAST_FHR
#                                  $highest_fhr -> $highest_input_fhr_found
#                                  In make_command_list:
#                                  $last_hr_to_process ->  $highest_input_fhr_found
#                                  $begin_fhr/$end_fhr -> $first_fhr_to_process, $last_fhr_to_process
#                                  $development_mode -> $command_line_mode
#                             - Fix tests for environment variables to test for if defined but empty.
#                             - Update log msg at end of load_list to include range
#                             - Add in "TO DO" comments from reviews to revisit later.
# Deyong Xu - 8/26/2016 - Redmine 20351: Introduced COMIN2 for ESTOFS model and hard-coded to find the 
#                                the highest fhr of ESTOFS model that satisfies its two input file patterns.
# Paula Freeman - 11/30/2016  - 25640: Update the NAM sim_radar exception to ignore the
#                               sim_radar_1km parameter instead of sim_radar 
# Deyong Xu - 12/5/2016 - Redmine 20920: Introduced COMIN3 for ESTOFS model and hard-coded to read
#                                GEMPAK input for east coast.
# Paula Freeman - 4/19/2017    - Add model soundings.  Get the list of stations for each area.
#                                If param is 'fcst_skewt', generate a command for each station in the area, 
#                                passing station name to the gempak script instead of the parameter.
# Paula Freeman - 7/21/2017    - Add model name to station area file
# Paula Freeman - 8/18/2017    - Updates for SREF-CLUSTER model:
# #                                1. Two more filename pattern designators:
#                                     FF (two digit fhr instead of 3) 
#                                     NNN (one-digit cluster number)
#                                     If a file exists with any cluster number (1-6)
#                                     then it satisfies the pattern.
#
#                                  2. Only create a command for one cluster.
#                                     Search through the files list and remove
#                                     keep only one per forecast hour 
#
#                                  3. Create transfer file after each batch of
#                                     commands have been executed, instead of
#                                     while building the list.  If SREF-CLUSTER
#                                     look for each of the 6 possible cluster
#                                     files and add to the transfer list if present.
# Paula Freeman - 3/14/2018    - Remove nam-sim-radar exception
# Paula Freeman - 3/14/2018    - Add ability for fhrs to vary by cycle
# Paula Freeman - 5/25/2018    - Update hard-coded kludge in get_file_list for 'polar' 
#                                to 'ice-drift'
# Paula Freeman - 8/23/2018    - Add ability to check for valid cycle by parameter.
#                                New valid-cycle tags under product exceptions.
#                                Add function check_valid_cycle_param and rename
#                                check_valid_cycle to check_valid_cycle_area
# Paula Freeman - 8/23/2018    - Fix for nbm.  It has exceptions for params where
#                                the first valid hour is not a multiple of the new
#                                timestep.  In make_command_list, when processing an
#                                exception, need to determine the offset of the
#                                first 1st valid fhr from multiples of the timestep.
#                                Then check that the fhr is the same offset hours 
#                                from the multiples of the timestep.  Update 
#                                get_fhrs_to_process to return the first valid hr
#                                of the model from the exception string.
# Paula Freeman - 11/29/2018   - Changes to keep running for longer, waiting for new fhrs to arrive:
#                                If any of the latest cycles is not complete (all fhrs found)
#                                then sleep and loop through again for $max_retries_for_more_fhrs
#                                (from MAG.xml for this model) times waiting for more fhrs.  
#                                Counter resets if another fhr arrives.
#                              - Fix bug:
#                                Rename the date/cycle variables passed in via -x and -y for
#                                testing. This was causing the last cycle to be reprocessed instead
#                                of processing the lastest_cycles array.
# Paula Freeman - 1/15/2019    - Phase 3 migration (MAG-111).  Use 'mpirun cfp' instead of 'mpirun.lsf cfp '
# Paula Freeman - 1/24/2019    - Restructure write_transfer_recs() to check for the existance of 
#                                large/small and minute files. MAG-121 (Bugzilla #839)
# Paula Freeman - 1/24/2019    - Check $KEEPDATA env variable before removing temp dir.
#                                Keep dir if YES|yes|1. MAG-120 (Bugzilla 840)
# Paula Freeman - 3/2019       - Add storm-tracks excepton to client call to requeue send2web job.
#                                Add better logging when attempting to requeue the send2web job.
# Paula Freeman - 3/2019       - Add a sleep before exiting to ensure that the job does not 
#                                complete and requeue before the sync jobs get a chance 
#                                get kicked off by the event. Also a ping before the event.
# Paula Freeman - 6/2019       - Add options to specify area and/or product to process: 
#                                -a <area> and -p <product>
#                              
#
#Anu Simon - 4/2020           Moved exmag-v3.pl from /scripts to /ush
#                            The corresponding changes  made in the script
#                            Renamed mag_processor_config to mag_processor_config.pl in /ush
#Anu Simon 4/24/2020         Changed script to read as mag_processor_config.pl in the appropriate places.
#Anu Simon 09/28/2020        With the estofs model upgrade there is no need to set $COMIN2 for estofs
##                           Removing dependency for $COMIN2 for ESTOFS
#Anu Simon 11/01/2021        Made changes pertaining to wcoss2 - mpiexec commands
#                            Job submission changes on wcoss2
#                            
#Anu Simon 07/2022           Perl version updates
#                            Changes to ecf job name to conform with production
#                            With imput from SPA, changes to sync jobs.
#
#
#Anu Simon 08/2022          Bugzilla ticket #1279.
#                           Status files are requitred to be written to subdirectory pertaining to
#                           $model and $PDY
#
#Anu Simon 10/2022          Changes to accomodate 2m_min_temp and 2m_max_temp having
#                           different fhrs for different valid cycles                           
#Anu Simon 11/2022          Status files written to dated directories                           
#
#Hieu Phung 01/2023        Add logic to check if a similar processor job is already running
#Anu Simon 03/2023         Revert status files and transfer files to write to common directories
#Anu Simon 08/2023         NAEFS precip products need GEFS ensembles which is defined in another directory and is called by COMIN3
#                          CMCE ensembles are in the same NAEFS directory. 
#Elliot Tardif 08/2024     Add logic for running NBM python scripts in the first forecast hour temp. directory
#                          $initval
#Anu Simon 08/2024         Incorporate the $initval logic for all the other model runs.                          
#E Tardiff 04/2025         Temp directories written with a date stamp
############################################################################3
use strict;
use warnings;
use File::Basename;
use File::Path qw(remove_tree);
use File::Copy;
use Switch;
use POSIX qw(strftime);     # for strftime
use FileHandle;
use Getopt::Std;
use Time::HiRes qw/usleep/;
use POSIX ":sys_wait_h" ;
use XML::LibXML;
use Time::HiRes;
use DateTime;		 

# Declare functions


sub int_handler();
sub make_temp_dir($);
sub get_model_fhr_ranges($$);
sub get_fhr_ranges($$);
sub get_params ($);
sub get_file_list ($$);
sub get_script ($$);
sub get_model_first_last_hr ($$);  
sub get_next_fhr($);
sub load_list($$$);
sub get_highest_fhr_found($$$$);
sub set_highest_fhr_processed($$$);
sub get_highest_fhr_processed($$);
sub make_command_list($$$$$$);
sub process_command_list($$$$);
sub get_fhrs_to_process ($$$$$);
sub write_transfer_recs($$$$$$$);
sub get_area_exception($$$);
sub launch_processes();
sub launch ();
sub check_for_completed_processes ();
sub logmsg($$);
sub logmsg_gempak($$);
sub already_running($$);
sub is_batch_job_running($);
sub mkdirs($);
sub make_subdir_levels_path($$);
sub check_and_create_output_dirs($$$);
sub silence_used_once_warnings();
sub check_valid_cycle_area($$$);
sub check_valid_cycle_param($$$);
sub get_area_stations($$);

# From magv3-xml-library.pl 
#sub valid_model
#sub get_max_time 
#sub get_max_threads 
#sub get_model_in_filename
#sub get_filepathpats 
#sub get_cycle_filepathpats 
#sub get_areas_by_model_name 
#sub get_params_by_model_name 
#sub get_fhr_ranges_by_model_name 
#sub get_fhr_ranges_by_model_cycle
#sub get_script_by_model_name 
#sub get_area_exception_by_name_param 
#sub get_cycles_by_model_name

# Declare lexically scoped variables only used in main

my $datecycle="";		# Current date/cycle being processed yyyymmddcc
my $date="";			# yyyymmdd portion of $date/cycle
my $cycle="";			# cc portion of $datecycle
my $highest_fhr_processed="";	# Highest forecast hour processed for current cycle
my $highest_input_fhr_found="";	# Highest forecast hour found for current cycle
my $MODEL_FIRST_FHR="";		# First forecast hour of the model
my $MODEL_LAST_FHR="";		# Last forecast hour of the model
my $first_fhr_to_process="";    # First forecast hour after the last one processed
my $process_file="";     	# name of id file containing job id or pid,used 
				# to make sure only one instance is running
my $proc_error=0;		# Global error status
my $step;			# timestep for current forecast hour range
my $highest_hr_processed;	# highest forecast hour processed for this cycle
my $found_more_fhrs=0;          # 1 if found some fhrs this pass.
my $starting_process_clock_time;# Job starting clock time
my $current_loop_starting_time;	# Job starting clock time
my $end_process_clock_time;	# Job ending clock time
my $job_duration;		# Job run duration
my $rsync_time_stamp=localtime(time);           # RSYNC event time stamp
my $num_cycles_to_process;      # Number of cycles to process. From
                                # xml, or overridden on command line
my $max_retries_for_more_fhrs=""; # max number of times to cycle through loop looking of more fhrs
                                # after no new files have been found (fetched from MAG.xml)
my $num_retries_for_more_fhrs=0;# counter for number of times through the loop
                                # with no new files found.
my $keep_running=1;             # 1 to keep looping, 0 to exit.
                                # Set to 0 if all fhrs found and processed, if max number
                                # of retries have been made, or if an error occurs.
my $latest_cycles_all_processed=0;# Set to 1 if all the fhrs expected for the latest 
                                # cycles have been processed
my $this_cycle_all_processed=0; # 1 if all fhrs expected have been processed for the current cycle
my $initval;                    # This val is for running the nbm python scripts in the first_fhr directory


# Package global variables accessed by functions

our $model="";                  # model to to make products for.
our $MAG_dir="";                # MAG root dir
our $MAG_tables="";             # directory for MAG tables
our $COMIN="";                  # Value of COMIN environment variable
our $COMIN2="";                 # Value of COMIN2 environment variable
our $COMIN3="";                 # Value of COMIN3 environment variable
our $COMOUT="";                 # Value of COMOUT environment variable
our $COMOUTstatus="";           # Value of COMOUTstatus environment variable
our $SYNC_TO_A="";              # Value of SYNC_TO_A environment variable
our $SYNC_TO_B="";              # Value of SYNC_TO_B environment variable
our $KEEPDATA="";              # If 'yes', overrides removing temp dir on exit
our $logname="";                # Base name of the log file (without the date at the end), 
our $logfile="";		# Filename of current log file $log_dir/$logname.<date>
our $job_tmp_dir="";            # temp dir for this run.  Removed at exit, unless it ends in fatal error
our $log=new FileHandle;	# Log file handle
our @forecast_hr_list=();	# List of forecast hours found for current cycle
our @model_cycles=();           # List of cycles valid for this model
our @latest_cycles=();		# Latest two cycles found
our @filepathpats=();		# Filepath patterns
our %cycle_filepathpats=();	# If defined, cycle-specific filepath patterns (cycle points to list of patterns)
our %all_filenames=();		# All filenames that match pattern for the current cycle
our $transfer_seq=0;            # Counter used in transfer file names.

our @command_list=();		# Queue of commands generated for new files found
our $max_threads;		# number of children to run concurrently
our $subdir_levels;		# defines if model cycle directories are further
                                # subdivided into areas, or areas/parameters. 
                                # value: none | areas | areasparams
                                # defaults to none
our $model_in_filename;         # model string used in GIF filenames
our $number_fhr_ranges=0;      	# Number of forecast hour ranges from model_fhr_ranges
our @model_fhr_ranges=[];		# One row for each forecast hour range: first hr, last hr, timestep
our $die=0;                    	# set to 1 if we receive INT or KILL signal
our $models_doc;                # parsed xml doc containing all model config info.
our $models_root;               # root of xml doc containing all model config info.
our $this_model;                # root of xml doc containing all model config info.
our @areas=();               	# valid areas for this model
our @params=();               	# parameters to make for this model
our $max_run_time;		# max run time for each command
our $exception_mode;            # if nam_sym_radar, then only make nam sim_radar_1km and
                                # ignore forecast hour ranges from xml file.
our $command_line_mode;         # For running at the command line.  Do not execute commands
                                # in poe, run sequentially.
our $logging_level;		# level of logging, default or overridden on command lie
our $reload_fhr_list=0;         # If an area exception is found for a parameter, the 
                                # forecast_hr_list is updated when processing that parameter.
                                # It needs to be reloaded after that for the next parameter.

##################################
#     Fetch options
##################################
#
# TO DO: All parameters that do not change should be stored in variables
#        in ALL CAPS. All values fetched from MAG.xml, all options, etc.
#  -m  <model>
#  -e  <exception mode -- no longer needed>
#  -l  <logging level>
#  -d  development model - commands executed sequentially without parallel environment
#  -c  <# cycles to process>
#  -x  <date>
#  -y  <cycle>
#  -a  <area>
#  -p  <param>
#  -s  simulation model - doesn't actually create products

my %options;
getopts('m:e:l:dc:x:y:s:a:p:', \%options);


# Model name to process

$model= defined $options{m} ? $options{m} : "";

# Exception mode - not currently used.
#
$exception_mode= defined $options{e} ? $options{e} : "none";

# Develoment mode - execute command list sequentially, do not use poe
# (intended for execution on the command line during development)

$command_line_mode= defined $options{d};

# Input date/cycle/area/product to process (used in testing)

my $process_date = defined $options{x} ? $options{x} : "";
my $process_cycle = defined $options{y} ? $options{y} : "";
my $process_area = defined $options{a} ? $options{a} : "";
my $process_param = defined $options{p} ? $options{p} : "";

# 

# Simulation mode (do not actually execute the command list, used in testing)
my $simulation_mode = defined $options{s};

# If model not specified, exit.

if (! $model)  {
   print "usage: $0 -m <model>  [-l <logging_level>] [ -c <num-cycles-to-process> ]\n";
   print "       development options:  [ -a <area> ] [ -p <product> ] [ -x <date> -y <cycle> ]\n";
   print "       simulation mode: -s \n";
   exit 1;
}


#--------------------------------------------------------------
# evaluate environment variables: 
#    HOMEmag, FIXmag, SYNC_TO_A, SYNC_TO_B, and KEEPDATA
#--------------------------------------------------------------


$MAG_dir=$ENV{'HOMEmag'};
(defined($MAG_dir) &&  $MAG_dir) or die  "Environment variable HOMEmag not set.\n";

$MAG_tables=$ENV{'FIXmag'};

(defined($MAG_tables) &&  $MAG_tables) or die  "Environment variable FIXmag not set.\n";


$COMIN=$ENV{'COMIN'};
$COMIN2=$ENV{'COMIN2'};
$COMIN3=$ENV{'COMIN3'};

(defined($COMIN) &&  $COMIN) or die  "Environment variable COMIN not set.\n";




# Check the 2nd input data location for models such as stofs or naefs.
if ( $model eq "naefs" |  $model eq "stofs" ) {
   (defined($COMIN3) || $COMIN3 ) or die  "Environment variable COMIN3 not set.\n";
}     

# Root of output GIF directory
#
$COMOUT=$ENV{'COMOUT'};
$COMOUTstatus=$ENV{'COMOUTstatus'};

$SYNC_TO_A=$ENV{'SYNC_TO_A'};  # If defined and "YES" then make transfer 
                               # files in Config::transfer_A_dir

$SYNC_TO_B=$ENV{'SYNC_TO_B'};  # If defined and "YES" then make transfer 
                               # files in Config::transfer_B_dir
                               #
# Overrides removing temp directory on exit:
#
$KEEPDATA=$ENV{'KEEPDATA'};

#--------------------------------------------------------------
# read in configuration file and XML routines
#--------------------------------------------------------------
require "${MAG_dir}/ush/mag_processor_config.pl" or die "Unable to open ${MAG_dir}/ush/mag_processor_config.pl";
require "${MAG_dir}/ush/magv3-xml-library.pl" or die "Unable to open ${MAG_dir}/ush/magv3-xml-library.pl";




#--------------------------------------------------------------
# read in models xml doc
#--------------------------------------------------------------
  my $parser = XML::LibXML->new();
  $models_doc    = $parser->parse_file("${MAG_tables}/MAG.xml");
  $models_root = $models_doc->getDocumentElement ;

# Set logging level to default if not set in command switch

$logging_level = defined $options{l} ? $options{l} : $Config::default_logging;
$logname="${Config::log_string}_${model}.log";

# Set number of cycles to process, overrides MAG.xml value if specified

$num_cycles_to_process = defined $options{c} ? $options{c} : 0;

logmsg($Config::info, "Begin exmag_processor.pl version $Config::version \n");

logmsg($Config::info, "XML Configuration file read successfully. \n");

if (defined($SYNC_TO_A) && "$SYNC_TO_A" eq "YES") {
   logmsg($Config::info, "Synching to site 'A'");
} else {
   logmsg($Config::info, "NOT synching to site 'A'");
   $SYNC_TO_A="NO";
}

if (defined($SYNC_TO_B) && "$SYNC_TO_B" eq "YES") {
   logmsg($Config::info, "Synching to site 'B'");
} else {
   logmsg($Config::info, "NOT synching to site 'B'");
   $SYNC_TO_B="NO";
}

#
# check for valid model 
#

logmsg($Config::info, "Checking for valid model $model ... \n");
if (! valid_model($models_doc, $model)) {
   logmsg($Config::fatal, "Unknown model name: $model so exit. \n ");
   exit 1;
}


#
#  Check for output directories and create them if necessary
#



our ($status_dir);
     $status_dir="$main::COMOUTstatus";
our $transfer_A_dir="${status_dir}/transfer_A";
our $transfer_B_dir="${status_dir}/transfer_B";

$proc_error=check_and_create_output_dirs($status_dir, $transfer_A_dir, $transfer_B_dir);


#--------------------------------------------------------------
# read in configuration file and XML routines
#--------------------------------------------------------------
require "${MAG_dir}/ush/mag_processor_config.pl" or die "Unable to open ${MAG_dir}/ush/mag_processor_config.pl";
require "${MAG_dir}/ush/magv3-xml-library.pl" or die "Unable to open ${MAG_dir}/ush/magv3-xml-library.pl";


#--------------------------------------------------------------
# read in models xml doc
#--------------------------------------------------------------
$parser = XML::LibXML->new();
$models_doc    = $parser->parse_file("${MAG_tables}/MAG.xml");
$models_root = $models_doc->getDocumentElement ;

# Set logging level to default if not set in command switch

$logging_level = defined $options{l} ? $options{l} : $Config::default_logging;
$logname="${Config::log_string}_${model}.log";


#
# get the max run time for each command (not currently used!)
# 

logmsg($Config::info, "Get the max run time for each command (not currently used!)... \n ");
$max_run_time=get_max_time($models_doc, $model);
logmsg($Config::info, "Get the max run time successful: $max_run_time \n");
#
# get the number of cycles to process
# 
if ($num_cycles_to_process > 0) {
   logmsg($Config::info, "Number of cycles to process: $num_cycles_to_process (defined by command line argument)... \n ");
} else {
   logmsg($Config::info, "Get the number of cycles to process... \n ");
   $num_cycles_to_process=get_model_cycles_to_process($models_doc, $model);
   logmsg($Config::info, "Get the number of cycles to process successful: $num_cycles_to_process \n");
}

# 
# get the max numer of retries for more fhrs
# 
logmsg($Config::info, "Get the max number for retries to wait for more fhrs \n ");

$max_retries_for_more_fhrs=get_model_max_retries_for_more_fhrs($models_doc, $model);
if ($max_retries_for_more_fhrs eq "") {
    $max_retries_for_more_fhrs=$Config::default_max_retries_for_more_fhrs;
    logmsg($Config::info, "Use default max retries to wait for more fhrs: $max_retries_for_more_fhrs \n");

}
logmsg($Config::info, "Get the max retries to wait for more fhrs successful: $max_retries_for_more_fhrs \n");

#
# get the model string used in the GIF filenames
# 

logmsg($Config::info, "Get the model string used in GIF filenames... \n ");
$model_in_filename=get_model_in_filename($models_doc, $model);
logmsg($Config::info, "Get the model string successful: $model_in_filename \n");

#
# get areas for this model
#

logmsg($Config::info, "Get the covered areas for $model ... \n ");
@areas = get_areas_by_model_name($models_doc, $model);
if ( $process_area ) {
   if ( grep( /^$process_area$/, @areas ) ) {
      logmsg( $Config::info, "Create images only for area $process_area. \n");
      @areas = ($process_area);
   }  else {
      logmsg( $Config::fatal, "$process_area is not a valid area for $model \n");
      $proc_error=1;
   }
}
if ( (scalar @areas) == 0) {
    logmsg( $Config::fatal, "No areas defined for model $model");
    $proc_error=1;
}
logmsg($Config::info, "Get the covered areas completed sucessfully. \n");

#
# get cycles for this model
#

logmsg($Config::info, "Get the defined cycles for $model ... \n ");
@model_cycles = get_cycles_by_model_name( $main::models_doc, $model);
if ( (scalar @model_cycles) == 0) {
    logmsg( $Config::fatal, "No cycles defined for model $model");
    $proc_error=1;
}
logmsg($Config::info, "Get the defined cycles completed successfully. \n");

#
# get parameters
#

logmsg($Config::info, "Get the parameters for model $model... \n");
@params = get_params_by_model_name($models_doc, $model);
if ( $process_param ) {
   if ( grep( /^$process_param$/, @params ) ) {
      logmsg( $Config::info, "Create images only for parameter $process_param. \n");
      @params = ($process_param);
   }  else {
      logmsg( $Config::fatal, "$process_param is not a valid product for $model \n");
      $proc_error=1;
   }
}
if ((scalar @params) == 0) {
    logmsg( $Config::fatal, "No parameters defined for model $model");
    $proc_error=1;
}
logmsg($Config::info, "Get parameters completed successfully. \n");

#
# get filepath patterns
#
# TO DO: Investigate looking for just initial hour.
#        Does it reduce the glob time?
#
logmsg($Config::info, "Get the file path patterns for model $model... ");

if ( $model eq "stofs" | $model eq "naefs" ) {
   # Models such as stofs or naefs get data from two input locations, so 
   # read in all the file patterns from two locations.
   @filepathpats = get_filepathpats_v2($models_doc, $model, $COMIN, $COMIN2, $COMIN3);
} else {
   @filepathpats = get_filepathpats($models_doc, $model, $COMIN);
}

if ((scalar @filepathpats) == 0) {
    logmsg( $Config::fatal, "No input filepath patterns defined for model $model");
    $proc_error=1;
}

logmsg($Config::info, "Get file path patterns completed successfully. \n");

#
# get optional cycle-specific filepath patterns
#

logmsg($Config::info, "Get the cycle-specific file path patterns for model $model... ");
  %cycle_filepathpats = get_cycle_filepathpats($models_doc, $model, $COMIN);
  if (keys %cycle_filepathpats == 0) {
      logmsg( $Config::info, "No cycle-specific input filepath patterns defined for model $model");
  }
logmsg($Config::info, "Get cycle file path patterns completed successfully. \n");

#
# Get max threads for this model 
#

logmsg($Config::info, "Get max threads for model $model... ");
$max_threads=get_max_threads($models_doc,$model);
if ($max_threads == 0) {
    logmsg( $Config::warn, "Max_threads undefined for model $model. Use default $Config::default_max_threads and proceed.");
    $max_threads = $Config::default_max_threads;
}
logmsg($Config::info, "Get max threads for model: $model is done. max_threads is $max_threads. \n");

        
#
# Get subdirectory definitions for this mode
#

logmsg($Config::info, "Get subdirectory levels for model $model... ");
$subdir_levels=get_model_subdir_levels($models_doc,$model);
if ($subdir_levels eq "") {
    logmsg( $Config::warn, "Subdirectory levels are undefined for model $model. Default to 'none' and proceed.");
    $subdir_levels = "none";
}
logmsg($Config::info, "Get subdirectory levels for model $model is done. subdir_levels = $subdir_levels. \n");

#
#  create id file string
#

$process_file="${model}_MAG";
logmsg($Config::info, "Id file string is: $process_file. \n");

#
# Check if already running elsewhere.  If so, exit.
#

already_running($status_dir, $process_file);

#
# Create a tmp dir for this run, and change working dir to it
#

$proc_error = $proc_error || make_temp_dir($model);

# set up for interrupts
$SIG{INT} = 'int_handler';
$SIG{QUIT} = 'int_handler';
$SIG{TERM} = 'int_handler';

# stamp job starting time
$starting_process_clock_time = Time::HiRes::time();

#
# Run until no new forecast hours are found, or signaled to die, or fatal error.
#


logmsg($Config::info, "Job starting time is: $starting_process_clock_time");

# Keep running while there is still more to do
#
$keep_running=1;

do {

   # Reset loop starting time
   $current_loop_starting_time = Time::HiRes::time();

   $found_more_fhrs=0;

   # Assume all latest cycles have been processed, set to 0 if all fhrs not processed.
   $latest_cycles_all_processed=1;


   # get the list of model files, determine latest cycles to process 

   $proc_error = $proc_error || get_file_list ($model, $num_cycles_to_process);

   # check for input date/cycle (Options intended for use in debugging only)

   if ($process_date && $process_cycle) {
      logmsg($Config::info, "Override dates/cycles found, process specified input date and cycle: $process_date $process_cycle");
      @latest_cycles=();
      $latest_cycles[0]=$process_date . $process_cycle;
   }

   # If no cycles were found, don't keep running.
   
   if (scalar @latest_cycles == 0 ) {
      $keep_running=0; 
   }
    
   # Process the latest cycles found
   #
  

   while ( ($datecycle=shift @latest_cycles) && ! $proc_error) {
 

      $datecycle =~ /(\d{8})(\d{2})/;
      $date=$1, $cycle=$2;
      logmsg($Config::info, "************************************");  
      logmsg($Config::info, "Processing date $date and cycle $cycle");
      logmsg($Config::info, "************************************\n"); 


      #
      # get the number of forecast hours expected for this cycle
      #

      ($MODEL_FIRST_FHR, $MODEL_LAST_FHR) = get_model_first_last_hr ($model, $cycle);

      # get highest forecast hour processed

      $highest_fhr_processed = get_highest_fhr_processed($date, $cycle);

      # Get the highest forecast hour received

      $highest_input_fhr_found = get_highest_fhr_found($date, $cycle, $MODEL_FIRST_FHR, $MODEL_LAST_FHR);
     logmsg($Config::info, "Check if we have processed all the forecast hours received..."); 
     logmsg($Config::info, "MODEL_FIRST_FHR (first hour of the model): $MODEL_FIRST_FHR ");
     logmsg($Config::info, "MODEL_LAST_FHR (last hour of the model): $MODEL_LAST_FHR");
     logmsg($Config::info, "highest_fhr_processed: $highest_fhr_processed ");
     logmsg($Config::info, "highest_input_fhr_found (highest forecast hour found): $highest_input_fhr_found ");
     

     if ($MODEL_LAST_FHR == $highest_fhr_processed) {

        logmsg($Config::info, "All forecast hours processed for cycle $date $cycle.\n");
         $this_cycle_all_processed=1;
     } else {
         $this_cycle_all_processed=0;

         logmsg($Config::info, "All forecast hours have NOT been processed for cycle $date $cycle.\n");
 
         # If we've gotten in some new forecast hours, submit the
         # jobs for them and update the highest processed.

         if ($highest_input_fhr_found <= $highest_fhr_processed )  {
            logmsg($Config::info, "All forecast hours found for cycle $date $cycle have been processed.\n");
         } else {
            logmsg($Config::info, "More forecast hours available... proceeding...");
            $found_more_fhrs=1;
   
            $first_fhr_to_process = get_next_fhr($highest_fhr_processed); 

            # load the list with the hours to be processed (
      
            load_list($model, $first_fhr_to_process, $highest_input_fhr_found);

            # Create the list of commands for hours > $highest_fhr_processed 
           
            $proc_error = make_command_list ($model, $date, $cycle, $highest_fhr_processed, $highest_input_fhr_found, $max_run_time);
  
	   
            if ( ! $simulation_mode ) { 
               $proc_error = $proc_error || process_command_list($model, $date, $cycle, $model_in_filename);	    

            # If simulation mode, do not actually execute the command list 
            } else {
               logmsg($Config::info, "In simulation mode... do not execute command list.");
            }
  
            # Store highest fhr processed.
            # Check if all the fhrs have been received and processed.
         }  
         if (! $proc_error && !$die) {

            set_highest_fhr_processed($highest_input_fhr_found, $date, $cycle);

            if  ($highest_input_fhr_found ==  $MODEL_LAST_FHR ) {
               logmsg($Config::info, "Fhrs just processed are all there are. set this_cycle_all_processed to 1 \n");
               $this_cycle_all_processed = 1;
            }
            if ($this_cycle_all_processed == 0) {
               $latest_cycles_all_processed=0;     # The latest cycles are not all processed
            }

            # Report POE stats
            #
            $end_process_clock_time = Time::HiRes::time();
            $job_duration = $end_process_clock_time - $current_loop_starting_time;
            logmsg($Config::info, "Current poe job submit information summary: $model $date $cycle: $highest_fhr_processed -> $highest_input_fhr_found ");
            logmsg($Config::info, "Current poe job time used (include overhead) is: $job_duration \n");

         }
      }
   }

   # Determine if it is time to exit the loop
   

   if ($latest_cycles_all_processed || $die || $proc_error) {
      $keep_running=0;
      if (! ($die || $proc_error)) {
         if ( $latest_cycles_all_processed ) {
            logmsg($Config::info, "Latest $num_cycles_to_process cycles are complete and processed.  Exit. \n");
         }
      }
   } else {

      # If some fhrs were found this time, reset the retries counter
      
      if ($found_more_fhrs) {
         $num_retries_for_more_fhrs=0;
      }
   
      # Otherwise, continue to run for up to $max_retries_for_more_fhrs times
      
      if ( $num_retries_for_more_fhrs > $max_retries_for_more_fhrs ) {
         $keep_running=0;
         logmsg($Config::info, "Max number for retries (${max_retries_for_more_fhrs}) looking for more forecast hours has been reached. Exit. \n");
      } else {
         logmsg($Config::info, "There are more forecast hours expected for one of the latest $num_cycles_to_process cycles. This was run #${num_retries_for_more_fhrs}. Sleep $Config::min_between_checks minute(s) and run again. \n");
         $num_retries_for_more_fhrs++;
         sleep  $Config::min_between_checks*60;
      }
  }


  $current_loop_starting_time = Time::HiRes::time();

} until (!$keep_running);

# Add a sleep to ensure that the job does not complete and requeue
# before the sync jobs get a chance to get kicked off in response
# to the event.

sleep  5; 

logmsg($Config::info, "Run has finished with proc_error set to: $proc_error \n ");

#
# Remove the temp directory 
#

if (! $proc_error) {
   if ( ! defined $KEEPDATA || $KEEPDATA !~ /^(YES|yes|1)$/ ) {
      logmsg($Config::info, "Remove the temp dir: $job_tmp_dir");
      if (! chdir $ENV{HOME} ) {
        logmsg($Config::fatal, "Cannot change to home dir $ENV{HOME} so I can remove the temp dir");
      } else {
         remove_tree($job_tmp_dir) ||  logmsg($Config::warn,"Cannot remove temporary dir (${job_tmp_dir}: $!");
         logmsg($Config::info, "Temp dir deleted successfully \n");
      }
   } else {
     logmsg($Config::info, "Do not delete temp dir $job_tmp_dir");
     logmsg($Config::info, "KEEPDATA set to $KEEPDATA");
   }
} else {
     logmsg($Config::info, "Do not delete temp dir $job_tmp_dir");
}


#
# Remove the process id file
#

logmsg($Config::info, "Unlink process file...");
unlink("${status_dir}/.${process_file}") || logmsg($Config::warn,"unlink(${status_dir}/.${process_file}  failed: $!");
logmsg($Config::info, "Unlink processing file done.");

#
# Report stats
#

$end_process_clock_time = Time::HiRes::time();
$job_duration = $end_process_clock_time - $starting_process_clock_time;
logmsg($Config::info, "Total time used: $job_duration \n");


#
# Report exit stats
#

if ($proc_error) {
   logmsg($Config::fatal, "Exiting on fatal error\n\n");
} elsif ($die) {
   logmsg($Config::warn, "Exiting on user interrupt\n\n");
} else {
   logmsg($Config::info, "Exiting successfully\n\n");
}

logmsg($Config::info, "Exiting MAG Processor; closing log.");
$main::log->close;

exit $proc_error;


#--------------------------------------------------------------
#                   Functions
#--------------------------------------------------------------

#--------------------------------------------------------------
# get_file_list: get the list of forecast hour files found 
# that match the patterns for this model, cycle, and date
# Determine the cycles that will be processed
#
# Input:
#    $model
#    $num_cycles_to_process
#
# Return:
#    0 - no error
#    1 - error getting filename patterns
#
# Changes globals:
#    %datecycles
#    @all_filenames
#    @filepathpats
#    @latest_cycles
#
# Paula Freeman 8/18/2017 - Update for SREF clusters:
#                             1. Two more filename pattern designators:
#                                FF (two digit fhr instead of 3) 
#                                NNN (one-digit cluster number)
#                                If a file exists with any cluster number (1-6)
#                                then it satisfies the pattern.
#                            2.  Keep only one cluster per fhr
#
sub get_file_list ($$) {

#--------------------------------------------------------------
# Read in the patterns file for this model.
# Replace YYYYMMDD and CC with date and cycle 
#--------------------------------------------------------------

   my $model=shift;
   my $num_cycles_to_process=shift;
   my ($date, $cycle, @wildcardpat, @REpat, @files, $file, $dc);
   my (@sorted_cycles, $pattern_file, $num_cycles, $npats);

   %$main::datecycles=();

   $npats=$#main::filepathpats + 1;
   %main::all_filenames=();
   logmsg($Config::info, "sub get_file_list proceeding...");
   # Loop through all the file patterns defined in MAG.xml
   # to make sure all the files matching the file patterns exist for the current cycle. 
   # For example, for STOFS there are 5 file patterns:
   #   stofs_pac.YYYYMMDD/stofs.pac_conus_YYYYMMDDCCfFFF
   #   stofs_pac.YYYYMMDD/stofs.pac_alaska_YYYYMMDDCCfFFF
   #   stofs_pac.YYYYMMDD/stofs.pac_hawaii_YYYYMMDDCCfFFF
   #   stofs.YYYYMMDD/stofsatl_YYYYMMDDCCfFFF
   #   gfs.YYYYMMDD/gfs_0p25_YYYYMMDDCCfFFF
   #   We need to find all the files availalbe on system for each of these file patterns. 
   for (my $i = 0; $i < $npats; $i++) {
      # 1. Search the system and return all the files that matches the current file pattern.
      chomp $main::filepathpats[$i];
      $wildcardpat[$i]=$main::filepathpats[$i];
      $wildcardpat[$i] =~ s/YYYYMMDD/\[0-9][0-9][0-9][0-9][0-9][0-9][0-9][0-9]/g;
      $wildcardpat[$i] =~ s/CC/\[0-9][0-9]/g;
      $wildcardpat[$i] =~ s/FFF/\[0-9][0-9][0-9]/g;
      # Updates for SREF-CLUSTER - cluster number and 2-digit fhr
      $wildcardpat[$i] =~ s/FF/\[0-9][0-9]/g;
      $wildcardpat[$i] =~ s/NNN/\[1-6]/g;
      @files=glob ($wildcardpat[$i]);

      # If SREF-CLUSTER, keep only one cluster per forecast fhr
      #
      if ($model eq "sref-cluster" ) {
         remove_extra_sref_clusters (\@files);
      }
            

      # 2. Define a regular expression
      $REpat[$i]=$main::filepathpats[$i];
      $REpat[$i] =~ s/YYYYMMDD/\(\\d\{8\}\)/;
      $REpat[$i] =~ s/YYYYMMDD/\\d\{8\}/;   # if there is a second YYYYMMDD, don't put in a backref to it.
      $REpat[$i] =~ s/CC/\(\\d\{2\}\)/g;
      $REpat[$i] =~ s/FFF/\(\\d\{3\}\)/g;
      # Updates for SREF-CLUSTER - cluster number and 2-digit fhr
      $REpat[$i] =~ s/FF/\(\\d\{2\}\)/g;
      $REpat[$i] =~ s/NNN/\\d/g;
      logmsg($Config::debug, "REpat[$i] = $REpat[$i]\n");
      logmsg($Config::debug, "wildcardpat[$i] = $wildcardpat[$i]\n");
      
      # 3. Check if the files returned above do match the RE, and retrieve 
      # date and cycle info from their filename if they do. 

      foreach $file (@files) {
        logmsg($Config::debug, "Looking for date and cycle to process in file: $file");
        $file =~ m/${REpat[$i]}/;  
        $date=$1; $cycle=$2;

        # Ice drift files do not contain the cycle anywhere in the filename,
        # It's only valid for cycle 00.
        
        if ($model eq "ice-drift") {
           $cycle="00";
        }

        if (check_valid_cycle_area($model, $cycle, "")) {
           # Use boolean 1/0 to indicate file existence via global array @all_filenames
           $main::all_filenames{$file} = 1;
           # Find a new cycle 
	   if (! exists $main::datecycles{"$date$cycle"} ) {
               logmsg($Config::debug, "Found date: $date and cycle: $cycle. Keep looking...");
               $main::datecycles{"$date$cycle"} = 1;
           }
        } else {
           logmsg($Config::debug, "Found date: $date and cycle: $cycle, but it's not valid for this model"); 
        }
      }
   }
  logmsg($Config::info, "All available dates and cycles found. \n");



# sort the date/time keys found, and get the latest cycles
#
   @sorted_cycles=( sort {$b <=> $a} keys( %main::datecycles ));
   $num_cycles=$#sorted_cycles + 1;

# If there are fewer cycles available than we want to process
# only process the ones available

   if ($num_cycles_to_process > $num_cycles) {
      $num_cycles_to_process = $num_cycles 
   } 

   for (my $i=1; $i <= $num_cycles_to_process; $i++) {
      $main::latest_cycles[$i-1]=$sorted_cycles[$i-1];
   } 

  logmsg($Config::info, "$num_cycles date/cycles found. \n");
  logmsg($Config::info, "Process $num_cycles_to_process cycles \n");

   return 0; 
}


#--------------------------------------------------------------
#
# get_highest_fhr_found: search the list of files for this $datecycle
# for the highest forecast hour
#
# Input:
#   $date
#   $cycle
#   $MODEL_FIRST_FHR
#   $MODEL_LAST_FHR
#
# Accesses globals:
#  @forecast_hr_list
#  @filepathpats
#  %cycle_filepathpats
#  @all_filenames
#
# returns: highest forecats hour found, or -1
#
# Paula Freeman 2/23/2016 - Update to use cycle-specific
#                           filepat list, if defined.
# Paula Freeman 8/18/2017 - Update for SREF clusters:
#                             1. Two more filename pattern designators:
#                                FF (two digit fhr instead of 3) 
#                                NNN (one-digit cluster number)
#                                If a file exists with any cluster number (1-6)
#                                then it satisfies the pattern.
#
#                             2. Only create a command for one cluster.
#                                Search through the files list and remove
#                                keep only one per forecast hour 
#
use strict;
use warnings;
use File::Basename;
use File::Path qw(remove_tree);
use File::Copy;
#
#
#--------------------------------------------------------------
sub get_highest_fhr_found ($$$$){
   use POSIX;
   
   my ($date, $cycle, $MODEL_FIRST_FHR, $MODEL_LAST_FHR) = @_;
   my ($highest_fhr, $got_all_this_pat, $keep_going, $hr, $checkforfile);
   my ($all_hours_in_one_file, $found_all_pats, $npats, $twodigit_hr);
   my ($cluster_file, $found);
   
   logmsg($Config::info, "Get the highest forecast hour received and forecast hours processed so far...");
   
   # Determine the filepaths to use for this cycle.
   # If there is a cycle-specific filepattern list, use it.
   #
   
   my @filepathpats_this_cycle;
   
   if (defined $main::cycle_filepathpats{$cycle}) {
      $npats= @{ $main::cycle_filepathpats{$cycle} };
      for (my $j = 0; $j < $npats; $j++) {
         $filepathpats_this_cycle[$j] = $main::cycle_filepathpats{$cycle}[$j];
      }
   } else {
      # file pattern exists
      $npats=$#main::filepathpats + 1;
      for (my $j = 0; $j < $npats; $j++) {
         $filepathpats_this_cycle[$j] = $main::filepathpats[$j];
      }
   }
   
   # Search through the files found, determine the
   # highest forecast hour for which we have all the
   # the patterns in the pattern file (usually just one)
   
   
   $highest_fhr=-1;
   logmsg($Config::debug, "find highest fhr received for cycle $date $cycle\n");
   
   # Load the forecast hour list with all the forecast hours.
   # It will be reloaded later with just the hours we'll process.
   
   load_list($model, $MODEL_FIRST_FHR, $MODEL_LAST_FHR);
   
   $keep_going=1;
   
   # Cycle through the forecast hours and filepatterns.
   # stop when we don't find all the filepatterns for a forecast hour.
      
   
   # TODO use while, e.g.  while ($keep_going) and increment in the block.
   #
   # Go thru all the fhrs saved in $main::forecast_hr_list and check 
   # to see if they are available on system.
   for (my $i=0; $keep_going; $i++) {
      $hr=$main::forecast_hr_list[$i];
      $twodigit_hr = $hr;  
      $twodigit_hr =~ s/^.//;
      $got_all_this_pat=1;

         
      $all_hours_in_one_file=0;
      $npats=$#filepathpats_this_cycle + 1;
      $found_all_pats=1;
      for (my $j = 0; $j < $npats; $j++) {
         chomp $main::filepathpats[$j];
         $checkforfile=$filepathpats_this_cycle[$j];
         $checkforfile =~ s/YYYYMMDD/${date}/g;
         $checkforfile =~ s/CC/${cycle}/g;
         # For models whose forecast hours are all in ONE file.
         # Input is like mod_YYYYMMDD, there is no FFF in file pattern.
         # Eg : wave.YYYYMMDD/nww3_YYYYMMDDCC
         # Only test $checkforfile to see if all hours are in one file withouth changing it.
         # Reason of not changing it is because we need to replace FFF with 
         # different GFS fhr in ESTOFS model.
         if ( $checkforfile !~ /FF/ ) {   # Could be 'FF' or 'FFF'
            # all hours are in the same file
            $all_hours_in_one_file=1;
         } 
   
         # For ESTOFS model, 
         # hourly ESTOFS input data is available up to 180, and 
         # hourly GFS input data is only available up to 120.
         # Between 123 and 240, GFS is available every 3 hrs.
         # ESTOFS fhrs 120, 121 and 122 use GFS fhr 120
         # ESTOFS fhrs 123, 124 and 125 use GFS fhr 123
         # and so on ......
         # Check if the current file patter is GFS input file pattern for ESTOFS model and fhr > 120.
         if ($main::model eq "stofs" && $checkforfile =~ /gfs/ && $hr > 120) {
             my $gfs_hr= floor($hr/3) * 3;
             $checkforfile =~ s/FFF/$gfs_hr/g;
         } else {
             $checkforfile =~ s/FFF/$hr/g;
             # Check for 2-digit hour
             $checkforfile =~ s/FFF/$hr/g;
             $checkforfile =~ s/FF/$twodigit_hr/g;
         }

         if (($main::model eq "hrrr" | $main::model eq "hrrr-subh") && $checkforfile =~ /alaska/ && $cycle%3 != 0) {
           logmsg($Config::info, "Skipping hrrr hrrr-subh alaska, cycle $cycle"); 
           next;
         }


         # For SREF-CLUSTER model,
         # pattern contains "NNN" for a single digit cluster number
         # Check 1 -> 6 to see if any of them are found
         # If any are found, that's enough.
         # If none are found, set found_all_pats to 0
         
         $found="no";
         if ($main::model eq "sref-cluster" ) {
            my $cluster_number=1;
            while ($found eq "no" && $cluster_number <= 6) {
                $cluster_file = $checkforfile;
                $cluster_file =~ s/NNN/$cluster_number/;
                logmsg($Config::debug, "Look for cluster file: $checkforfile\n");
                if ( exists $main::all_filenames{$cluster_file}) {
                   $found="yes";
                }
                $cluster_number++;
            }
            if ($found eq "no") {
               $found_all_pats=0;
            }
         } else {
   
            # Not SREF-CLUSTER
   
            logmsg($Config::debug, "look for $checkforfile\n");
            if (! exists $main::all_filenames{$checkforfile}) {
               $found_all_pats=0;
	       last;
            } else {
               logmsg($Config::debug, "$checkforfile found in list\n");
               my $lsmsg = `ls -l $checkforfile`;
               logmsg($Config::debug, "ls -l $checkforfile: $lsmsg \n");
            }
         }
      }
      if ($found_all_pats) {
         $highest_fhr=$hr;
         if ($all_hours_in_one_file) {
            $highest_fhr= $main::forecast_hr_list[-1];
          }
   
      } else {
         $keep_going=0;
         logmsg($Config::info, "Did not find all filename patterns for forecast hour $hr\n");
         logmsg($Config::info, "Missing file : $checkforfile \n\n");
      }
      $keep_going = 0 if ($i > $#main::forecast_hr_list-1);
   }
   
   logmsg($Config::debug, "highest_fhr (file received)=$highest_fhr\n");

return $highest_fhr;

}

#------------------------------------------------------------------
#  get_next_fhr: get the next forecast hour after this one
#
# Input:
#  $highest_fhr_processed
# Return:
#  $next_fhr
#
# Access global:
#  @forecast_hr_list

#------------------------------------------------------------------
sub get_next_fhr($) {

   my $highest_fhr_processed = shift;

   my ($next_fhr, $done, $i);

   $next_fhr=0;
   $done=0;
   $i=0;

# TO DO: $forecast_hr_list[$i] is global, should be $main::forecast_hr_list[$i]
   while (! $done) {
      if ( $forecast_hr_list[$i] <= $highest_fhr_processed) {
         $i++;
         next;
      } 
      $done=1;
      $next_fhr=$forecast_hr_list[$i];
   }
      
   return $next_fhr;
}

#------------------------------------------------------------------
#  load_list: Generate the list of forecast hours 
#
# Input:
#  $model
#  $first_fhr_to_process
#  $last_fhr_to_process
#
# Access global:
#  @forecast_hr_list
#  @model_fhr_ranges

#------------------------------------------------------------------
sub load_list($$$) {

   my ($model, $first_fhr_to_process, $last_fhr_to_process) = @_;

   my ($last_fhr_this_range, $hr, $ihr, $done, $found, $i);

   logmsg($Config::info, "Loading the forecast hour list with hours from $first_fhr_to_process to $last_fhr_to_process ...");


   # Build list of of hours 
   # Process from $first_fhr_to_process to $last_fhr_to_process
   #
   logmsg($Config::debug, "first_fhr_to_process=$first_fhr_to_process last_fhr_to_process=$last_fhr_to_process\n"); 
   $ihr=$first_fhr_to_process;


   my $nhrs=0;

   # Find the forecast hour range of the first fhr to process.
   # get the step and last fhr of the range. 
   
   $i=0;
   $done=0;
   while ( ($i<$main::number_fhr_ranges) && (!$done) ) {
       if ( ( $first_fhr_to_process >= $main::model_fhr_ranges[$i][0]) &&
            ( $first_fhr_to_process <= $main::model_fhr_ranges[$i][1]) ) {
             $done=1;
             $step=$main::model_fhr_ranges[$i][2];
             $last_fhr_this_range=$main::model_fhr_ranges[$i][1];
       }
       $i++;
   }


   # build the list by successively adding the $step.
   # if it exceeds the last fhr of the current range, find the
   # next range

   $done=0;
   @main::forecast_hr_list=();
   while ($ihr <= $last_fhr_to_process && (! $done)) {

      $hr= sprintf("%03d", $ihr);  #pad to 3 chars
      logmsg($Config::debug, "sub load_list add fcst hour to forecast_hr_list list: $hr\n");   
      push(@main::forecast_hr_list, $hr);
      $nhrs++;

      if ($ihr == $last_fhr_to_process) {
         $done=1;
      } else {

         if ( ($ihr + $step) > $last_fhr_this_range) {
            $found=0;

            # find the next forecast hour range to get the 
            # step and $last_fhr_this_range

            $i=0; 
            while ( ($i<$main::number_fhr_ranges) && (!$found) ) {
               if ( (($ihr + $main::model_fhr_ranges[$i][2]) >= $main::model_fhr_ranges[$i][0]) && 
                    (($ihr + $main::model_fhr_ranges[$i][2]) <= $main::model_fhr_ranges[$i][1]) ) {
                  $found=1;
                  $step=$main::model_fhr_ranges[$i][2];
                  $ihr = $ihr + $step; 
                  $last_fhr_this_range=$main::model_fhr_ranges[$i][1];
               }
               $i++;
            }
            if (! $found) {      # This shouldn't happen, fhr ranges were checked in
                                 # get_model_first_last_hr. But if it does, we're done. 
               $done=1; 
            }
         } else {
           $ihr = $ihr + $step;
         }
      }

   }
   logmsg($Config::info, "Number of possible forecast hours to process between $first_fhr_to_process and $last_fhr_to_process: $nhrs"); 
   
}


#------------------------------------------------------------------
# Get the forecast hour range info, set first and last forecast hour,
# and the time step for the first forecast hour range 
#
# Input: 
#    $model
#    $cycle
#
# Return:
#    $first_fhr
#    $step 
#    $last_fhr
#
# Access global:
#    @model_fhr_ranges
#------------------------------------------------------------------

sub get_model_first_last_hr ($$) {

   my ($model, $cycle) = @_;
   my ($first_fhr, $step, $last_fhr);

   logmsg($Config::info, "Get the number of forecast hours expected...");

   # get model info for first_fhr_this_range for the hours requested. 

   $main::number_fhr_ranges= get_model_fhr_ranges($model, $cycle);

   logmsg($Config::debug, "number of fhr ranges=$main::number_fhr_ranges\n");

   # load up the info for the first forecast hour range

   if ($main::number_fhr_ranges > 0) {
      $first_fhr           = $main::model_fhr_ranges[0][0];
      $step                = $main::model_fhr_ranges[0][2];
      $last_fhr            = $main::model_fhr_ranges[$main::number_fhr_ranges-1][1];
   } 
   logmsg($Config::debug, "get_model_first_last_hr: first_fhr=$first_fhr last_fhr=$last_fhr\n");

   logmsg($Config::debug, "last hour of the model=$last_fhr\n");
   return ($first_fhr, $last_fhr);
   
}



#------------------------------------------------------------------
# get_model_fhr_ranges  - retrieve the forecast hour range and timestep
#                   for model and cycle from the MAG.xml file.  
#  Input: 
#     model_name
#     cycle_name
#  Return:
#     nrange - number of forecast hour ranges
#
# Changes global:
#     @model_fhr_ranges 
#
#------------------------------------------------------------------

sub get_model_fhr_ranges ($$) {

   my $model = shift;
   my $cycle = shift;

   my $nrange;
   $nrange=0;

  get_fhr_ranges($model,$cycle);
  $nrange=scalar @model_fhr_ranges;
  return $nrange;
}

#---------------------------------------------------------
#
# Return the highest forecast hour processed for this model.
# Fetch from the stat file in tmp.
#
# Input:
#   date
#   cycle
#
# Accesses global:
#   tmp_dir
#
#---------------------------------------------------------
sub get_highest_fhr_processed($$) {

   my ($date , $cycle) = @_;
   my ($last_processed, $fhrs_processed_fname);
   $last_processed=-1;
  
   $fhrs_processed_fname="${status_dir}/${model}_${date}${cycle}";
   logmsg($Config::info, "Get the highest fhr processed in $fhrs_processed_fname \n");
   if ( -e $fhrs_processed_fname && -s $fhrs_processed_fname ) {
      $last_processed=`cat $fhrs_processed_fname`;
   }

   logmsg($Config::debug, "highest fhr processed=$last_processed\n");

   return $last_processed;
}

#---------------------------------------------------------
#
# Set the highest forecast hour processed for this model.
# Put in the stat file in tmp.
#
# Input:
#    hr
#    date
#    cycle
#
# Accesses global:
#   tmp_dir
#
#---------------------------------------------------------

sub set_highest_fhr_processed($$$) {

   my ($hr, $date, $cycle)=@_;
   my $fhrs_processed_fname;
   
   logmsg($Config::info, "Set the highest forecast hour processed to $hr");

   $fhrs_processed_fname="${status_dir}/${model}_${date}${cycle}";

   if (! open (FILE, ">$fhrs_processed_fname") ) {
      logmsg($Config::fatal, "Cannot open file $fhrs_processed_fname: $! \n");
      return -1;
   } else {
      logmsg($Config::info, "Update file $fhrs_processed_fname with highest processed: $hr\n");
      print FILE "$hr";
      close FILE;
   }

}

#------------------------------------------------------------------
# make_command_list     
# Create the commands to create a GIF file for each forecast hour 
# from $highest_fhr_processed to $highest_input_fhr_found.
#
# For a fcst_snd parameter, generate one command for each station.
#
#  Input: 
#     $model                      - model
#     $date                       - date to process
#     $cycle                      - cycle to process
#     $highest_fhr_processed      - highest hour processed so far
#     $highest_input_fhr_found    - highest input forecast hour found
#     $max_run_time               - max run time for each command
#  Return:
#     1 if error 
#     0 if no error
#
#  Accesses globals:
#     @main::forecast_hr_list  - forecast hours found 
#     @main::params             - params for this model
#     @main::areas              - areas for this  model
#     @main::command_list      - commands genereated for new hours
#     $load_balancing_script
#------------------------------------------------------------------


sub make_command_list($$$$$$) {
   my ($model, $date, $cycle, $highest_fhr_processed, $highest_input_fhr_found, $max_run_time)=@_;
   my ($first_hr_after_last_processed, $first_fhr_to_process, $last_fhr_to_process, $timestep);
   my ($keep_going, $hr, $base_hour);
   my ($index_first_hr, $param, $script, $area, $fhr, $i, $command);
   my (@params);
   my $reg_string="";  
   $keep_going=1;


   logmsg($Config::info, "\n");  
   logmsg($Config::info, "--------------------------------------------"); 
   logmsg($Config::info, " Create the list of commands to execute...");
   logmsg($Config::info, "--------------------------------------------\n"); 


   #-------------------------------------------------------------------
   # find the first hour in the list that is greater than the highest
   # hour processed.
   #-------------------------------------------------------------------
  
   
   logmsg($Config::debug, "find first hour after last processed\n");

   $index_first_hr=0;
   $keep_going=1;
   $i=0;
   while ($keep_going) {
      logmsg($Config::debug, "keep looking for the last processed..., $main::forecast_hr_list[$i] > $highest_fhr_processed ? \n");
      $first_hr_after_last_processed = $main::forecast_hr_list[$i];
      $index_first_hr=$i;
      $keep_going = 0 if ( $main::forecast_hr_list[$i] > $highest_fhr_processed);
      $i++;
   }
 

  #----------------------------------------------------------
  # Write the command to create the gifs for each
  # parameter, for each area
  #----------------------------------------------------------
      
   logmsg($Config::info, "Submit jobs for  $first_hr_after_last_processed -> $highest_input_fhr_found\n");
																
   foreach $param (@main::params) { 


      $script=get_script_by_model_name($main::models_doc, $model,$param);
      if ( $script =~ /^$/  ) {
          $script=get_script_by_model_name($main::models_doc, $model,"all"); 
          if ($script =~ /^$/ ) {
            logmsg($Config::info, "No script defined in xml for $model and $param\n");
            return 1;
          }
      }
     
      logmsg($Config::info, "\n\n");
      logmsg($Config::info, "====================================================");
      logmsg($Config::info, " Now processing parameter : $param ");
      logmsg($Config::info, "====================================================\n");


      # Check if this cycle is valid for this param

      if (check_valid_cycle_param($model, $cycle, $param)) {

         foreach $area (@main::areas) { 
            logmsg($Config::info, "   -------------------------------------------------------------------");
            logmsg($Config::info, "    Generating command list for parameter : $param for area : $area  ");
            logmsg($Config::info, "   -------------------------------------------------------------------");
   
            # Check if this cycle is valid for this area
   
            if (check_valid_cycle_area($model, $cycle, $area)) {
   
               logmsg($Config::debug, "Create command: check $param $area for exceptions, currently  $first_hr_after_last_processed, $highest_input_fhr_found\n");
   
               # Get the first and last fhr to process for this parameter/area.
               # Exceptions may be applied to alter the first/last fhr or timestep
               # If $first_fhr_to_process=-99 then do not create any commands for this $param and $area
               # If timestep=-99, then no alteration was made to the timestep.
               # so there is no need to skip over any of the forecast hours.
               
               ($first_fhr_to_process, $last_fhr_to_process, $timestep, $base_hour) = get_fhrs_to_process($model, $area, $param, $first_hr_after_last_processed, $highest_input_fhr_found);

               if (($model eq "nbm") && ($cycle==00 ) && ($param eq "2m_max_temp" )) {
									  
                    $first_fhr_to_process="024";
                    $last_fhr_to_process="264";
                    $timestep="024";
                    $base_hour="024"
                    }

               if (($model eq "nbm") && ($cycle==06 ) && ($param eq "2m_max_temp" )) {
                    $first_fhr_to_process="018";
                    $last_fhr_to_process="258";
                    $timestep="024";
                    $base_hour="018"
                    }
               if (($model eq "nbm") && ($cycle==12 ) && ($param eq "2m_max_temp" )) {
                    $first_fhr_to_process="012";
                    $last_fhr_to_process="252";
                    $timestep="024";
                    $base_hour="012"
                    }	
	   
               if (($model eq "nbm") && ($cycle==00 ) && ($param eq "2m_min_temp" )) {
                    $first_fhr_to_process="012";
                    $last_fhr_to_process="252";
											   
                    $timestep="024";
                    $base_hour="012"
                    }

               if (($model eq "nbm") && ($cycle==06 ) && ($param eq "2m_min_temp" )) {
                    $first_fhr_to_process="030";
                    $last_fhr_to_process="246";
                    $timestep="024";
                    $base_hour="030"
                    }
	   
               if (($model eq "nbm") && ($cycle==12 ) && ($param eq "2m_min_temp" )) {
                    $first_fhr_to_process="024";
                    $last_fhr_to_process="264";
                    $timestep="024";
                    $base_hour="024"
                    }
	   
   # TO DO: Define a constant for undefined value and use it instead of
   #        inconsistent use of -99, -1, etc.

               if ( $first_fhr_to_process != -99) {
                  logmsg($Config::debug, "Create command: After get_fhrs_to_process: first_fhr_to_process -> last_fhr_to_process = $first_fhr_to_process -> $last_fhr_to_process timestep=$timestep\n");
                  $keep_going=1;
               } else {
                  logmsg($Config::debug, "Do not create any commands for $param $area hours: $first_hr_after_last_processed -> $highest_input_fhr_found\n");
                  $keep_going=0;
               }
               $i=0;
      
               while ($keep_going && $i<=$#main::forecast_hr_list) {
      
                  if ( ($main::forecast_hr_list[$i] >= $first_fhr_to_process) &&
                       ($main::forecast_hr_list[$i] <= $last_fhr_to_process) ) {
   
                     # If a timestep is defined for this parameter, it overrides the default value
                     # (that was used to generate the hours in @forecast_hr_list)
                     # If $timestep is not -99, then 
                     #    Determine the offset of the 1st valid fhr of the model from multiples of 
                     #    the timestep.
                     #    (usually the 1st valid fhr is a multiple of the timestep, so the
                     #    offset is 0, but not always..., nbm has a product that starts at 36
                     #    with a timestep of 24))
                     #    Then check that fhr is the same offset hours from the multiples 
                     #    of the timestep .
   
                     my $validfhr_offset = $base_hour % $timestep;

                     if (($timestep != -99) && (($main::forecast_hr_list[$i] % $timestep) != $validfhr_offset)) {
                        logmsg($Config::debug, "Skip fhr $main::forecast_hr_list[$i]");
                     } else {
                        $fhr=$main::forecast_hr_list[$i];
  
			if ( $main::forecast_hr_list[$i] == $first_fhr_to_process) {
				$initval = 1;
			} else {
				 $initval = 0;
			} 	 

                        # if this is a fcst_snd, get the station list and generate one command for each station
			#

			if ($model ne "aigfs") {
			
			if ($model ne "nbm") {
				if ($param eq "fcst_snd") {
                           	my @stations=get_area_stations($model, $area);
                           	# Check if stations exist for the model/area by testing array size
                           	my $arr_size = @stations;
                           	if ( $arr_size > 0 ) {
   			   		foreach my $station (@stations) { 
                                 # Parse a station's full information to extract its name, lat and lon
                                 	my $stn=get_station_ID($station);         
						   
						
   			      		logmsg($Config::info, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $area $param $stn");
   			      		push (@main::command_list, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $area $param $stn");
   			   	}
	 
                           	}
	  
                           
                        	} else {

							
                           	logmsg($Config::info, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $area $param");
                           	push (@main::command_list, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $area $param");
                        	}


			} elsif ($initval == 1) {
								
			 logmsg($Config::info, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $area $param $initval");
		         push (@main::command_list, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $area $param $initval");
		}


		} elsif (($model eq "aigfs") && ($param eq "1000_500_thick")) {

			logmsg($Config::info, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $area $param");
                        push (@main::command_list, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $area $param");

		}


			push (@main::transfer_list, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $area $param");

		  }
	   
					 
		}  elsif ($main::forecast_hr_list[$i] >  $last_fhr_to_process) {
                     $keep_going=0;
                  }

                  $i++;
	  }	

   
            } else {
               logmsg($Config::info, "Cycle $cycle is not valid for model area $model $area\n");
            }
																																					  
         } #  foreach $area (@main::areas)

      } else {
         logmsg($Config::info, "Cycle $cycle is not valid for parameter $param\n");
      }
   
      logmsg($Config::info, "\n");
      logmsg($Config::info, "====================================================");
      logmsg($Config::info, " Done processing parameter : $param ");
      logmsg($Config::info, "====================================================\n\n");

   }  #  foreach $param (@main::params) 
	  

   logmsg($Config::info, "\n");
   logmsg($Config::info, "==============================");
   logmsg($Config::info, " Commands creation is done!!! ");
   logmsg($Config::info, "==============================\n\n");

   return 0;   
}

   
#------------------------------------------------------------------
#  get_fhrs_to_process - Adjust the first and last forecast
#     hour to process if any exceptions exist for this
#     model/area/param.
#     Return the adjusted first hr, last hour, timestep, and base hour.
#     The base hour is the first valid forecast hour.  This is
#     needed to calculate the valid hour when they are not
#     multiples of the new timestep (a NBM product starts
#     at 36 with a timestep of 24)
#
# Input:
#    $model
#    $area
#    $param
#    $first_hour_to_submit
#    $last_hour_to_submit
# Return:
#    (new-first-hour, new-last-hour, new-timestep, base-hour)
#  Accesses globals:
#     @forecast_hr_list
#  
#------------------------------------------------------------------
sub get_fhrs_to_process($$$$$) { 
   my ($model, $area, $param, $first_hr_to_submit, $last_hr_to_submit)=@_;

   my ($new_first_hr, $new_last_hr, $new_timestep, $pinterval, $hr, $keep_going);

   my @exception_hour_list=();
   my $exception=0;

   # First, check exceptions list for this model, area, and parameter.
   # If not found, check this model, parameter, for all areas.
   # If a rec exists, replace the first and last hour.

   # If new_first_hr=-99, then no exception was found to alter the forecast hours.

   logmsg($Config::debug,  "call get_area_exception: $model, $param, $area");
   ($new_first_hr,$new_last_hr,$new_timestep) = get_area_exception($model, $param, $area);
   logmsg($Config::debug,  "returned $new_first_hr,$new_last_hr,$new_timestep");
   if ($new_first_hr != -99) {  # found an exception
      $exception=1;
   } else {
      logmsg($Config::debug,  "call get_area_exception: $model, $param, all");
      ($new_first_hr,$new_last_hr,$new_timestep) = get_area_exception($model, $param, "all");
      logmsg($Config::debug,  "returned $new_first_hr,$new_last_hr,$new_timestep");
      if ($new_first_hr != -99) {  # found an exception
         $exception=1;
      }
   }

# TO DO: Identify all the possible patterns and explicitly test for them,
#        This will clarify the code and ensure that all exception combinations
#        are accounted for.
#        <tns:all>003, 240 , 3<..> 
#        <tns:all>003, all , 3<..>
#        <tns:all>003, all<..>
#        <tns:namer>999, -999<..>
#        <tns:namer>003, 12 , 3<..>
#        <tns:namer>003, all, 3<..>
#
   my $base_hour=$new_first_hr;

   if ($exception) {
   

      # NOTE: If this parameter is not to be made for this area at all, then
      # table entry will return $new_first_hr greater than $new_last_hr. 
      # It will pass the checks below for a product that has already been
      # created for all it's hours.

      if ( ($new_last_hr ne "all") && $new_first_hr > $new_last_hr) {
         logmsg($Config::debug, "Do not create any $param products for $model $area");
         ($first_hr_to_submit, $last_hr_to_submit) = ($new_first_hr,$new_last_hr);

      # If new last hour is less than the first hour to submit, then there are no forecast hours to process.

      } elsif (($new_last_hr ne "all") && $new_last_hr < $first_hr_to_submit) {

         logmsg($Config::debug, "New first hour ($new_first_hr) for $model $area is less than the first hour to submit ($first_hr_to_submit) ");
         ($first_hr_to_submit, $last_hr_to_submit) = ( 999, -999);

      } else {
   
         # if new last hour is less than the last hour to submit, replace it. 
   
         if ($new_last_hr ne "all" && $new_last_hr < $last_hr_to_submit) {
            logmsg($Config::debug, "replace the last hour with $new_last_hr");
            $last_hr_to_submit = $new_last_hr;
         }

      
         # If the first hour to submit is less than the first hour 
         # for this param, then replace it. 
      
         if ( $new_first_hr > $first_hr_to_submit) {
            logmsg($Config::debug,"replace the first hour with $new_first_hr");
            $first_hr_to_submit = sprintf("%03d", $new_first_hr);
         } else {

            # Otherwise, get the list of forecast hours for the exception
            # and repopulate the forecast hour list. 
            # Find the first forecast hour that is greater than or equal 
            # to first_hr_to_submit, this is the new $first_hr_to_submit.
   
            logmsg($Config::debug,"Reload forecast_hr_list with hours for the exception."); 
            load_list($model, $first_hr_to_submit, $last_hr_to_submit);
            $main::reload_fhr_list=1;
      
            $keep_going=1;
            my $i=0;
            while (($hr=$main::forecast_hr_list[$i]) && $keep_going) {
               if ($hr >= $first_hr_to_submit) {
                 $first_hr_to_submit=$hr;
                 $keep_going=0;
               }
               $i++;
            }
         }

      }


      logmsg($Config::debug, "now first_hr_to_submit = $first_hr_to_submit");
      logmsg($Config::debug, "now last_hr_to_submit = $last_hr_to_submit");
      logmsg($Config::debug, "new timestep = $new_timestep");
      logmsg($Config::debug, "base_hour = $base_hour");

      # If this parameter is not to be made at all for this area, then 
      # the exception table set $first_hr_to_submit > $last_hr_to_submit)

      # Or, now that the first hour may have been adjusted, there 
      # is the possibility that we have not yet
      # received the first valid hour for this parameter.
      # If so, return -99 for first_hr_to_submit.

      if ( $first_hr_to_submit > $last_hr_to_submit) {

         logmsg($Config::debug, "After adjustment for $param, $first_hr_to_submit > $last_hr_to_submit ");
         $first_hr_to_submit=-99;
      }  

   # If no exception found, check if the forecast_hr_list was altered due to an exception
   # found in the previous parameter.  If so, reload the forecast_hr_list

   } else {
      if ( $main::reload_fhr_list ) {
         logmsg($Config::debug, "forecast_hr_list was altered due exception in previous parameter, reload it with default.");
         load_list($model, $first_hr_to_submit, $last_hr_to_submit);
         $main::reload_fhr_list=0;
      }

   }

   return ($first_hr_to_submit, $last_hr_to_submit, $new_timestep, $base_hour);
   
}
   
   
#------------------------------------------------------------------
#  Interrupt handler 
#------------------------------------------------------------------
sub int_handler () {
   logmsg($Config::fatal, "Interrupt--start to shut down.\n");
   $die=1;
}


#------------------------------------------------------------------
#  Print to a dated log file. Only print if logging level of msg
#  is greater than or equal to logging level.
#
#  $logging_level - global level of logging enabled
#                   0=debug, 1=info, 2=warn, 3=fatal
#  $log_msg_level - log level of this msg
#
#  Accesses globals:
#  $log -  file handle
#  $log_dir -  log directory
#  $logfile - name of current log file.
#------------------------------------------------------------------
sub logmsg($$) {
   my ($log_msg_level, $msg) = @_;
   my ($foo, $line);

   my $date=strftime("%Y%m%d", gmtime(time));
   my $time=strftime("%H:%M:%S", gmtime(time));
   my $this_script =  basename ( (caller(0) )[1]);
   # my $parent = (caller(1) )[3] || "main";
   ($foo, $foo, $line) = caller;
   my $called_from_line_num = sprintf("%04d", $line);
   my ($new_log, $new_msg);
   return if $log_msg_level < $main::logging_level;

# Open the logfile, if we are making one
#
   if ($Config::make_log_file) {
 
      # Use PDY for log date, if defined.

      $date=$ENV{'PDY'};
      if (! defined($date))  {
        $date=strftime("%Y%m%d", gmtime(time));
      }

      if ( $main::logfile ne $new_log ) {
         $main::log->close if ($main::logfile ne "");
         $main::logfile=$new_log;
         $main::log->open (">> $main::logfile") or die "Error opening log file $main::logfile: $?";
      }
   }

   
   #my $dateInfo=strftime("%Y/%m/%d", gmtime(time));
   #$new_msg = "$dateInfo ${time} : $Config::severity{$log_msg_level} : [Line# $called_from_line_num]  ${msg}\n";
   $new_msg = "$Config::severity{$log_msg_level}: $time [L#$called_from_line_num]: ${msg}\n";


   print $new_msg;
   if ($Config::make_log_file) {
      print $main::log $new_msg or print "Could not write to $main::log: $?\n";
   }
}

#------------------------------------------------------------------
#  Print to a GEMPAK log file. Only print if logging level of msg
#  is greater than or equal to logging level.
#  Input:
#     $log_msg_level - log level
#     $msg           - msg to be printed out
#
#  Accesses globals:
#     $main::logging_level
#------------------------------------------------------------------
sub logmsg_gempak($$) {
   my ($log_msg_level, $msg) = @_;

   return if $log_msg_level < $main::logging_level;

   my $new_msg;
   $new_msg = "$Config::severity{$log_msg_level}: ${msg}";

   print $new_msg;
}


#------------------------------------------------------------
#  already_running: check if this script is already running for this model.
#  
#  Check for the process id file.  This file contains the LSF_JOBID
#  if the environment variable exists, the pid otherwise.  
#  
#   If process id file does not exist,  
#      create it and return false.
#   else 
#       if the process from the id file is actually running.  
#          If it is, return true. 
#       else
#          replace id file pid for this process
#          return false 
#------------------------------------------------------------
 
sub already_running ($$) {
   my ($status_dir, $process_file) = @_;
   my ($id, $exists, $jobid, $use_jobid);
   my $error;

logmsg($Config::info, "Check if $process_file is already running elsewhere... ");
# Get the job id for this job

   $jobid=$ENV{'PBS_JOBID'};
   $use_jobid = (defined $jobid && $jobid);

   logmsg($Config::debug, "PBS_JOBID? >$jobid<");

   (defined($jobid) || !$jobid) or print "Not running in batch, so use pid instead of jobid\n";

   if (-s "${status_dir}/.${process_file}") {
      if (! open(RUNNING, "< ${status_dir}/.${process_file}")) {
         logmsg($Config::fatal, "Pid file $status_dir/.$process_file exists, but cannot open it to check if the other process is still running: $?");
         exit 1;
      }
      $id=<RUNNING>;

  #   Assume this is a batch job id.  See if the job is running.

     $exists = is_batch_job_running($id) ;
      if ($exists) {
         logmsg($Config::info,"Process is already running - job found for contents of  ${status_dir}/.${process_file} = $id");
        exit 0; 
      } 

  #   If not found, use it as a pid (only if it is numeric) and check for that
      if ($id =~ /^\d+$/) {
         $exists = ( kill 0, $id );
         if ($exists) {
            logmsg($Config::fatal,"Process is already running - process found for contents of  ${status_dir}/.${process_file} = $id");
            exit 1;
         }
      }

      logmsg($Config::warn,"Id file $status_dir/.$process_file exists, but no job or process found for $id so replace with new id file and continue.");
      if (! unlink "${status_dir}/.$process_file") {
         logmsg($Config::fatal, "Could not remove pid file ${status_dir}/.${process_file} to replace with this one so execution cannot continue: $?");
            exit 1;
      }
   }
       
 
   if ( ! open(RUNNING, "> ${status_dir}/.${process_file}") ) {
      logmsg($Config::fatal,"open(${status_dir}/.${process_file}) failed: $!");
      exit 1;
   }

   if ($use_jobid) {
      print RUNNING $jobid;
   } else {
      print RUNNING $$;
   }
   close(RUNNING) || logmsg($Config::fatal,"close(${status_dir}/.${process_file}) failed: $!");

   logmsg($Config::info, "Done checking. Return from already_running function. Proceeding. \n");
   return;
} 

#------------------------------------------------------
# mkdirs - make a directory and parents
#------------------------------------------------------

sub mkdirs ($) {
  my ( $dir ) = @_;
  my ( @dirs, $d, $path );

  # Very often the directory does exist - so return now
  return 1 if -d $dir;

  # Make sure that the target directory exists
  @dirs = split( '/', $dir );

  # the root directory always exists
  $path = '';
  if( $dirs[ 0 ] eq '' ){
    shift( @dirs );
    $path = '/';
  }

  foreach $d ( @dirs ){
    $path = $path . $d;
    if( ! -d $path ){
       return 0 if  !mkdir($path, 0775);
     }
     $path .= "/";
  }
  return 1;
}


#------------------------------------------------------------------
# get_fhr_ranges: Get the forecast hour ranges for model and cycle
# Check of definitions exists for cycle, first, if not, get
# def for model.
#
# Access globals:
#    $main::models_root
#
#------------------------------------------------------------------
sub get_fhr_ranges($$) {

   my $model=shift;
   my $cycle=shift;
   my $error=0;
   my @ranges=();
   my $i;

   @ranges = get_fhr_ranges_by_model_cycle( $main::models_doc, $model, $cycle);
   if  ((scalar @ranges) == 0) {
      @ranges = get_fhr_ranges_by_model_name( $main::models_doc, $model);
   }

   if ((scalar @ranges) == 0) {
      logmsg($Config::fatal, "No forecast hours defined for $model");
      $error=1;
   }

   $i=0;
   foreach my $range (@ranges) {

      if ($range =~ /(\d+)-(\d+), (\d+)/) {
         logmsg($Config::debug, "next range: $range\n");
         $i++;
         $main::model_fhr_ranges[$i-1][0]=$1;
         $main::model_fhr_ranges[$i-1][1]=$2;
         $main::model_fhr_ranges[$i-1][2]=$3;
      } else {
         logmsg($Config::fatal, "Error in format of <forecast-hrs><range> tag $range");
         $error=1;
      }
   }
        
  return $error;
}

   
   
sub get_area_exception($$$) {

   my $model=shift;
   my $param=shift;
   my $area=shift;
   my $new_first_hr=-99;
   my $new_last_hr=-99;
   my $new_timestep=-99;
   my @new_fhrs_strings;
   my $query;
  
   @new_fhrs_strings=get_area_exception_by_name_param( $main::models_doc, $model, $area, $param); 
   if ( (scalar @new_fhrs_strings) == 0) {
      logmsg($Config::debug, "No exception found for $param except-$area");
   } else {
      if ( $new_fhrs_strings[0] =~ /(.*),(.*),(.*)/) {
         $new_first_hr=$1;
         $new_last_hr=$2;
         $new_timestep=$3;
      } else {
        if ( $new_fhrs_strings[0] =~ /(.*),(.*)/) {
           $new_first_hr=$1;
           $new_last_hr=$2;
        }
      }
   }
   logmsg($Config::debug, "Return exception for $param except-$area: new_first_hr=$new_first_hr new_last_hr=$new_last_hr new_timestep=$new_timestep");
   return ($new_first_hr, $new_last_hr, $new_timestep);
}

#------------------------------------------------------------------
# silence_used_once_warnings
#
# I like to use a configuration file, but with use strict, if something 
# is only used once it produces a warning, thinking it's a typo.
# After verifying that the once-use is intention, add another use
# here to silence the warning.

#------------------------------------------------------------------
sub silence_used_once_warnings() {

   $Config::log_string = $Config::log_string;
   $Config::areaHash = $Config::areaHash;
   $Config::load_balancing = $Config::load_balancing;
   $Config::log_dir = $Config::log_dir;
   $Config::warn = $Config::warn;
   $Config::severity = $Config::severity;
   $Config::default_logging = $Config::default_logging;
   $Config::load_balancing_script = $Config::load_balancing_script;
   $Config::min_between_checks = $Config::min_between_checks;
   $Config::paramHash = $Config::paramHash;
   $Config::use_dated_dirs = $Config::use_dated_dirs;
   $Config::default_max_time = $Config::default_max_time;
   $Config::default_max_threads = $Config::default_max_threads;
   $Config::capture_output_to_log_script = $Config::capture_output_to_log_script;
   $Config::version = $Config::version;
   $Config::transfer = $Config::transfer;
   $Config::default_max_retries_for_more_fhrs=$Config::default_max_retries_for_more_fhrs;

}

#------------------------------------------------------------------
## make_tmp dir - create a temp dir for this  job and cd to it.
#                 if environment variable PBS_JOBID is defined
#                 use that to make the dir name unique.  If not,
#                 use $$.
##
## return 0 if error, 1 if no err.
#
#  Globals accessed:
#  $Config::tmp_dir
#  $main::job_tmp_dir
#------------------------------------------------------------------
sub make_temp_dir($) {

logmsg($Config::info, "Create a temporary dir for this run, and change working dir to it... ");

my $model=shift;

my $jobid=$ENV{'PBS_JOBID'};
my $use_jobid = (defined $jobid && $jobid);

if ($use_jobid) {
   $main::job_tmp_dir="$Config::tmp_dir/MAG_processor_${model}_$jobid";
} else {
   $main::job_tmp_dir="$Config::tmp_dir/MAG_processor_${model}_$$";
}
if (!mkdirs($main::job_tmp_dir)) {
   logmsg($Config::fatal, "Could not create temp dir $main::job_tmp_dir: $?");
   return 1;
}

if (! chdir $main::job_tmp_dir) {
   logmsg($Config::fatal, "Could not change working directory to $main::job_tmp_dir: $?");
   return 1;
}

logmsg($Config::info, "Create tmp dir is done: $main::job_tmp_dir \n");

return 0;

}

#


#------------------------------------------------------------------
##
## process_command_list
##
## create the job files to submit to poe . Process the command list
#  in chunks.  Put $main::max_threads commands in a poe job  and
#  execute it.  Repeat until error or all the commands queue has
#  been processed.
#
#  For each poe job, write the filenames each command will create
#  into a transfer file.
#
#  Input:
#     $model, $date,  $cycle used for log file
#     $model_in_filename used to create the filenames generated
#                        by the commands (for the transfer file)
##
#  Globals accessed:
#     @main::command_list  - list of commands
#     $main::max_threads - number of commands to put in each job
#     $main::transfer_seq - counter for transfer file names
#     $main::transfer_A_dir - directory for transfer files
#     $main::transfer_B_dir - directory for site 'B' transfer files
##
##
##------------------------------------------------------------------

sub process_command_list ($$$$) {

my ($model, $date, $cycle, $model_in_filename) = @_;


# Determine how many poe jobs there will be.  Put no more than
# $main::max_threads commands into any one poe job.

my $num_commands=$#main::command_list + 1;
my $use_cfp=(exists $ENV{'USE_CFP'} ? $ENV{'USE_CFP'} : "NO");
my $poe_jobs = (uc $use_cfp eq "YES" ? 1 : int (($num_commands / $main::max_threads) + 1));
my $commands_processed = 0;
my @commands_this_job=();
my $ncommands_this_job;
my $command;
my $poe_run_script_file;
my $poe_script_file;
my $error=0;
my $random;
my $local_process_start_time;
my $local_process_end_time;
my $local_process_duration;
my $transfer_A_file;
my $transfer_A_file_hidden;
my $transfer_B_file;
my $transfer_B_file_hidden;
my $trans_A_filehandle;
my $trans_B_filehandle;
my $err;
my $transfer_list;
my $timemarker;
my $epochmarker;				

logmsg($Config::info, "================================================================================");
logmsg($Config::info, " Start process_command_list for model: $model, date: $date, and cycle: $cycle");
logmsg($Config::info, "================================================================================");


logmsg($Config::info, "Number of commands in list: $num_commands");
logmsg($Config::info, "Number of concurrent tasks: $main::max_threads");

#
# Create a processing subdir for each script
#
my $tmp_subdir;  
my $job=0;

while ( ($commands_processed < $num_commands) && ! $error ) {

   $job++;
   $timemarker = DateTime->now;
   $epochmarker = $timemarker->epoch;							   
																
   $local_process_start_time = Time::HiRes::time();
   $tmp_subdir="${main::job_tmp_dir}/job_${job}_${epochmarker}";
   logmsg($Config::debug, "Create temp directory for job: ${job}_${epochmarker}..."); 
   logmsg($Config::info, "\n");
   logmsg($Config::info, "---------------------------------------");
   logmsg($Config::info, " Running commands for job ${job}_${epochmarker} : ");
   logmsg($Config::info, "---------------------------------------");
   if (! mkdirs ($tmp_subdir) ) {
      logmsg($Config::fatal, "Unable to create $tmp_subdir: $? ");
      return 1;
    }
   logmsg($Config::info, "Temp directory for job $job: $tmp_subdir");
   chdir($tmp_subdir);
   $poe_run_script_file = "$tmp_subdir/run_poe_script";
   $poe_script_file = "$tmp_subdir/poe_script";
   $ncommands_this_job=0;
   logmsg($Config::info, "poe_run_script_file : $poe_run_script_file");
   logmsg($Config::info, "poe_script_file : $poe_script_file");

   # calculate the number of commands in this job.
   
   if (uc $use_cfp eq "YES" ) {
      $ncommands_this_job=$num_commands;
   } elsif ( ($commands_processed + $main::max_threads) >= $num_commands) {
      $ncommands_this_job= $num_commands - $commands_processed;
   } else {
      $ncommands_this_job=$main::max_threads;
   }
   logmsg($Config::info, "There are $ncommands_this_job commands for this job.");

   # create the run poe script file
   #
   if (! open (POERUNFILE, ">$poe_run_script_file")) {
      logmsg($Config::fatal, "Cannot open list file $poe_run_script_file: $! \n");
	
      return 1;
   }
   # create the poe script file
   
   if (! open (POEFILE, ">$poe_script_file")) {
      logmsg($Config::fatal, "Cannot open list file $poe_script_file: $! \n");
	
      return 1;
   }
							   
   print POERUNFILE "cd ${tmp_subdir}\n";
   print POERUNFILE "set -x\n";
   if (uc $use_cfp eq "YES" ) {
      logmsg($Config::info, "Using the cfp utility to process poe_script\n");
      print POERUNFILE "mpiexec --cpu-bind core --configfile ./poe_script\n";
   } else {
      print POERUNFILE "mpiexec --cpu-bind core --configfile poe_script\n";
   }
   #print POERUNFILE "error=$?\n";
   #print POERUNFILE "exit $error\n";
   close POERUNFILE;

   # write the commands to the poe script
   #
   # Build the list of commands for this poe job, use it
   # to create the transfer file after it runs.
   
   logmsg($Config::debug, "Write the commands to poe script file and filenames to transfer file...");
   @commands_this_job=();
   for (my $i=0; $i<$ncommands_this_job; $i++) {
      logmsg($Config::debug, "Write number $i command to poe script file...");

      # write command to poe list and to the commands_this_job array

      $command = shift @main::command_list;
      $commands_processed++;
      push @commands_this_job, $command;

      print POEFILE "${main::MAG_dir}/ush/${Config::capture_output_to_log_script} ${main::job_tmp_dir} ${command}\n";
      logmsg($Config::debug, "Finished write command to poe script file commands_processed is $commands_processed and loop count is $i");

      
   }

   # 
   #  if less than max threads, dummy up the rest.
   # 
   for (my $i=$ncommands_this_job; $i< $main::max_threads; $i++) {
      print POEFILE  "ls /dev/null\n";
   }


   close POEFILE;

   # set script permissions
   #
   chmod 0775,$poe_run_script_file;
   chmod 0775,$poe_script_file;

   #-----------------------------------------------
   # while testing...
   #-----------------------------------------------
   # Use local lexcial variable to easy turn on/off the need 
   # of showing poe script.
   my $log_level=$Config::debug;
   logmsg($log_level, "execute: $poe_run_script_file");
   logmsg($log_level, "\n");
   logmsg($log_level, "---------------------------------");
   logmsg($log_level, "File run_poe_script contains:");
   logmsg($log_level, "---------------------------------");
   open (POEFILE, "<$poe_run_script_file");
   while (<POEFILE>) {
      logmsg($log_level, "$_");
   }
   close POEFILE;
   logmsg($log_level, "---------------------------------");
   logmsg($log_level, "\n");

   logmsg($log_level, "---------------------------------");
   logmsg($log_level, "File poe_script contains:");
   logmsg($log_level, "---------------------------------");
   open (POEFILE, "<$poe_script_file");
   while (<POEFILE>) {
      logmsg($log_level, "$_");
   }
   close POEFILE;
   logmsg($log_level, "---------------------------------");
   logmsg($log_level, "\n");

   #-----------------------------------------------

   #  error in open for pipe indicates problem forking, not the execution of the script

   if (! $main::command_line_mode) {
      if (! open (RUN_JOB, '-|', $poe_run_script_file)) {
         logmsg($Config::fatal, "Error opening pipe for $poe_run_script_file: $!");
         $error=1;
      }
   } else {
      print "running in development mode/n";
      if (! open (RUN_JOB, '-|', $poe_script_file)) {
         logmsg($Config::fatal, "Error opening pipe for $poe_script_file: $!");
         $error=1;
      }
   }

   # print output to standard output
  
   if (! $error) { 

      logmsg($Config::info, "Output of $poe_run_script_file:\n");
      while (<RUN_JOB>) {
         # Output from RUN_JOB has unique heading to identify GEMPAK running message, 
         # No need to have timestamp as heading like regular message does. 
         logmsg_gempak($Config::info, $_);
      }


      # Exit status is returned when the pipe is closed. $? is updated, but 
      # apparently $! isn't, my tests show an empty string in $! on error.
      # In my tests the error was in syserr, printed above with the job output.
      if (! close RUN_JOB) {
	 logmsg($Config::fatal, "Error executing $poe_run_script_file (look for error string above). $!");
         $error=1;
      } else {

         # Job was successful
         #
         # Make the transfer files for files created in this job
         #
         #
	 logmsg($Config::info, "Poe job completed successfully, so make the transfer file(s)");
         
         # Create the empty hidden files
         
         my $sys_timestamp=strftime("%Y_%m_%d_%H%M%S", gmtime(time));  # Get system timetamp
        
         $main::transfer_seq++;
         $transfer_A_file="${transfer_A_dir}/${model}_${date}${cycle}_${job}_${main::transfer_seq}_${sys_timestamp}.transfer";
         $transfer_A_file_hidden="${transfer_A_dir}/.${model}_${date}${cycle}_${job}_${main::transfer_seq}_${sys_timestamp}.transfer";
         $transfer_B_file="${transfer_B_dir}/${model}_${date}${cycle}_${job}_${main::transfer_seq}_${sys_timestamp}.transfer";
         $transfer_B_file_hidden="${transfer_B_dir}/.${model}_${date}${cycle}_${job}_${main::transfer_seq}_${sys_timestamp}.transfer";
          
         if ("$SYNC_TO_A" eq "YES") {
	    logmsg($Config::info, "Make transfer file A (hidden): $transfer_A_file_hidden");
            if (! open ($trans_A_filehandle, "> $transfer_A_file_hidden")) {
             logmsg($Config::fatal, "Error opening transfer file: $transfer_A_file_hidden: $!");
                return 1;
            }
         } else {
            $trans_A_filehandle = "";
         }
         if ("$SYNC_TO_B" eq "YES") {
	    logmsg($Config::info, "Make transfer file B (hidden): $transfer_B_file_hidden");
            if (! open ($trans_B_filehandle, "> $transfer_B_file_hidden")) {
             logmsg($Config::fatal, "Error opening transfer B file: $transfer_B_file_hidden: $!");
                return 1;
            }
         } else {
            $trans_B_filehandle = "";
         }

         # For each command in the job, determine the gif files created by the job and 
         # write the filenames to the transfer file(s)
       
	 if (($model eq "nbm") or ($model eq "aigfs")) { 
         foreach $command (@main::transfer_list) {

           $err= write_transfer_recs($trans_A_filehandle, $trans_B_filehandle, $model, $date, $cycle, $model_in_filename, $command);
           if ($err != 0) {
              return 1;
           }
        }
	} else {
		foreach $command (@commands_this_job) {
			$err= write_transfer_recs($trans_A_filehandle, $trans_B_filehandle, $model, $date, $cycle, $model_in_filename, $command);
		}
		if ($err != 0) {
		     return 1;
		}

	}
 

	$rsync_time_stamp = strftime "%a %b %e %H:%M:%S %Y", localtime;  ## Time::HiRes::time();
	 logmsg($Config::info, "MAG processor time stamp for $model rsync event is: $rsync_time_stamp ");

	 # Rename the transfer files from hidden files to visible.

         if ( $SYNC_TO_A eq "YES" ) {
            close($trans_A_filehandle);
	    logmsg($Config::info, "Make the transfer file A visible: ${transfer_A_file}");
	    if (! rename $transfer_A_file_hidden, $transfer_A_file) {
	       logmsg($Config::fatal, "Error renaming the hidden transfer file to visible name: ${transfer_A_file}");
	       return 1;
	    } 
	 } 
         if ( $SYNC_TO_B eq "YES" ) {
        close($trans_B_filehandle);
	    logmsg($Config::info, "Make transfer file B visible: ${transfer_B_file}");
	    if (! rename $transfer_B_file_hidden, $transfer_B_file) {
	       logmsg($Config::fatal, "Error renaming the hidden transfer B file to visible name: ${transfer_B_file}");
	       return 1;
	    } 
	 } 

	 if (! $main::command_line_mode) {   # when not running at the command prompt
	    # Requeue the send2web jobs to transfer the images created by the current processor task
	    my $ECF_NAME=$ENV{'ECF_NAME'};
	    $ECF_NAME =~ s/processor$/sync/;             # change "processor" suffix in task name to "sync"
	    $ECF_NAME =~ s/hrrr_subh_sync$/hrrr_sync/;   # special case where both hrrr and hrrr_subh processors feed the hrrr_sync job
            $ECF_NAME =~ s/\/jmag_[^\/]*?storm_tracks_sync/\/jmag_storm_tracks_sync/;  # special case where multiple storm trask jobs feed the storm_tracks_sync job
	    if ( $SYNC_TO_A eq "YES" ) {
	       my $send2web_A_task = $ECF_NAME;
	      #$send2web_A_task =~ s/\/processor\//\/send2web_A\//;  # look in send2web_A family
	       $send2web_A_task =~ s/\/processor\//\/send2web\/cprk\//;  # look in send2web/cprk family
               my $send2web_A_status = `ecflow_client --get_state $send2web_A_task | grep task`;
               if ( $send2web_A_status =~ /state:complete/ ) {
                  logmsg($Config::info, "Requeue ecFlow task $send2web_A_task");
                  `ecflow_client --requeue $send2web_A_task`;      # requeue the job if it was complete
               } elsif ( $send2web_A_status =~ /state:queued/ ) {
                     logmsg($Config::info, "ecFlow sync task is already queued: $send2web_A_task");
               } else {
                     logmsg($Config::info, "ecflow task $send2web_A_task is not queued or complete, cannot requeue: $send2web_A_status");
               }
            }
	    if ( $SYNC_TO_B eq "YES" ) {
	       my $send2web_B_task = $ECF_NAME;
	      #$send2web_B_task =~ s/\/processor\//\/send2web_B\//;  # look in send2web_B family
	       $send2web_B_task =~ s/\/processor\//\/send2web\/bldr\//;  # look in send2web/bldr family
               my $send2web_B_status = `ecflow_client --get_state $send2web_B_task | grep task`;

             if ( $send2web_B_status =~ /state:complete/ ) {
                  logmsg($Config::info, "Requeue ecFlow task $send2web_B_task");
                  `ecflow_client --requeue $send2web_B_task`;      # requeue the job if it was complete
               } elsif ( $send2web_B_status =~ /state:queued/ ) {
                     logmsg($Config::info, "ecFlow sync task is already queued: $send2web_B_task");
               } else {
                     logmsg($Config::info, "ecflow task $send2web_B_task is not queued or complete, cannot requeue: $send2web_B_status");
               }
            }

            # Ping the ecflow server
            my $ping = `ecflow_client --ping`;
            logmsg($Config::info, "$0: Error returned from 'ecflow_client --ping': \n $!") if $?;
            logmsg($Config::info, "$0: Output from 'ecflow_client --ping': $ping");

           # Set the ecflow event

            my $set_event = `ecflow_client --event processing`;
            logmsg($Config::info, "$0: Error returned from 'ecflow_client --event processing': \n $!") if $?;
            logmsg($Config::info, "$0: Output from 'ecflow_client --event processing': $set_event");
            
         }
      }
   }
   
   $local_process_end_time = Time::HiRes::time();
   $local_process_duration = $local_process_end_time-$local_process_start_time;
   logmsg($Config::info, "Total time used for the current $ncommands_this_job jobs is $local_process_duration. \n");
   logmsg($Config::info, "\n");
   logmsg($Config::info, "---------------------------------------");
   logmsg($Config::info, " Done running commands for job $job : ");
   logmsg($Config::info, "---------------------------------------\n\n\n");

}
   logmsg($Config::info, "End of process_command_list. Exit status = $error ");

return  $error;

}

#------------------------------------------------------------------
##
## is_batch_job_running
##
## Return the running status of an lsb job
#  Return 1 if running, 0 otherwise
#
#  Input:
#     jobid 
##
##------------------------------------------------------------------

sub is_batch_job_running($) {
   my $jobid=shift;
   chomp($jobid);

   open(BJOBS, "-|","qstat" ) or die "Can't open pipe from bjobs: $!\n";
   while( <BJOBS> ) {
#      logmsg($Config::info, "$jobid---Checking qstat line: $_");
   if ($_ =~ /^${jobid}\b.*mag.*\bE\b/) {
       logmsg($Config::info, "jobid $jobid has Exited");
       return 0;
     }
     if ($_ =~ /^${jobid}\b.*mag.*\b(Q|R)\b/) {
       logmsg($Config::info, "jobid $jobid is $1 (Queued or Running)");
       return 1;
     }
   }
   close( BJOBS );
   return 0;

}

#------------------------------------------------------------------
##
## write_transfer_recs      
##
## Write the filenames that the command created  into the
#  file connected to the file handle.
#  Check that the files actually exist and are not empty.
#  If any do not exist (except sref-cluster), return error.
#  Do not make transfer files for skewt.
#
#  Input:
#     $trans_A_filehandle, $trans_B_filehandle, $model, $date,  $cycle, $model_in_filename, $command
#
## Globals accessed:
#     $main::COMOUT
##
##------------------------------------------------------------------

sub write_transfer_recs($$$$$$$) {


my ($trans_A_filehandle, $trans_B_filehandle, $model, $date, $cycle, $model_in_filename, $command) = @_;
my ($file, @files, $sm_file, $lg_file, $fhr, $area, $param, $stn, $make_fmins, @fmins, $sizes, $subdir_path, $GIF_dir, $error, $missing_file);

$error=0;
$missing_file="";


if ($model ne "skewt" ) {
 
   # Extract the fhr, area, and param field from the command
   if ($model eq "gfs-snd" || $model eq "nam-snd"  ) {
      if ($command !~ /^\S*\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)/) {
	 logmsg($Config::fatal, "Illegal pattern for command, cannot extract parameters: $command");
	 return 1;
      }
      $fhr=$4;
      $area=$5;
      $param=$6;
      $stn=$7; 
      # make the subdirectory levels to be used for this file
      $subdir_path = make_subdir_levels_path($area, $stn);
   } else {
      if ($command !~ /^\S*\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)\s*(\S*)/) {
	 logmsg($Config::fatal, "Illegal pattern for command, cannot extract parameters: $command");
	 return 1;
      }
      $fhr=$4;
      $area=$5;
      $param=$6;
      # make the subdirectory levels to be used for this file
      $subdir_path = make_subdir_levels_path($area, $param);
   }



   # Check if forecast minutes are defined for this parameter.
   # If so, 
   #    - we must use the minutes field in the filename.
   #    - If make_fmins=yes, then make the forecast minutes for the minutes defined.
   # else
   #      make_fmins=no, just make the hourly, no  minutes.
   #

       
   @fmins = get_forecast_mins_for_model($main::models_doc, $model);

   if ( @fmins > 0 ) {
      $make_fmins=get_make_fmins_by_model_param( $main::models_doc, $model, $param);
   } else {
     $make_fmins = "no";
   }
   
   # Get sizes for this area
   
   my $sizes=get_image_size_model_area( $main::models_doc, $model, $area);
   
   if ((! defined $sizes) || ($sizes eq "")) {
      logmsg($Config::fatal, "No image size defined for $model $area");
      return 1;
   }
        
   # if make_fmins eq 'no', we are not making forecast minutes.

   if ($make_fmins eq 'no') {
   
     # If uair, then there is no forecast hour field in the name.
     # If forecast minutes are defined, then must add forecast minutes field "00" 
   
      my  $fhr_part="${fhr}_";
      if (@fmins > 0 ) {
         $fhr_part="${fhr}00_";
      } elsif ( $model eq "uair") {
        $fhr_part="";
      }
   
     #
     # Create the filenames to write to the transfer file.
     # If this model is sref-cluster, then there may be a file for each cluster, up to 6.
     # Add large/small filenames, if made.
     #
   
      @files=(); 
      my $file=""; 
      if ($model eq "sref-cluster" ) {
         for (my $i=1; $i <= 6; $i++) {
            push (@files, "${subdir_path}${model_in_filename}_${area}_${fhr_part}${param}_cluster${i}.gif");
         }
      } else {
         push (@files, "${subdir_path}${model_in_filename}_${area}_${fhr_part}${param}.gif");
																																  
         if ($sizes eq "SML") {
            push (@files, "${subdir_path}${model_in_filename}_${area}_${fhr_part}${param}_s.gif");
            push (@files, "${subdir_path}${model_in_filename}_${area}_${fhr_part}${param}_l.gif");
         }

      }
   } else {
  
      # forecast minutes are defined for this model.
      
   
      foreach my $fmin (@fmins) {
      
         push (@files, "${subdir_path}${model_in_filename}_${area}_${fhr}${fmin}_${param}.gif");
      
         if ($sizes eq "SML") {
            push (@files, "${subdir_path}${model_in_filename}_${area}_${fhr}${fmin}_${param}_s.gif");
            push (@files, "${subdir_path}${model_in_filename}_${area}_${fhr}${fmin}_${param}_l.gif");
         }
      }
   }


   # Construct the name of the directory where the files reside
      
   $GIF_dir="$main::COMOUT/$model_in_filename/$date/$cycle";

   # 
   #  Write filenames to transfer files.
   #  Check that they actually exist.
   #

   my $missing_file_count = 0;
   foreach $file (@files) {

      # Check if it exists first (sref-cluster may not make all the clusters)

      if ( -s "$GIF_dir/$file" ) {
         if ( $trans_A_filehandle ne "" ) {   
            logmsg($Config::debug, "Write filename to transfer file A: $file");
            if (! print $trans_A_filehandle "$file\n") {
               logmsg($Config::debug, "Error writing filename to transfer file A: $file");
               $error=1;
            }
         }
         if ( $trans_B_filehandle ne "" ) {   
            logmsg($Config::debug, "Write filename to transfer file B: $file");
            if (! print $trans_B_filehandle "$file\n") {
               logmsg($Config::debug, "Error writing filename to transfer file B: $file");
               $error=1;
            }
         }

      } else {
         if ($model ne "sref-cluster") {
            $missing_file_count++;
            logmsg($Config::warn, "Expected file is empty or was not created: $GIF_dir/$file");
         }
      }
   }

   if ($missing_file_count > 0) {
       logmsg($Config::warn, "Total number of empty/missing files that were expected: ${missing_file_count}");
   }
   return $error;
}
#------------------------------------------------------------------
#
## make_subdir_levels_path
#
# return the subdirectory path levels defined by
# $subdir levels
# Access globals:
#    $main::subdir_levels

sub make_subdir_levels_path($$) {
   my ($area, $param) = @_;
   switch($main::subdir_levels) {
      case "none"      {return ""; }
      case "area"      {return "$area/"; }
      case "areaparam" {return "$area/$param/";}
      case "station"   {return "$param/"; }
      else             {return "";}
   }
}
#------------------------------------------------------------------
##
## check_and_create_output_dirs
##
## Check for the existance of status and transfer directories and
#  create them if necessary.  The GEMPAK scripts check for and
#  and create the output GIF directory if necessary.
#
##
##------------------------------------------------------------------

sub check_and_create_output_dirs($$$) {
my ($status_dir, $transfer_A_dir, $transfer_B_dir) = @_;
  if (! -d $status_dir) {
      if (!mkdirs($status_dir)) {
         logmsg($Config::fatal, "Could not create status dir $status_dir: $?");
         return 1;
      }
   }

   if (! -d "$transfer_A_dir") {
      if (!mkdirs("$transfer_A_dir")) {
         logmsg($Config::fatal, "Could not create transfer dir $transfer_A_dir: $?");
         return 1;
      }
   }
   if (! -d "$transfer_B_dir") {
      if (!mkdirs("$transfer_B_dir")) {
         logmsg($Config::fatal, "Could not create transfer B dir $transfer_B_dir: $?");
         return 1;
      }
   }
}

return 0;
}

#------------------------------------------------------------------
# Check_valid_cycle - check if cycle is valid 
# If area specified, check for area-specific cycle defs
#
# Input:
# model
# cycle
# area
#
# Access globals:
#    $main::models_root
#    @main::model_cycles
#
# Log:
# 8/2018 PJF Renamed to check_valid_cycle_area
#------------------------------------------------------------------
sub check_valid_cycle_area ($$$) {

   my $model=shift;
   my $cycle=shift;
   my $area=shift;
   my @cycles=();
   my @areacycles=();

   my $valid_for_model=grep{ /^$cycle/ } @main::model_cycles;

   if ( $area ne "") {

      @areacycles = get_cycles_by_model_area_name( $main::models_doc, $model, $area);

      # If cycles are defined for the area, it overrides
      #
      if (@areacycles > 0 ) {
         $valid_for_model= scalar grep{ /^$cycle/ } @areacycles;
      }
   }


   return $valid_for_model;
}

#------------------------------------------------------------------
# Check_valid_cycle_param - check if cycle is valid for this param. 
#
# Input:
# model
# cycle
# param
#
# Access globals:
#    $main::models_root
#    @main::model_cycles
#
#------------------------------------------------------------------
sub check_valid_cycle_param ($$$) {

   my $model=shift;
   my $cycle=shift;
   my $param=shift;
   my @param_exception_cycles=();

   return 0 if ( $param eq "");

   # First make sure that the cycle is valid for the model

   my $valid_for_model=grep{ /^$cycle/ } @main::model_cycles;

   my $valid_for_param=$valid_for_model;

   if ($valid_for_model) {

      @param_exception_cycles = get_cycle_exception_by_model_param( $main::models_doc, $model, $param);

      #  if nothing returned, there was no param exception defined, so  it's valid
      #  if array returned, then the param is only valid for the cycles listed
      #  so check for this cycle in the list. 
     
      if (@param_exception_cycles == 0) { 
          $valid_for_param=1;
      } else {
         $valid_for_param= scalar grep{ /^$cycle/ } @param_exception_cycles;
      }
   }

   return $valid_for_param;
}

#------------------------------------------------------------------
## get_area_stations - get a list of stations for a model/area.
##
## Input:
## model
## area
##
## Access globals:
##   MAG_dir
##
##------------------------------------------------------------------
sub get_area_stations($$) {
   my $model_type=shift;
   my $area=shift;
   my @stations=();

   # Build station filename for each model/area.
   # Remove "-snd" from model type to get model name

   my $model = substr($model_type, 0, index($model_type, '-'));
   my $stn_file="${main::MAG_tables}/snstns_${model}_${area}.tbl";
   my $fh; # file handler  

   # If a station file exists. 
   if ( -e $stn_file ) {
      # Read station file for the area.
      if (! open ($fh, "<", $stn_file)) { return 1; }

      while (<$fh>){
	 chomp;
	 push  @stations , $_;
      }

      close $fh;
     
      return @stations;
   } else {
      return;
   }
}

#------------------------------------------------------------------
### get_station_ID - extract station's digital ID.
###
### Input:
### station line ( char_id, digit_id, ctr, lat, lon )
###
### Access globals:
###   
###
###------------------------------------------------------------------
sub get_station_ID($){
   my $station=shift;
   chomp($station);
   my @words=split / /, $station;
   my $char_id=$words[0];
   my $digit_id=$words[1];
   my $ctr=$words[2];
   my $lat=$words[3]/100;
   my $lon=$words[4]/100;

   return $digit_id;
}


#------------------------------------------------------------------
## remove_extra_sref_clusters - Iterate through list of files and
##       keep only one cluster per forecast hour.
##       
## For instance, the (delete) files would be removed from
## the list:
##
## sref_mean_cluster3_ncep_2017080921f60
## sref_mean_cluster3_ncep_2017080921f63
## sref_mean_cluster1_ncep_2017080921f66
## sref_mean_cluster2_ncep_2017080921f66 (delete)
## sref_mean_cluster3_ncep_2017080921f66 (delete)
## sref_mean_cluster1_ncep_2017080921f69
## sref_mean_cluster2_ncep_2017080921f69 (delete)
## sref_mean_cluster3_ncep_2017080921f69 (delete)
## sref_mean_cluster1_ncep_2017080921f72
## sref_mean_cluster2_ncep_2017080921f72 (delete)
##
##
## Input:
##    array reference to array of files
##
## Output:
##
##    Same array reference to edited array of files
##
## Access globals:
##
##------------------------------------------------------------------
#
sub remove_extra_sref_clusters {

   my @files = @{$_[0]}; # Dereference the array argument
   my ($file, $compare_file, $prev_file, $index);

   # perform a fast sort by last 13 chars: YYYYMMDDCCfHH
 
   @files = map { # Get original line back
        $_->[0]
      } sort { # Compare last 13 char fields fields
        $a->[1] cmp $b->[1]
      } map { # Turn each line into [original line, last 13 chars]
        [ $_, substr($_, -13) ] 
   } @files;


   # Iterate through files, removing any extra clusters
   # For the same forecast hour.  Index points to the
   # current array element.
   #
   # For each filename in the list, create a string to compare by
   # replacing  the cluster number in the filename with 'n'
   # and compare it to the previous filename.  If they match, delete 
   # the current array element. 
   #
   $index=0;
   foreach $file (@files) {
      $compare_file=$file;
      $compare_file =~ s/_cluster._/_clustern_/;
      if ( $index > 0 ) { # first rec, no previous to compare it to.
         if ($compare_file eq $prev_file) {
            splice (@files, $index, 1); # remove this element of the array
         } else {
            $index++;
         }
      } else {
         $index++;
      }
      $prev_file=$compare_file;
  }
}