gen_single_ob.py

Go to the documentation of this file.

#!/usr/bin/env python3
#
# (C) Copyright 2020 NOAA NWS NCEP EMC
#
# This software is licensed under the terms of the Apache Licence Version 2.0
# which can be obtained at http://www.apache.org/licenses/LICENSE-2.0.
#
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
import numpy as np
from pathlib import Path
from datetime import datetime
import yaml
import sys
 
IODA_CONV_PATH = Path(__file__).parent/"@SCRIPT_LIB_PATH@"
if not IODA_CONV_PATH.is_dir():
    IODA_CONV_PATH = Path(__file__).parent/'..'/'lib-python'
sys.path.append(str(IODA_CONV_PATH.resolve()))
 
import meteo_utils
import ioda_conv_ncio as iconv
from orddicts import DefaultOrderedDict
 
 
class singleob(object):
    loctype = {
        float: 'float',
        str: 'string',
        int: 'integer',
    }
 
    def __init__(self, yamlfile):
        # read a YAML file and generate a single observation file
        # using the configuration in the YAML file
        # read the YAML config into a dictionary
        with open(yamlfile, 'r') as stream:
            yamlconfig = yaml.safe_load(stream)
        self.filenamefilename = yamlconfig['obsdataout']
        self.locKeyListlocKeyList = []
        self.AttrDataAttrData = {
            'date_time_string': yamlconfig['lockeys']['datetime']
        }
        locKeys = []
        # set up location keys based on YAML
        for key in yamlconfig['lockeys'].keys():
            value = yamlconfig['lockeys'][key]
            self.locKeyListlocKeyList.append((key, self.loctypeloctype[type(value)]))
            locKeys.append(value)
        self.keyDictkeyDict = DefaultOrderedDict(lambda: DefaultOrderedDict(dict))
 
        self.writerwriter = iconv.NcWriter(self.filenamefilename, self.locKeyListlocKeyList)
 
        # set up variable name
        key = 'oneob'
        value = yamlconfig['variable']['name']
        self.keyDictkeyDict[key]['valKey'] = value, self.writerwriter.OvalName()
        self.keyDictkeyDict[key]['errKey'] = value, self.writerwriter.OerrName()
        self.keyDictkeyDict[key]['qcKey'] = value, self.writerwriter.OqcName()
        # set up the data
        self.datadata = DefaultOrderedDict(lambda: DefaultOrderedDict(dict))
        locKey = tuple(locKeys)
        self.datadata[0][locKey][self.keyDictkeyDict[key]['valKey']] = float(yamlconfig['variable']['obsvalue'])
        self.datadata[0][locKey][self.keyDictkeyDict[key]['errKey']] = float(yamlconfig['variable']['obserr'])
        self.datadata[0][locKey][self.keyDictkeyDict[key]['qcKey']] = int(yamlconfig['variable']['preqc'])
 
        # call the IODA API and write the file
        (ObsVars, LocMdata, VarMdata) = self.writerwriter.ExtractObsData(self.datadata)
        self.writerwriter.BuildNetcdf(ObsVars, LocMdata, VarMdata, self.AttrDataAttrData)
        return
 
 
def main():
 
    desc = 'Generate single observation IODA file from YAML input'
    parser = ArgumentParser(
        description=desc,
        formatter_class=ArgumentDefaultsHelpFormatter)
    parser.add_argument(
        '-y', '--yaml', help='path to input YAML file', type=str, required=True)
 
    args = parser.parse_args()
 
    yamlfile = args.yaml
 
    singleob(yamlfile)
 
 
if __name__ == '__main__':
    main()