Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def find_trigger_urls(channel, etg, gpsstart, gpsend, verbose=False):
"""Find the paths of trigger files that represent the given
observatory, channel, and ETG (event trigger generator) for a given
GPS [start, end) segment.
"""
if etg.lower().startswith('omicron'):
etg = '?' + etg[1:]
# construct search
gpsstart = to_gps(gpsstart).seconds
gpsend = to_gps(gpsend).seconds
span = Segment(gpsstart, gpsend)
ifo, channel = channel.split(':', 1)
trigtype = "%s_%s" % (channel, etg.lower())
epoch = '*'
searchbase = os.path.join(TRIGFIND_BASE_PATH, epoch, ifo, trigtype)
gpsdirs = range(int(str(gpsstart)[:5]), int(str(gpsend)[:5])+1)
trigform = ('%s-%s_%s-%s-*.xml*'
% (ifo, re_dash.sub('_', channel), etg.lower(), '[0-9]'*10))
# test for channel-level directory
if not glob.glob(searchbase):
raise ValueError("No channel-level directory found at %s. Either the "
"channel name or ETG names are wrong, or this "
"channel is not configured for this ETG."
% searchbase)
def find_latest(observatory, frametype, gpstime=None, allow_tape=False,
connection=None, **connection_kw):
"""Find the path of the latest file of a given data type.
See also
--------
gwdatafind.http.HTTPConnection.find_latest
FflConnection.find_latest
for details on the underlying method(s)
"""
observatory = observatory[0]
try:
if gpstime is not None:
gpstime = int(to_gps(gpstime))
path = find_urls(observatory, frametype, gpstime, gpstime+1,
on_gaps='ignore', connection=connection)[-1]
else:
path = connection.find_latest(observatory, frametype,
on_missing='ignore')[-1]
except (IndexError, RuntimeError):
raise RuntimeError(
"no files found for {}-{}".format(observatory, frametype))
path = urlparse(path).path
if not allow_tape and on_tape(path):
raise IOError("Latest frame file for {}-{} is on tape "
"(pass allow_tape=True to force): "
"{}".format(observatory, frametype, path))
return path
def _auto_epoch(self, axis):
# use the lower data/view limit as the epoch
epoch = round(self._lim(axis)[0])
# round epoch in successive units for large scales
unit = self.get_unit()
date = from_gps(epoch)
fields = ('second', 'minute', 'hour', 'day')
for i, u in enumerate(fields[1:]):
if unit < units.Unit(u):
break
if u in ('day',):
date = date.replace(**{fields[i]: 1})
else:
date = date.replace(**{fields[i]: 0})
return int(to_gps(date))
>>> print(DataQualityFlag.fetch_open_data('H1_DATA', 'Jan 1 2010',
... 'Jan 2 2010'))
"""
start = to_gps(start).gpsSeconds
end = to_gps(end).gpsSeconds
known = [(start, end)]
active = timeline.get_segments(flag, start, end, **kwargs)
return cls(flag.replace('_', ':', 1), known=known, active=active,
label=flag)
# otherwise unpack two arguments as a segment
if len(args) == 1:
args = args[0]
# if not two arguments, panic
try:
start, end = args
except ValueError as exc:
exc.args = ('{0}() takes 2 arguments for start and end GPS time, '
'or 1 argument containing a Segment or SegmentList'.format(
func.__name__),)
raise
# return list with one Segment
return SegmentList([Segment(to_gps(start), to_gps(end))])
pad : `float`, optional
value with which to fill gaps in the source data, if not
given gaps will result in an exception being raised
Returns
-------
dict : :class:`~gwpy.timeseries.TimeSeriesDict`
dict of (channel, `TimeSeries`) data pairs
"""
# import the frame library here to have any ImportErrors occur early
import_gwf_library(library)
# -- from here read data
if start:
start = to_gps(start)
if end:
end = to_gps(end)
# parse output format kwargs -- DEPRECATED
if resample is not None:
warnings.warn('the resample keyword for is deprecated, instead '
'you should manually resample after reading',
DeprecationWarning)
if not isinstance(resample, dict):
resample = dict((c, resample) for c in channels)
if dtype is not None:
warnings.warn('the dtype keyword for is deprecated, instead '
'you should manually call astype() after reading',
DeprecationWarning)
if not isinstance(dtype, dict):
dtype = dict((c, dtype) for c in channels)
given gaps will result in an exception being raised
Returns
-------
dict : :class:`~gwpy.timeseries.TimeSeriesDict`
dict of (channel, `TimeSeries`) data pairs
"""
# import the frame library here to have any ImportErrors occur early
import_gwf_library(library)
# -- from here read data
if start:
start = to_gps(start)
if end:
end = to_gps(end)
# parse output format kwargs -- DEPRECATED
if resample is not None:
warnings.warn('the resample keyword for is deprecated, instead '
'you should manually resample after reading',
DeprecationWarning)
if not isinstance(resample, dict):
resample = dict((c, resample) for c in channels)
if dtype is not None:
warnings.warn('the dtype keyword for is deprecated, instead '
'you should manually call astype() after reading',
DeprecationWarning)
if not isinstance(dtype, dict):
dtype = dict((c, dtype) for c in channels)
# format gap handling
def arg_xaxis(cls, parser):
"""Add an `~argparse.ArgumentGroup` for X-axis options.
This method includes the standard X-axis options, as well as a new
``--epoch`` option for the time axis.
"""
group = super(TimeDomainProduct, cls).arg_xaxis(parser)
group.add_argument('--epoch', type=to_gps,
help='center X axis on this GPS time, may be'
'absolute date/time or delta')
return group