from typing import List, Optional, Tuple, Generator, Union
from datetime import datetime
from .._private.dto_converters._data_dto_converter import DataDtoConverter
from .._private._data_api import DataAPI
from ..datetime_utils import datetime_to_str
from ..unit_enum import UnitEnum
import requests
[docs]
class DataCommandService:
"""Python wrapper for a Data Command Service.
.. note:: Should not be instantiated directly.
"""
[docs]
def __init__(self, data_api: DataAPI, data_dto_converter: DataDtoConverter):
self.__write_chunk_size = 30000
self.__data_api = data_api
self.__data_dto_converter = data_dto_converter
[docs]
def remove_range(self, vector_id: str) -> None:
"""Clears the specified dataset.
Clears the KAPPA Automate data specified by the vector id.
Parameters
----------
vector_id:
The id of the KAPPA Automate vector to clear.
"""
self.__data_api.delete_vector_data(vector_id)
[docs]
def add_values(self, vector_id: str, x_vector: List[datetime], y_vector: Union[List[int], List[float], List[Optional[float]]], unit: Optional[UnitEnum] = None) -> None:
"""Populates the specified dataset.
Populates the KAPPA Automate data specified by the vector id with values of x-vector and y-vector.
Parameters
----------
vector_id:
The id of the KAPPA Automate vector.
x_vector:
The list of datetime values for the x-axis.
y_vector:
The list of float values for the y-axis.
unit:
Convert values to internal units from a specific unit.
.. note:: If the Kappa Automate vector is point data then None values will be automatically removed.
"""
def get_chunks(seq_x: List[datetime], seq_y: Union[List[int], List[float], List[Optional[float]]], size: int) -> Tuple[Generator[List[datetime], None, None], Generator[Union[List[int], List[float], List[Optional[float]]], None, None]]:
return (seq_x[pos:pos + size] for pos in range(0, len(seq_x), size)), (seq_y[pos:pos + size] for pos in range(0, len(seq_y), size))
if unit is not None:
y_vector = [self.__data_dto_converter.unit_converter.convert_to_internal(unit, value) for value in y_vector]
chunks_x, chunks_y = get_chunks(x_vector, y_vector, self.__write_chunk_size)
metadata = self.__data_api.get_metadata(vector_id)
for chunk_x, chunk_y in zip(chunks_x, chunks_y):
dto = self.__data_dto_converter.get_append_dto(metadata, chunk_x, chunk_y)
self.__data_api.append_vector(vector_id, dto)
[docs]
def set_first_x(self, vector_id: str, reference_date: Optional[datetime]) -> None:
"""Updates the reference date for the specified dataset.
Updates the reference date value for KAPPA Automate data specified by the vector id.
Parameters
----------
vector_id:
The id of the KAPPA Automate vector.
reference_date:
The new reference date.
"""
dto = {"firstX": datetime_to_str(reference_date)}
self.__data_api.update_first_x(vector_id, dto)
[docs]
def replace_data(self, vector_id: str, x_vector: List[datetime], y_vector: Union[List[int], List[float], List[Optional[float]]], first_x: Optional[datetime] = None, unit: Optional[UnitEnum] = None) -> None:
"""Replaces the content of the specified dataset.
Replaces the content of KAPPA Automate data specified by the vector id with values of x-vector and y-vector.
Parameters
----------
vector_id:
The id of the KAPPA Automate vector.
x_vector:
The list of datetime values for the x-axis.
y_vector:
The list of float values for the y-axis.
first_x:
The first_x of the data if is_by_step
unit:
Convert values to internal units from a specific unit.
.. note:: If the Kappa Automate vector is point data then None values will be automatically removed.
"""
if first_x is None:
vector_metadata = self.__data_api.get_metadata(vector_id)
if vector_metadata and vector_metadata.is_by_step:
first_x = vector_metadata.first_x
self.remove_range(vector_id)
if first_x is not None:
self.set_first_x(vector_id, first_x)
try:
self.add_values(vector_id, x_vector, y_vector, unit)
except requests.exceptions.InvalidJSONError:
raise requests.exceptions.InvalidJSONError("You probably have nan values in your data, convert it to None values")
[docs]
def convert_nan(self, values: Union[List[Optional[float]], List[float]]) -> List[Optional[float]]:
for i, val in enumerate(values):
if str(val).lower() == "nan":
values[i] = None # type:ignore[call-overload]
return values # type:ignore[return-value]
