from datetime import datetime, timedelta
from itertools import chain
_epoch = datetime(1970, 1, 1, 0, 0, 0)
def _to_list(iter, tolist):
if tolist:
return list(iter)
else:
return iter
[docs]class Temporal(object):
"""
Keep track of time in an experiment and convert times between seconds since experiment start, datetime objects and
seconds since the epoch.
"""
def __init__(self, base_time, t_ens_start, t_ens_end, t_ens_step):
"""
Constructor for the Temporal object
Args:
base_time (datetime): A datetime object specifying the reference time of the experiment.
t_ens_start (int): An integer specifying the start time for the experiment.
t_ens_end (int): An integer specifying the end time for the experiment.
t_ens_step (int): An integer specifying the time between history dumps for the experiment.
"""
self._base_time = base_time
self._t_ens_start = t_ens_start
self._t_ens_end = t_ens_end
self._t_ens_step = t_ens_step
return
[docs] def __iter__(self):
"""
Iterate over the experiment times.
Yields:
int: The next time in seconds since the experiment reference time.
Examples:
>>> dt = datetime(2011, 5, 24, 18)
>>> temp = Temporal(dt, 0, 1800, 300)
>>> [ t for t in temp ]
[ 0, 300, 600, 900, 1200, 1500, 1800 ]
"""
for t_ens in xrange(self._t_ens_start, self._t_ens_end + self._t_ens_step, self._t_ens_step):
yield t_ens
[docs] def __len__(self):
"""
Get the number of times in the experiment.
Returns:
int: The number of times in the experiment.
Examples:
>>> dt = datetime(2011, 5, 24, 18)
>>> temp = Temporal(dt, 0, 1800, 300)
>>> len(temp)
7
"""
return len(xrange(self._t_ens_start, self._t_ens_end + self._t_ens_step, self._t_ens_step))
[docs] def __getitem__(self, index):
"""
Get the time at a particular index.
Args:
index: The index at which to return the time.
Returns:
If `index` is an integer, return the number of seconds at `index` since the experiment reference time. If
`index` is a slice, return a new Temporal object with subsetted according to the slice.
Examples:
>>> dt = datetime(2011, 5, 24, 18)
>>> temp = Temporal(dt, 0, 1800, 300)
>>> temp[3]
900
"""
times = list(self)
if type(index) == slice:
subset_times = times[index]
if len(subset_times) == 1:
ret_val = Temporal(self._base_time, subset_times[0], subset_times[-1], times[1] - times[0])
else:
ret_val = Temporal(self._base_time, subset_times[0], subset_times[-1], subset_times[1] - subset_times[0])
else:
ret_val = times[index]
return ret_val
[docs] def get_times(self):
"""
Get the list of times represented by the Temporal object.
Returns:
A list of times in seconds since the experiment reference time.
"""
return list(self)
[docs] def get_datetimes(self, aslist=False):
"""
Get a list of datetimes represented by the Temporal object.
Args:
aslist (bool): Whether to return a list (True) or a Python generator (False). They can be used mostly
interchangeably, but a generator is more efficient. Default is False (return a generator).
Returns:
A Python generator or list of datetime objects.
"""
dts = (self._base_time + timedelta(seconds=float(t_ens)) for t_ens in self)
return _to_list(dts, aslist)
[docs] def get_epochs(self, aslist=False):
"""
Get a list of times represented by the Temporal object.
Args:
aslist (bool): Whether to return a list (True) or a Python generator (False). They can be used mostly
interchangeably, but a generator is more efficient. Default is False (return a generator).
Returns:
A Python generator or list of times in seconds since the epoch.
"""
epochs = (self.sec_to_epoch(t_ens) for t_ens in self)
return _to_list(epochs, aslist)
[docs] def get_strings(self, str_format, aslist=False):
"""
Get a list of times as formatted strings.
Args:
str_format (str): The string format to use.
aslist (bool): Whether to return a list (True) or a Python generator (False). They can be used mostly
interchangeably, but a generator is more efficient. Default is False (return a generator).
Returns:
A Python generator or list of strings containing the formatted times.
"""
strings = (dt.strftime(str_format) for dt in self.get_datetimes())
return _to_list(strings, aslist)
[docs] def sec_to_epoch(self, exp_seconds):
"""
Convert a time in seconds since experiment reference time to a time in seconds since the epoch.
Args:
exp_seconds (int): The number of seconds since the experiment reference time.
Returns:
int: A number of seconds since the epoch.
"""
base_epoch = dt_to_epoch(self._base_time)
return base_epoch + exp_seconds
[docs] def sec_to_datetime(self, exp_seconds):
"""
Convert a number of seconds since the experiment reference time to a datetime object.
Args:
exp_seconds (int): The number of seconds since the experiment reference time.
Returns:
datetime: A datetime object representing that time.
"""
return self._base_time + timedelta(seconds=float(exp_seconds))
[docs]class PatchedTemporal(object):
"""
Keep track of time in an experiment and do time conversions, but for timelines in which the time step is not
constant.
"""
def __init__(self, *args):
"""
Constructor for the PatchedTemporal object.
Args:
*args (Temporal): A variable length argument list. Each argument must be a Temporal object.
"""
if not all(isinstance(a, Temporal) for a in args):
raise TypeError("All arguments to PatchedTemporal must be of type Temporal.")
self._temporals = [ args[0] ] + [ a[1:] for a in args[1:] ] # Chop out the first element of each Temporal
self._base_time = self._temporals[0]._base_time
return
[docs] def __iter__(self):
"""
Iterate over the experiment times.
Yields:
int: The next time in seconds since the experiment reference time.
Examples:
>>> temp1 = Temporal(0, 1200, 300)
>>> temp2 = Temporal(1200, 1800, 150)
>>> temp = PatchedTemporal(temp1, temp2)
>>> [ t for t in temp ]
[ 0, 300, 600, 900, 1200, 1350, 1500, 1650, 1800 ]
"""
return chain(*self._temporals)
[docs] def __len__(self):
"""
Get the number of times in the experiment.
Returns:
int: The number of times in the experiment.
Examples:
>>> temp1 = Temporal(0, 1200, 300)
>>> temp2 = Temporal(1200, 1800, 150)
>>> temp = PatchedTemporal(temp1, temp2)
>>> len(temp)
9
"""
return sum( len(t) for t in self._temporals )
[docs] def __getitem__(self, index):
"""
Get the time at a particular index.
Args:
index (int): The index at which to return the time.
Returns:
int: The number of seconds at that index since the experiment reference time.
Examples:
>>> temp1 = Temporal(0, 1200, 300)
>>> temp2 = Temporal(1200, 1800, 150)
>>> temp = PatchedTemporal(temp1, temp2)
>>> temp[6]
1500
"""
return list(self)[index]
[docs] def get_times(self):
"""
Get the list of times represented by the Temporal object.
Returns:
A list of times in seconds since the experiment reference time.
"""
return list(self)
[docs] def get_datetimes(self, aslist=False):
"""
Get a list of datetimes represented by the Temporal object.
Args:
aslist (bool): Whether to return a list (True) or a Python generator (False). They can be used mostly
interchangeably, but a generator is more efficient. Default is False (return a generator).
Returns:
A Python generator or list of datetime objects.
"""
dts = chain(*[temp.get_datetimes() for temp in self._temporals])
return _to_list(dts, aslist)
[docs] def get_epochs(self, aslist=False):
"""
Get a list of times represented by the Temporal object.
Args:
aslist (bool): Whether to return a list (True) or a Python generator (False). They can be used mostly
interchangeably, but a generator is more efficient. Default is False (return a generator).
Returns:
A Python generator or list of times in seconds since the epoch.
"""
epochs = chain(*[temp.get_epochs() for temp in self._temporals])
return _to_list(epochs, aslist)
[docs] def get_strings(self, format, aslist=False):
"""
Get a list of times as formatted strings.
Args:
format (str): The string format to use.
aslist (bool): Whether to return a list (True) or a Python generator (False). They can be used mostly
interchangeably, but a generator is more efficient. Default is False (return a generator).
Returns:
A Python generator or list of strings containing the formatted times.
"""
strings = chain(*[temp.get_strings(format) for temp in self._temporals])
return _to_list(strings, aslist)
[docs] def sec_to_epoch(self, exp_seconds):
"""
Convert a time in seconds since experiment reference time to a time in seconds since the epoch.
Args:
exp_seconds (int): The number of seconds since the experiment reference time.
Returns:
int: A number of seconds since the epoch.
"""
base_epoch = dt_to_epoch(self._base_time)
return base_epoch + exp_seconds
[docs] def sec_to_dt(self, exp_seconds):
"""
Convert a number of seconds since the experiment reference time to a datetime object.
Args:
exp_seconds (int): The number of seconds since the experiment reference time.
Returns:
datetime: A datetime object representing that time.
"""
return self._base_time + timedelta(seconds=float(exp_seconds))
[docs]def temporal(base_time, *slices):
"""
Create an object to keep track of time in an experiment and do time-related conversions.
Args:
base_time (datetime): The base time in the experiment (i.e. inittime in the ARPS input file) as a datetime
object.
*slices (slice): One or more slice objects, each specifying start and end times (in seconds since the base time)
and the time step length (in seconds) for a segment. Multiple segments are chained together under the hood
to form a continuous timeline.
Returns:
A Temporal or PatchedTemporal object (both have the same interface).
"""
temp_objs = [Temporal(base_time, s.start, s.stop, s.step) for s in slices]
if len(temp_objs) == 0:
raise ValueError("temporal() needs one or more slices.")
elif len(temp_objs) == 1:
temp = temp_objs[0]
else:
temp = PatchedTemporal(*temp_objs)
return temp
[docs]def dt_to_epoch(dt):
"""
Convert a Python datetime object to a number of seconds since the epoch (00 UTC 1 January 1970).
Args:
dt (datetime): A Python datetime object
Returns:
Time since the epoch in seconds.
"""
return (dt - _epoch).total_seconds()
[docs]def epoch_to_dt(epoch):
"""
Convert a number of seconds since the epoch (00 UTC 1 January 1970) to a Python datetime object.
Args:
epoch (float): Time since the epoch in seconds
Returns:
A Python datetime object.
"""
return _epoch + timedelta(seconds=float(epoch))
if __name__ == "__main__":
# This returns a temporal object with two segments.
temp_test = temporal(datetime(2011, 5, 24, 18, 0, 0), slice(0, 1800, 300), slice(1800, 3600, 150))
print "Temporal testing ..."
print type(temp_test)
print list(temp_test) # For when you need a list, not an iterator
print temp_test.get_datetimes(aslist=True) # Get a list of datetimes
print temp_test.get_epochs(aslist=True) # Get a list of epochs
for t_ens in temp_test.get_strings("%H%M"):
print t_ens
print "Done!"