Source code for kappa_sdk.user_tasks.data_query_service
from datetime import datetime
from .. import UnitConverter
from .._private._data_api import DataAPI
from ..vector import Vector
from ..metadata import Metadata
from ..unit_enum import UnitEnum
from typing import Optional, cast
[docs]
class DataQueryService:
"""Python wrapper for a Data Query Service.
.. note:: Should not be instantiated directly.
"""
[docs]
def __init__(self, data_api: DataAPI, unit_converter: UnitConverter):
self.__data_api = data_api
self.__max_read_size = 100000
self.__unit_converter = unit_converter
[docs]
def read_vector(self, vector_id: str, from_time: Optional[datetime] = None, to_time: Optional[datetime] = None, count: int = -1,
last: bool = False, unit: Optional[UnitEnum] = None) -> Vector:
""" Reads the data values as :class:`Vector`.
All parameters are optional, if none are given it will read all values.
Parameters
----------
vector_id:
The id of the KAPPA Automate vector.
from_time:
Date to start reading from.
to_time:
Date to read the data up to.
count:
Maximum count of points to return, regardless of from/to settings.
last:
Will return last (count) of points if set to true.
unit:
Convert values from internal units from a specific unit.
Returns
-------
:class:`Vector`:
Vector that contains the requested data values.
"""
metadata = self.__data_api.get_metadata(vector_id)
dates, values = self.__data_api.read_vector(vector_id, from_time, to_time, count, last)
if unit is not None:
values = [cast(float, self.__unit_converter.convert_to_internal(unit, value)) for value in values]
if metadata.is_by_step:
return Vector(dates, values, metadata.first_x, vector_id)
else:
return Vector(dates, values, vector_id=vector_id)
