in curator/helpers/date_ops.py [0:0]
def date_range(unit, range_from, range_to, epoch=None, week_starts_on='sunday'):
"""
This function calculates a date range with a distinct epoch time stamp of both the
start time and the end time in counts of ``unit`` relative to the time at
execution, if ``epoch`` is not set.
If ``unit`` is ``weeks``, you can also determine when a week begins using
``week_starts_on``, which can be either ``sunday`` or ``monday``.
:param unit: One of ``hours``, ``days``, ``weeks``, ``months``, or ``years``.
:param range_from: Count of ``unit`` in the past/future is the origin?
:param range_to: Count of ``unit`` in the past/future is the end point?
:param epoch: An epoch timestamp used to establish a point of reference for
calculations.
:param week_starts_on: Either ``sunday`` or ``monday``. Default is ``sunday``
:type unit: str
:type range_from: int
:type range_to: int
:type epoch: int
:type week_starts_on: str
:returns: The epoch start time and end time of a date range
:rtype: tuple
"""
logger = logging.getLogger(__name__)
start_date = None
start_delta = None
acceptable_units = ['hours', 'days', 'weeks', 'months', 'years']
if unit not in acceptable_units:
raise ConfigurationError(f'"unit" must be one of: {acceptable_units}')
if not range_to >= range_from:
raise ConfigurationError(
'"range_to" must be greater than or equal to "range_from"'
)
if not epoch:
epoch = time.time()
epoch = fix_epoch(epoch)
raw_point_of_ref = datetime.fromtimestamp(epoch, timezone.utc)
logger.debug('Raw point of Reference = %s', raw_point_of_ref)
# Reverse the polarity, because -1 as last week makes sense when read by
# humans, but datetime timedelta math makes -1 in the future.
origin = range_from * -1
# These if statements help get the start date or start_delta
if unit == 'hours':
point_of_ref = datetime(
raw_point_of_ref.year,
raw_point_of_ref.month,
raw_point_of_ref.day,
raw_point_of_ref.hour,
0,
0,
)
start_delta = timedelta(hours=origin)
if unit == 'days':
point_of_ref = datetime(
raw_point_of_ref.year, raw_point_of_ref.month, raw_point_of_ref.day, 0, 0, 0
)
start_delta = timedelta(days=origin)
if unit == 'weeks':
point_of_ref = datetime(
raw_point_of_ref.year, raw_point_of_ref.month, raw_point_of_ref.day, 0, 0, 0
)
sunday = False
if week_starts_on.lower() == 'sunday':
sunday = True
weekday = point_of_ref.weekday()
# Compensate for ISO week starting on Monday by default
if sunday:
weekday += 1
logger.debug('Weekday = %s', weekday)
start_delta = timedelta(days=weekday, weeks=origin)
if unit == 'months':
point_of_ref = datetime(
raw_point_of_ref.year, raw_point_of_ref.month, 1, 0, 0, 0
)
year = raw_point_of_ref.year
month = raw_point_of_ref.month
if origin > 0:
for _ in range(0, origin):
if month == 1:
year -= 1
month = 12
else:
month -= 1
else:
for _ in range(origin, 0):
if month == 12:
year += 1
month = 1
else:
month += 1
start_date = datetime(year, month, 1, 0, 0, 0)
if unit == 'years':
point_of_ref = datetime(raw_point_of_ref.year, 1, 1, 0, 0, 0)
start_date = datetime(raw_point_of_ref.year - origin, 1, 1, 0, 0, 0)
if unit not in ['months', 'years']:
start_date = point_of_ref - start_delta
# By this point, we know our start date and can convert it to epoch time
start_epoch = datetime_to_epoch(start_date)
logger.debug('Start ISO8601 = %s', epoch2iso(start_epoch))
# This is the number of units we need to consider.
count = (range_to - range_from) + 1
# We have to iterate to one more month, and then subtract a second to get
# the last day of the correct month
if unit == 'months':
month = start_date.month
year = start_date.year
for _ in range(0, count):
if month == 12:
year += 1
month = 1
else:
month += 1
end_date = datetime(year, month, 1, 0, 0, 0)
end_epoch = datetime_to_epoch(end_date) - 1
# Similarly, with years, we need to get the last moment of the year
elif unit == 'years':
end_date = datetime((raw_point_of_ref.year - origin) + count, 1, 1, 0, 0, 0)
end_epoch = datetime_to_epoch(end_date) - 1
# It's not months or years, which have inconsistent reckoning...
else:
# This lets us use an existing method to simply add unit * count seconds
# to get hours, days, or weeks, as they don't change
end_epoch = get_point_of_reference(unit, count * -1, epoch=start_epoch) - 1
logger.debug('End ISO8601 = %s', epoch2iso(end_epoch))
return (start_epoch, end_epoch)