"""!Post-processes GSI input and output for data assimilation diagnostics.""" ##@var __all__ # The list of symbols to export via "from hwrf.gsipost import *" __all__ = ["GSIPost",'F_ORG','F_GES','F_ANL','F_ALL'] import os, glob, shutil import produtil.fileop, produtil.run, produtil.cd import hwrf.gsi, hwrf.hwrftask, hwrf.post, hwrf.exceptions from produtil.run import run, checkrun, alias, mpi, mpirun, exe, bigexe from produtil.cd import NamedDir from produtil.fileop import deliver_file, make_symlink, remove_file from hwrf.post import EGRIB1Product, check_post, link_post_fix ##@var F_ORG # Flag to indiciate the original, downscaled parent model data is desired F_ORG=1 ##@var F_GES # Flag to indicate the vortex-relocated data (first guess) is desired F_GES=2 ##@var F_ANL # Flag to indiciate the final WRF input (after GSI and merge) is desired F_ANL=4 ##@var F_ALL # Flag to indicate all initialization steps are desired F_ALL=F_ORG|F_GES|F_ANL class GSIPost(hwrf.hwrftask.HWRFTask): """!Post-processes GSI inputs and outputs. Runs the hwrf_post on GSI inputs and outputs from one or more of three stages. * original data (F_ORG) - the original, downscaled, parent model data * first guess (F_GES) - the original data, with the vortex relocated * analysis (F_ANL) - the final data input to the forecast This class is compatible with the hwrf.regrib and hwrf.gribtask, and can be used in place of an hwrf.post.PostManyWRF. """ def __init__(self,dstore,conf,section,**kwargs): """!Constructor for GSIPost @param dstore the produtil.datastore.Datastore database to use @param conf the hwrf.config.HWRFConfig configuration information @param section the section to use in conf @param kwargs keyword arguments passed to the superclass constructor""" super(GSIPost,self).__init__(dstore,conf,section,**kwargs) self._domains=list() self._wrfout=dict() self._wrforg=dict() self._wrfges=dict() self._wrfanl=dict() self._org_products=dict() self._ges_products=dict() self._anl_products=dict() def add_case(self,domain,wrfout,org=None,ges=None,anl=None): """!Adds another case to the list of those to post-process @param domain a WRFDomain to be processed @param wrfout a wrfout file for internal use. It will be copied and its fields will e replaced with the ges and/or analysis file as needed @param org Optional: original down-scaled parent domain data @param ges Optional: a first guess wrfinput or restart file, generally after vortex relocation @param anl Optional: an analysis (GSI output) wrfinput or restart file @note At least one of org, ges or anl must be given.""" assert(not (org is None and ges is None and anl is None)) assert(wrfout is not None) assert(domain is not None) self._domains.append(domain) #self._domsize[domain]=[int(mdeglat),int(mdeglon),int(mdegres)] self._wrfout[domain]=wrfout if org is not None: self._wrforg[domain]=org self._org_products[domain]=EGRIB1Product( self.dstore,category=self.taskname, prodname=domain.name+'_org_'+org.prodname) if ges is not None: self._wrfges[domain]=ges self._ges_products[domain]=EGRIB1Product( self.dstore,category=self.taskname, prodname=domain.name+'_ges_'+ges.prodname) if anl is not None: self._wrfanl[domain]=anl self._anl_products[domain]=EGRIB1Product( self.dstore,category=self.taskname, prodname=domain.name+'_anl_'+anl.prodname) def run(self): """!Executes the GSI post for all cases on all domains.""" logger=self.log() good=True for domain in self._domains: logger.info('%s: process this domain'%(str(domain),)) wrfout=self._wrfout[domain] if domain in self._org_products: logger.info('%s: process org'%(str(domain),)) self._process(self._wrforg[domain], domain,'org',wrfout,self._org_products[domain]) else: logger.info('%s: has no org data'%(str(domain),)) if domain in self._ges_products: logger.info('%s: process ges'%(str(domain),)) self._process(self._wrfges[domain], domain,'ges',wrfout,self._ges_products[domain]) else: logger.info('%s: has no ges data'%(str(domain),)) if domain in self._anl_products: logger.info('%s: process anl'%(str(domain),)) self._process(self._wrfanl[domain], domain,'anl',wrfout,self._anl_products[domain]) else: logger.info('%s: has no anl data'%(str(domain),)) logger.info('Done processing domains.') def _process(self,inprod,domain,why,wrfout,outprod): """!Internal function that implements most of run() Do not call directly. This is the implementation of self.run: it runs the post for one case, on one domain.""" assert(inprod is not None) assert(outprod is not None) assert(wrfout is not None) assert(domain is not None) logger=self.log() #assert(outprod.location) # should already be set in constructor assert(wrfout.location) wrfthere=wrfout.location assert(wrfthere) if not produtil.fileop.isnonempty(wrfthere): raise hwrf.exceptions.PostHasNoInput( '%s: is empty or nonexistent'%(wrfthere,)) if not inprod.location: logger.info('%s: not available (location unknown)'%( inprod.did,)) elif not inprod.available: logger.info('%s (%s): not available according to database'%( inprod.did, inprod.location)) shortname=domain.name+'_'+why+'_'+inprod.prodname workdir=os.path.join(self.workdir,shortname) if os.path.exists(workdir): logger.info('%s: exists; will delete'%(workdir,)) shutil.rmtree(workdir) with NamedDir(workdir,keep=not self.scrub,logger=logger): diffwrf=alias(bigexe(self.getexe('hwrf_3dvar'))) post=alias(mpi(self.getexe('post'))) # Copy the fields into the wrfout file: deliver_file(wrfthere,'postinput',keep=True) make_symlink(inprod.location,'ghost',force=True) cmd=diffwrf['storm_relocate','ghost','flnm3','new_ght'] checkrun(cmd >= 'storm_relocate.log',logger=logger) cmd=diffwrf['3dvar_update','postinput','new_ght'] checkrun(cmd >= '3dvar_update.log',logger=logger) # Delete any stray fort.* files: for filename in glob.glob('fort.*'): produtil.fileop.remove_file(filename,logger=logger) # Run the post: datestamp=self.conf.cycle.strftime('%Y-%m-%d_%H:%M:%S') with open('itag','wt') as itag: itag.write("""postinput netcdf %s NMM NEST """ % (datestamp,)) cmd=mpirun(post,allranks=True) >= 'vpost.log' needcrtm=self.confbool('needcrtm',False) logger.info('link post fix files') link_post_fix(self.getdir('FIXhwrf'),needcrtm,logger=logger) fort14=self.confstr('control') logger.info('%s: use this control file for gsi post'%(fort14,)) produtil.fileop.make_symlink(fort14,'fort.14',force=True, logger=logger) logger.info('Run post, log to vpost.log.') ret=run(cmd,logger=logger) # Check to see if the post succeeded and get the center # lat & lon of the domain for post-processing: (ok,cenla,cenlo,filename)=check_post(ret,shortname,logger) if not ok: raise hwrf.exceptions.PostFailed('GSI post on '+shortname) #toloc=outprod.location toloc=os.path.join(self.outdir,shortname) outprod.deliver(toloc,{'CENLA':cenla,'CENLO':cenlo, 'fromloc':filename},logger=logger) return True def products(self,domains=None,domain=None,which_step=F_ALL,**kwargs): """!Iterates over EGRIB1Product objects produced by this task. @param domains a list of WRFDomain objects. Only these will be iterated. @param domain a single WRFDomain; only its products will be iterated @param which_step which steps are of interest: * F_ORG - original parent model data * F_GES - first guess (vortex-relocated parent model data) * F_ANL - analysis, the final input to the forecast * any integer or of the above * F_ALL - all products, the default @param kwargs ignored""" if domains is None and domain is not None: domains=[domain] elif domains is None: domains=self._domains for d in domains: if which_step&F_ORG and d in self._org_products: yield self._org_products[d] if which_step&F_ANL and d in self._anl_products: yield self._anl_products[d] if which_step&F_GES and d in self._ges_products: yield self._ges_products[d]