from typing import Optional, List, Any, Dict, Union
from datetime import datetime, timezone
from ._private._cluster_apis import ClusterAPIS
from .ka_context import KAContext
from .vector import Vector
from .datetime_utils import str_to_datetime, datetime_to_str
from ._private.dto_converters._well_property_dto_converter import WellPropertyDtoConverter
from .field_well_properties_catalog import FieldWellPropertiesCatalog
from .unit_enum import UnitEnum
from .measure_enum import MeasureEnum
from .base_vector import BaseVector
from .well_property_type_enum import WellPropertyTypeEnum
from .non_numeric_vector import NonNumericVector
T = Optional[Union[str, bool, float, datetime]]
[docs]
class WellPropertyContainer:
""" Well property container object.
Presents a KAPPA Automate well property container object.
"""
[docs]
def __init__(self, context: KAContext, well_property_container_id: str, name: str, well_properties_catalog: FieldWellPropertiesCatalog, labels: List[str], is_master: bool, cluster_apis: ClusterAPIS, dto_converter: WellPropertyDtoConverter):
self.__context: KAContext = context
self.__id: str = well_property_container_id
self.__name: str = name
self.__well_properties_catalog: FieldWellPropertiesCatalog = well_properties_catalog
self.__labels: List[str] = labels
self.__is_master: bool = is_master
self.__cluster_apis: ClusterAPIS = cluster_apis
self.__dto_converter: WellPropertyDtoConverter = dto_converter
@property
def id(self) -> str:
""" Gets the id of the :class:`WellPropertyContainer` object.
"""
return self.__id
@property
def name(self) -> str:
""" Gets the name of the :class:`WellPropertyContainer`.
"""
return self.__name
@property
def is_master(self) -> bool:
""" Gets a value indicating whether this contains is a master one.
"""
return self.__is_master
@property
def labels(self) -> List[str]:
""" Gets the labels of the :class:`WellPropertyContainer` object.
"""
return self.__labels
[docs]
def get_well_properties(self, validity_date: Optional[datetime] = None) -> Dict[str, Any]:
""" Gets a dictionary of alias/value pairs for all well properties in this :class:`WellPropertyContainer`.
Parameters
----------
validity_date:
The validity date of values. If not specified, latest values will be returned.
"""
well_properties: Dict[str, Union[str, bool, float, datetime]] = dict()
ids = ["{}/{}".format(self.__id, x.alias) for x in self.__well_properties_catalog]
if validity_date is not None:
dto = {"ids": ids, "atX": datetime_to_str(validity_date)}
well_properties_dto = self.__cluster_apis.data_api.get_well_properties_at_date(dto)
else:
well_properties_dto = self.__cluster_apis.data_api.get_well_properties_last_value(ids)
for value in well_properties_dto:
if value.valueWithOrigin is not None and value.valueWithOrigin.value is not None:
if value.valueWithOrigin.value.type == "String":
well_properties[value.id.split('/')[1]] = str(value.valueWithOrigin.value.value)
elif value.valueWithOrigin.value.type == "Boolean":
well_properties[value.id.split('/')[1]] = bool(value.valueWithOrigin.value.value)
elif value.valueWithOrigin.value.type == "Double":
well_properties[value.id.split('/')[1]] = float(value.valueWithOrigin.value.value)
elif value.valueWithOrigin.value.type == "DateTimeOffset":
well_properties[value.id.split('/')[1]] = str_to_datetime(str(value.valueWithOrigin.value.value)) # type:ignore[assignment]
return well_properties
[docs]
def get_well_property_values(self,
well_property_alias: str,
from_time: Optional[datetime] = None,
to_time: Optional[datetime] = None,
count: int = -1,
last: bool = False,
unit: Optional[UnitEnum] = None) -> Vector:
""" Gets a vector of values for a given well property from this :class:`WellPropertyContainer`.
Only works for numeric well properties.
Parameters
----------
well_property_alias:
The alias of the well property.
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 to a specific unit.
Returns
-------
:class:`Vector`:
Vector that contains the requested data values.
"""
alias = self.__dto_converter.get_alias(well_property_alias)
well_property = next(x for x in self.__well_properties_catalog if x.alias == alias)
if well_property.type != WellPropertyTypeEnum.numeric:
raise ValueError(f"Well property {well_property_alias} is not a numeric well property, you should use get_non_numeric_well_property_values method instead")
dates, values = self.__cluster_apis.data_api.read_vector(self.__id + '/' + alias, from_time, to_time, count, last)
if unit is not None:
values = [self.__dto_converter.unit_converter.convert_from_internal(unit, value) for value in values]
return Vector(dates, values, vector_id=self.__id + '/' + alias)
[docs]
def get_non_numeric_well_property_values(self,
well_property_alias: str,
from_time: Optional[datetime] = None,
to_time: Optional[datetime] = None,
count: int = -1,
last: bool = False) -> NonNumericVector:
"""
Retrieves non-numeric well property values based on the provided parameters.
This method fetches non-numeric well property values associated with the specified
well property alias. The data retrieval can be filtered by specifying a time range,
a count of values to fetch, or by retrieving the most recent values.
Parameters
----------
well_property_alias : str
The alias of the well property for which values are to be retrieved.
from_time : Optional[datetime], optional
The starting point of the time range for retrieving values (default is None,
which implies no lower boundary for the time).
to_time : Optional[datetime], optional
The ending point of the time range for retrieving values (default is None,
which implies no upper boundary for the time).
count : int, optional
The maximum number of values to be retrieved. Use -1 for no limit (default is -1).
last : bool, optional
Whether to retrieve the most recent values. If True, retrieves the most recent
data up to the specified count (default is False).
Returns
-------
NonNumericVector
A vector containing non-numeric well property values, along with their associated dates.
"""
alias = self.__dto_converter.get_alias(well_property_alias)
well_property = next(x for x in self.__well_properties_catalog if x.alias == alias)
if well_property.type == WellPropertyTypeEnum.numeric:
raise ValueError(f"Well property {well_property_alias} is a numeric well property, you should use get_well_property_values method instead")
dates, values = self.__cluster_apis.data_api.read_non_numeric_well_properties_vector(self.__id + '/' + alias, from_time, to_time, count, last)
return NonNumericVector(dates, values, vector_id=self.__id + '/' + alias)
[docs]
def delete_well_property_values(self, well_property_alias: str, from_time: Optional[datetime] = None, to_time: Optional[datetime] = None) -> None:
""" Deletes values of a given well property for a given range in this :class:`WellPropertyContainer`,
all values will be deleted if from_time and to_time are undefined.
Parameters
----------
well_property_alias:
The alias of well property to update.
from_time:
The date from where values have to be removed, can be None
to_time
The date until where values have to be removed, can be None
"""
self.__cluster_apis.data_api.delete_well_property_values(self.__id, well_property_alias, from_time, to_time)
[docs]
def set_well_property_value(self, well_property_alias: str, value: Optional[Union[str, bool, float, datetime]], timestamp: Optional[datetime] = None, is_step_at_start: bool = True, first_x: Optional[datetime] = None) -> None:
"""
Sets a value for a well property.
This method sets the given value for a specified well property. If the property does
not exist in the current set of well properties, it will be created using additional
parameters. Otherwise, it updates the well property values based on the provided
arguments.
Parameters
----------
well_property_alias : str
Alias of the well property for which the value is to be set.
value : Optional[Union[str, bool, float]]
Value to be set for the well property. The format and type
of the value depend on the specific property.
timestamp : Optional[datetime], optional
Timestamp at which the value is to be recorded. If not provided,
the current UTC time will be used, by default None.
is_step_at_start : bool, optional
Determines whether the step is at the start or end when recording
the value, by default True.
first_x : Optional[datetime], optional
Indicates the starting timestamp of the time series for the property,
used when the well property does not yet exist, by default None, only used when you are in step at end.
"""
if timestamp is None:
timestamp = datetime.now(timezone.utc)
if well_property_alias not in self.get_well_properties().keys():
try:
well_property = next(x for x in self.__well_properties_catalog if x.alias == well_property_alias)
except StopIteration:
raise ValueError(f"There is no well property with the alias {well_property_alias} in the field well properties catalog")
measure = str(well_property.measure.value) if isinstance(well_property.measure, MeasureEnum) else well_property.measure
creation_dto = self.__dto_converter.get_well_property_creation_dto(self.__context, [timestamp], [value], well_property.name, measure, is_step_at_start, first_x, well_property.type)
self.__cluster_apis.data_api.create_well_property(self.__id, well_property_alias, creation_dto)
else:
dto = self.__dto_converter.get_add_well_properties_dto([timestamp], [value])
self.__cluster_apis.data_api.add_well_property_values(self.__id, well_property_alias, dto)
[docs]
def add_well_property_values(self, well_property_alias: str, vector: BaseVector[T], is_step_at_start: bool = True, first_x: Optional[datetime] = None) -> None:
"""
Adds well property values for a specified well property alias, creating the well property
if it does not already exist and attaching the provided data to it.
Only works for numeric well properties.
Parameters
----------
well_property_alias : str
The alias of the well property to which the values will be added. If the well property
does not already exist, it will be created using the provided data.
vector : BaseVector
The vector containing the values to be added.
is_step_at_start : bool, optional
Indicates whether the values should be treated as step changes starting from the
given date. Default is True.
first_x : datetime, optional
The starting date for the well property. This is used if the property needs to be
created and applies only in that context.
Returns
-------
None
This method does not return any value. It performs the operation of creating or
updating the well property values in the backend system.
"""
if well_property_alias not in self.get_well_properties().keys():
well_property = next(x for x in self.__well_properties_catalog if x.alias == well_property_alias)
measure = str(well_property.measure.value) if isinstance(well_property.measure, MeasureEnum) else well_property.measure
creation_dto = self.__dto_converter.get_well_property_creation_dto(self.__context, vector.dates, vector.values, well_property.name, measure, is_step_at_start, first_x, well_property.type)
self.__cluster_apis.data_api.create_well_property(self.__id, well_property_alias, creation_dto)
else:
dto = self.__dto_converter.get_add_well_properties_dto(vector.dates, vector.values)
self.__cluster_apis.data_api.add_well_property_values(self.__id, well_property_alias, dto)
