6489602581

Committed 12 Oct 2023 12:25AM UTC coverage: 45.573% (+2.8%) from 42.772%

Build # 6489602581

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #363 from jo-basevi/358-date-based-frequency

Add support for date-based restart frequency

Run Details

111 of 147 new or added lines in 10 files covered. (75.51%)

2 existing lines in 1 file now uncovered.

1580 of 3467 relevant lines covered (45.57%)

1.37 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

58.46

/payu/calendar.py

import datetime
import re

from dateutil.relativedelta import relativedelta
import cftime

NOLEAP, GREGORIAN = range(2)


def int_to_date(date):
    """
    Convert an int of form yyyymmdd to a python date object.
    """

    year = date // 10**4
    month = date % 10**4 // 10**2
    day = date % 10**2

    return datetime.date(year, month, day)


def date_to_int(date):
    return date.year * 10**4 + date.month * 10**2 + date.day


def runtime_from_date(start_date, years, months, days, seconds, caltype):
    """
    Get the number of seconds from start date to start date + date_delta.

    Ignores Feb 29 for caltype == NOLEAP.
    """

    end_date = start_date + relativedelta(
        years=years, months=months, days=days
    )
    runtime = end_date - start_date

    if caltype == NOLEAP:
        runtime -= get_leapdays(start_date, end_date)

    return int(runtime.total_seconds() + seconds)


def date_plus_seconds(init_date, seconds, caltype):
    """
    Get a new_date = date + seconds.

    Ignores Feb 29 for no-leap days.
    """

    end_date = init_date + datetime.timedelta(seconds=seconds)

    if caltype == NOLEAP:
        end_date += get_leapdays(init_date, end_date)
        if end_date.month == 2 and end_date.day == 29:
            end_date += datetime.timedelta(days=1)

    return end_date


def get_leapdays(init_date, final_date):
    """
    Find the number of leap days between arbitrary dates. Returns a
    timedelta object.

    FIXME: calculate this instead of iterating.
    """

    curr_date = init_date
    leap_days = 0

    while curr_date != final_date:
        if curr_date.month == 2 and curr_date.day == 29:
            leap_days += 1

        curr_date += datetime.timedelta(days=1)

    return datetime.timedelta(days=leap_days)


def calculate_leapdays(init_date, final_date):
    """Currently unsupported, it only works for differences in years."""

    leap_days = (final_date.year - 1) // 4 - (init_date.year - 1) // 4
    leap_days -= (final_date.year - 1) // 100 - (init_date.year - 1) // 100
    leap_days += (final_date.year - 1) // 400 - (init_date.year - 1) // 400

    # TODO: Internal date correction (e.g. init_date is 1-March or later)

    return datetime.timedelta(days=leap_days)


def add_year_start_offset_to_datetime(initial_dt, n):
    """Return a cftime datetime at the start of the year, that is n years
    from the initial datetime"""
    return cftime.datetime(
        year=initial_dt.year + n,
        month=1,
        day=1,
        hour=0,
        minute=0,
        second=0,
        calendar=initial_dt.calendar,
    )


def add_month_start_offset_to_datetime(initial_dt, n):
    """Return a cftime datetime of the start of the month, that is n months
    from the initial datetime"""
    years_to_add = (initial_dt.month + n - 1) // 12
    months_to_add = n - years_to_add * 12

    return cftime.datetime(
        year=initial_dt.year + years_to_add,
        month=initial_dt.month + months_to_add,
        day=1,
        hour=0,
        minute=0,
        second=0,
        calendar=initial_dt.calendar,
    )


def add_timedelta_fn(timedelta):
    """Returns a function that takes initial datetime and multiplier n,
    and returns a datetime that is n * offset from the initial datetime"""
    return lambda initial_dt, n: initial_dt + n * timedelta


class DatetimeOffset:
    """A utility class for adding various time offsets to cftime datetimes.

    Parameters:
        unit (str): The unit of the time offset. Supported units are:
            - "YS" for years (start of the year)
            - "MS" for months (start of the month)
            - "W" for weeks
            - "D" for days
            - "H" for hours
            - "T" for minutes
            - "S" for seconds
        magnitude (int): The magnitude of the time offset.

    Methods:
        - `add_to_datetime(initial_dt: cftime.datetime) -> cftime.datetime`:
          Adds the specified time offset to the given cftime datetime and
          returns the resulting datetime.

    Attributes:
        - unit (str): The unit of the time offset.
        - magnitude (int): The magnitude of the time offset.
    """

    def __init__(self, unit, magnitude):
        supported_datetime_offsets = {
            "YS": add_year_start_offset_to_datetime,
            "MS": add_month_start_offset_to_datetime,
            "W": add_timedelta_fn(datetime.timedelta(weeks=1)),
            "D": add_timedelta_fn(datetime.timedelta(days=1)),
            "H": add_timedelta_fn(datetime.timedelta(hours=1)),
            "T": add_timedelta_fn(datetime.timedelta(minutes=1)),
            "S": add_timedelta_fn(datetime.timedelta(seconds=1)),
        }
        if unit not in supported_datetime_offsets:
            raise ValueError(
                f"Unsupported datetime offset: {unit}. "
                "Supported offsets: YS, MS, W, D, H, T, S"
            )
        self.unit = unit
        self.magnitude = magnitude
        self._add_offset_to_datetime = supported_datetime_offsets[unit]

    def add_to_datetime(self, initial_dt):
        """Takes an initial cftime datetime,
        and returns a datetime with the offset added"""

        if not (isinstance(initial_dt, cftime.datetime)):
            raise TypeError(
                f"Invalid initial datetime type: {type(initial_dt)}. "
                "Expected type: cftime.datetime"
            )

        return self._add_offset_to_datetime(
            initial_dt=initial_dt, n=self.magnitude
        )


def parse_date_offset(offset):
    """Parse a given string date offset string and return an DatetimeOffset"""
    match = re.search("[0-9]+", offset)
    if match is None:
        raise ValueError(
            f"No numerical value given for offset: {offset}"
        )
    n = match.group()
    unit = offset.lstrip(n)
    return DatetimeOffset(unit=unit, magnitude=int(n))

1	import datetime	3✔
2	import re	3✔
3
4	from dateutil.relativedelta import relativedelta	3✔
5	import cftime	3✔
6
7	NOLEAP, GREGORIAN = range(2)	3✔
8
9
10	def int_to_date(date):	3✔
11	"""
12	Convert an int of form yyyymmdd to a python date object.
13	"""
14
15	year = date // 10**4	×
16	month = date % 104 // 102	×
17	day = date % 10**2	×
18
19	return datetime.date(year, month, day)	×
20
21
22	def date_to_int(date):	3✔
NEW 23	return date.year * 10*4 + date.month 10**2 + date.day	×
24
25
26	def runtime_from_date(start_date, years, months, days, seconds, caltype):	3✔
27	"""
28	Get the number of seconds from start date to start date + date_delta.
29
30	Ignores Feb 29 for caltype == NOLEAP.
31	"""
32
NEW 33	end_date = start_date + relativedelta(	×
34	years=years, months=months, days=days
35	)
UNCOV 36	runtime = end_date - start_date	×
37
38	if caltype == NOLEAP:	×
39	runtime -= get_leapdays(start_date, end_date)	×
40
41	return int(runtime.total_seconds() + seconds)	×
42
43
44	def date_plus_seconds(init_date, seconds, caltype):	3✔
45	"""
46	Get a new_date = date + seconds.
47
48	Ignores Feb 29 for no-leap days.
49	"""
50
51	end_date = init_date + datetime.timedelta(seconds=seconds)	×
52
53	if caltype == NOLEAP:	×
54	end_date += get_leapdays(init_date, end_date)	×
55	if end_date.month == 2 and end_date.day == 29:	×
56	end_date += datetime.timedelta(days=1)	×
57
58	return end_date	×
59
60
61	def get_leapdays(init_date, final_date):	3✔
62	"""
63	Find the number of leap days between arbitrary dates. Returns a
64	timedelta object.
65
66	FIXME: calculate this instead of iterating.
67	"""
68
69	curr_date = init_date	×
70	leap_days = 0	×
71
72	while curr_date != final_date:	×
UNCOV 73	if curr_date.month == 2 and curr_date.day == 29:	×
74	leap_days += 1	×
75
76	curr_date += datetime.timedelta(days=1)	×
77
78	return datetime.timedelta(days=leap_days)	×
79
80
81	def calculate_leapdays(init_date, final_date):	3✔
82	"""Currently unsupported, it only works for differences in years."""
83
84	leap_days = (final_date.year - 1) // 4 - (init_date.year - 1) // 4	×
85	leap_days -= (final_date.year - 1) // 100 - (init_date.year - 1) // 100	×
86	leap_days += (final_date.year - 1) // 400 - (init_date.year - 1) // 400	×
87
88	# TODO: Internal date correction (e.g. init_date is 1-March or later)
89
90	return datetime.timedelta(days=leap_days)	×
91
92
93	def add_year_start_offset_to_datetime(initial_dt, n):	3✔
94	"""Return a cftime datetime at the start of the year, that is n years
95	from the initial datetime"""
96	return cftime.datetime(	3✔
97	year=initial_dt.year + n,
98	month=1,
99	day=1,
100	hour=0,
101	minute=0,
102	second=0,
103	calendar=initial_dt.calendar,
104	)
105
106
107	def add_month_start_offset_to_datetime(initial_dt, n):	3✔
108	"""Return a cftime datetime of the start of the month, that is n months
109	from the initial datetime"""
110	years_to_add = (initial_dt.month + n - 1) // 12	3✔
111	months_to_add = n - years_to_add * 12	3✔
112
113	return cftime.datetime(	3✔
114	year=initial_dt.year + years_to_add,
115	month=initial_dt.month + months_to_add,
116	day=1,
117	hour=0,
118	minute=0,
119	second=0,
120	calendar=initial_dt.calendar,
121	)
122
123
124	def add_timedelta_fn(timedelta):	3✔
125	"""Returns a function that takes initial datetime and multiplier n,
126	and returns a datetime that is n * offset from the initial datetime"""
127	return lambda initial_dt, n: initial_dt + n * timedelta	3✔
128
129
130	class DatetimeOffset:	3✔
131	"""A utility class for adding various time offsets to cftime datetimes.
132
133	Parameters:
134	unit (str): The unit of the time offset. Supported units are:
135	- "YS" for years (start of the year)
136	- "MS" for months (start of the month)
137	- "W" for weeks
138	- "D" for days
139	- "H" for hours
140	- "T" for minutes
141	- "S" for seconds
142	magnitude (int): The magnitude of the time offset.
143
144	Methods:
145	- `add_to_datetime(initial_dt: cftime.datetime) -> cftime.datetime`:
146	Adds the specified time offset to the given cftime datetime and
147	returns the resulting datetime.
148
149	Attributes:
150	- unit (str): The unit of the time offset.
151	- magnitude (int): The magnitude of the time offset.
152	"""
153
154	def __init__(self, unit, magnitude):	3✔
155	supported_datetime_offsets = {	3✔
156	"YS": add_year_start_offset_to_datetime,
157	"MS": add_month_start_offset_to_datetime,
158	"W": add_timedelta_fn(datetime.timedelta(weeks=1)),
159	"D": add_timedelta_fn(datetime.timedelta(days=1)),
160	"H": add_timedelta_fn(datetime.timedelta(hours=1)),
161	"T": add_timedelta_fn(datetime.timedelta(minutes=1)),
162	"S": add_timedelta_fn(datetime.timedelta(seconds=1)),
163	}
164	if unit not in supported_datetime_offsets:	3✔
165	raise ValueError(	3✔
166	f"Unsupported datetime offset: {unit}. "
167	"Supported offsets: YS, MS, W, D, H, T, S"
168	)
169	self.unit = unit	3✔
170	self.magnitude = magnitude	3✔
171	self._add_offset_to_datetime = supported_datetime_offsets[unit]	3✔
172
173	def add_to_datetime(self, initial_dt):	3✔
174	"""Takes an initial cftime datetime,
175	and returns a datetime with the offset added"""
176
177	if not (isinstance(initial_dt, cftime.datetime)):	3✔
178	raise TypeError(	3✔
179	f"Invalid initial datetime type: {type(initial_dt)}. "
180	"Expected type: cftime.datetime"
181	)
182
183	return self._add_offset_to_datetime(	3✔
184	initial_dt=initial_dt, n=self.magnitude
185	)
186
187
188	def parse_date_offset(offset):	3✔
189	"""Parse a given string date offset string and return an DatetimeOffset"""
190	match = re.search("[0-9]+", offset)	3✔
191	if match is None:	3✔
192	raise ValueError(	3✔
193	f"No numerical value given for offset: {offset}"
194	)
195	n = match.group()	3✔
196	unit = offset.lstrip(n)	3✔
197	return DatetimeOffset(unit=unit, magnitude=int(n))	3✔

payu-org / payu / 6489602581

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous