<commit_before>#!/usr/bin/env python
# __BEGIN_LICENSE__
#  Copyright (c) 2009-2013, United States Government as represented by the
#  Administrator of the National Aeronautics and Space Administration. All
#  rights reserved.
#
#  The NGT platform is licensed under the Apache License, Version 2.0 (the
#  "License"); you may not use this file except in compliance with the
#  License. You may obtain a copy of the License at
#  http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
# __END_LICENSE__

# Top level program to process all of the Icebridge data.
# - This program is not sophisticated enough to handle everything and will need to be
#   superceded by another script.

import os, sys, argparse, datetime, time, subprocess, logging, multiprocessing
import re, shutil, glob

import os.path as P

# The path to the ASP python files and tools
basepath      = os.path.dirname(os.path.realpath(__file__)) # won't change, unlike syspath
pythonpath    = os.path.abspath(basepath + '/../Python')     # for dev ASP
libexecpath   = os.path.abspath(basepath + '/../libexec')    # for packaged ASP
icebridgepath = os.path.abspath(basepath + '/../IceBridge')  # IceBridge tools

# Prepend to Python path
sys.path.insert(0, basepath)
sys.path.insert(0, pythonpath)
sys.path.insert(0, libexecpath)
sys.path.insert(0, icebridgepath)

import icebridge_common, pbs_functions, archive_functions, run_helper
import asp_system_utils, generate_flight_summary

asp_system_utils.verify_python_version_is_supported()

# Prepend to system PATH
os.environ["PATH"] = basepath       + os.pathsep + os.environ["PATH"]
os.environ["PATH"] = pythonpath     + os.pathsep + os.environ["PATH"]
os.environ["PATH"] = libexecpath    + os.pathsep + os.environ["PATH"]
os.environ["PATH"] = icebridgepath  + os.pathsep + os.environ["PATH"]

#=========================================================================
# Parameters

# Constants used in this file

CAMGEN_PBS_QUEUE   = 'normal'
BATCH_PBS_QUEUE    = 'normal'
BLEND_PBS_QUEUE    = 'normal'
ORTHOGEN_PBS_QUEUE = 'normal'
LABEL_PBS_QUEUE = 'normal'

GROUP_ID = 's1827'

g_start_time = -1
g_stop_time  = -1

def start_time():
    global g_start_time, g_stop_time
    g_start_time = time.time()
    
def stop_time(job, logger):
    global g_start_time, g_stop_time
    g_stop_time = time.time()
    wall_s = float(g_stop_time - g_start_time)/3600.0
    logger.info( ("Elapsed time for %s is %g hours." % (job, wall_s) ) )


PFE_NODES = ['san', 'ivy', 'has', 'bro']

#=========================================================================

# 'wes' = Westmere = 12 cores/24 processors, 48 GB mem, SBU 1.0, Launch from mfe1 only!
# 'san' = Sandy bridge = 16 cores,  32 GB mem, SBU 1.82
# 'ivy' = Ivy bridge   = 20 cores,  64 GB mem, SBU 2.52
# 'has' = Haswell      = 24 cores, 128 GB mem, SBU 3.34
# 'bro' = Broadwell    = 28 cores, 128 GB mem, SBU 4.04

def getParallelParams(nodeType, task):
    '''Return (numProcesses, numThreads, tasksPerJob, maxHours) for running a certain task on a certain node type'''

    # Define additional combinations and edit as needed.

    if task == 'camgen':
        if nodeType == 'san': return (8,  3, 300, 3)
        if nodeType == 'ivy': return (10, 3, 400, 3)
        if nodeType == 'bro': return (14, 4, 500, 3)
        if nodeType == 'wes': return (10, 4, 400, 8)
    
    if task == 'dem':
        if nodeType == 'san': return (2, 8, 70,  8)
        if nodeType == 'ivy': return (4, 8, 80,  8)
        if nodeType == 'bro': return (6, 8, 100, 8)
        if nodeType == 'wes': return (3, 8, 75,  8)
    
    if task == 'blend':
        if nodeType == 'san': return (8,  3,  800, 4)
        if nodeType == 'ivy': return (10, 3, 1000, 4)
        if nodeType == 'bro': return (14, 4, 1400, 4) # 200 seems to finish in 10 minutes
        if nodeType == 'wes': return (10, 3,  800, 8) 
    
    if task == 'orthogen':
        if nodeType == 'san': return (8,  2, 350, 6)
        if nodeType == 'ivy': return (10, 2, 400, 5)
        if nodeType == 'bro': return (14, 4, 500, 4)
        if nodeType == 'wes': return (10, 4, 400, 8)

    # TODO: All guesses!
    if task == 'label':
        if nodeType == 'san': return (16, 1, 350, 6)
        if nodeType == 'ivy': return (30, 1, 400, 5)
        if nodeType == 'bro': return (28, 1, 500, 4)
        if nodeType == 'wes': return (12, 1, 400, 8)


    raise Exception('No params defined for node type ' + nodeType + ', task = ' + task)

#=========================================================================

def getEmailAddress(userName):
    '''Return the email address to use for a user'''
    
    if userName == 'smcmich1':
        return '<EMAIL>'
    if userName == 'oalexan1':
        return '<EMAIL>'

def sendEmail(address, subject, body):
    '''Send a simple email from the command line'''
    # Remove any quotes, as that confuses the command line.
    subject = subject.replace("\"", "")
    body    = body.replace("\"", "")
    try:
        cmd = 'mail -s "' + subject + '" ' + address + ' <<< "' + body + '"'
        #print(cmd) # too verbose to print
        os.system(cmd)
    except Exception, e:
        print("Could not send mail.")
        
#---------------------------------------------------------------------

def readRunList(path, options):
    '''Reads a list of runs in this format: GR 20110411.
       Output is a list of (site, yyyymmdd) pairs.'''

    runList = []
    with open(path, 'r') as f:
        for line in f:
            if line[0] == '#': # Skip comments
                continue
            parts = line.split() # Site, yyyymmdd
            # Set up a class object to help manage the run
            runList.append(run_helper.RunHelper(parts[0], parts[1], options.unpackDir))
    return runList

def addToRunList(path, run):
    '''Append a run to a list file'''
    with open(path, 'a') as f:
        f.write(path[0]+' '+path[1]+'\n')
    
def getRunsToProcess(allRuns, skipRuns, doneRuns):
    '''Go through the run lists to get the list of runs that
       should be processed.'''
       
    runList = []
    for run in allRuns:
        if (run in skipRuns) or (run in doneRuns):
            continue
        runList.append(run)
    return runList

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


def runFetch(run, options, logger):
    '''Fetch all the data for a run if it is not already available'''

    logger.info("Checking all data is present.")

    # Check if already done
    allIsFetched = False
    if options.skipCheckInputs:
        allIsFetched = True
    else:
        try:
            # Note that the check below is incomplete, it does not check for Fireball
            allIsFetched = run.allSourceDataFetched(options.noNavFetch)
        except Exception, e:
            allIsFetched = False
            logger.warning('Caught error checking fetch status.\n'
                       + str(e))

    # Fetch the archive from lfe, only in the case the directory is not present
    if not options.skipTapeFetch:
        archive_functions.retrieveRunData(run, options.unpackDir, options.useTar, logger)

    pythonPath = asp_system_utils.which('python')
    
    # Fetch whatever is missing directly from NSIDC, and force to have the indices
    # regenerated in this case. Hopefully just a few files are missing.
    # - This is likely to have to fetch the large nav data file(s).
    # Note that we do all conversions as well, sans camera generation. But hopefully the run is clean and all
    # has been done by now. 
    if not options.noRefetch:
        logger.info("Fetch from NSIDC.")
        cmd = (pythonPath + ' ' + icebridge_common.fullPath('full_processing_script.py') + ' --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --output-folder %s --refetch --stop-after-fetch --start-frame %d --stop-frame %d' % (options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, run.getFolder(), options.startFrame, options.stopFrame))

        if options.noNavFetch:
            cmd += ' --no-nav'
            
        if options.skipValidate or options.parallelValidate:
            # We will do validation in parallel later
            cmd += ' --skip-validate'

        if not options.noRefetchIndex:
            cmd += ' --refetch-index'
            
        logger.info(cmd)
        os.system(cmd)

    # Don't need to check results, they should be cleaned out in conversion call.
    run.setFlag('fetch_complete')

def runConversion(run, options, logger):
    '''Run the conversion tasks for this run on the supercomputer nodes.
       This will also run through the fetch step to make sure we have everything we need.'''

    # Check if already done. If we want to validate in parallel, we must still
    # go through this step, though hopefully it will be very fast then.
    # TODO: Can have a check if all files have been validated (jpeg, ortho, image, etc),
    # and then, if all cameras are also present, this step can be skipped.
    try:
        if run.conversionIsFinished(options.startFrame, options.stopFrame)   and \
           run.checkForImages(options.startFrame, options.stopFrame, logger) and \
           (not options.parallelValidate):
            logger.info('Conversion is already complete.')
            return
    except Exception, e:
        logger.warning('Caught error checking conversion status, re-running conversion.\n' + str(e))
        
    logger.info('Converting data for run ' + str(run))

    # Run just the nav data --> estimated camera conversion on the PFE machine.
    # - This is single threaded and may take a while but is primarily IO driven.
    # - We can't start the slower ortho2pinhole processes until this is finished.
    # - If all of the camera files already exist this call will finish up very quickly.
    if not options.noNavFetch:
        logger.info("Generating estimated camera files from the navigation files.")
        pythonPath = asp_system_utils.which('python')
        cmd = (pythonPath + ' ' + icebridge_common.fullPath('full_processing_script.py') + ' --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --output-folder %s --skip-fetch --stop-after-convert --no-lidar-convert --no-ortho-convert --skip-fast-conversions --start-frame %d --stop-frame %d' % (options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, run.getFolder(), options.startFrame, options.stopFrame))
    
        if options.skipValidate or options.parallelValidate:
            cmd += ' --skip-validate'
            
        logger.info(cmd)
        os.system(cmd)    
        logger.info("Finished generating estimated camera files from nav.")
    
    # Retrieve parallel processing parameters
    (numProcesses, numThreads, tasksPerJob, maxHours) = getParallelParams(options.nodeType, 'camgen')
    
    # Split the conversions across multiple nodes using frame ranges
    # - The first job submitted will be the only one that converts the lidar data.
    # We will also run validation in parallel.
    numFrames    = options.stopFrame - options.startFrame + 1
    numCamgenJobs = numFrames / tasksPerJob
    if numCamgenJobs < 1:
        numCamgenJobs = 1
        
    outputFolder = run.getFolder()

    scriptPath = icebridge_common.fullPath('full_processing_script.py')
    args       = (' --skip-fetch --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --stop-after-convert --num-threads %d --num-processes %d --output-folder %s' 
                  % ( options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, numThreads, numProcesses, outputFolder))
    if options.noNavFetch:
        args += ' --no-nav'
    if options.simpleCameras:       # This option greatly decreases the conversion run time
        args += ' --simple-cameras' # - Camera conversion could be local but image conversion still takes time.

    baseName = run.shortName() # SITE + YYMMDD = 8 chars, leaves seven for frame digits.

    # Get the location to store the logs    
    pbsLogFolder = run.getPbsLogFolder()
    os.system('mkdir -p ' + pbsLogFolder)

    # Submit all the jobs
    currentFrame = options.startFrame
    jobList = []
    for i in range(0, numCamgenJobs):
        jobName    = ('%s%06d%s' % ('C', currentFrame, baseName) ) # C for camera
        startFrame = currentFrame
        stopFrame  = currentFrame+tasksPerJob-1
        if (i == numCamgenJobs - 1):
            stopFrame = options.stopFrame # Make sure nothing is lost at the end

        thisArgs = (args + ' --start-frame ' + str(startFrame)
                         + ' --stop-frame  ' + str(stopFrame ) )
        if i != 0: # Only the first job will convert lidar files
            thisArgs += ' --no-lidar-convert'

        logPrefix = os.path.join(pbsLogFolder, 'convert_' + jobName)
        logger.info('Submitting camera generation job: ' + scriptPath + ' ' + thisArgs)
        pbs_functions.submitJob(jobName, CAMGEN_PBS_QUEUE, maxHours, logger,
                                options.minutesInDevelQueue,
                                GROUP_ID,
                                options.nodeType, '/usr/bin/python2.7',
                                scriptPath + " " + thisArgs, logPrefix)
        jobList.append(jobName)
        currentFrame += tasksPerJob

    # Wait for conversions to finish
    pbs_functions.waitForJobCompletion(jobList, logger)

    if options.parallelValidate:

        # Consolidate various validation files done in parallel, to make the subsequent
        # step faster.
        validFilesList = icebridge_common.validFilesList(outputFolder,
                                                         options.startFrame, options.stopFrame)
        validFilesSet = set()
        validFilesSet = icebridge_common.updateValidFilesListFromDisk(validFilesList, validFilesSet)
        allValidFiles = glob.glob(os.path.join(outputFolder,
                                               icebridge_common.validFilesPrefix() + '*'))
        for fileName in allValidFiles:
            validFilesSet = icebridge_common.updateValidFilesListFromDisk(fileName, validFilesSet)
        icebridge_common.writeValidFilesList(validFilesList, validFilesSet)

        # Now we will refetch and reprocess all files that were not
        # valid so far. Hopefully not too many. This must be on a head
        # node to be able to access the network. We don't do any orthoconvert
        # here as that one is too time-consuming
        logger.info("Refetch and process any invalid files.")
        pythonPath = asp_system_utils.which('python')
        cmd = (pythonPath + ' ' + icebridge_common.fullPath('full_processing_script.py') + ' --refetch --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --output-folder %s --stop-after-convert --no-ortho-convert --start-frame %d --stop-frame %d --num-threads %d --num-processes %d' % (options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, run.getFolder(), options.startFrame, options.stopFrame, numThreads, numProcesses))
        if options.noNavFetch:
            cmd += ' --no-nav'
        logger.info(cmd)
        os.system(cmd)

        # See if there are any new files for which we need to run ortho2pinhole on some node
        logger.info("See for which files to redo ortho.")
        (orthoList, numFrames) = icebridge_common.orthoListToRerun(validFilesSet, outputFolder)

        logger.info("Number of cameras to regenerate using ortho2pinhole: " + str(numFrames))
        if numFrames > 0:
            cmd = (icebridge_common.fullPath('full_processing_script.py') + ' --skip-fetch --skip-validate --skip-fast-conversions --no-nav --stop-after-convert --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --output-folder %s --start-frame %d --stop-frame %d --num-threads %d --num-processes %d --frames-file %s' % (options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, run.getFolder(), options.startFrame, options.stopFrame, numThreads, numProcesses, orthoList))
            #logger.info(cmd)
            jobList = []
            jobName    = ('%s%06d%s' % ('C', 0, baseName) ) # C for camera
            logPrefix = os.path.join(pbsLogFolder, 'convert_' + jobName)
            logger.info('Submitting camera generation job: ' + cmd)
            pbs_functions.submitJob(jobName, CAMGEN_PBS_QUEUE, maxHours, logger,
                                    options.minutesInDevelQueue,
                                    GROUP_ID,
                                    options.nodeType, '/usr/bin/python2.7',
                                    cmd, logPrefix)
            jobList.append(jobName)
            
            # Wait for conversions to finish
            pbs_functions.waitForRunCompletion(baseName, jobList, logger)

        logger.info("Finished refetching and reprocessing.")

    # Check the results
    # - If we didn't get everything keep going and process as much as we can.
    success = False
    try:
        success = run.conversionIsFinished(options.startFrame, options.stopFrame, verbose = True)
    except Exception, e:
        logger.warning('Caught error checking conversion status.\n' + str(e))

    if not success:
        #raise Exception('Failed to convert run ' + str(run))
        logger.warning('Could not fully convert run ' + str(run))

    run.setFlag('conversion_complete')

def generateBatchList(run, options, listPath, logger):
    '''Generate a list of all the processing batches required for a run'''

    logger.info("Generate batch list: " + listPath)
    
    refDemName = icebridge_common.getReferenceDemName(run.site)
    refDemPath = os.path.join(options.refDemFolder, refDemName)

    # Retrieve parallel processing parameters
    (numProcesses, numThreads, tasksPerJob, maxHours) = getParallelParams(options.nodeType, 'dem')

    # No actual processing is being done here so it can run on the PFE
    # - No start/stop frames here, always generate the full list so it is stable.
    pythonPath = asp_system_utils.which('python')
    scriptPath = icebridge_common.fullPath('full_processing_script.py')
    cmd       = ('%s %s --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --skip-fetch --skip-convert --num-threads %d --num-processes %d --output-folder %s --bundle-length %d --log-batches ' 
                  % (pythonPath, scriptPath, options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, numThreads, numProcesses, run.getFolder(), options.bundleLength))

    # For full runs we must cleanup, as otherwise we'll run out of space
    if not options.skipCleanup:
        cmd += ' --cleanup'

    if options.noNavFetch:
        cmd += ' --no-nav'

# TODO: Find out what takes so long here!
# - Also fix the logging!

    logger.info(cmd)
    os.system(cmd)


def getOutputFolderFromBatchCommand(batchCommand):
    '''Extract the output folder from a line in the batch file'''
    # TODO: Integrate with the more robust logic in icebridge_common.getFrameRangeFromBatchFolder()
    # which can parse a batch command.
    parts        = batchCommand.split()
    outputFolder = parts[10] # This needs to be kept up to date with the file format!
    return outputFolder

# TODO: Share code with the other function
def filterBatchJobFile(run, batchListPath, logger):
    '''Make a copy of the batch list file which only contains incomplete batches.'''
    <commit_msg>Bug fix<commit_after>#!/usr/bin/env python
# __BEGIN_LICENSE__
#  Copyright (c) 2009-2013, United States Government as represented by the
#  Administrator of the National Aeronautics and Space Administration. All
#  rights reserved.
#
#  The NGT platform is licensed under the Apache License, Version 2.0 (the
#  "License"); you may not use this file except in compliance with the
#  License. You may obtain a copy of the License at
#  http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
# __END_LICENSE__

# Top level program to process all of the Icebridge data.
# - This program is not sophisticated enough to handle everything and will need to be
#   superceded by another script.

import os, sys, argparse, datetime, time, subprocess, logging, multiprocessing
import re, shutil, glob

import os.path as P

# The path to the ASP python files and tools
basepath      = os.path.dirname(os.path.realpath(__file__)) # won't change, unlike syspath
pythonpath    = os.path.abspath(basepath + '/../Python')     # for dev ASP
libexecpath   = os.path.abspath(basepath + '/../libexec')    # for packaged ASP
icebridgepath = os.path.abspath(basepath + '/../IceBridge')  # IceBridge tools

# Prepend to Python path
sys.path.insert(0, basepath)
sys.path.insert(0, pythonpath)
sys.path.insert(0, libexecpath)
sys.path.insert(0, icebridgepath)

import icebridge_common, pbs_functions, archive_functions, run_helper
import asp_system_utils, generate_flight_summary

asp_system_utils.verify_python_version_is_supported()

# Prepend to system PATH
os.environ["PATH"] = basepath       + os.pathsep + os.environ["PATH"]
os.environ["PATH"] = pythonpath     + os.pathsep + os.environ["PATH"]
os.environ["PATH"] = libexecpath    + os.pathsep + os.environ["PATH"]
os.environ["PATH"] = icebridgepath  + os.pathsep + os.environ["PATH"]

#=========================================================================
# Parameters

# Constants used in this file

CAMGEN_PBS_QUEUE   = 'normal'
BATCH_PBS_QUEUE    = 'normal'
BLEND_PBS_QUEUE    = 'normal'
ORTHOGEN_PBS_QUEUE = 'normal'
LABEL_PBS_QUEUE = 'normal'

GROUP_ID = 's1827'

g_start_time = -1
g_stop_time  = -1

def start_time():
    global g_start_time, g_stop_time
    g_start_time = time.time()
    
def stop_time(job, logger):
    global g_start_time, g_stop_time
    g_stop_time = time.time()
    wall_s = float(g_stop_time - g_start_time)/3600.0
    logger.info( ("Elapsed time for %s is %g hours." % (job, wall_s) ) )


PFE_NODES = ['san', 'ivy', 'has', 'bro']

#=========================================================================

# 'wes' = Westmere = 12 cores/24 processors, 48 GB mem, SBU 1.0, Launch from mfe1 only!
# 'san' = Sandy bridge = 16 cores,  32 GB mem, SBU 1.82
# 'ivy' = Ivy bridge   = 20 cores,  64 GB mem, SBU 2.52
# 'has' = Haswell      = 24 cores, 128 GB mem, SBU 3.34
# 'bro' = Broadwell    = 28 cores, 128 GB mem, SBU 4.04

def getParallelParams(nodeType, task):
    '''Return (numProcesses, numThreads, tasksPerJob, maxHours) for running a certain task on a certain node type'''

    # Define additional combinations and edit as needed.

    if task == 'camgen':
        if nodeType == 'san': return (8,  3, 300, 3)
        if nodeType == 'ivy': return (10, 3, 400, 3)
        if nodeType == 'bro': return (14, 4, 500, 3)
        if nodeType == 'wes': return (10, 4, 400, 8)
    
    if task == 'dem':
        if nodeType == 'san': return (2, 8, 70,  8)
        if nodeType == 'ivy': return (4, 8, 80,  8)
        if nodeType == 'bro': return (6, 8, 100, 8)
        if nodeType == 'wes': return (3, 8, 75,  8)
    
    if task == 'blend':
        if nodeType == 'san': return (8,  3,  800, 4)
        if nodeType == 'ivy': return (10, 3, 1000, 4)
        if nodeType == 'bro': return (14, 4, 1400, 4) # 200 seems to finish in 10 minutes
        if nodeType == 'wes': return (10, 3,  800, 8) 
    
    if task == 'orthogen':
        if nodeType == 'san': return (8,  2, 350, 6)
        if nodeType == 'ivy': return (10, 2, 400, 5)
        if nodeType == 'bro': return (14, 4, 500, 4)
        if nodeType == 'wes': return (10, 4, 400, 8)

    # TODO: All guesses!
    if task == 'label':
        if nodeType == 'san': return (16, 1, 350, 6)
        if nodeType == 'ivy': return (30, 1, 400, 5)
        if nodeType == 'bro': return (28, 1, 500, 4)
        if nodeType == 'wes': return (12, 1, 400, 8)


    raise Exception('No params defined for node type ' + nodeType + ', task = ' + task)

#=========================================================================

def getEmailAddress(userName):
    '''Return the email address to use for a user'''
    
    if userName == 'smcmich1':
        return '<EMAIL>'
    if userName == 'oalexan1':
        return '<EMAIL>'

def sendEmail(address, subject, body):
    '''Send a simple email from the command line'''
    # Remove any quotes, as that confuses the command line.
    subject = subject.replace("\"", "")
    body    = body.replace("\"", "")
    try:
        cmd = 'mail -s "' + subject + '" ' + address + ' <<< "' + body + '"'
        #print(cmd) # too verbose to print
        os.system(cmd)
    except Exception, e:
        print("Could not send mail.")
        
#---------------------------------------------------------------------

def readRunList(path, options):
    '''Reads a list of runs in this format: GR 20110411.
       Output is a list of (site, yyyymmdd) pairs.'''

    runList = []
    with open(path, 'r') as f:
        for line in f:
            if line[0] == '#': # Skip comments
                continue
            parts = line.split() # Site, yyyymmdd
            # Set up a class object to help manage the run
            runList.append(run_helper.RunHelper(parts[0], parts[1], options.unpackDir))
    return runList

def addToRunList(path, run):
    '''Append a run to a list file'''
    with open(path, 'a') as f:
        f.write(path[0]+' '+path[1]+'\n')
    
def getRunsToProcess(allRuns, skipRuns, doneRuns):
    '''Go through the run lists to get the list of runs that
       should be processed.'''
       
    runList = []
    for run in allRuns:
        if (run in skipRuns) or (run in doneRuns):
            continue
        runList.append(run)
    return runList

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


def runFetch(run, options, logger):
    '''Fetch all the data for a run if it is not already available'''

    logger.info("Checking all data is present.")

    # Check if already done
    allIsFetched = False
    if options.skipCheckInputs:
        allIsFetched = True
    else:
        try:
            # Note that the check below is incomplete, it does not check for Fireball
            allIsFetched = run.allSourceDataFetched(options.noNavFetch)
        except Exception, e:
            allIsFetched = False
            logger.warning('Caught error checking fetch status.\n'
                       + str(e))

    # Fetch the archive from lfe, only in the case the directory is not present
    if not options.skipTapeFetch:
        archive_functions.retrieveRunData(run, options.unpackDir, options.useTar, logger)

    pythonPath = asp_system_utils.which('python')
    
    # Fetch whatever is missing directly from NSIDC, and force to have the indices
    # regenerated in this case. Hopefully just a few files are missing.
    # - This is likely to have to fetch the large nav data file(s).
    # Note that we do all conversions as well, sans camera generation. But hopefully the run is clean and all
    # has been done by now. 
    if not options.noRefetch:
        logger.info("Fetch from NSIDC.")
        cmd = (pythonPath + ' ' + icebridge_common.fullPath('full_processing_script.py') + ' --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --output-folder %s --refetch --stop-after-fetch --start-frame %d --stop-frame %d' % (options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, run.getFolder(), options.startFrame, options.stopFrame))

        if options.noNavFetch:
            cmd += ' --no-nav'
            
        if options.skipValidate or options.parallelValidate:
            # We will do validation in parallel later
            cmd += ' --skip-validate'

        if not options.noRefetchIndex:
            cmd += ' --refetch-index'
            
        logger.info(cmd)
        os.system(cmd)

    # Don't need to check results, they should be cleaned out in conversion call.
    run.setFlag('fetch_complete')

def runConversion(run, options, logger):
    '''Run the conversion tasks for this run on the supercomputer nodes.
       This will also run through the fetch step to make sure we have everything we need.'''

    # Check if already done. If we want to validate in parallel, we must still
    # go through this step, though hopefully it will be very fast then.
    # TODO: Can have a check if all files have been validated (jpeg, ortho, image, etc),
    # and then, if all cameras are also present, this step can be skipped.
    try:
        if run.conversionIsFinished(options.startFrame, options.stopFrame)   and \
           run.checkForImages(options.startFrame, options.stopFrame, logger) and \
           (not options.parallelValidate):
            logger.info('Conversion is already complete.')
            return
    except Exception, e:
        logger.warning('Caught error checking conversion status, re-running conversion.\n' + str(e))
        
    logger.info('Converting data for run ' + str(run))

    # Run just the nav data --> estimated camera conversion on the PFE machine.
    # - This is single threaded and may take a while but is primarily IO driven.
    # - We can't start the slower ortho2pinhole processes until this is finished.
    # - If all of the camera files already exist this call will finish up very quickly.
    if not options.noNavFetch:
        logger.info("Generating estimated camera files from the navigation files.")
        pythonPath = asp_system_utils.which('python')
        cmd = (pythonPath + ' ' + icebridge_common.fullPath('full_processing_script.py') + ' --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --output-folder %s --skip-fetch --stop-after-convert --no-lidar-convert --no-ortho-convert --skip-fast-conversions --start-frame %d --stop-frame %d' % (options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, run.getFolder(), options.startFrame, options.stopFrame))
    
        if options.skipValidate or options.parallelValidate:
            cmd += ' --skip-validate'
            
        logger.info(cmd)
        os.system(cmd)    
        logger.info("Finished generating estimated camera files from nav.")
    
    # Retrieve parallel processing parameters
    (numProcesses, numThreads, tasksPerJob, maxHours) = getParallelParams(options.nodeType, 'camgen')
    
    # Split the conversions across multiple nodes using frame ranges
    # - The first job submitted will be the only one that converts the lidar data.
    # We will also run validation in parallel.
    numFrames    = options.stopFrame - options.startFrame + 1
    numCamgenJobs = numFrames / tasksPerJob
    if numCamgenJobs < 1:
        numCamgenJobs = 1
        
    outputFolder = run.getFolder()

    scriptPath = icebridge_common.fullPath('full_processing_script.py')
    args       = (' --skip-fetch --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --stop-after-convert --num-threads %d --num-processes %d --output-folder %s' 
                  % ( options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, numThreads, numProcesses, outputFolder))
    if options.noNavFetch:
        args += ' --no-nav'
    if options.simpleCameras:       # This option greatly decreases the conversion run time
        args += ' --simple-cameras' # - Camera conversion could be local but image conversion still takes time.

    baseName = run.shortName() # SITE + YYMMDD = 8 chars, leaves seven for frame digits.

    # Get the location to store the logs    
    pbsLogFolder = run.getPbsLogFolder()
    os.system('mkdir -p ' + pbsLogFolder)

    # Submit all the jobs
    currentFrame = options.startFrame
    jobList = []
    for i in range(0, numCamgenJobs):
        jobName    = ('%s%06d%s' % ('C', currentFrame, baseName) ) # C for camera
        startFrame = currentFrame
        stopFrame  = currentFrame+tasksPerJob-1
        if (i == numCamgenJobs - 1):
            stopFrame = options.stopFrame # Make sure nothing is lost at the end

        thisArgs = (args + ' --start-frame ' + str(startFrame)
                         + ' --stop-frame  ' + str(stopFrame ) )
        if i != 0: # Only the first job will convert lidar files
            thisArgs += ' --no-lidar-convert'

        logPrefix = os.path.join(pbsLogFolder, 'convert_' + jobName)
        logger.info('Submitting camera generation job: ' + scriptPath + ' ' + thisArgs)
        pbs_functions.submitJob(jobName, CAMGEN_PBS_QUEUE, maxHours, logger,
                                options.minutesInDevelQueue,
                                GROUP_ID,
                                options.nodeType, '/usr/bin/python2.7',
                                scriptPath + " " + thisArgs, logPrefix)
        jobList.append(jobName)
        currentFrame += tasksPerJob

    # Wait for conversions to finish
    pbs_functions.waitForJobCompletion(jobList, logger)

    if options.parallelValidate:

        # Consolidate various validation files done in parallel, to make the subsequent
        # step faster.
        validFilesList = icebridge_common.validFilesList(outputFolder,
                                                         options.startFrame, options.stopFrame)
        validFilesSet = set()
        validFilesSet = icebridge_common.updateValidFilesListFromDisk(validFilesList, validFilesSet)
        allValidFiles = glob.glob(os.path.join(outputFolder,
                                               icebridge_common.validFilesPrefix() + '*'))
        for fileName in allValidFiles:
            validFilesSet = icebridge_common.updateValidFilesListFromDisk(fileName, validFilesSet)
        icebridge_common.writeValidFilesList(validFilesList, validFilesSet)

        # Now we will refetch and reprocess all files that were not
        # valid so far. Hopefully not too many. This must be on a head
        # node to be able to access the network. We don't do any orthoconvert
        # here as that one is too time-consuming
        logger.info("Refetch and process any invalid files.")
        pythonPath = asp_system_utils.which('python')
        cmd = (pythonPath + ' ' + icebridge_common.fullPath('full_processing_script.py') + ' --refetch --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --output-folder %s --stop-after-convert --no-ortho-convert --start-frame %d --stop-frame %d --num-threads %d --num-processes %d' % (options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, run.getFolder(), options.startFrame, options.stopFrame, numThreads, numProcesses))
        if options.noNavFetch:
            cmd += ' --no-nav'
        logger.info(cmd)
        os.system(cmd)

        # See if there are any new files for which we need to run ortho2pinhole on some node
        logger.info("See for which files to redo ortho.")
        (orthoList, numFrames) = icebridge_common.orthoListToRerun(validFilesSet, outputFolder)

        logger.info("Number of cameras to regenerate using ortho2pinhole: " + str(numFrames))
        if numFrames > 0:
            cmd = (icebridge_common.fullPath('full_processing_script.py') + ' --skip-fetch --skip-validate --skip-fast-conversions --no-nav --stop-after-convert --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --output-folder %s --start-frame %d --stop-frame %d --num-threads %d --num-processes %d --frames-file %s' % (options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, run.getFolder(), options.startFrame, options.stopFrame, numThreads, numProcesses, orthoList))
            #logger.info(cmd)
            jobList = []
            jobName    = ('%s%06d%s' % ('C', 0, baseName) ) # C for camera
            logPrefix = os.path.join(pbsLogFolder, 'convert_' + jobName)
            logger.info('Submitting camera generation job: ' + cmd)
            pbs_functions.submitJob(jobName, CAMGEN_PBS_QUEUE, maxHours, logger,
                                    options.minutesInDevelQueue,
                                    GROUP_ID,
                                    options.nodeType, '/usr/bin/python2.7',
                                    cmd, logPrefix)
            jobList.append(jobName)
            
            # Wait for conversions to finish
            pbs_functions.waitForRunCompletion(baseName, jobList, logger)

        logger.info("Finished refetching and reprocessing.")

    # Check the results
    # - If we didn't get everything keep going and process as much as we can.
    success = False
    try:
        success = run.conversionIsFinished(options.startFrame, options.stopFrame, verbose = True)
    except Exception, e:
        logger.warning('Caught error checking conversion status.\n' + str(e))

    if not success:
        #raise Exception('Failed to convert run ' + str(run))
        logger.warning('Could not fully convert run ' + str(run))

    run.setFlag('conversion_complete')

def generateBatchList(run, options, listPath, logger):
    '''Generate a list of all the processing batches required for a run'''

    logger.info("Generate batch list: " + listPath)
    
    refDemName = icebridge_common.getReferenceDemName(run.site)
    refDemPath = os.path.join(options.refDemFolder, refDemName)

    # Retrieve parallel processing parameters
    (numProcesses, numThreads, tasksPerJob, maxHours) = getParallelParams(options.nodeType, 'dem')

    # No actual processing is being done here so it can run on the PFE
    # - No start/stop frames here, always generate the full list so it is stable.
    pythonPath = asp_system_utils.which('python')
    scriptPath = icebridge_common.fullPath('full_processing_script.py')
    cmd       = ('%s %s --camera-calibration-folder %s --reference-dem-folder %s --site %s --yyyymmdd %s --skip-fetch --skip-convert --num-threads %d --num-processes %d --output-folder %s --bundle-length %d --log-batches ' 
                  % (pythonPath, scriptPath, options.inputCalFolder, options.refDemFolder, run.site, run.yyyymmdd, numThreads, numProcesses, run.getFolder(), options.bundleLength))

    # For full runs we must cleanup, as otherwise we'll run out of space
    if not options.skipCleanup:
        cmd += ' --cleanup'

    if options.noNavFetch:
        cmd += ' --no-nav'

# TODO: Find out what takes so long here!
# - Also fix the logging!

    logger.info(cmd)
    os.system(cmd)


def getOutputFolderFromBatchCommand(batchCommand):
    '''Extract the output folder from a line in the batch file'''
    # TODO: Integrate with the more robust logic in icebridge_common.getFrameRangeFromBatchFolder()
    # which can parse a batch command.
    parts        = batchCommand.split()
    outputFolder = parts[12] # This needs to be kept up to date with the file format!
    return outputFolder

# TODO: Share code with the other function
def filterBatchJobFile(run, batchListPath, logger):
    '''Make a copy of the batch list file which only contains incomplete batches.'''
    
{'fraction_non_alphanumeric': 0.1232023496050233, 'fraction_numerical': 0.012203767470123557, 'mean_word_length': 2.6496950656070966, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1771272', 'n_tokens_mistral': 11977, 'n_tokens_neox': 11387, 'n_words': 3767}