Module ocean_science_utilities.tools.time

Copyright (C) 2022 Sofar Ocean Technologies

Authors: Pieter Bart Smit

"""
Copyright (C) 2022
Sofar Ocean Technologies

Authors: Pieter Bart Smit


"""
import numpy as np
import pandas as pd
import xarray

from datetime import datetime, timezone, timedelta
from numbers import Number
from typing import Optional, Union, Sequence


scalar_input_types = Union[str, float, int, datetime, np.datetime64]
input_types = Union[scalar_input_types, Sequence[scalar_input_types]]


def to_datetime_utc(time: input_types) -> Union[datetime, Sequence[datetime], None]:
    """
    Output datetimes are garantueed to be in the UTC timezone. For timezone naive input
    the timezone is assumed to be UTC. None as input is translated to None as output
    to allow for cases where time is optional. Note that the implementation works
    with heterogeneous sequences.

    :param time: Time, is either a valid scalar time type or a sequence of time types.
    :return: If the input is a sequence, the output is a sequence of datetimes,
    otherwise it is a scalar datetime.

    """

    if time is None:
        return None

    if isinstance(time, (xarray.DataArray, list, tuple, np.ndarray, pd.Series)):
        # if this is a sequence type, recursively call to datetime on the sequence
        if isinstance(time, (xarray.DataArray, pd.Series)):
            time = time.values

        return [to_datetime_utc(x) for x in time]  # type: ignore

    else:
        # if this is a scalar type, do the appropriate conversion.
        if isinstance(time, datetime):
            if time.tzinfo is None:
                time = time.replace(tzinfo=timezone.utc)
            return time.astimezone(timezone.utc)

        elif isinstance(time, str):
            if time[-1] == "Z":
                # From isoformat does not parse "Z" as a valid timezone designator.
                # This should be fixed in Python 3.11.
                time = time[:-1] + "+00:00"
            try:
                dt = datetime.fromisoformat(time)
                if dt.tzinfo is None:
                    dt = dt.replace(tzinfo=timezone.utc)

                return dt.astimezone(timezone.utc)
            except ValueError:
                return datetime.strptime(time, "%Y-%m-%dT%H:%M:%S.%f%z").astimezone(
                    timezone.utc
                )

        elif isinstance(time, np.datetime64):
            # We first cast datetime64 explicitly to seconds and then as a float to
            # allow for fractional seconds.
            return datetime.fromtimestamp(
                np.datetime64(time, "s").astype("float64"), tz=timezone.utc
            ).replace(tzinfo=timezone.utc)

        elif isinstance(time, Number):
            return datetime.fromtimestamp(time, tz=timezone.utc)  # type: ignore

        else:
            raise ValueError(f"Unknown time type: {type(time)} with value {time}")


def datetime_to_iso_time_string(time: Optional[input_types]) -> Optional[str]:
    if time is None:
        return None

    time = to_datetime_utc(time)
    return time.strftime("%Y-%m-%dT%H:%M:%S.%fZ")  # type: ignore


def datetime64_to_timestamp(time: Sequence[np.datetime64]) -> Sequence[np.datetime64]:
    return (time - np.datetime64(0, "s")) / np.timedelta64(1, "s")  # type: ignore


def to_datetime64(time) -> Union[None, np.datetime64, np.typing.NDArray[np.datetime64]]:
    """
    Convert time input to numpy np.ndarrays.
    :param time:
    :return:
    """
    if time is None:
        return None

    # Convert to a (sequence of) UTC datetime(s)
    time = to_datetime_utc(time)

    if isinstance(time, datetime):
        # If a datetime- do conversion
        return np.datetime64(int(time.timestamp()), "s").astype("<M8[ns]")
    else:
        # if  a sequence, do list comprehension and return an array
        return np.array(
            [np.datetime64(int(x.timestamp()), "s").astype("<M8[ns]") for x in time]
        )


def time_from_timeint(t: int) -> timedelta:
    """
    unpack a timedelta from a time given as an integer in the form "hhmmss"
    e.g. 201813 for 20:18:13
    """
    if t >= 10000:
        hours = t // 10000
        minutes = (t - hours * 10000) // 100
        seconds = t - hours * 10000 - minutes * 100
    elif t >= 100:
        hours = t // 100
        minutes = t - hours * 100
        seconds = 0
    else:
        hours = t
        minutes = 0
        seconds = 0

    return timedelta(seconds=(hours * 3600 + minutes * 60 + seconds))


def date_from_dateint(t: int) -> datetime:
    """
    unpack a datetime from a date given as an integer in the form "yyyymmdd" or "yymmdd"
    e.g. 20221109 for 2022-11-09 or 221109 for 2022-11-09
    """

    if t > 1000000:
        years = t // 10000
        months = (t - years * 10000) // 100
        days = t - years * 10000 - months * 100
    else:
        years = t // 10000
        months = (t - years * 10000) // 100
        days = t - years * 10000 - months * 100
        years = years + 2000

    return datetime(years, months, days, tzinfo=timezone.utc)


def datetime_from_time_and_date_integers(
    date_int: int, time_int: int, as_datetime64=False
) -> Union[datetime, None, np.datetime64, np.typing.NDArray[np.datetime64]]:
    """
    Convert a date and time given as integed encoded in the form "yyyymmdd" and "hhmm"
    _or_ "hhmmss" to a datetime
    :param date_int: integer of the form yyyymmdd
    :param time_int: time of the form "hhmm" or "hhmmss"
    :return:
    """
    dt = date_from_dateint(date_int) + time_from_timeint(time_int)
    if as_datetime64:
        return to_datetime64(dt)
    else:
        return dt

Functions

def date_from_dateint(t: int) ‑> datetime.datetime

unpack a datetime from a date given as an integer in the form "yyyymmdd" or "yymmdd" e.g. 20221109 for 2022-11-09 or 221109 for 2022-11-09

Expand source code

def date_from_dateint(t: int) -> datetime:
    """
    unpack a datetime from a date given as an integer in the form "yyyymmdd" or "yymmdd"
    e.g. 20221109 for 2022-11-09 or 221109 for 2022-11-09
    """

    if t > 1000000:
        years = t // 10000
        months = (t - years * 10000) // 100
        days = t - years * 10000 - months * 100
    else:
        years = t // 10000
        months = (t - years * 10000) // 100
        days = t - years * 10000 - months * 100
        years = years + 2000

    return datetime(years, months, days, tzinfo=timezone.utc)

def datetime64_to_timestamp(time: Sequence[numpy.datetime64]) ‑> Sequence[numpy.datetime64]

Expand source code

def datetime64_to_timestamp(time: Sequence[np.datetime64]) -> Sequence[np.datetime64]:
    return (time - np.datetime64(0, "s")) / np.timedelta64(1, "s")  # type: ignore

def datetime_from_time_and_date_integers(date_int: int, time_int: int, as_datetime64=False) ‑> Union[datetime.datetime, ForwardRef(None), numpy.datetime64, numpy.ndarray[Any, numpy.dtype[numpy.datetime64]]]

Convert a date and time given as integed encoded in the form "yyyymmdd" and "hhmm" or "hhmmss" to a datetime :param date_int: integer of the form yyyymmdd :param time_int: time of the form "hhmm" or "hhmmss" :return:

Expand source code

def datetime_from_time_and_date_integers(
    date_int: int, time_int: int, as_datetime64=False
) -> Union[datetime, None, np.datetime64, np.typing.NDArray[np.datetime64]]:
    """
    Convert a date and time given as integed encoded in the form "yyyymmdd" and "hhmm"
    _or_ "hhmmss" to a datetime
    :param date_int: integer of the form yyyymmdd
    :param time_int: time of the form "hhmm" or "hhmmss"
    :return:
    """
    dt = date_from_dateint(date_int) + time_from_timeint(time_int)
    if as_datetime64:
        return to_datetime64(dt)
    else:
        return dt

def datetime_to_iso_time_string(time: Union[str, float, int, datetime.datetime, numpy.datetime64, Sequence[Union[str, float, int, datetime.datetime, numpy.datetime64]], ForwardRef(None)]) ‑> Optional[str]

Expand source code

def datetime_to_iso_time_string(time: Optional[input_types]) -> Optional[str]:
    if time is None:
        return None

    time = to_datetime_utc(time)
    return time.strftime("%Y-%m-%dT%H:%M:%S.%fZ")  # type: ignore

def time_from_timeint(t: int) ‑> datetime.timedelta

unpack a timedelta from a time given as an integer in the form "hhmmss" e.g. 201813 for 20:18:13

Expand source code

def time_from_timeint(t: int) -> timedelta:
    """
    unpack a timedelta from a time given as an integer in the form "hhmmss"
    e.g. 201813 for 20:18:13
    """
    if t >= 10000:
        hours = t // 10000
        minutes = (t - hours * 10000) // 100
        seconds = t - hours * 10000 - minutes * 100
    elif t >= 100:
        hours = t // 100
        minutes = t - hours * 100
        seconds = 0
    else:
        hours = t
        minutes = 0
        seconds = 0

    return timedelta(seconds=(hours * 3600 + minutes * 60 + seconds))

def to_datetime64(time) ‑> Union[ForwardRef(None), numpy.datetime64, numpy.ndarray[Any, numpy.dtype[numpy.datetime64]]]

Convert time input to numpy np.ndarrays. :param time: :return:

Expand source code

def to_datetime64(time) -> Union[None, np.datetime64, np.typing.NDArray[np.datetime64]]:
    """
    Convert time input to numpy np.ndarrays.
    :param time:
    :return:
    """
    if time is None:
        return None

    # Convert to a (sequence of) UTC datetime(s)
    time = to_datetime_utc(time)

    if isinstance(time, datetime):
        # If a datetime- do conversion
        return np.datetime64(int(time.timestamp()), "s").astype("<M8[ns]")
    else:
        # if  a sequence, do list comprehension and return an array
        return np.array(
            [np.datetime64(int(x.timestamp()), "s").astype("<M8[ns]") for x in time]
        )

def to_datetime_utc(time: Union[str, float, int, datetime.datetime, numpy.datetime64, Sequence[Union[str, float, int, datetime.datetime, numpy.datetime64]]]) ‑> Union[datetime.datetime, Sequence[datetime.datetime], ForwardRef(None)]

Output datetimes are garantueed to be in the UTC timezone. For timezone naive input the timezone is assumed to be UTC. None as input is translated to None as output to allow for cases where time is optional. Note that the implementation works with heterogeneous sequences.

:param time: Time, is either a valid scalar time type or a sequence of time types. :return: If the input is a sequence, the output is a sequence of datetimes, otherwise it is a scalar datetime.

Expand source code

def to_datetime_utc(time: input_types) -> Union[datetime, Sequence[datetime], None]:
    """
    Output datetimes are garantueed to be in the UTC timezone. For timezone naive input
    the timezone is assumed to be UTC. None as input is translated to None as output
    to allow for cases where time is optional. Note that the implementation works
    with heterogeneous sequences.

    :param time: Time, is either a valid scalar time type or a sequence of time types.
    :return: If the input is a sequence, the output is a sequence of datetimes,
    otherwise it is a scalar datetime.

    """

    if time is None:
        return None

    if isinstance(time, (xarray.DataArray, list, tuple, np.ndarray, pd.Series)):
        # if this is a sequence type, recursively call to datetime on the sequence
        if isinstance(time, (xarray.DataArray, pd.Series)):
            time = time.values

        return [to_datetime_utc(x) for x in time]  # type: ignore

    else:
        # if this is a scalar type, do the appropriate conversion.
        if isinstance(time, datetime):
            if time.tzinfo is None:
                time = time.replace(tzinfo=timezone.utc)
            return time.astimezone(timezone.utc)

        elif isinstance(time, str):
            if time[-1] == "Z":
                # From isoformat does not parse "Z" as a valid timezone designator.
                # This should be fixed in Python 3.11.
                time = time[:-1] + "+00:00"
            try:
                dt = datetime.fromisoformat(time)
                if dt.tzinfo is None:
                    dt = dt.replace(tzinfo=timezone.utc)

                return dt.astimezone(timezone.utc)
            except ValueError:
                return datetime.strptime(time, "%Y-%m-%dT%H:%M:%S.%f%z").astimezone(
                    timezone.utc
                )

        elif isinstance(time, np.datetime64):
            # We first cast datetime64 explicitly to seconds and then as a float to
            # allow for fractional seconds.
            return datetime.fromtimestamp(
                np.datetime64(time, "s").astype("float64"), tz=timezone.utc
            ).replace(tzinfo=timezone.utc)

        elif isinstance(time, Number):
            return datetime.fromtimestamp(time, tz=timezone.utc)  # type: ignore

        else:
            raise ValueError(f"Unknown time type: {type(time)} with value {time}")