Source code for kappa_sdk.vector

from __future__ import annotations
import uuid
from datetime import datetime, timedelta, timezone
from typing import List, Optional, Union
from .datetime_utils import datetime_to_utc




class VectorElement:
    def __init__(self, date: datetime, value: float, elapsed_time: float):
        self.date = date
        self.value = value
        self.elapsed_time = elapsed_time





class Vector:
    """ Vector object.

    Container of the data that supports:

    * Conversion of X values between the absolute and elapsed times.
    * Conversion of Y values from the internal units to the given ones.

    .. important::
        When initializing the :class:`Vector`, Y values must be given in ``internal units``.
        This conversion can be done with the help of the :class:`UnitConverter`.

    Parameters
    ----------
    dates:
        List of X (date) values.
    values:
        List of Y (data) values.
    first_x:
        first_x of a step data
    """

    def __init__(self, dates: List[datetime], values: List[float], first_x: Optional[datetime] = None, vector_id: Optional[str] = None):
        self.__first_x: Optional[datetime] = first_x if first_x is not None else dates[0] if len(dates) > 0 else None
        self.__dates: List[datetime] = dates
        self.__elapsed_times: Optional[List[float]] = None
        self.__values: List[float] = values
        self.__id: str = vector_id if vector_id is not None else str(uuid.uuid4())

    @property
    def id(self) -> str:
        """ Gets the :class:`Vector` id.

        """
        return self.__id

    @property
    def first_x(self) -> Optional[datetime]:
        """Gets the first_x/ start time of the elapsed times of this :class:`Vector` """
        return self.__first_x

    @property
    def dates(self) -> List[datetime]:
        """ List of X (date) values in this :class:`Vector`.

        """
        return self.__dates

    @property
    def elapsed_times(self) -> List[float]:
        """ List of X (date) values as elapsed time (in hours), counted from the first date in the :class:`Vector`.
        """
        if self.__elapsed_times is None:
            self.__elapsed_times = [(Vector.__to_elapsed_time(date) - Vector.__to_elapsed_time(self.__first_x)) for date in self.dates]
        return self.__elapsed_times

    @property
    def values(self) -> List[float]:
        """ List of Y (data) values in this :class:`Vector`.

        """
        return self.__values

    def set_first_x(self, first_x: datetime) -> None:
        self.__elapsed_times = None
        self.__first_x = first_x

    def set_elapsed_times(self, elapsed_times: List[float], reference_date: Optional[datetime] = None) -> None:
        """ Sets the elapsed times.

        Sets the elapsed times and recalculates absolute times based on the provided reference date.

        Parameters
        ----------
        elapsed_times:
            List of X (date) values as elapsed time (in hours)
        reference_date:
            Reference date used to calculate absolute time values. Default value is ``datetime.now()``.
        """
        reference_date = datetime_to_utc(reference_date)
        if reference_date is None:
            reference_date = datetime.now().replace(tzinfo=timezone.utc)
        self.__first_x = reference_date
        self.__elapsed_times = elapsed_times
        self.__dates = [reference_date + timedelta(hours=h) for h in self.__elapsed_times]

    @staticmethod
    def __to_elapsed_time(date: Optional[datetime]) -> float:
        date = datetime_to_utc(date)
        if date is None:
            return 0
        else:
            return ((date - datetime(1970, 1, 1).replace(tzinfo=timezone.utc)).total_seconds()) / 3600

    def __iter__(self) -> Vector:
        self._index = 0
        return self

    def __next__(self) -> VectorElement:
        if self._index < len(self.__dates):
            element = VectorElement(self.__dates[self._index], self.__values[self._index], self.elapsed_times[self._index])
            self._index += 1
            return element
        else:
            raise StopIteration

    def __getitem__(self, index: Union[int, slice]) -> Union[VectorElement, List[VectorElement]]:
        if isinstance(index, slice):
            return [VectorElement(self.__dates[i], self.__values[i], self.elapsed_times[i]) for i in range(*index.indices(len(self.__dates)))]
        else:
            return VectorElement(self.__dates[index], self.__values[index], self.elapsed_times[index])

    def __len__(self) -> int:
        return len(self.__values)