#!/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
#     -l logging_level   - overrides $Config::default_logging_level
#           0 - debug
#           1 - info
#           2 - warn
#           3 - fatal
#  
# 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 - 9/05/2013  - Update get_file_list to return the latest cycles in @latest_cycles, and
#                              the array of storms for each of the cycles in %storm_latest_cycles (hash of arrays).
#                              In the main program, process the @latest_cycles, and get list of storms
#                              from %storm_latest_cycles.
# Paula Freeman - 10/6/2014  - Move the trigger to start the transfer  job to after the files  have been processed.
#                              This will ensure the transfer job will find some files and not exit prematurely..
# 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 - 5/07/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 - 5/07/2015  - Redmine 5978: In poe job, set MP_PULSE to set a max number of seconds for job 
#                              to run. Set to 1200 (20 min).  
# Paula Freeman - 6/1/2015   - Add storm name to transfer_file name to make it unique.  Add storm to 
#                              argument  list for process_command_list
# Paula Freeman - 6/9/2015   - Require $COMIN environment variable, add as argument
# Paula Freeman - 6/15/2015  - Move transfer files to "transfer" subdirectory of status directory
# 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 - 10/29/2015 - Add counter and random number to transfer file name to ensure it is unique..
# Paula Freeman - 11/04/2015 - Use 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 - 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_dir 
#                                if SYNC_TO_B == YES, make in Config::transfer_B_dir
#                 6/15/2016 - Call get_make-fmins_by_model_param instead of 
#                             get_fmins_by_model_param
# 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. 
# 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 - 6/23/2016 - Replace ghm with hmon
# Paula Freeman - 6/05/2017 - Directory pattern for input files has changed.
#                             In get_file_list, make backreferences in pattern  
#                             for first YYYYMMDD and first CC only. Then
#                             grab date, cycle, and storm from $1, $2, and $3
# Paula Freeman - 1/15/2019 - Phase 3 migration (MAG-111).  Use 'mpirun cfp' instead of 'mpirun.lsf cfp '
# 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 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.
# Anu Simon - 4/2020          Moved exmag-v3.pl scripts to /ush
#                             Renamed mag_processor_config to mag_processor_config.pl in /ush
#                             The change is made in the script
#Anu Simon 11/01/2021         Made changes pertaining to wcoss2 - mpiexec command
#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
#
#Hieu Phung 01/2023        Add logic to check if a similar processor job is already running
#A Simon 03/22023          hwrf replaced as hafsa and hmon replaced as hafsb
#A Simon 03/2023           Changes made to read in the storm name from the holdvars.txt and changes for the proper transfer_list names.
#A Simon 04/2024           Change to read hafsa/hafsb as hfsa/hfsb
#A Simon 04/2024           SPA suggested change to ignore an existing but empty hidden process_file
#A Simon 04/2024           Changes to read the correct directory path.
##############################################################################
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 Data::Dumper;

# Declare functions


sub int_handler();
sub make_temp_dir($);
sub get_model_info($);
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 check_for_exceptions ($$$$$);
sub write_command_filenames($$$$$$$);
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 silence_used_once_warnings();
sub check_and_create_output_dirs($$$);
sub get_stormname($$$$$);

# 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_areas_by_model_name 
#sub get_params_by_model_name 
#sub get_fhr_ranges_by_model_name 
#sub get_script_by_model_name 
#sub get_area_exception_by_name_param 

# Declare lexically scoped variables only used in main

my $model;                      # model to to make products for.
my $date="";			# yyyymmdd portion of $date/cycle
my $cycle="";			# cc portion of $datecycle
my $storm="";                   # current storm - is stormid
my $STORM="";                   # to grep stromname ifrom strom?.holdvars.txt
my $stormS="";                  # to grep for stormid from storm?.holdvars.txt defiedn in upper case
my $stormname="";               # Current stormname 
my $highest_fhr_processed="";	# Highest forecast hour processed for current cycle
my $highest_fhr="";		# Highest forecast hour found for current cycle
my $first_fhr="";		# First forecast hour of the model
my $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 $num_cycles_to_process;      # Number of cycles to process. From
                                # xml, or overridden on command line

# Package global variables accessed by functions

our $MAG_dir;                   # MAG root dir
our $MAG_tables;                # directory for MAG tables
our $COMIN="";                  # Value of COMIN 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 @latest_cycles=();          # Latest cycles to process
our %storms_latest_cycles=();   # Storms associated with latest_cycles
our @filepathpats=();		# Filepath patterns
our %all_filenames=();		# All filenames that match pattern for the current cycle
our $transfer_seq=0;            # Counter used in transfer file names.
our $datecycle="";		# Current date/cycle being processed yyyymmddcc

our @command_list=();		# Queue of commands generated for new files found
our $max_threads;		# number of children to run concurrently
our $model_in_filename;         # model string used in GIF filenames
our $number_fhr_ranges=0;      	# Number of forecast hour ranges from model_info
our @model_info=[];		# 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 @storms=();               	# storms found for this model
our @params=();               	# parameters to make for this model
our $max_run_time;		# max run time for each command
our $exception_mode=0;          # used in MAG_processor.pl. No exception mode for hurricanes.
                                # ignore forecast hour ranges from xml file.
our $development_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 $COMOUTstatus="";           # Value of COMOUTstatus environment variable
our $COMOUT="";                 # Value of COMOUT environment variable

# Get options

my %options;
getopts('m:l:dc:', \%options);
$model= defined $options{m} ? $options{m} : "";
$development_mode= defined $options{d};

# Get model name

if (! $model)  {
   print "usage: $0 -m <model> [-l <logging_level>]\n";
   exit 1;
}

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

$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'};
(defined($COMIN) || !$COMIN) or die  "Environment variable COMIN not set.\n";


$COMOUT=$ENV{'COMOUT'};
$COMOUTstatus=$ENV{'COMOUTstatus'};
$SYNC_TO_A=$ENV{'SYNC_TO_A'};  # If defined and "YES" then make transfer 
                               # files in Config::transfer_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}/scripts/mag_processor_config.pl";
require "${MAG_dir}/ush/magv3-xml-library.pl" or die "Unable to open ${MAG_dir}/scripts/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 
#

if ( $model !~ /^((atcf)|(hfsa|hfsb)-(nested|parent))$/) {
   logmsg($Config::fatal, "This is not a hurricane model: $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);
#
# 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 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 successfully: $model_in_filename \n");

#
# get parameters
#

logmsg($Config::info, "Get the parameters for model $model... \n");
@params = get_params_by_model_name($models_doc, $model);
if ((scalar @params) == 0) {
    logmsg( $Config::fatal, "No parameters defined for model $model");
    $proc_error=1;
}
logmsg($Config::info, "Get parameters completed successful. \n");

#
# get filepath patterns
#

logmsg($Config::info, "Get the file path patterns for model $model... ");
@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;
}

#
# 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");

#
#  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");
do {

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

   $found_more_fhrs=0;

   # get the list of model files and storm names.
   # $datecycle is the latest cycle, @storms is the list of storms
   
   $proc_error = $proc_error || get_file_list ($model, $num_cycles_to_process);

   # Process the latest cycles found
   #
   # @latest_cycles is the array of cycles
   # %storms_latest_cycles is an hash of arrays.  The key is the cycle, the value is an array of storms.
   
   # print Dumper \%main::storms_latest_cycles;

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

      if ( @storms = (@ {$storms_latest_cycles{$datecycle}})) {

      $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
      #
   
      ($first_fhr, $last_fhr) = get_model_first_last_hr ($model, $cycle);

      logmsg($Config::debug, "Storms for this cycle:\n");
      foreach $storm (@storms) {
         logmsg($Config::debug, "$storm\n");
      


      ###Get stormname from storm*.holdvars.txt"
      $stormname = get_stormname($model, $storm, $date, $cycle, $stormname);
      logmsg($Config::info, "$model $storm $date $cycle $stormname\n");
      ###########
      }
      
      # process each storm
   
      foreach $storm (@storms) {
         logmsg($Config::info, "Process storm: $storm\n");


         # get highest forecast hour processed
   
         $highest_fhr_processed = get_highest_fhr_processed($date, $cycle, $storm);
   
         # Get the highest forecast hour received
   
         $highest_fhr = get_highest_fhr_found($date, $cycle, $storm,$first_fhr, $last_fhr);
   
         # Check if we have processed all the forecast hours received
         
        logmsg($Config::info, "Check if we have processed all the forecast hours received..."); 
        logmsg($Config::info, "last_fhr: $last_fhr and highest_fhr_processed: $highest_fhr_processed ");
   
        if ($last_fhr == $highest_fhr_processed) {
   
           logmsg($Config::info, "All forecast hours processed for cycle $date $cycle is done. Exiting loop.\n");
        } else {
   
            if ($highest_fhr <= $highest_fhr_processed )  {
               logmsg($Config::info, "All forecast hours received have been processed. Exiting.\n");
            } else {
               logmsg($Config::info, "More forecast hours available... proceeding...");

               # If we've gotten in some new forecast hours, submit the
               # jobs for them and update the highest processed.
               #
               $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_fhr);
   
               # Create the list of commands for hours > $highest_fhr_processed 
               # up to $highest_fhr and execute them.
     

               $proc_error = make_command_list ($model, $storm, $highest_fhr_processed, $highest_fhr, $max_run_time);
   	    
   
               $proc_error = $proc_error || process_command_list($model, $date, $cycle, $storm, $model_in_filename);
      
               if (! $proc_error && !$die) {
                  set_highest_fhr_processed($highest_fhr, $date, $cycle, $storm);
               }
   
               # 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_fhr ");
               logmsg($Config::info, "Current poe job time used (include overhead) is: $job_duration \n");
            }
         }
      }
      } else {
          logmsg($Config::debug, "no storms found for datetype: $datecycle \n");
      }

      # if we found more forecast hours this time, sleep and try again.

      if ($found_more_fhrs && (! $die && !$proc_error)) {
         logmsg($Config::info, "Sleep if no error found..");
         sleep  $Config::min_between_checks*60;
      }
  }

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

} until  (!$found_more_fhrs || $die || $proc_error );


# Sleep to ensure that this job does not get requeued 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
# Return latest cycle with storms in datecycle
# Return the list of storms in datecycle in global @storms
# Return the list of cycles to be processed in @latest_cycles
#
# Input:
#    $model
#    $num_cycles_to_process
#
# Return:
#    0 - no error
#    1 - error getting filename patterns
#
# Changes globals:
#    %datecycles  (doesn't need to be global anymore, but left it alone)
#    @all_filenames
#    @filepathpats
#    @latest_cycles
#    %storms_latest_cycles
#
#--------------------------------------------------------------
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...");
   for (my $i = 0; $i < $npats; $i++) {
      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;
      $wildcardpat[$i] =~ s/STORM/\*/g;
      @files=glob ($wildcardpat[$i]);
      $REpat[$i]=$main::filepathpats[$i];
      $REpat[$i] =~ s/YYYYMMDD/\(\\d\{8\}\)/; # Backreference on first YYYYMMDD
      $REpat[$i] =~ s/YYYYMMDD/\\d\{8\}/g;    
      $REpat[$i] =~ s/CC/\(\\d\{2\}\)/;       # Backreference on first CC
      $REpat[$i] =~ s/CC/\\d\{2\}/g; 
      $REpat[$i] =~ s/FFF/\(\\d\{3\}\)/g;
      $REpat[$i] =~ s/STORM/\(\.\*\)/g;
      logmsg($Config::debug, "REpat[$i] = $REpat[$i]\n");
      logmsg($Config::debug, "wildcardpat[$i] = $wildcardpat[$i]\n");
      
      foreach $file (@files) {
         logmsg($Config::debug, "Looking for date and cycle to process in file: $file");
         if ($Config::use_dated_dirs) {    
            logmsg($Config::debug, "using a dated directory pattern: $REpat[$i]");
            # WCOSS /com/hur/prod/hwrf.YYYYMMDDCC/nawips/STORM/hwrfp_YYYYMMDDCCfFFF_STORM
            $file =~ m/${REpat[$i]}/;  
            # Backreferences correspond to these fields:
            # /com/hur/prod/hwrf.(YYYYMMDD)(CC)/nawips/(STORM)/hwrfp_YYYYMMDDCCfFFF_STORM
            $date=$1; $cycle=$2; $storm=$3;
         } else {
            logmsg($Config::debug, "using a NAWIPS-style directory pattern: $REpat[$i]");
            # netapp .../model/grid/gempak/hwrf/hwrfn_YYYYMMDDCCfFFF_STORM
            $file =~ m/${REpat[$i]}/;  
            $date=$1; $cycle=$2; $storm=$4;
         }


         # ignore invest storms
         
         if (! exists $main::storms{"$storm"}) {
             logmsg($Config::info, "Found storm: $storm");
         }
         if ($storm !~ /invest.*/) {

            if ( exists $main::datecycles{"$date$cycle"}) {

               # add storm to list for date/time, if not already there.
               
               if (grep {$_ eq $storm} @{ $main::datecycles{"$date$cycle"}}) {
                  logmsg($Config::debug, "Storm $storm already in the list for $date$cycle");
               } else {
                  push @{ $main::datecycles{"$date$cycle"}}, $storm;
               }
            } else {
               logmsg($Config::debug, "Found date: $date and cycle: $cycle. Keep looking...");
               $main::datecycles{"$date$cycle"} = ();
               push @{ $main::datecycles{"$date$cycle"}}, $storm;
            }
            $main::all_filenames{$file} = 1;
            #print Dumper(\%main::datecycles);
         }
      }
      logmsg($Config::info, "All available dates and cycles found. \n");
   }

   #print "datecycles:\n"; 

   #print Dumper \%main::datecycles;

# sort the date/time keys found, and get the latest  
# Load @latest_cycles with the latest cycles to process
# Load %storms_latest_cycles with the array of storms for the 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++) {
     if ($num_cycles >= $i) {
        $main::latest_cycles[$i-1]=$sorted_cycles[$i-1];
        foreach my $storm  (@{$main::datecycles{"$sorted_cycles[$i-1]"}}) {
           push @{ $main::storms_latest_cycles{"$sorted_cycles[$i-1]"}}, $storm;
        }
     }
   } 

  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
#
# Accesses globals:
#  @forecast_hr_list
#  @filepathpats
#  @all_filenames
#
# returns: highest forecats hour found, or -1
#
#--------------------------------------------------------------
sub get_highest_fhr_found ($$$$$){

   logmsg($Config::info, "Get the highest forecast hour received and forecast hours processed so far...");

# 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)

   my ($date, $cycle, $storm, $first_fhr, $last_fhr) = @_;
   my ($highest_fhr, $got_all_this_pat, $keep_going, $hr, $checkforfile);
   my ($all_hours_in_one_file, $found_all_pats, $npats);


   $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, $first_fhr, $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.
   
   for (my $i=0; $keep_going; $i++) {
      $hr=$main::forecast_hr_list[$i];
      $got_all_this_pat=1;
      
      $all_hours_in_one_file=0;
      $npats=$#main::filepathpats + 1;
      $found_all_pats=1;
      for (my $j = 0; $j < $npats; $j++) {
         chomp $main::filepathpats[$j];
         $checkforfile=$main::filepathpats[$j];
         $checkforfile =~ s/YYYYMMDD/${date}/g;
         $checkforfile =~ s/CC/${cycle}/g;
         $checkforfile =~ s/STORM/${storm}/g;

         if ( $checkforfile !~ s/FFF/${hr}/g) {
           # all hours are in the same file
           $all_hours_in_one_file=1;
         } 
         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");
         }
      }
      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 the 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;
   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_info

#------------------------------------------------------------------
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_info[$i][0]) &&
            ( $first_fhr_to_process <= $main::model_info[$i][1]) ) {
             $done=1;
             $step=$main::model_info[$i][2];
             $last_fhr_this_range=$main::model_info[$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_info[$i][2]) >= $main::model_info[$i][0]) && 
                    (($ihr + $main::model_info[$i][2]) <= $main::model_info[$i][1]) ) {
                  $found=1;
                  $step=$main::model_info[$i][2];
                  $ihr = $ihr + $step; 
                  $last_fhr_this_range=$main::model_info[$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 forecast hours found: $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_info
#------------------------------------------------------------------

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_info($model);

   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_info[0][0];
      $step                = $main::model_info[0][2];
      $last_fhr            = $main::model_info[$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_info  - retrieve the forecast hour range and timestep
#                   from the MAG.xml file.  
#                   If exception mode is nam_sim_radar, 
#                   hardcode with timestep 03 for 000-036 and
#                   06 for 042-084.
#  Input: 
#     model_name
#  Return:
#     nrange - number of forecast hour ranges
#
# Changes global:
#     @model_info 
#
#------------------------------------------------------------------

sub get_model_info ($) {

   my $model = shift;

   my $nrange;
   my @cycles;
   $nrange=0;

   logmsg($Config::debug, "call get_fhr_ranges");
   get_fhr_ranges($model);
   $nrange=scalar @model_info;
   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, $storm) = @_;
   my ($last_processed, $fhrs_processed_fname);
   $last_processed=-1;
  
   $fhrs_processed_fname="${status_dir}/${model}_${storm}_${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
#   exception_mode
#
#---------------------------------------------------------

sub set_highest_fhr_processed($$$$) {

   my ($hr, $date, $cycle, $storm)=@_;
   my $fhrs_processed_fname;

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

   
   logmsg($Config::info, "Set the highest forecast hour processed to $hr");

   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     
#  Input: 
#     $model                      - model
#     $storm                      - storm
#     $highest_fhr_processed     - highest hour processed so far
#     $last_hr_to_submit          - last hour to submit
#     $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
#     $main::exception_mode    - nam_sim_radar or 'none'
#     $load_balancing_script
#------------------------------------------------------------------


sub make_command_list($$$$$) {

   my ($model, $storm, $highest_fhr_processed, $last_hr_to_submit, $max_run_time)=@_;
   my ($first_hr_after_last_processed, $begin_hr, $end_hr);
   my $hr;
   my ($index_first_hr, $param, $script, $area, $fhr, $i, $command);
   my (@params);
   my $reg_string="";  
   my $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++;
   }
 

   logmsg($Config::info, "Submit jobs for  $first_hr_after_last_processed -> $last_hr_to_submit\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;
          }
      }
     

 
      #----------------------------------------------------------
      # Write the command to create the gifs for the 
      # parameter, for each area
      #
      #----------------------------------------------------------
      
      logmsg($Config::info, "\n\n");
      logmsg($Config::info, "====================================================");
      logmsg($Config::info, " Now processing parameter : $param ");
      logmsg($Config::info, "====================================================\n");

      ($begin_hr, $end_hr) = ($first_hr_after_last_processed, $last_hr_to_submit);

      $keep_going=1;
      $i=0;
      while ($keep_going && $i<=$#main::forecast_hr_list) {

         if ( ($main::forecast_hr_list[$i] >= $begin_hr) &&
              ($main::forecast_hr_list[$i] <= $end_hr) ) {

            $fhr=$main::forecast_hr_list[$i];

            if ( $Config::load_balancing ) {
                $command = "${main::MAG_dir}/${Config::load_balancing_script} $command";
             }


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

            push (@main::command_list, "$main::MAG_dir/ush/$script $model $cycle $date $fhr $storm $param");
         } elsif ($main::forecast_hr_list[$i] >  $end_hr) {
            $keep_going=0;
         }
         $i++;
      }

      logmsg($Config::info, "\n");
      logmsg($Config::info, "====================================================");
      logmsg($Config::info, " Done processing parameter : $param ");
      logmsg($Config::info, "====================================================\n\n");

   }   

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

   return 0;   
 
}

   
#------------------------------------------------------------------
#  check_for_exceptions - Adjust the first and last forecast
#     hour to process if any exceptions exist for this
#     model/area/param.
#
# Input:
#    $model
#    $area
#    $param
#    $first_hour_to_submit
#    $last_hour_to_submit
#
#  Accesses globals:
#     @forecast_hr_list
#  
#------------------------------------------------------------------
sub check_for_exceptions($$$$$) { 
   my ($model, $area, $param, $first_hr_to_submit, $last_hr_to_submit)=@_;

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

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

   #print "look for exceptions for $check_area, then $check_all\n";
   # 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.
   logmsg($Config::debug,  "call get_area_exception: $model, $param, $area");
   ($new_first_hr,$new_last_hr) = get_area_exception($model, $param, $area);
   logmsg($Config::debug,  "returned $new_first_hr,$new_last_hr");
   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) = get_area_exception($model, $param, "all");
      logmsg($Config::debug,  "returned $new_first_hr,$new_last_hr");
      if ($new_first_hr != -99) {  # found an exception
         $exception=1;
      }
   }

   if ($exception) {
   

      # 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, don't make any

      } 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
            # parameter and cycle through the hours until the first one >= to
            # first_hr_to_submit is found.
   
            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");

      # 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 prevous 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);
   
}
   
   
#------------------------------------------------------------------
#  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));
      }
      $new_log="${Config::log_dir}/${main::logname}.${date}";
      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 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
#
# Access globals:
#    $main::models_root
#
#------------------------------------------------------------------
sub get_fhr_ranges($) {

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

   @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_info[$i-1][0]=$1;
         $main::model_info[$i-1][1]=$2;
         $main::model_info[$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_exception=shift;
   my $new_first_hr=-99;
   my $new_last_hr=-99;
   my @new_fhrs_strings;
   my $query;
  
   @new_fhrs_strings=get_area_exception_by_name_param( $main::models_doc, $model, $area_exception, $param); 
   if ( (scalar @new_fhrs_strings) == 0) {
      logmsg($Config::debug, "No exception found for $param except-$area_exception");
   } else {
      if ( $new_fhrs_strings[0] =~ /(.*),(.*)(,.*)?/) {
         $new_first_hr=$1;
         $new_last_hr=$2;
         if ( $model eq "gfs" && $param =~ /(dom_precip_type|precip_p.*)/){
            if ( $new_first_hr =~ /(.*),(.*)/) {
                $new_first_hr=$1;
                $new_last_hr=$2;
            }
         }
       }
   }
   logmsg($Config::debug, "Return exception for $param except-$area_exception: new_first_hr=$new_first_hr new_last_hr=$new_last_hr");
   return ($new_first_hr, $new_last_hr);
}

#------------------------------------------------------------------
# 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;

}

#------------------------------------------------------------------
## 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, $storm, $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 $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;


logmsg($Config::info, "--------------------------------------------------------------------------------");
logmsg($Config::info, " Start process_command_list for model: $model, date: $date, cycle: $cycle and storm: $storm");
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++;
   $local_process_start_time = Time::HiRes::time();
   $tmp_subdir="${main::job_tmp_dir}/job_$job";
   logmsg($Config::debug, "Create temp directory for job: $job..."); 
   logmsg($Config::info, "\n");
   logmsg($Config::info, "---------------------------------------");
   logmsg($Config::info, " Running commands for job $job : ");
   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_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;

   # Create the transfer lists in hidden files
   # Get system time info
   my $sys_timestamp=strftime("%Y_%m_%d_%H%M%S", gmtime(time));

   $main::transfer_seq++;
   $transfer_A_file="${transfer_A_dir}/${model}_${date}${cycle}_${stormname}${storm}_${job}_${main::transfer_seq}_${sys_timestamp}.transfer";
   $transfer_A_file_hidden="${transfer_A_dir}/.${model}_${date}${cycle}_${stormname}${storm}_${job}_${main::transfer_seq}_${sys_timestamp}.transfer";
   $transfer_B_file="${transfer_B_dir}/${model}_${date}${cycle}_${stormname}${storm}_${job}_${main::transfer_seq}_${sys_timestamp}.transfer";
   $transfer_B_file_hidden="${transfer_B_dir}/.${model}_${date}${cycle}_${stormname}${storm}_${job}_${main::transfer_seq}_${sys_timestamp}.transfer";


   if ("$SYNC_TO_A" eq "YES") {
      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") {
      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 = "";
   }

   # write the commands to the poe script
   # For each command, write the files it will create to the hidden transfer list
   
   logmsg($Config::debug, "Write the commands to poe script file and filenames to transfer file...");
   # write the commands to the poe script
   #
   my @copy_of_command_list;
   logmsg($Config::debug, "Write the commands to poe script file...");
   for (my $i=0; $i<$ncommands_this_job; $i++) {
      logmsg($Config::debug, "Write number $i command to poe script file...");
      $command = shift @main::command_list;;
      $commands_processed++;
      print POEFILE "-n 1 ${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");

      push(@copy_of_command_list, $command)
      #$err= write_command_filenames($trans_A_filehandle, $trans_B_filehandle, $model, $date, $cycle, $model_in_filename, $command);
      #if ($err != 0) {
	      #   return 1;
	      #}
   }
   #  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::development_mode) {
      if (! open (RUN_JOB, '-|', $poe_run_script_file)) {
         logmsg($Config::fatal, "Error opening pipe for $poe_run_script_file: $!");
         $error=1;
      }
   } else {
      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 {

         my $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 ");
   
         # Poe job has completed successfully
         # Rename the transfer lists from hidden files to visible.
         
         logmsg($Config::info, "Poe job completed successfully, so make the transfer file ${transfer_A_file} visible:");

         for my $cmd (@copy_of_command_list){
	     $err= write_command_filenames($trans_A_filehandle, $trans_B_filehandle, $model, $date, $cycle, $model_in_filename, $cmd);
	     if ($err != 0) {
	        return 1;
	     }
	}


         if ( $SYNC_TO_A eq "YES" ) {
            if ( -s $transfer_A_file_hidden ) {
               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;
               }
            } else {
               logmsg($Config::warn, "skip empty hidden transfer file: ${transfer_A_file_hidden}");
            }
         }
         if ( $SYNC_TO_B eq "YES" ) {
            if ( -s $transfer_B_file_hidden ) {
               logmsg($Config::info, "Poe job completed successfully, so make the transfer file ${transfer_B_file} visible:");
               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;
               }
            } else {
               logmsg($Config::warn, "skip empty hidden transfer file: ${transfer_B_file_hidden}");
            }
         }

         if (! $main::development_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
	    if ( $SYNC_TO_A eq "YES" && -s $transfer_A_file ) {
	       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, "ecFlow task $send2web_A_task is being requeued");
	          `ecflow_client --requeue $send2web_A_task`;      # requeue the job if it was complete
	       } else {
	          logmsg($Config::info, "Unable to requeue ecFlow task $send2web_A_task");
	       }
        }
	    if ( $SYNC_TO_B eq "YES" && -s $transfer_B_file ) {
	       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, "ecFlow task $send2web_B_task is being requeued");
	          `ecflow_client --requeue $send2web_B_task`;      # requeue the job if it was complete
	       } else {
	          logmsg($Config::info, "Unable to requeue ecFlow task $send2web_B_task");
	       }
        }
            `ecflow_client --event processing`;
         }
      }
   }
   
   $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_command_filenames      
##
## Write the filenames that the command will create into the
#  file connected to the file handles.
#
#  Input:
#     $trans_A_filehandle, $trans_B_filehandle, $model, $date,  $cycle, $model_in_filename, $command
##
##
##------------------------------------------------------------------

sub write_command_filenames($$$$$$$) {


my ($trans_A_filehandle, $trans_B_filehandle, $model, $date, $cycle, $model_in_filename, $command) = @_;
my ($file, $sm_file, $lg_file, $fhr, $area, $param, $make_fmins, @fmins,  $sizes);
my ($stormS, $stormname);

# extract the fhr, area, and param field from the command

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;
$storm=$5;




# 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";
}

# if make_fmins eq 'no', we are not making forecast minutes.
  
if ($make_fmins eq 'no' ) {
   if (@fmins > 0 ) {
     $fhr="${fhr}00";
   }
   my $file="${model_in_filename}_${area}_${fhr}_${param}.gif";
   my $GIF_dir="$main::COMOUT/$model_in_filename/$date/$cycle";
   if ( $trans_A_filehandle ne "" ) {
      $stormname=get_stormname($model, $storm, $date, $cycle, $stormname);	   
      my $newfile="${model_in_filename}_${stormname}${area}_${fhr}_${param}.gif";
      logmsg($Config::debug, "Write filename to transfer file A: $newfile");
      if ( -s "$GIF_dir/$newfile" ) {
      print $trans_A_filehandle "$newfile\n";
      } else {
	      logmsg($Config::info, "Transfer file not present");
     }	      
   }
   if ( $trans_B_filehandle ne "" ) {
      $stormname=get_stormname($model, $storm, $date, $cycle, $stormname);
      my $newfile="${model_in_filename}_${stormname}${area}_${fhr}_${param}.gif";
      logmsg($Config::debug, "Write filename to transfer file B: $newfile");
      if ( -s "$GIF_dir/$newfile" ) {
      print $trans_B_filehandle "$newfile\n";
      } else {
	      logmsg($Config::info, "Transfer file not present");
	      
   }
 }

   # Get the sizes made for this area

   my $sizes=get_image_size_model_area( $main::models_doc, $model, "atl-pac");
   if ((! defined $sizes) || ($sizes eq "")) {
      logmsg($Config::fatal, "No image size defined for $model $area");
      return 1;
   }
   if ($sizes eq "SML") {
      $sm_file="${model_in_filename}_${area}_${fhr}_${param}_s.gif";
      $lg_file="${model_in_filename}_${area}_${fhr}_${param}_l.gif";
      if ( $trans_A_filehandle ne "" ) {
         logmsg($Config::debug, "Write filename to transfer file A: $sm_file");
         print $trans_A_filehandle "$sm_file\n";
         logmsg($Config::debug, "Write filename to transfer file A: $lg_file");
         print $trans_A_filehandle "$lg_file\n";
      }
      if ( $trans_B_filehandle ne "" ) {
         logmsg($Config::debug, "Write filename to transfer file B: $sm_file");
         print $trans_B_filehandle "$sm_file\n";
         logmsg($Config::debug, "Write filename to transfer file B: $lg_file");
         print $trans_B_filehandle "$lg_file\n";
      }
   }
   return 0;
} else {

  # forecast minutes are defined for this model.

   foreach my $fmin (@fmins) {
   
      my $file="${model_in_filename}_${area}_${fhr}${fmin}_${param}.gif";
      if ( $trans_A_filehandle ne "" ) {
         logmsg($Config::debug, "Write filename to transfer file A: $file");
         print $trans_A_filehandle "$file\n";
      }
      if ( $trans_B_filehandle ne "" ) {
         logmsg($Config::debug, "Write filename to transfer file B: $file");
         print $trans_B_filehandle "$file\n";
      }


      # Get the sizes made for this area

      my $sizes=get_image_size_model_area( $main::models_doc, $model, "atl-pac");
   
      if ($sizes eq "SML") {

         $sm_file="${model_in_filename}_${area}_${fhr}${fmin}_${param}_s.gif";
         $lg_file="${model_in_filename}_${area}_${fhr}${fmin}_${param}_l.gif";
         if ( $trans_A_filehandle ne "" ) {
             logmsg($Config::debug, "Write filename to transfer file A: $sm_file");
             print $trans_A_filehandle "$sm_file\n";
             logmsg($Config::debug, "Write filename to transfer file A: $lg_file");
             print $trans_A_filehandle "$lg_file\n";
          }
          if ( $trans_B_filehandle ne "" ) {
             logmsg($Config::debug, "Write filename to transfer file B: $sm_file");
             print $trans_B_filehandle "$sm_file\n";
             logmsg($Config::debug, "Write filename to transfer file B: $lg_file");
             print $trans_B_filehandle "$lg_file\n";
          }

      }
   
   }
   return 0;
}

## Get the stormname
## Input model, data, cycle and name of storm

sub get_stormname ($$$$$) {
     my ($model, $storm, $date, $cycle) = @_;
     my ($stormname);
     my ($stormS);
      $stormS = uc($storm);
      if ($model eq "hfsa-parent" or $model eq "hfsa-nested" or "atcf") {
           $stormname=lc(`grep -B 1 "STORMID=$stormS" ${COMIN}/hfsa.${date}/${cycle}/storm*.holdvars.txt | grep STORM= | cut -d = -f 2 | tr -d "[:space:]"`); 
	 } else {
		  $stormname=lc(`grep -B 1 "STORMID=$stormS" ${COMIN}/hfsb.${date}/${cycle}/storm*.holdvars.txt | grep STORM= | cut -d = -f 2 | tr -d "[:space:]"`);}
  }
     return ($stormname);



#------------------------------------------------------------------
##
## 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;
}
