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 .field_well_properties_catalog import FieldWellPropertiesCatalog
from .unit_enum import UnitEnum
from .measure_enum import MeasureEnum
[docs]
class WellPropertyContainer:
""" Well property container object.
Presents a KAPPA Automate well property container object.
"""
[docs]
def __init__(self, field_id: str, well_id: str, well_property_container_id: str, name: str, well_properties_catalog: FieldWellPropertiesCatalog, 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.__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`.
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)
[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]], 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.__field_id, self.__well_id, [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, dates: List[datetime], values: List[float], is_step_at_start: bool = True, first_x: Optional[datetime] = None) -> None:
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.__field_id, self.__well_id, dates, 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(dates, values)
self.__cluster_apis.data_api.add_well_property_values(self.__id, well_property_alias, dto)
