Harmony TEMPO Subset

get_temposubset.py

#!/usr/bin/python
__doc__ = """
Relies on NASA credentials in .netrc and tested with
harmony-py==1.3.4
pandas==2.3.1
"""
import logging
from harmony import BBox, Client, Collection, Request, Environment
import os
import pandas as pd

# atmos has an ENVIRONMENT variable, and it breaks harmony
if 'ENVIRONMENT' in os.environ:
    del os.environ['ENVIRONMENT']

# Standard RSIG Inputs
bbox = (-180, -90, 180, 90)  # Phoenix Area
_concept_ids = {
    'no2': 'C3685896872-LARC_CLOUD',  # TEMPO_L2_NO2_V04
    'hcho': 'C3685912035-LARC_CLOUD'  # TEMPO_L2_HCHO_V04
}
_no2vlist = [
    'geolocation/time',
    'geolocation/longitude',
    'geolocation/latitude',
    'geolocation/solar_zenith_angle',
    'support_data/eff_cloud_fraction',
    'support_data/fitted_slant_column',
    'support_data/amf_troposphere',
    'support_data/amf_stratosphere',
    'product/main_data_quality_flag',
    'product/vertical_column_troposphere',
    'product/vertical_column_stratosphere',
]
_hchovlist = [
    'geolocation/time',
    'geolocation/longitude',
    'geolocation/latitude',
    'geolocation/solar_zenith_angle',
    'support_data/eff_cloud_fraction',
    'support_data/fitted_slant_column',
    'support_data/amf',
    'product/main_data_quality_flag',
    'product/vertical_column',
]
_vlists = {
    'no2': _no2vlist,
    'hcho': _hchovlist
}
harmony_client = Client(env=Environment.PROD)


def getdate(spc, start):
    """
    Download all granules for start date 00-23:59:59.
    Creates a log. If there are noo lines that start ERROR
    and the log ends with INFO:...:Complete, then it was
    successfully applied.

    Arguments
    ---------
    start : date-like
        Processed by pandas.to_datetime as a date

    Returns
    -------
    None
    """
    concept_id = _concept_ids[spc]
    vlist = _vlists[spc]
    start = pd.to_datetime(start)
    dt = pd.to_timedelta('86399s')
    end = (start + dt).to_pydatetime()
    start = start.to_pydatetime()
    odir = f'data/{start:%Y/%m}'
    os.makedirs(odir, exist_ok=True)
    spcu = spc.upper()

    # Check for existing log and successful completion
    # if successful complete and no errors, do not reprocess.
    logpath = f'{odir}/harmony_{spc}_{start:%Y-%m-%d}.log'
    if os.path.exists(logpath):
        with open(logpath, 'r') as logf:
            loglines = logf.read().strip().split('\n')
        errs = [logl for logl in loglines if logl.startswith('ERROR')]
        if len(errs) == 0:
            if loglines[-1].endswith(':Complete'):
                print(f'Skip {start:%Y-%m-%d}: successfully completed')
                return

    logger = logging.getLogger(start.strftime('%Y-%m-%d'))
    logger.setLevel(logging.INFO)
    hndlr = logging.FileHandler(logpath)
    hndlr.setFormatter(logging.Formatter('%(levelname)s:%(name)s:%(message)s'))
    logger.addHandler(hndlr)

    logger.info(f'Started {start:%Y-%m-%dT%H%M%S} {end:%Y-%m-%dT%H%M%S}')
    request = Request(
        collection=Collection(id=concept_id),
        granule_name=f'TEMPO_{spcu}_L2_V04_{start:%Y%m%d}T*.nc',
        spatial=BBox(*bbox),
        # temporal={'start': start, 'stop': end},
        variables=vlist
    )
    assert request.is_valid()

    # submit the request and make note of the job id
    jid = harmony_client.submit(request)
    logger.info(f'submitted {jid}')
    # Attempt to download results to the output directory
    try:
        results = harmony_client.download_all(
            jid, directory=odir, overwrite=False
        )
    except Exception as e:
        logger.error('download_all failed - ' + str(e))
        return  # try the next time

    # if successful, rename outputs to omit harmony numeric id
    paths = [r.result() for r in results]
    logger.info(f'downloaded_all got {paths}')
    for ri, rawpath in enumerate(list(paths)):
        rid = os.path.split(rawpath)[-1].split('_')[0]
        cleanpath = rawpath.replace(f'{rid}_', '')
        cleanpath = cleanpath.replace('_subsetted.nc', '.nc')
        try:
            os.rename(rawpath, cleanpath)
            logger.info(f'rename {rawpath} to {cleanpath}')
            paths[ri] = cleanpath
        except Exception as e:
            logger.error(f'rename failed for {rawpath} ' + str(e))
    logger.info('Complete')


if __name__ == '__main__':
    import argparse
    now = pd.to_datetime('now').floor('1d')
    defend = (now - pd.to_timedelta('1d')).strftime('%Y-%m-%d')
    defstart = (now - pd.to_timedelta('8d')).strftime('%Y-%m-%d')
    spcopt = {'no2', 'hcho'}
    desc = """Caches TEMPO V04 granules for NO2 or HCHO in data/%Y/%m
directories. Each day is processed and creates its own log at
data/%Y/%m/harmony_{spc}_{date}.log. If the log exists and has no ERROR lines,
then the day will not be redownloaded. Otherwise, the day will be reprocessed.
At this time, artifacts of an uncomplete run are not automatically cleaned up.
Failed runs should be cleaned up by the user.
"""
    prsr = argparse.ArgumentParser(description=desc)
    helpstr = 'default no2'
    prsr.add_argument('--spc', default='no2', choices=spcopt, help=helpstr)
    helpstr = f'YYYY-MM-DD (default 8 days ago; {defstart})'
    prsr.add_argument('start', nargs='?', default=defstart, help=helpstr)
    helpstr = f'YYYY-MM-DD (default yesterday; {defend})'
    prsr.add_argument('end', nargs='?', default=defend, help=helpstr)
    args = vars(prsr.parse_args())
    spc = args.pop('spc')
    dates = pd.date_range(**args, freq='1d')
    for start in dates:
        getdate(spc, start)