from datetime import datetime, timedelta
from ._private._kw_api import KWAPI
from ._private._data_api import DataAPI
from ._private._well_dto import LogLogPreviewDto, EmbeddedChannelDto, DoubleValuesDto
from ._private._data_dto import ShutInCategory, ShutInType, ShutInAttributesUpdateDto, ShutInIntervalAttributesUpdateDto, ShutInVectorAttributesUpdateDto
from typing import Optional, List, cast
from .document_vector import DocumentVector
from .datetime_utils import datetime_to_str
from .shut_in_types_enum import ShutInTypesEnum
from .shut_in_categories_enum import ShutInCategoriesEnum
[docs]
class ShutIn:
""" Class to store any shutin """
[docs]
def __init__(self, start_date: datetime, end_date: datetime, is_reference: bool, pressure_id: Optional[str], shut_in_category: ShutInCategoriesEnum,
shut_in_type: ShutInTypesEnum, rate_multiplier: float, comment: Optional[str], kw_api: KWAPI,
vector_id: Optional[str] = None, data_api: Optional[DataAPI] = None):
self.__start_date: datetime = start_date
self.__end_date: datetime = end_date
self.__is_reference: bool = is_reference
self.__pressure_id: Optional[str] = pressure_id
self.__kw_api: KWAPI = kw_api
self.__shut_in_category: ShutInCategoriesEnum = shut_in_category
self.__shut_in_type: ShutInTypesEnum = shut_in_type
self.__rate_multiplier: float = rate_multiplier
self.__comment: Optional[str] = comment
self.__vector_id: Optional[str] = vector_id
self.__data_api: Optional[DataAPI] = data_api
@property
def start_date(self) -> datetime:
""" Returns the start date of the :class:`ShutIn` object"""
return self.__start_date
@property
def end_date(self) -> datetime:
""" Returns the end date of the :class:`ShutIn` object"""
return self.__end_date
@property
def is_reference(self) -> bool:
""" Returns whether the ShutIn is reference or not"""
return self.__is_reference
@property
def fit_for_use(self) -> Optional[bool]:
""" Returns whether the ShutIn is fit for use or not, returns None if the category is Unknown"""
return None if self.__shut_in_category == ShutInCategoriesEnum.unknown else self.__shut_in_category == ShutInCategoriesEnum.fit_for_use
@property
def hard_shutin(self) -> Optional[bool]:
"""Returns whether the ShutIn is an hard shutin or not, returns None if the type is Unknown"""
return None if self.__shut_in_type == ShutInTypesEnum.unknown else self.__shut_in_type == ShutInTypesEnum.hard
@property
def duration(self) -> timedelta:
""" Returns the duration of the :class:`ShutIn` object"""
return self.__end_date - self.__start_date
@property
def rate_multiplier(self) -> float:
""" Returns the rate multiplier of the :class:`ShutIn` object"""
return self.__rate_multiplier
@property
def comment(self) -> Optional[str]:
""" Returns the comment of the :class:`ShutIn` object"""
return self.__comment
@property
def shut_in_category(self) -> ShutInCategoriesEnum:
""" Returns the category of the :class:`ShutIn` object"""
return self.__shut_in_category
@property
def shut_in_type(self) -> ShutInTypesEnum:
""" Returns the type of the :class:`ShutIn` object"""
return self.__shut_in_type
[docs]
def update(self, shut_in_category: Optional[ShutInCategoriesEnum] = None, shut_in_type: Optional[ShutInTypesEnum] = None,
rate_multiplier: Optional[float] = None, comment: Optional[str] = None, is_reference: Optional[bool] = None) -> None:
""" Updates the attributes of this :class:`ShutIn` and persists the change immediately.
Only the arguments you provide are modified; the others are left unchanged. To clear an
existing comment, pass an empty string.
The :class:`ShutIn` must have been obtained through :meth:`ShutInList.from_data`.
Parameters
----------
shut_in_category:
New :class:`ShutInCategoriesEnum` (fit for use / not fit for use / unknown).
shut_in_type:
New :class:`ShutInTypesEnum` (hard / soft / unknown).
rate_multiplier:
New rate multiplier applied to the reference rate of this shut-in.
comment:
New comment. Pass an empty string to clear the existing comment.
is_reference:
Whether this shut-in should be the reference shut-in.
"""
if self.__data_api is None or self.__vector_id is None:
raise ValueError("This ShutIn cannot be edited, it must be obtained through ShutInList.from_data")
if shut_in_category is not None:
self.__shut_in_category = shut_in_category
if shut_in_type is not None:
self.__shut_in_type = shut_in_type
if rate_multiplier is not None:
self.__rate_multiplier = rate_multiplier
if comment is not None:
self.__comment = comment
if is_reference is not None:
self.__is_reference = is_reference
dto = ShutInVectorAttributesUpdateDto(attributes=[ShutInIntervalAttributesUpdateDto(
start=cast(str, datetime_to_str(self.__start_date)),
end=cast(str, datetime_to_str(self.__end_date)),
attributes=ShutInAttributesUpdateDto(
reference=self.__is_reference,
category=ShutInCategory(self.__shut_in_category.value),
type=ShutInType(self.__shut_in_type.value),
rateMultiplier=self.__rate_multiplier,
comment=self.__comment))])
self.__data_api.update_shutin_attributes(self.__vector_id, dto)
[docs]
def get_log_log_preview_from_shut_in_dates(self, pressure_id: Optional[str] = None, to_date: Optional[datetime] = None, tp: float = 10) -> List[DocumentVector]:
"""
Get Loglog preview :class:`DocumentVector` from shut-in dates
Parameters
----------
pressure_id:
Pressure vector id to use
to_date:
End date of the shut-in, if None it will use the initial end date of the shut-in
tp:
Horner Time on production
Returns
-------
List[:class:`DocumentVector`]:
List of document vectors which contains the Loglog dP values and the Bourdet derivative values
"""
to_date_str = datetime_to_str(to_date)
pressure_id = pressure_id if pressure_id is not None else self.__pressure_id
if pressure_id is None:
raise ValueError("Missing pressure vector Id to get the loglog preview")
dto = LogLogPreviewDto(pressureId=pressure_id, startDate=cast(str, datetime_to_str(self.__start_date)),
endDate=cast(str, datetime_to_str(self.__end_date)) if to_date_str is None else to_date_str, tp=tp)
plot_instance = self.__kw_api.get_log_log_preview(dto)
document_vectors = []
for channel in plot_instance.panes[0].channels:
if isinstance(channel, EmbeddedChannelDto):
document_vector = DocumentVector([], channel.yValues.values, channel.name, channel.yValues.dimension, True if "derivative" in channel.name else False, False, channel.isByStep)
document_vector.set_elapsed_times(cast(DoubleValuesDto, channel.xValues).values, self.__start_date)
document_vectors.append(document_vector)
return document_vectors
