api ¶

def convert_timestamps_to_utc(
    user_requested_timezone: ZoneInfo | None = None,
    user_requested_timestamps: Timestamp | DatetimeIndex | None = None,
) -> Timestamp | DatetimeIndex:
    """ """
    if user_requested_timestamps is None:
        user_requested_timestamps = Timestamp.now()

    logger.debug(
        f"Input time zone : {user_requested_timezone}",
        alt=f"Input time zone : [code]{user_requested_timezone}[/code]",
    )
    utc_timestamps = user_requested_timestamps  # Fallback if already UTC

    # naive timestamps
    if user_requested_timestamps.tz is None:
        utc_timestamps = user_requested_timestamps.tz_localize(ZONEINFO_UTC)
        logger.debug(
            f"Naive input timestamps\n({user_requested_timestamps})\nlocalized to UTC aware for all internal calculations :\n{utc_timestamps}"
        )

    # timezone aware timestamps
    elif user_requested_timestamps.tz != ZONEINFO_UTC:
        utc_timestamps = user_requested_timestamps.tz_convert(ZONEINFO_UTC)
        logger.debug(
            f"Input zone\n{user_requested_timezone}\n& timestamps :\n{user_requested_timestamps}\n\nconverted for all internal calculations to :\n{utc_timestamps}",
            alt=f"Input zone : [code]{user_requested_timezone}[/code]\n& timestamps :\n{user_requested_timestamps}\n\nconverted for all internal calculations to :\n{utc_timestamps}",
        )

    return utc_timestamps

def generate_datetime_series(
    start_time: str | None = None,
    end_time: str | None = None,
    periods: str | None = None,
    frequency: str | None = TIMESTAMPS_FREQUENCY_DEFAULT,
    timezone: ZoneInfo | None = None,
    name: str | None = None,
) -> Timestamp | DatetimeIndex:
    """Generate a fixed frequency DatetimeIndex

    Generates a range of equally spaced timestamps wrapping over Pandas'
    date_range() function. The timestamps satisfy `start_time <[=] x <[=]
    end_time`, where the first and last stamps fall on the boundary of the
    requested ``frequency`` string. The difference between any two timestamps
    is specified by the requested ``frequency``.

    If exactly one of ``start_time``, ``end_time``, or ``frequency`` is *not*
    specified, it can be computed by the ``periods``, the number of timesteps
    in the range.

    Parameters
    ----------
    start_time : str, datetime, date, pandas.Timestamp, or period-like, default None
        The starting time (if str in ISO format), also described as the left
        bound for generating periods.
    end_time : str, datetime, date, pandas.Timestamp, or period-like, default None
        The ending time in ISO format. In Pandas described as the right bound
        for generating periods.
    periods : int, default None
        Number of periods to generate.
    frequency : str or DateOffset, optional
        Frequency alias of the timestamps to generate, e.g., 'h' for hourly.
        By default the frequency is taken from start_time or
        end_time if those are Period objects. Otherwise, the default is "h" for
        hourly frequency.
    name : str, default None
        Name of the resulting PeriodIndex.

    Returns
    -------
    DatetimeIndex
        A Pandas DatetimeIndex at the specified frequency.

    See Also
    --------
    pandas.date_range
        Return a fixed frequency DatetimeIndex.

    Notes
    -----
    Of the four parameters ``start_time``, ``end_time``, ``periods``, and
    ``frequency``, exactly three must be specified. If ``frequency`` is
    omitted, the resulting ``DatetimeIndex`` will have ``periods`` linearly
    spaced elements between ``start`` and ``end`` (closed on both sides).

    Common time series frequencies are indexed via a set of string (also
    referred to as offset) aliases described at
    <https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#offset-aliases>`__.

    Example
    -------
    >>> start_time = '2010-06-01 06:00:00'
    >>> end_time = '2010-06-01 08:00:00'
    >>> frequency = 'h'  # 'h' for hourly
    >>> generate_datetime_series(start_time, end_time, frequency)
    DatetimeIndex(['2010-06-01 06:00:00', '2010-06-01 07:00:00', '2010-06-01 08:00:00'], dtype='datetime64[ns]', freq=None)

    Using the periods input parameter to define the number of timesteps to generate :

    >>> generate_datetime_series(start_time=start_time, periods=4, frequency=frequency)
    DatetimeIndex(['2010-06-01 06:00:00', '2010-06-01 07:00:00',
               '2010-06-01 08:00:00', '2010-06-01 09:00:00'],
              dtype='datetime64[ns]', freq='H')
    """
    # Validate input parameters --
    # Can we do this with a callback and at the Context level ?
    number_of_parameters = sum(
        parameter is not None for parameter in [start_time, end_time, periods]
    )
    if number_of_parameters < 2:
        error_message = (
            f"Insufficient parameters to generate timestamps. "
            f"User input is : start_time={start_time}, end_time={end_time}, periods={periods}. "
            f"Please provide at least two or at most three out of the timestamp relevant parameters!"
        )
        logger.error(error_message)
        raise ValueError(error_message)

    elif start_time is None and end_time is None:
        timestamps = Timestamp.now(tz="UTC")

    else:
        try:
            timestamps = date_range(
                start=start_time,
                end=end_time,
                periods=periods,
                freq=frequency,
                tz=timezone,
                name=name,
            )
        except Exception as e:
            logger.exception("Failed to generate datetime series.")
            raise ValueError(f"Failed to generate datetime series: {str(e)}")

        if timestamps.empty:
            error_message = "The generated DatetimeIndex is empty! You might want to check the relevant timestamp parameters for accuracy."
            logger.error(error_message)
            raise ValueError(error_message)

    return timestamps

def generate_timestamps(
    data_file: Path |  None,
    time_offset: Timedelta | None = None,
    start_time: Timestamp | None = None,
    end_time: Timestamp | None = None,
    periods: str | None = None,
    frequency: str | None = TIMESTAMPS_FREQUENCY_DEFAULT,
    timezone: ZoneInfo | None = None,
    name: str | None = None,
) -> DatetimeIndex:
    """
    Parameters
    ----------
    data_file: Path

    time_offset: Timedelta
        Time delta to add to all timestamps in the DatetimeIndex. This may be
        especially important for instantaneous data (i.e. SATAHx solar
        irradiance time series data) where the solar positioning has to
        coincide with the acquisition of the data. 

    start_time: Timestamp

    """
    if start_time == end_time:
        raise ValueError(
            "The `start_time` and `end_time` cannot be identical. If you intend to use a single timestamp, please specify it directly, e.g., '2121-12-22 12:21:12'."
        )

    # Extract timestamps from first available space-time data file
    if data_file is not None:
        logger.debug(
            f"Retrieving timestamps from input time series data {data_file}",
            alt=f"[bold]Retrieving[/bold] timestamps from input time series data [code]{data_file}[/code]",
        )
        if isinstance(data_file, Path | str):
            timestamps = read_data_array_or_set(data_file).time  # type: ignore
        else:
            timestamps = data_file.time  # type: ignore

        logger.info(
                f"timestamps retrieved from {data_file} :\n{timestamps}",
                alt=f"timestamps retrieved from [code]{data_file}[/code] :\n{timestamps}"
                )
        # Implement Me ? --------------------------------------------------- #
        #                                                                    #

        # if check_for_duplicate_timestamps:
        #     if timestamps.indexes['time'].duplicated().any():
        #         logger.error(
        #                 f"Duplicate timestamps detected.",
        #                 alt= f"[red]Duplicate timestamps detected![/red]"
        #                 )

        #                                                                    #
        # ----------------------------------------------------- Implement Me #

        if timestamps is None:
            logger.error(
                "No timestamps found in the provided data file!",
                alt="[red]No timestamps found in the provided data file![/red]",
            )
            raise ValueError("Unable to extract timestamps from the data file.")

        if start_time and end_time:
            if start_time > end_time:  # type: ignore
                logger.error(
                    f"Timestamp {start_time=} should be later than {end_time=}!",
                    alt=f"[red]Timestamp {start_time=} should be later than {end_time=}![/red]",
                )
                raise ValueError(
                    f"Timestamp {start_time=} should be later than {end_time=}!"
                )

        # # convert back to Xarray !
        # from xarray import DataArray
        # timestamps = DataArray(timestamps, dims=["time"], name="time")

        # Filter timestamps based on start_time and end_time
        if start_time:

            if start_time.tzinfo:
                start_time = start_time.tz_localize(None)

        if end_time:

            if end_time.tzinfo:
                end_time = end_time.tz_localize(None)

        if start_time or end_time:

            logger.debug(
                f"> Slicing timestamps from {start_time} to {end_time}",
                alt=f"> [bold]Slicing[/bold] timestamps from {start_time} to {end_time}",
            )
            timestamps = timestamps.sel(time=slice(start_time, end_time))
            logger.debug(
                f"  : Slice of timestamps :\n{timestamps}",
                alt=f"  [blue]:[/blue] Slice of timestamps :\n{timestamps}",
            )

        if start_time and periods and not end_time:
            if frequency:
                timestamps = timestamps.resample(time=frequency).nearest()
            timestamps = timestamps.isel(time=slice(0, periods))

        elif end_time and periods and not start_time:
            if frequency:
                timestamps = timestamps.resample(time=frequency).nearest()
            timestamps = timestamps.isel(time=slice(-periods, None))

        elif start_time and end_time and periods:
            logger.error(
                f"Best if you provide a `start_time` OR an `end_time` along with `periods`, not both!",
                alt=f"[bold]Best if you provide a[/bold] `start_time` [bold][italics yellow]or[/italics yellow] an[/bold] `end_time` [bold]along with[/bold] `periods`, [bold red]not both![/bold red]",
            )
            raise ValueError(
                "Best if you provide a `start_time` or an `end_time` along with `periods`, not both! Else, I cannot decide which periods to return, from the start or the the end.. ;-?"
            )

        elif frequency and not periods and (start_time or end_time):
            # resampled_timestamps = DatetimeIndex(
            #     timestamps.resample(time=frequency).nearest()
            # )
            # resampled_timestamps.intersection(timestamps)
            # logger.debug(
            #     f"Resampled timestamps at frequency = {frequency} :\n{timestamps}",
            #     alt=f"Resampled timestamps at frequency = {frequency} :\n{timestamps}",
            # )
            logger.warning(
                f"Resampling the timestamps retrieved from the data would eventually introduce new timestamps! Skipping...",
                alt=f"[red on white]Resampling the timestamps retrieved from the data would eventually introduce new timestamps! Skipping...[/red on white]",
            )

        timestamps = DatetimeIndex(timestamps)

    else:
        logger.debug(
            f"  + Generating timestamps based on user-requested time series parameters",
            alt=f"  [magenta]+[/magenta] [bold]Generating[/bold] timestamps based on user-requested time series parameters",
        )
        timestamps = generate_datetime_series(
            start_time=start_time,
            end_time=end_time,
            periods=periods,
            frequency=frequency,
            timezone=timezone,
            name=name,
        )
        # from pandas import to_datetime
        # -----------------------------------------------------------------------
        # If we do the following, we need to take care of external naive time series!
        # timezone_aware_timestamps = [
        #     attach_requested_timezone(timestamp, timezone) for timestamp in timestamps
        # ]
        # return to_datetime(timezone_aware_timestamps, format="mixed")
        # -----------------------------------------------------------------------
    logger.warning(
        f"  < Returning timestamps :\n{timestamps}",
        alt=f"  [green]<[/green] Returning timestamps :\n{timestamps}",
    )

    if time_offset is not None:
        timestamps += time_offset

    return timestamps

def get_day_from_hour_of_year(year: int, hour_of_year: int):
    """Get day of year from hour of year."""
    start_of_year = np.datetime64(f"{year}-01-01")
    date_and_time = start_of_year + np.timedelta64(hour_of_year, "h")
    date_and_time = date_and_time.astype(datetime.datetime)
    day_of_year = int(date_and_time.strftime("%j"))
    # month = int(date_and_time.strftime('%m'))  # Month
    # day_of_month = int(date_and_time.strftime('%d'))
    # hour_of_day = int(date_and_time.strftime('%H'))

    return day_of_year

def get_days_in_year(year):
    """Calculate the number of days in a given year, accounting for leap years.

    Parameters
    ----------
    year : int
        The year for which to calculate the number of days.

    Returns
    -------
    int
        The number of days in the given year.

    Examples
    --------
    >>> get_days_in_year(2020)
    366

    >>> get_days_in_year(2021)
    365
    """
    start_date = datetime(year, 1, 1)  # First day of the year
    end_date = datetime(year + 1, 1, 1)  # First day of the next year
    return (end_date - start_date).days

def get_days_in_years(years):
    """Calculate the number of days in a given year, accounting for leap years.

    Parameters
    ----------
    years : DatetimeIndex
        The years series for which to calculate the number of days.

    Returns
    -------
    DatetimeIndex :
        The number of days series for the given years series

    Examples
    --------
    >>> get_days_in_years_series(pd.DatetimeIndex(['2000-12-22 21:12:12', '2001-11-11 11:11:11']))
    Index([366, 365], dtype='int64')
    """

    years = years.to_numpy().astype(int)
    is_leap_year = (years % 4 == 0) & ((years % 100 != 0) | (years % 400 == 0)) # Vectorized calculation for leap years
    days_in_year = np.where(is_leap_year, 366, 365)

    return Index(days_in_year, dtype='int32')

def now_datetime() -> datetime:
    """Returns the current datetime in UTC.

    Return an aware timestamp using the local system time, however defaulting
    to UTC timezone.
    """
    return datetime.now()

def now_local_datetimezone():
    """Get current local date and time and zone"""
    print("[yellow]i[/yellow] Runnning the now_local_datetimezone() function!")
    return datetime.now().astimezone()

def now_utc_datetimezone() -> datetime:
    """Returns the current datetime in UTC.

    Return an aware timestamp using the local system time, however defaulting
    to UTC timezone.
    """
    return datetime.now(ZoneInfo("UTC"))

def random_datetimezone() -> tuple:
    """
    Generate a random datetime and timezone object
    """
    year = datetime.now().year
    month = randint(1, 12)
    _, days_in_month = monthrange(year, month)
    day = randint(1, days_in_month)
    hour = randint(0, 23)
    minute = randint(0, 59)
    second = randint(0, 59)
    datetimestamp = datetime(
        year, month, day, hour, minute, second, tzinfo=ZoneInfo("UTC")
    )
    timezone_str = choice(list(available_timezones()))
    timezone = ZoneInfo(timezone_str)

    return datetimestamp, timezone

def random_day_of_year(days_in_year: int) -> int:
    """
    Generate a random datetime and timezone object
    """
    return randint(1, days_in_year)

def attach_requested_timezone(
    timestamp: Timestamp,
    timezone: ZoneInfo,
) -> Timestamp | None:
    """
    Attaches the requested timezone to a naive datetime. Attention : Defaults to UTC if no timezone requested!
    """
    # print(f'[green]i[/green] Callback function attach_requested_timezone()')

    if timestamp.tzinfo is not None:
        raise ValueError(
            f"  [yellow]>[/yellow] The provided timestamp '{timestamp}' already has a timezone!  Expected a [yellow]naive[/yellow] [bold]datetime[/bold] or [bold]Timestamp[/bold] object."
        )

    if timezone:
        try:
            # print(f'[yellow]i[/yellow] Attaching the requested zone [bold]{timezone}[/bold] to {timestamp}')
            return timestamp.tz_localize(timezone)
        except Exception as e:
            print(
                f"[red]x[/red] Failed to attach the requested timezone '{timezone}' to the timestamp: {e}!"
            )
    else:
        zoneinfo_utc = ZoneInfo("UTC")
        print(
            f"  [yellow]i[/yellow] Timezone not requested! Setting to [red]{zoneinfo_utc}[/red]."
        )
        return timestamp.tz_localize(zoneinfo_utc)

def attach_timezone(
    timestamp: Timestamp | None = None, timezone: str | None = None
) -> datetime | None:
    """Convert datetime object to timezone-aware."""
    if timestamp is None:
        timestamp = Timestamp.now(tz=ZoneInfo("UTC"))  # Default

    if isinstance(timezone, str):
        try:
            tzinfo = generate_a_timezone(ZoneInfo(timezone))
            timestamp = timestamp.replace(tzinfo=tzinfo)
        except Exception as e:
            raise ValueError(f"Could not convert timezone: {e}")

    return timestamp

def callback_generate_a_timezone(timezone: str) -> ZoneInfo:
    """Convert string to `tzinfo` timezone object"""
    return generate_a_timezone(timezone)

def ctx_attach_requested_timezone(
    ctx: typer.Context,
    timestamp: Timestamp,
    # param: typer.CallbackParam,
) -> Timestamp | None:
    """Returns the current datetime in the user-requested timezone."""

    # print(f'[yellow]i[/yellow] Context: {ctx.params}')
    # print(f'[yellow]i[/yellow] typer.CallbackParam: {param}')
    # print(f'  [yellow]>[/yellow] Executing ctx_attach_requested_timezone()')
    timezone = ctx.params.get("timezone")
    # print(f'  [yellow]>[/yellow] User requested input parameter [code]timezone[/code] = [bold]{timezone}[/bold]')
    # print(f'  [green]>[/green] Callback function returns : {attach_requested_timezone(timestamp, timezone)}')

    return attach_requested_timezone(timestamp, ZoneInfo(timezone))

def generate_a_timezone(timezone: ZoneInfo) -> ZoneInfo:
    """ """
    context_message = f"> Executing callback function callback_generate_a_timezone()"
    context_message_alternative = f"[yellow]>[/yellow] Executing [underline]callback function[/underline] callback_generate_a_timezone()"
    logger.debug(context_message, alt=context_message_alternative)

    warning_message = warning_message_alternative = str()
    if not timezone:
        warning_message += (
            f"No timezone requested. Assuming and setting {ZONEINFO_UTC}."
        )
        warning_message_alternative += f"[red]No timezone requested.[/red] [bold yellow]Assuming and setting [code]{ZONEINFO_UTC}[/code]."
        logger.warning(warning_message, alt=warning_message_alternative)
        timezone = ZONEINFO_UTC

    if timezone == "local":
        warning_message += (
            f"Local timezone is requested. Retrieving it from the current system."
        )
        warning_message_alternative += f"[bold]Local timezone[/bold] is requested. [bold yellow]Retrieving it from the current system.[/bold yellow]"
        logger.warning(warning_message, alt=warning_message_alternative)
        try:
            timezone = ZoneInfo(Timestamp(datetime.now().astimezone()).tzinfo)

        except (ZoneInfoNotFoundError, Exception):
            logger.error(
                f"{x_mark} Requested zone {timezone} not found. Setting it to UTC.",
                alt=f"[x_mark][red]Requested zone {timezone} not found. Setting it to [code]UTC[/code][/red].",
            )
            raise ValueError("The requested time zone {timezone} is not valid!")

    context_message = f"  < Returning object : {type(timezone)} : {timezone}"
    context_message_alternative = (
        f"  [green]<[/green] Returning object : {type(timezone)} : {timezone}"
    )
    logger.debug(context_message, alt=context_message_alternative)

    return timezone

def parse_timezone(
    timezone: str,
) -> ZoneInfo | str:
    """ """
    context_message = f"> Executing parser function : parse_timestamp()"
    # context_message += f'\ni Callback parameter : {typer.CallbackParam}'
    context_message += f"\n  - Parameter input : {type(timezone)} : {timezone}"
    # context_message += f'\ni Context : {ctx.params}'

    context_message_alternative = f"[yellow]>[/yellow] Executing [underline]parser function[/underline] : parse_timezone()"
    # context_message_alternative += f'\n[yellow]i[/yellow] Callback parameter : {typer.CallbackParam}'
    context_message_alternative += (
        f"\n  - Parameter input : {type(timezone)} : {timezone}"
    )
    # context_message_alternative += f'\n[yellow]i[/yellow] Context: {ctx.params}'

    if not timezone:
        timezone = str()

    elif timezone != "local":
        timezone = ZoneInfo(timezone)

    context_message += f"\n  < Returning object : {type(timezone)} : {timezone}"
    context_message_alternative += (
        f"\n  [green]<[/green] Returning object : {type(timezone)} : {timezone}"
    )
    logger.debug(context_message, alt=context_message_alternative)

    return timezone