Fetch data from multiple sources. More...

Detailed Description

Fetch data from multiple sources.

This class knows how to fetch data from remote clusters, or the local machine. The data locations are specified by a several DataCatalog sections, each of which is given a priority, a valid set of dates and a file transfer mechanism. Data catalogs are tried in priority order. Files are obtained in multiple threads at once, and several file transfer mechanisms are understood:

file:// — obtain files on disk
ftp:// — contact an FTP server
sftp:// — contact a server over SSH. SSH-based rsync is used.
htar:// — use the proprietary htar program to get a tape archive

However, only one DataCatalog is examined at a time. All threads work on that one DataCatalog until all data that can be obtained from it is done. Then the threads exit, and new ones are spawned to examine the next DataCatalog.

For example, suppose you are on the Jet supercomputer running a HISTORY (retrospective) simulation. You set up this configuration section in your hwrf.conf config file:

 [jet_sources_prod2014]
 jet_hist_PROD2014%location  = file:///
 jet_hist_PROD2014%histprio=90
 jet_hist_PROD2014%fcstprio=90
 
 prod15_data_sp%location=htar://
 prod15_data_sp%histprio=59
 prod15_data_sp%dates=2015011218-2015123118
 
 [jet_hist_PROD2014]
 @inc=gfs2014_naming
 inputroot2014=/lfs3/projects/hwrf-data/hwrf-input
 gfs={inputroot2014}/HISTORY/GFS.{aYYYY}/{aYMDH}/
 gfs_sfcanl = gfs.t{aHH}z.sfcanl
 
 [prod15_data_sp]
 inputroot=/NCEPPROD/2year/hpssprod/runhistory/rh{aYYYY}/{aYYYY}{aMM}/{aYMD}
 gfs={inputroot}/
 gfs_sfcanl = {gfs_tar}#./gfs.t{aHH}z.sfcanl
 
 [hwrfdata]
 inputroot=/pan2/projects/hwrfv3/John.Doe/hwrfdata
 gfs={inputroot}/hwrf.{aYMDH}/
 gfs_sfcanl = gfs.t{aHH}z.sfcanl

and this is the code:

 is=InputSource(conf,"jet_sources_prod2014","2015071806")
 hwrfdata=DataCatalog(conf,"hwrfdata")
 is.get([
    {"dataset":"gfs", "item":"gfs_sfcanl","atime"="2015071800"},
    {"dataset":"gfs", "item":"gfs_sfcanl","atime"="2015071806"},
    {"dataset":"gfs", "item":"gfs_sfcanl","atime"="2015071812"} ],
    hwrfdata,realtime=False)

In this example, the InputSource will look for three GFS surface analysis files. It will search two possible locations for them: the on-disk Jet "PROD2014" history location and the NCO production tape files. The disk location will be searched first because its history priority is 90, while the tape area has a priority of 59.

Three files will show up eventually:

/pan2/projects/hwrfv3/John.Doe/hwrfdata/hwrf.2015071800/gfs.t00z.sfcanl
/pan2/projects/hwrfv3/John.Doe/hwrfdata/hwrf.2015071806/gfs.t06z.sfcanl
/pan2/projects/hwrfv3/John.Doe/hwrfdata/hwrf.2015071812/gfs.t12z.sfcanl

Each file will come from either here:

/lfs3/projects/hwrf-data/hwrf-input/HISTORY/GFS.2015071800/gfs.t00z.sfcanl
/lfs3/projects/hwrf-data/hwrf-input/HISTORY/GFS.2015071806/gfs.t06z.sfcanl
/lfs3/projects/hwrf-data/hwrf-input/HISTORY/GFS.2015071812/gfs.t12z.sfcanl

or here:

htar -xf /NCEPPROD/2year/hpssprod/runhistory/rh2015/201507/20150718/2015071800gfs.tar ./gfs.t00z.sfcanl
htar -xf /NCEPPROD/2year/hpssprod/runhistory/rh2015/201507/20150718/2015071806gfs.tar ./gfs.t06z.sfcanl
htar -xf /NCEPPROD/2year/hpssprod/runhistory/rh2015/201507/20150718/2015071812gfs.tar ./gfs.t12z.sfcanl

Definition at line 301 of file input.py.

Inheritance diagram for hwrf.input.InputSource:

Public Member Functions
def	__init__
	InputSource constructor. More...

def	add
	Adds a DataCatalog to this InputSource. More...

def	open_ftp
	Opens an FTP connection. More...

def	rsync_check_access
	Checks to see if rsync can even access a remote server. More...

def	fetch_file
	Internal implementation function that fetches one file. More...

def	list_for
	Returns the list of DataCatalog objects for FORECAST or HISTORY mode. More...

def	priotable (self, dclist)
	Generates a string containing a human-readable, prioritized list of data sources. More...

def	get
	Transfers the specified set of data to the specified target. More...

def	get_one (self, dataset, item, dest, logger=None, timeout=20, realtime=True, kwargs)
	This is a simple wrapper around fetch_file that gets only one file. More...

Public Attributes
	conf
	The hwrf.config.HWRFConfig object used for configuration info.

	section
	The section in conf that contains the data catalog list and relevant info.

	anltime
	The default analysis time. More...

	forecast
	List of forecast mode DataCatalog objects. More...

	history
	List of history mode DataCatalog objects. More...

	locks
	Lock objects to restrict access to FTP servers to one thread at a time. More...

	htar
	A produtil.prog.ImmutableRunner that runs htar. More...

	hsi
	A produtil.prog.ImmutableRunner that runs hsi. More...

	valid
	Data source validitiy information. More...

Constructor & Destructor Documentation

def hwrf.input.InputSource.__init__	(	self,
		conf,
		section,
		anltime,
		htar = `None`,
		logger = `None`,
		hsi = `None`
	)

InputSource constructor.

Parameters

conf	the hwrf.config.HWRFConfig to use for configuration info
section	the section that specifies the list of data catalogs
anltime	the default analysis time
htar	the produtil.prog.Runner that runs htar
logger	a logging.Logger for log messages
hsi	the produtil.prog.Runner that runs hsi

Definition at line 384 of file input.py.

Member Function Documentation

def hwrf.input.InputSource.add	(	self,
		dc,
		location,
		fcstprio = `None`,
		histprio = `None`,
		dates = `None`
	)

Adds a DataCatalog to this InputSource.

Called automatically from the constructor to add a DataCatalog to this InputSource. The list of add() calls is generated from the config section specified in the constructor. You should never need to call this function unless you want to explicitly add more DataCatalog objects that are not listed in the config files.

The location parameter is a URL from file, sftp, ftp or htar. Examples:

local files: file:///lfs3/projects/hwrf-data/hwrf-input/
scp: sftp://Some.User.nosp@m.name.nosp@m.@dtn-.nosp@m.zeus.nosp@m..rdhp.nosp@m.cs.n.nosp@m.oaa.g.nosp@m.ov/
ftp: ftp://anonymous@ftpprd.ncep.noaa.gov/
htar: htar:///NCEPPROD/1year/hpssprod/runhistory/rh2012/201204/20120418/

Warning: Bad things will happen if you add the same source twice. Bad things.

Note: If fcstprio and histprio are both None, this call has no effect.

Parameters

dc	the DataCatelog object
location	the URL of the data source, including the username if needed.
fcstprio	the priority for using this source in FORECAST (real-time) mode. If missing or None, the source will not be used in FORECAST mode.
histprio	the priority for using this source in HISTORY (retrospective) mode. If missing or None,the source will not be used in HISTORY mode.
dates	Dates for which this source is valid. This is passed to the trange argument of in_date_range(t,trange)

Definition at line 511 of file input.py.

Referenced by produtil.fileop.FileWaiter.add().

def hwrf.input.InputSource.fetch_file	(	self,
		streams,
		dc,
		dsurl,
		urlmore,
		dest,
		logger = `None`,
		timeout = `20`,
		realtime = `True`
	)

Internal implementation function that fetches one file.

You should not call this directly; it is meant to be called by "get" and re-implemented in subclasses. This grabs one file, potentially from a remote location. The URL for the base directory of some dataset is in dsurl, while the specific file is in urlmore. The urlmore will be appended to the file part of dsurl via urljoin, and the resulting file will be transferred.

Parameters

streams	a list used to store opened streams
dc	the DataCatalog being obtained
dsurl	the URL of the DataCatalog
urlmore	additional parts of the URL such as the reference or HTTP Get
dest	The local disk destination
logger	the logging.Logger for log messages
timeout	the connection timeout in seconds
realtime	True for FORECAST mode, False for HISTORY mode.

Returns: True if successful, False if not

Definition at line 611 of file input.py.

Referenced by hwrf.input.InputSource.get_one(), hwrf.input.InputSource.list_for(), and hwrf.input.InputSource.rsync_check_access().

def hwrf.input.InputSource.get	(	self,
		data,
		target_dc,
		realtime = `False`,
		logger = `None`,
		skip_existing = `True`
	)

Transfers the specified set of data to the specified target.

The "target_dc" is a DataCatalog that specifies the destination filenames. The "realtime" argument is True for FORECAST (real-time) mode runs, and False for HISTORY (retrospective) mode runs. The "data" argument should be an iterable (list, tuple, etc.) where each element is a dict-like object that describes one file to obtain. Each dict contains:

dataset - string name of the dataset (gfs, gdas1, gefs, enkf, etc.) item - string name of the object (ie.: sf, sfcanl, bufr) atime - Optional: a datetime.datetime specifying the analysis time. Default is the atime from the InputSource's constructor. ftime - Optional: a datetime.datetime specifying the forecast time. ...others... - any other keyword arguments will be sent to the .location functions in any of this InputSource's DataCatalog objects.

Definition at line 944 of file input.py.

Referenced by hwrf.wrfbase.WRFDomains.__contains__(), hwrf.wrfbase.WRFDomains.__getitem__(), hwrf.wrfbase.WRFDomains.add(), hwrf.wrf.WRFSimulation.analysis_name(), produtil.datastore.UpstreamFile.check(), hwrf.regrib.GRIB1Product.getgrib1grbindex(), hwrf.regrib.GRIB1Product.getgrib1grid(), hwrf.regrib.GRIB1Product.getgrib1index(), hwrf.regrib.GRIB2Product.getgrib2grid(), hwrf.regrib.GRIB2Product.getgrib2index(), and hwrf.input.InputSource.priotable().

def hwrf.input.InputSource.get_one	(	self,
		dataset,
		item,
		dest,
		logger = `None`,
		timeout = `20`,
		realtime = `True`,
		kwargs
	)

This is a simple wrapper around fetch_file that gets only one file.

It will fail if the file requires pulling an archive.

Parameters

dataset	the dataset to transfer
item	the desired item in the dataset
dest	the on-disk destination filename
logger	a logging.Logger for log messages
timeout	the connection timeout in seconds
realtime	True for FORECAST mode, False for HISTORY mode
kwargs	extra keyword arguments are passed to DataCatalog.locate()

Definition at line 1068 of file input.py.

Referenced by hwrf.input.InputSource.get().

def hwrf.input.InputSource.list_for	(	self,
		realtime = `True`
	)

Returns the list of DataCatalog objects for FORECAST or HISTORY mode.

Parameters

realtime True for FORECAST mode, False for HISTORY

Returns: self.forecast or self.history

Postcondition: _sort() has been called, sorting self.forecast and self.history in order of priority

Definition at line 787 of file input.py.

Referenced by hwrf.input.InputSource.get(), and hwrf.input.InputSource.get_one().

def hwrf.input.InputSource.open_ftp	(	self,
		netpart,
		logger = `None`,
		timeout = `20`
	)

Opens an FTP connection.

Opens the specified ftp://user@host/... request subject to the specified timeout, logging to the specified logger (if present and non-Null).

Parameters

netpart	The netpart portion of the URL
logger	the logging.Logger for log messages
timeout	the connection timeout in seconds

Definition at line 556 of file input.py.

Referenced by hwrf.input.InputSource.fetch_file().

def hwrf.input.InputSource.priotable	(	self,
		dclist
	)

Generates a string containing a human-readable, prioritized list of data sources.

Parameters

dclist The data source list from list_for()

Returns: A multi-line string containing the table.

Example: Prioritized list of data sources: PRIO- LOCATION = SOURCE @ DATES 100 - file:/// = DataCatalog(conf,'wcoss_fcst_PROD2014',2015080518) @ '1970010100-2038011818' 098 - file:/// = DataCatalog(conf,'wcoss_prepbufrnr_PROD2014',2015080518) @ '1970010100-2038011818' 097 - file:// = DataCatalog(conf,'zhan_gyre',2015080518) @ '2011060718-2011111200,2013051800-2013091018'

Definition at line 922 of file input.py.

Referenced by hwrf.input.InputSource.get().