from typing import Optional, List, Any, Dict, Union
from datetime import datetime, timezone
from ._private._cluster_apis import ClusterAPIS
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 .unit_enum import UnitEnum
[docs]
class WellPropertyContainer:
""" Well property container object.
Presents a KAPPA Automate well property container object.
"""
def __init__(self, field_id: str, well_id: str, well_property_container_id: str, name: str, aliases: List[str], labels: List[str], is_master: bool, cluster_apis: ClusterAPIS, dto_converter: WellPropertyDtoConverter):
self.__field_id: str = field_id
self.__well_id: str = well_id
self.__id: str = well_property_container_id
self.__name: str = name
self.__aliases: List[str] = aliases
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
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) for x in self.__aliases]
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
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`.
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)
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)
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)
def set_well_property_value(self, well_property_alias: str, value: Optional[Union[str, bool, float]], timestamp: Optional[datetime] = datetime.now(tz=timezone.utc)) -> None:
""" Updates a value of a given well property in this :class:`WellPropertyContainer`.
Parameters
----------
well_property_alias:
The alias of well property to update.
value:
The new value of the well property.
timestamp:
The timestamp that will be applied to the property value. If not specified, it will be set to the current date/time.
"""
dto = self.__dto_converter.get_well_property_dto(self.__id, well_property_alias, value, timestamp)
self.__cluster_apis.field_api.set_well_property_value(self.__field_id, self.__well_id, self.__id, dto)
