monkey_wrench.date_time package

The package providing all datetime utilities.

class monkey_wrench.date_time.ChimpFilePathParser[source]

Bases: DateTimeParserBase

Static parser class for CHIMP-compiliant input and output file paths.

regex = '[0-9A-Za-z]+_([0-9]{4})([0-9]{2})([0-9]{2})_([0-9]{2})_([0-9]{2})'
static parse(filepath: Annotated[Path | T, AfterValidator(func=_validate_path)]) datetime[source]

Parse the given filepath into a datetime object.

Parameters:

filepath – The filepath to parse. It can be either an absolute path or a relative path (e.g. just the base name). For the parsing to be successful, the filepath must have the following format: <optional_path><prefix>_<YYYY>_<mm>_<DD>_<HH>_<MM><optional_extension>, where <prefix> is mandatory but can be anything except for an empty string. See the examples below.

Examples

>>> from pathlib import Path
>>>
>>> # Input is an absolute path of type `Path`.
>>> ChimpFilePathParser.parse(Path("/home/user/dir/seviri_20150731_22_12.extension"))
datetime.datetime(2015, 7, 31, 22, 12, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> # Input is an absolute path of type `Path`.
>>> ChimpFilePathParser.parse(Path("/home/user/dir/seviri_20150110_00_01.extension"))
datetime.datetime(2015, 1, 10, 0, 1, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> # Input is a relative path of type `Path`.
>>> ChimpFilePathParser.parse(Path("chimp_20150731_22_12.extension"))
datetime.datetime(2015, 7, 31, 22, 12, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> # Input is an absolute path of type `str`.
>>> ChimpFilePathParser.parse("/home/user/dir/prefix_20150731_22_12.extension")
datetime.datetime(2015, 7, 31, 22, 12, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> # Input is a relative path of type `str` and does not have an extension.
>>> ChimpFilePathParser.parse("seviri_20150731_22_12")
datetime.datetime(2015, 7, 31, 22, 12, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> # Input is a relative path of type `str` and its extension is numeric, i.e. `72`.
>>> ChimpFilePathParser.parse("p_20150731_22_1272")
datetime.datetime(2015, 7, 31, 22, 12, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> # Input is invalid (missing prefix). The following will raise an exception!
>>> # FilePathParser.parse("20150731_22_12")
>>> # Input is invalid (empty prefix). The following will raise an exception!
>>> # FilePathParser.parse("_20150731_22_12")
class monkey_wrench.date_time.DateTimeParserBase[source]

Bases: object

A static base class for parsing items, e.g. product IDs or file paths, into datetime objects.

static _raise_value_error(item: Any) Never[source]

Helper function to raise a ValueError when the given item cannot be parsed.

static parse_by_regex(item: str, regex: str, timezone: ZoneInfo | None = None) datetime[source]

Parse the given item into a datetime object using a regular expression.

Parameters:
  • item – The item to parse.

  • regex – The regular expression to match against.

  • timezone – The timezone to add to the datetime object. Defaults to None, which means UTC will be used.

Returns:

The parsed datetime object, if successful.

Raises:

ValueError – If the given item cannot be parsed.

Example

>>> regex = r"^(19|20\d{2})(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])_(0\d|1\d|2[0-3])_([0-5]\d)$"
>>> DateTimeParserBase.parse_by_regex("20230102_22_30", regex)
datetime.datetime(2023, 1, 2, 22, 30, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
static parse_by_format_string(datetime_string: str, datetime_format_string: str, timezone: ZoneInfo | None = None) datetime[source]

Parse the given datetime string into a datetime object using the given format string.

Parameters:
  • datetime_string – The datetime string to parse.

  • datetime_format_string – The format string using which the parsing is done, e.g. "%Y%m%d_%H_%M".

  • timezone – The timezone to add to the datetime object. Defaults to None, which means UTC will be used.

Returns:

The parsed datetime object, if successful.

Raises:

ValueError – If the given datetime string cannot be parsed.

Example

>>> DateTimeParserBase.parse_by_format_string("20230101_22_30", "%Y%m%d_%H_%M")
datetime.datetime(2023, 1, 1, 22, 30, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
classmethod parse_collection(items: list[Any] | set[Any] | tuple[Any, ...] | Generator) list[datetime] | set[datetime] | tuple[datetime, ...] | Generator[datetime, None, None][source]

Parse the given collection of items into a collection of datetime objects.

Parameters:

items – The collection (list/set/tuple or generator) of items to parse.

Returns:

A collection of datetime objects. The type of collection matches the type of the input collection, e.g. a list as input results in a list of datetime objects.

static parse(item: Any) Any[source]

Return the given item as is.

Warning

Oerride this static method for each derived class.

class monkey_wrench.date_time.DateTimePeriod(*, end_datetime: AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None, start_datetime: AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None)[source]

Bases: StartDateTime, EndDateTime

property datetime_period: DateTimePeriod
property span: timedelta

Return the span between the start and end datetimes.

as_tuple(sort: bool = False) tuple[datetime, datetime][source]

Return the datetime period as a 2-tuple.

Parameters:

sort – Determines whether the returned tuple should be first sorted or not. Defaults to False. If it is set to True, the first element of the 2-tuple is always the minimum of the start_datetime and end_datetime.

Returns:

The datetime period as a 2-tuple.

assert_both_or_neither_datetime_instances_are_none()[source]

Assert that if one of the datetime instances is None, the other one is also None.

assert_datetime_instances_are_not_none()[source]

Assert that none of the datetime instances are None.

class monkey_wrench.date_time.DateTimePeriodStrict(*, end_datetime: AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None, start_datetime: AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None)[source]

Bases: DateTimePeriod

Same as DateTimePeriod but does not allow fields to have None values.

property datetime_period: DateTimePeriodStrict
validate_datetime_instances() Self[source]

Ensure that datetime instances are not None.

class monkey_wrench.date_time.DateTimeRange(*, end_datetime: ~typing.Annotated[~pydantic.types.AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None, start_datetime: ~typing.Annotated[~pydantic.types.AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None, interval: ~datetime.timedelta | ~typing.Annotated[dict[~typing.Literal['weeks', 'days', 'hours', 'minutes', 'seconds'], float], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1), MaxLen(max_length=5)]), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)])[source]

Bases: DateTimePeriodStrict

Pydantic model for datetime ranges.

Note

This can be used both as a model and also as a generator. See the examples below.

Example

>>> from datetime import UTC, datetime, timedelta
>>>
>>> # as a data model
>>> dt_range = DateTimeRange(
...  start_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...  end_datetime=datetime(2022, 1, 8, tzinfo=UTC),
...  interval=timedelta(days=2)
... )
>>> dt_range.start_datetime
datetime.datetime(2022, 1, 1, 0, 0, tzinfo=datetime.timezone.utc)
>>> dt_range.end_datetime
datetime.datetime(2022, 1, 8, 0, 0, tzinfo=datetime.timezone.utc)
>>> dt_range.interval
datetime.timedelta(days=2)
>>> # as a generator with a positive interval
>>> dt_range = DateTimeRange(
...  start_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...  end_datetime=datetime(2022, 1, 8, tzinfo=UTC),
...  interval=timedelta(days=2)
... )
>>> for datetime_instance in dt_range:
...     print(datetime_instance.isoformat())
2022-01-01T00:00:00+00:00
2022-01-03T00:00:00+00:00
2022-01-05T00:00:00+00:00
2022-01-07T00:00:00+00:00
>>> # Negative interval
>>> dt_range = DateTimeRange(
...  start_datetime=datetime(2022, 1, 8, tzinfo=UTC),
...  end_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...  interval=timedelta(days=-2)
... )
>>> for datetime_instance in dt_range:
...     print(datetime_instance.isoformat())
2022-01-08T00:00:00+00:00
2022-01-06T00:00:00+00:00
2022-01-04T00:00:00+00:00
2022-01-02T00:00:00+00:00
>>> # The interval is negative, but the end datetime is after the start datetime.
>>> # This leads to a generator with no items.
>>> dt_range = DateTimeRange(
...  start_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...  end_datetime=datetime(2022, 1, 8, tzinfo=UTC),
...  interval=timedelta(days=-2)
... )
>>> list(dt_range)
[]
>>> # The interval is positive, but the end datetime is before the start datetime.
>>> # This leads to a generator with no items.
>>> dt_range = DateTimeRange(
...  start_datetime=datetime(2022, 1, 8, tzinfo=UTC),
...  end_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...  interval=timedelta(days=2)
... )
>>> list(dt_range)
[]
>>> # The start and the end datetime are the same.
>>> # This leads to a generator with no items.
>>> dt_range = DateTimeRange(
...  start_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...  end_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...  interval=timedelta(days=2)
... )
>>> list(dt_range)
[]
interval: <lambda>)]

The interval between two consecutive datetime instances. It can be both positive and negative.

class monkey_wrench.date_time.DateTimeRangeInBatches(*, end_datetime: ~typing.Annotated[~pydantic.types.AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None, start_datetime: ~typing.Annotated[~pydantic.types.AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None, batch_interval: ~datetime.timedelta | ~typing.Annotated[dict[~typing.Literal['weeks', 'days', 'hours', 'minutes', 'seconds'], float], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1), MaxLen(max_length=5)]), ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)])[source]

Bases: DateTimePeriodStrict

Pydantic model for a datetime range in batches.

Note

This can be used both as a model and also as a generator. See the examples below.

Warning

Note that end_datetime is inclusive, i.e. it will show up in the last batch. Even if the start and the end datetime are equal, we still get one batch. This is different from DateTimeRange, which treats end_datetime as exclusive.

Warning

Depending on the value of batch_interval, the batches can differ. See the examples below.

Examples

>>> from datetime import UTC, datetime, timedelta
>>>
>>> # A positive interval, means batches are returned in ascending order.
>>> # This is with respect to both the start and the end datetime.
>>> batches = DateTimeRangeInBatches(
...  start_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...  end_datetime=datetime(2022, 1, 8, tzinfo=UTC),
...  batch_interval=timedelta(days=2)
... )
>>> for batch in batches:
...     start = batch.start_datetime.isoformat()
...     end = batch.end_datetime.isoformat()
...     print(f"(start={start}, end={end})")
(start=2022-01-01T00:00:00+00:00, end=2022-01-03T00:00:00+00:00)
(start=2022-01-03T00:00:00+00:00, end=2022-01-05T00:00:00+00:00)
(start=2022-01-05T00:00:00+00:00, end=2022-01-07T00:00:00+00:00)
(start=2022-01-07T00:00:00+00:00, end=2022-01-08T00:00:00+00:00)
>>> # Compare with the following example, where the interval is negative.
>>> # The batches are returned in descending order.
>>> batches = DateTimeRangeInBatches(
...   start_datetime=datetime(2022, 1, 8, tzinfo=UTC),
...   end_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...   batch_interval=timedelta(days=-2)
... )
>>> for batch in batches:
...     start = batch.start_datetime.isoformat()
...     end = batch.end_datetime.isoformat()
...     print(f"(start={start}, end={end})")
(start=2022-01-06T00:00:00+00:00, end=2022-01-08T00:00:00+00:00)
(start=2022-01-04T00:00:00+00:00, end=2022-01-06T00:00:00+00:00)
(start=2022-01-02T00:00:00+00:00, end=2022-01-04T00:00:00+00:00)
(start=2022-01-01T00:00:00+00:00, end=2022-01-02T00:00:00+00:00)
>>> # The interval is positive, but the end datetime is before the start datetime.
>>> # This leads to a generator with no items.
>>> batches = DateTimeRangeInBatches(
...   start_datetime=datetime(2022, 1, 8, tzinfo=UTC),
...   end_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...   batch_interval=timedelta(days=2)
... )
>>> list(batches)
[]
>>> # The interval is negative, but the end datetime is after the start datetime.
>>> # This leads to a generator with no items.
>>> batches = DateTimeRangeInBatches(
...   start_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...   end_datetime=datetime(2022, 1, 8, tzinfo=UTC),
...   batch_interval=timedelta(days=-2)
... )
>>> list(batches)
[]
>>> # The end datetime is inclusive.
>>> # Although the start and the end datetime are equal, we still get one batch.
>>> batches = DateTimeRangeInBatches(
...   start_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...   end_datetime=datetime(2022, 1, 1, tzinfo=UTC),
...   batch_interval=timedelta(days=-2)
... )
>>> for batch in batches:
...     start = batch.start_datetime.isoformat()
...     end = batch.end_datetime.isoformat()
...     print(f"(start={start}, end={end})")
(start=2022-01-01T00:00:00+00:00, end=2022-01-01T00:00:00+00:00)
batch_interval: <lambda>)]

The datetime interval of a single batch. It can be both positive and negative.

This is defined as the difference between the two datetime instances in each batch.

Note

As a rule of thumb this parameter can be set to 30 days. A smaller value for batch_interval means a larger number of batches which increases the overall time needed to fetch all the products. A larger value for batch_interval shortens the total time to fetch all the products, however, you might get an error regarding sending too many requests to the server.

Note

The interval of each batch, is equal to batch_interval, except for the last batch if end_datetime - start_datetime is not divisible by batch_interval.

property datetime_range_in_batches: DateTimeRangeInBatches
class monkey_wrench.date_time.EndDateTime(*, end_datetime: AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None)[source]

Bases: Model

end_datetime: <lambda>)] | None
class monkey_wrench.date_time.FCIIDParser[source]

Bases: DateTimeParserBase

Static parser class for FCI product IDs.

static parse(fci_product_id: str) datetime[source]

Parse the given FCI product ID into a datetime object.

Example

>>> FCIIDParser.parse(
... "W_XX-EUMETSAT-Darmstadt,IMG+SAT,MTI1+FCI-1C-RRAD-FDHSI-FD--x-x---x_C_EUMT_"
... "20251216091032_IDPFI_OPE_20251216091007_20251216091923_N__O_0056_0000"
... )
datetime.datetime(2025, 12, 16, 9, 10, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> FCIIDParser.parse(
... "W_XX-EUMETSAT-Darmstadt,IMG+SAT,MTI1+FCI-1C-RRAD-HRFI-FD--x-x---x_C_EUMT_"
... "20250102102250_IDPFI_OPE_20250102102007_20250102102924_N__O_0063_0000"
... )
datetime.datetime(2025, 1, 2, 10, 20, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
class monkey_wrench.date_time.HritFilePathParser[source]

Bases: DateTimeParserBase

Static parser class for HRIT file paths.

static parse(filepath: Annotated[Path | T, AfterValidator(func=_validate_path)]) datetime[source]

Parse the given filepath into a datetime object.

Parameters:

filepath – The HRIT filepath to parse. It can be either an absolute path or a relative path (e.g. just the base name). For the parsing to be successful, the filepath must have the following format: <optional_path><optional_prefix><YYYYmmDDHHMM>-__. See the examples below.

Examples

>>> from pathlib import Path
>>>
>>> # Input is an absolute path of type `Path`.
>>> HritFilePathParser.parse(
...  Path("/home/user/dir/H-000-MSG3__-MSG3________-WV_073___-000008___-202503041900-__")
... )
datetime.datetime(2025, 3, 4, 19, 0, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> # Input is a relative path of type `Path`.
>>> HritFilePathParser.parse(Path("H-000-MSG3__-MSG3________-WV_073___-000008___-202503041900-__"))
datetime.datetime(2025, 3, 4, 19, 0, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> # Input is a relative path of type `str` without a prefix.
>>> HritFilePathParser.parse(Path("202503041900-__"))
datetime.datetime(2025, 3, 4, 19, 0, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
>>> # Input is invalid as it misses the mandatory trailing `-__`. The following will raise an exception!
>>> # HritFilePathParser.parse(Path("202503041900"))
class monkey_wrench.date_time.SeviriIDParser[source]

Bases: DateTimeParserBase

Static parser class for SEVIRI product IDs.

regex = '[0-9A-Za-z]+-SEVI-[0-9A-Za-z]+-[0-9]+-NA-([0-9]{4})([0-9]{2})([0-9]{2})([0-9]{2})([0-9]{2})[0-9]{2}\\.[0-9]+Z-NA'
static parse(seviri_product_id: str) datetime[source]

Parse the given SEVIRI product ID into a datetime object.

Example

>>> SeviriIDParser.parse("MSG3-SEVI-MSG15-0100-NA-20150731221240.036000000Z-NA")
datetime.datetime(2015, 7, 31, 22, 12, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
class monkey_wrench.date_time.StartDateTime(*, start_datetime: AwareDatetime, ~pydantic.functional_validators.AfterValidator(func=~monkey_wrench.date_time.models._base.<lambda>)] | None = None)[source]

Bases: Model

start_datetime: <lambda>)] | None
monkey_wrench.date_time.assert_datetime_has_past(datetime_instance: datetime, silent: bool = False) bool[source]

Assert that the datetime_instance is in not in the future.

Note

This function relies on assert_().

Examples

>>> # The following will not raise an exception.
>>> assert_datetime_has_past(datetime(2020, 1, 1, tzinfo=UTC))
True
>>> assert_datetime_has_past(datetime(2100, 1, 2, tzinfo=UTC), silent=True)
False
>>> # The following will raise an exception!
>>> # assert_has_datetime_past(datetime(2100, 1, 2, tzinfo=UTC))
monkey_wrench.date_time.assert_datetime_is_timezone_aware(datetime_object: datetime, silent: bool = False) bool[source]

Assert that the datetime_object is timezone-aware.

Note

This function relies on assert_().

Examples

>>> assert_datetime_is_timezone_aware(datetime.now(), silent=True)
False
>>> assert_datetime_is_timezone_aware(datetime.now(UTC))
True
monkey_wrench.date_time.assert_start_precedes_end(start_datetime: datetime, end_datetime: datetime, silent: bool = False) bool[source]

Assert that the start_datetime is not later than the end_datetime.

Note

This function relies on assert_().

Examples

>>> # The following will not raise an exception.
>>> assert_start_precedes_end(datetime(2020, 1, 1), datetime(2020, 12, 31))
True
>>> # The following will raise an exception!
>>> assert_start_precedes_end(datetime(2020, 1, 2), datetime(2020, 1, 1), silent=True)
False
>>> # The following will raise an exception!
>>> # assert_start_precedes_end(datetime(2020, 1, 2), datetime(2020, 1, 1))
monkey_wrench.date_time.floor_datetime_minutes_to_specific_snapshots(datetime_instance: datetime, snapshots: list[Annotated[int, Ge(ge=0), FieldInfo(annotation=NoneType, required=True, metadata=[Lt(lt=60)])]] | None = None) datetime[source]

Floor the given datetime instance to the closest minute from the given list of snapshots.

Parameters:
  • datetime_instance – The datetime instance to floor.

  • snapshots – A (sorted) list of minutes. Defaults to None, which means the given datetime instance will be returned as it is, without any modifications. As an example, for SEVIRI we have one snapshot per 15 minutes, starting from the 12th minute. As a result, we have [12, 27, 42, 57] for SEVIRI snapshots in an hour. The snapshots will be first sorted in increasing order.

Returns:

A datetime instance which is smaller than or equal to the given datetime_instance, such that the minute matches the closest minute from the snapshots.

Examples

>>> floor_datetime_minutes_to_specific_snapshots(
...  datetime(2020, 1, 1, 0, 3), [12, 27, 42, 57]
... )
datetime.datetime(2019, 12, 31, 23, 57)
>>> floor_datetime_minutes_to_specific_snapshots(
...  datetime(2020, 1, 1, 0, 58), [12, 27, 42, 57]
... )
datetime.datetime(2020, 1, 1, 0, 57)
>>> floor_datetime_minutes_to_specific_snapshots(
...  datetime(2020, 1, 1, 1, 30), [12, 27, 42, 57]
... )
datetime.datetime(2020, 1, 1, 1, 27)
>>> floor_datetime_minutes_to_specific_snapshots(
...  datetime(2020, 1, 1, 1, 27), [12, 27, 42, 57]
... )
datetime.datetime(2020, 1, 1, 1, 27)
>>> floor_datetime_minutes_to_specific_snapshots(
...  datetime(2020, 1, 1, 1, 26)
... )
datetime.datetime(2020, 1, 1, 1, 26)
monkey_wrench.date_time.number_of_days_in_month(year: Annotated[int, Gt(gt=0), FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=1950), Le(le=2100)])], month: Annotated[int, Gt(gt=0), FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=1), Le(le=12)])]) Annotated[int, Gt(gt=0), FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=1), Le(le=31)])][source]

Return the number of days in a month, taking into account both common and leap years.

Examples

>>> # `2018` was a common year.
>>> number_of_days_in_month(2018, 2)
28
>>> # `2020` was a leap year.
>>> number_of_days_in_month(2020, 2)
29

Subpackages

Submodules