In [1]:

    

import subprocess
import os
import sys
from dask_jobqueue import PBSCluster
from distributed import Client, progress
from datetime import datetime, timedelta
from pkg_resources import load_entry_point
from distributed import progress


    

In [2]:

    

def exec_adi(info_dict):
    """
    This function will call adi_cmac2 from within Python. It takes in a dictionary where the inputs to adi_cmac2 are
    stored.
    
    Parameters
    ----------
    info_dict: dict
        A dictionary with the following keywords:
        'facility' = The facility marker (i.e. 'sgp', 'nsa', etc.)
        'site' = The site marker (i.e. i4, i5, i6)
        'start_date' = The start date as a string formatted YYYYMMDD
        'end_date' = The end date as a string formatted YYYYMMDD
    """
    facility = info_dict['facility']
    site = info_dict['site']
    start_date = info_dict['start_date']
    end_date = info_dict['end_date']
    
    # Change this directory to where you want your adi logs stored
    logs_dir = "/home/rjackson/adi_logs"
    
    # Set the path to your datasteam here! 
    os.environ["DATASTREAM_DATA"] = "/lustre/or-hydra/cades-arm/rjackson/"
    
    logs_dir += logs_dir + "/" + site + start_date + "_" + end_date
    if not os.path.isdir(logs_dir):
        os.makedirs(logs_dir)
    os.environ["LOGS_DATA"] = logs_dir
    os.environ["PROJ_LIB"] = "/home/rjackson/anaconda3/envs/adi_env3/share/proj/"
    
    # Set the path to the clutter file here!
    os.environ["CMAC_CLUTTER_FILE"] = "/home/rjackson/cmac2.0/scripts/clutter201901.nc"
    subprocess.call(("/home/rjackson/anaconda3/envs/adi_env3/bin/adi_cmac2 -D 1 -f " + 
                     facility + " -s " + site + " -b " + start_date + " -e "+  end_date), shell=True)


    

In [3]:

    

the_cluster = PBSCluster(processes=6, cores=36, queue="arm_high_mem", 
                         walltime="3:00:00", resource_spec="qos=std",
                         job_extra=["-A arm", "-W group_list=cades-arm"],
                         env_extra=[". /home/rjackson/anaconda3/etc/profile.d/conda.sh", "conda activate adi_env3"])
the_cluster.scale(36)


    

In [4]:

    

client = Client(the_cluster)
client


    

    
Out[4]:







Client

  
Scheduler: tcp://10.23.216.81:41764
  
Dashboard: http://10.23.216.81:8787/status



Cluster

  
Workers: 0
  
Cores: 0
  
Memory: 0 B

In [5]:

    

client


    

    
Out[5]:







Client

  
Scheduler: tcp://10.23.216.81:41764
  
Dashboard: http://10.23.216.81:8787/status



Cluster

  
Workers: 0
  
Cores: 0
  
Memory: 0 B

In [6]:

    

def make_date_list_dict_list(start_day, end_day):
    """
    This automatically generates a list of day inputs for the exec_adi function.
    
    Parameters
    ----------
    start_day: datetime
        The start date
    end_day:
        The end date
    Returns
    -------
    the_list: A list of dictionary inputs for exec_adi
    """
    cur_day = start_day
    the_list = []
    while(cur_day < end_day):
        next_day = cur_day + timedelta(days=1)
        temp_dict = {}
        # Change these next two lines to fit your facility
        temp_dict['facility'] = "I5"
        temp_dict['site'] = "sgp"
        temp_dict['start_date'] = cur_day.strftime("%Y%m%d")
        temp_dict['end_date'] = next_day.strftime("%Y%m%d")
        the_list.append(temp_dict)
        cur_day = cur_day + timedelta(days=1)
    return the_list

# Here we specify the dates that we want to process
date_list = make_date_list_dict_list(datetime(2019, 1, 1), datetime(2019,2,6))


    

In [7]:

    

# Run the cluster
futures = client.map(exec_adi, date_list)


    

In [8]:

    

# Put up a little progress bar!
progress(futures)


    

In [94]:

    

# This will make the tasks quit
del futures
cluster.stop_all_jobs()