from ._private.dto_converters._production_folder_dto_converter import ProductionFolderDtoConverter
from .data import Data
from typing import List, Optional, cast
from .unit_enum import UnitEnum
from .time_format_enum import TimeFormatEnum
from datetime import datetime
from .measure_enum import MeasureEnum
from ._private._cluster_apis import ClusterAPIS
from .production_folder_kind_enum import ProductionFolderKindEnum
[docs]
class ProductionFolder:
[docs]
def __init__(self, field_id: str, well_id: str, id: str, name: str, phases: List[Data], production_folder_type: ProductionFolderKindEnum, dto_converter: ProductionFolderDtoConverter, cluster_apis: ClusterAPIS):
self.__field_id = field_id
self.__well_id = well_id
self.__id: str = id
self.__name: str = name
self.__data: List[Data] = phases
self.__production_folder_type: ProductionFolderKindEnum = production_folder_type
self.__dto_converter: ProductionFolderDtoConverter = dto_converter
self.__cluster_apis: ClusterAPIS = cluster_apis
@property
def id(self) -> str:
"""
The id of the production folder
Returns
-------
str: id of the production folder
"""
return self.__id
@property
def name(self) -> str:
"""
The name of the production folder
Returns
-------
str: name of the production folder
"""
return self.__name
@property
def oil(self) -> Optional[Data]:
"""
The oil data contains in the production folder, returns None if there is no production gauge available
Returns
-------
Optional[Data]: Oil data
"""
return next((x for x in self.__data if x.data_type == "qo"), None)
@property
def gas(self) -> Optional[Data]:
"""
The gas data contains in the raw production folder, returns None if there is no production gauge available
Returns
-------
Optional[Data]: Gas data
"""
return next((x for x in self.__data if x.data_type == "qg"), None)
@property
def water(self) -> Optional[Data]:
"""
The water data contains in the raw production folder, returns None if there is no production gauge available
Returns
-------
Optional[Data]: water Data
"""
return next((x for x in self.__data if x.data_type == "qw"), None)
@property
def data(self) -> List[Data]:
""" Gets the list of data contained in this :class:`ProductionFolder`.
"""
return self.__data
@property
def type(self) -> ProductionFolderKindEnum:
""" Get the type of Production Folder"""
return self.__production_folder_type
[docs]
def load_production_gauge(self, datasource_name: str, datasource_gauge_name: str, dimension: MeasureEnum, data_type: str,
unit: UnitEnum, time_format: TimeFormatEnum, gauge_name: Optional[str] = None, children_datasource_names: Optional[List[str]] = None,
last_step_duration_hours: Optional[float] = None,
is_high_frequency: bool = True, gauge_model: str = "", measure_depth: float = 0, true_vertical_depth: float = 0, true_vertical_depth_sub_sea: float = 0, serial_number: str = "", labels: Optional[List[str]] = None,
comment: str = "", read_from: Optional[datetime] = None, read_to: Optional[datetime] = None) -> Data:
"""
Load a production gauge.
Parameters
----------
datasource_name:
name of the datasource
datasource_gauge_name:
name of the gauge in the datasource
dimension:
Measure of the gauge
data_type:
Type of the gauge data
unit:
Unit to load the gauge data
time_format:
time format of the gauge data, could be points, time at start or time at end
gauge_name:
Name of the gauge that will be created under the well, if None then it will use the name of the gauge under the datasource
children_datasource_names:
Names of the different sub-datasources, could be None if there is no sub-datasources
last_step_duration_hours:
Last step duration in hours
is_high_frequency:
Enable it if you have high frequency data
gauge_model:
Type of the gauge
measure_depth:
Depth of the gauge, must be converted to internal units, use convert_to_internal method from :class:`kappa_sdk.UnitConverter` to do so
true_vertical_depth:
TVD of the gauge, must be converted to internal units, use convert_to_internal method from :class:`kappa_sdk.UnitConverter` to do so
true_vertical_depth_sub_sea:
TVDss of the gauge, must be converted to internal units, use convert_to_internal method from :class:`kappa_sdk.UnitConverter` to do so
serial_number:
Serial number of the gauge
labels:
Add custom labels to the gauge
comment:
Add any comment to the gauge description
read_from:
start date to read the data
read_to:
end data to read the data
"""
if self.__production_folder_type != ProductionFolderKindEnum.production:
raise ValueError("Production Folder is not raw production, cannot load a production gauge in a Corrected production Folder")
if gauge_name is None:
gauge_name = datasource_gauge_name
datasource_id, tag_id = self.__cluster_apis.external_data_api.get_datasource_id_and_tag_id(datasource_name, datasource_gauge_name, children_datasource_names)
payload = self.__dto_converter.data_dto_converter.get_gauge_dto(self.__field_id, self.__well_id, datasource_id, tag_id, gauge_name, dimension, data_type,
unit,
time_format, last_step_duration_hours, is_high_frequency, gauge_model, measure_depth, true_vertical_depth, true_vertical_depth_sub_sea,
serial_number, labels, comment, True, read_from, read_to)
production_folder_dto = self.__cluster_apis.automation_api.load_production_gauge(self.__id, payload)
self.__data = [self.__dto_converter.build_data(self.__field_id, self.__well_id, y.data) for y in production_folder_dto.phases if y.data is not None]
if data_type == "qo":
return cast(Data, self.oil)
elif data_type == "qg":
return cast(Data, self.gas)
else:
return cast(Data, self.water)
