Source code for kappa_sdk.shutin

from datetime import datetime, timedelta
from ._private._kw_api import KWAPI
from ._private._well_dto import LogLogPreviewDto, EmbeddedChannelDto, DoubleValuesDto
from typing import Optional, List, cast
from .document_vector import DocumentVector
from .datetime_utils import datetime_to_str




class ShutIn:
    """ Class to store any shutin """



    def __init__(self, start_date: datetime, end_date: datetime, is_reference: bool, pressure_id: Optional[str], fit_for_use: str, shut_in_type: str, kw_api: KWAPI):
        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.__fit_for_use: Optional[bool] = None if fit_for_use == "Unknown" else fit_for_use == "FitForUse"
        self.__hard_shutin: Optional[bool] = None if shut_in_type == "Unknown" else shut_in_type == "Hard"


    @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 self.__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 self.__hard_shutin

    @property
    def duration(self) -> timedelta:
        """ Returns the duration of the :class:`ShutIn` object"""
        return self.__end_date - self.__start_date



    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