from typing import Optional, Dict, List, Any
from ._private._cluster_apis import ClusterAPIS
from ._private.dto_converters._well_group_dto_converter import WellGroupDtoConverter
from .file import File
from .document import Document
from .pvt.pvt import PVT
from .well_production_enum import WellProductionTypeEnum
from .well import Well
from .field_data_types_catalog import FieldDataTypesCatalog
from .field_well_properties_catalog import FieldWellPropertiesCatalog
from .file_folder import FileFolder
from .file_folders_extensions import find_file_folder_recursively_by_id
from datetime import datetime
from .gas_oil_type_enum import GasOilTypeEnum
from .unit_system_pvt_enum import UnitSystemPvtEnum
from .user_task import UserTask
from .ka_context import KAContext
from .pvt_format_enum import PVTFormatEnum
[docs]
class WellGroup:
""" Well group object.
Presents a KAPPA Automate well group information object.
"""
[docs]
def __init__(self, context: KAContext, name: str, uwi: Optional[str], datum: Optional[float], wells: List[Well], file_folders: List[FileFolder], pvts: List[PVT], files: List[File], user_tasks: List[UserTask], well_group_dto_converter: WellGroupDtoConverter, data_types_catalog: FieldDataTypesCatalog,
well_properties_catalog: FieldWellPropertiesCatalog, cluster_apis: ClusterAPIS):
self.__context: KAContext = context
self.__name: str = name
self.__uwi: Optional[str] = uwi
self.__datum: Optional[float] = None
self.__wells: List[Well] = wells
self.__dto_converter: WellGroupDtoConverter = well_group_dto_converter
self.__file_folders: List[FileFolder] = file_folders
self.__data_type_catalog: FieldDataTypesCatalog = data_types_catalog
self.__well_properties_catalog: FieldWellPropertiesCatalog = well_properties_catalog
self.__pvts: List[PVT] = pvts
self.__files: List[File] = files
self.__user_tasks: List[UserTask] = user_tasks
self.__cluster_apis: ClusterAPIS = cluster_apis
@property
def id(self) -> str:
""" Gets the id of the :class:`WellGroup` object."""
if self.__context.well_group_id is None:
raise ValueError("Well group id is missing in the context")
return self.__context.well_group_id
@property
def context(self) -> KAContext:
""" Get the context of the :class:`WellGroup` object."""
return self.__context
@property
def name(self) -> str:
""" Gets the name of the :class:`WellGroup`."""
return self.__name
@property
def uwi(self) -> Optional[str]:
""" Gets the UWI of the :class:`WellGroup` object."""
return self.__uwi
@property
def datum(self) -> Optional[float]:
""" Gets the Datum of the :class:`WellGroup` object."""
return self.__datum
@property
def wells(self) -> List[Well]:
""" Gets the list of all wells contained in the :class:`WellGroup`, including contained in the sub-well groups."""
return self.__wells
@property
def file_folders(self) -> List[FileFolder]:
""" Gets the list of file folders contained in this :class:`WellGroup`.
.. note:: This property is populated on-demand and is cached for the duration of the :class:`Connection`.
"""
return self.__file_folders
@property
def pvts(self) -> List[PVT]:
""" Gets the list of PVTs contained in this :class:`WellGroup`.
.. note:: This property is populated on-demand and is cached for the duration of the :class:`Connection`.
"""
return self.__pvts
@property
def files(self) -> List[File]:
""" Gets the list of files contained in this :class:`WellGroup`."""
return self.__files
@property
def documents(self) -> List[Document]:
""" Gets the list of KW documents contained in this :class:`Field`."""
document_list = list()
for file in self.files:
try:
document = file.as_kw_document()
except ValueError:
document = None
if document is not None:
document_list.append(document)
return document_list
@property
def user_tasks(self) -> List[UserTask]:
""" Gets the list of user tasks contained in this :class:`WellGroup`."""
return self.__user_tasks
[docs]
def upload_file(self, file_path: str, file_folder_id: Optional[str] = None, overwrite: bool = False) -> File:
""" Uploads a file to this :class:`WellGroup`.
Parameters
----------
file_path:
Full path and name of the file to upload.
overwrite:
A value indicating whether to overwrite a file with the same name if it already exists in the well group.
file_folder_id:
Id of the file folder to upload the file
Returns
-------
:class:`File`:
An uploaded file object.
"""
file = next((x for x in self.files if x.name == file_path.split("/")[-1]), None) if overwrite else None
file_folder = find_file_folder_recursively_by_id(file_folder_id, self.file_folders) if file_folder_id is not None else None
if file_folder is None and file_folder_id is not None:
raise ValueError(f"Missing File folder {file_folder_id} in well group {self.__name}")
if file is not None:
file_dto = self.__cluster_apis.field_api.update_file(self.__context, file.file_id, file_path)
elif file_folder is not None:
file_dto = self.__cluster_apis.field_api.upload_file_to_file_folder_in_field(self.__context, file_folder.id, file_path)
else:
file_dto = self.__cluster_apis.field_api.upload_file_to_field(self.__context, file_path)
file = self.__dto_converter.file_dto_converter.build_file_from_file_dto(self.__context, file_dto)
return file
[docs]
def create_well(self, name: str, uwi: Optional[str] = None, comment: Optional[str] = None, production_type: WellProductionTypeEnum = WellProductionTypeEnum.unknown, labels: Optional[List[str]] = None,
well_properties_values: Optional[List[Dict[str, Any]]] = None) -> Well:
"""
Create a new well under the field associated to this :class:`WellGroup`
Parameters
----------
name:
Name of the new well
uwi:
Unique well identifier of the new well
comment:
Any description
production_type:
Production type of the new well, unknow by default
labels:
Labels of the new well
well_properties_values:
You can fill the well properties values, it has to be a dictionary following this format {'alias_of_the_well_property':value}
Returns
-------
:class:`Well`:
The new well
"""
payload = self.__dto_converter.get_create_well_payload(name, uwi, comment, production_type, labels, well_properties_values)
return self.__dto_converter.build_well(self.__context, self.__cluster_apis.field_api.create_well(self.__context, payload), self.__data_type_catalog, self.__well_properties_catalog)
[docs]
def create_pvt_from_file(self, pvt_name: str, file_id: str, pvt_format: PVTFormatEnum, start_date: Optional[datetime] = None,
reservoir_pressure: Optional[float] = None, reservoir_temperature: Optional[float] = None,
gas_oil_type: Optional[GasOilTypeEnum] = None, unit_system: Optional[UnitSystemPvtEnum] = None) -> PVT:
"""
Creates a PVT (Pressure-Volume-Temperature) object from a file. You can define fallback parameters when the gas oil type is undetermined.
Parameters
----------
pvt_name : str
The name of the PVT object to be created.
file_id : str
The identifier of the file from which the PVT object will be created.
pvt_format: PVTFormatEnum
Format of the PVT document
start_date : datetime, optional
The start date for the PVT data coverage. Defaults to None.
reservoir_pressure : float, optional
The pressure of the reservoir associated with the PVT object. Defaults to None.
reservoir_temperature : float, optional
The temperature of the reservoir associated with the PVT object. Defaults to None.
gas_oil_type : GasOilTypeEnum, optional
The type of gas or oil associated with the PVT object, as per the enumerated
GasOilTypeEnum. Defaults to None.
unit_system : UnitSystemPvtEnum, optional
The unit system used for the PVT object, as per the enumerated UnitSystemPvtEnum.
Defaults to None.
Returns
-------
PVT
An instance of the PVT object created using the provided parameters and data from
the specified text file.
"""
file = next(x for x in self.files if x.file_id == file_id)
dto = self.__dto_converter.well_dto_converter.get_command_pvt_from_text_file_dto(pvt_name, self.__context, file, pvt_format, start_date, reservoir_pressure, reservoir_temperature, gas_oil_type, unit_system)
pvt = self.__dto_converter.well_dto_converter.build_pvt(self.__context, self.__cluster_apis.automation_api.create_pvt_from_text_file_wellgroup(self.id, dto))
self.pvts.append(pvt)
return pvt
[docs]
def create_pvt_from_kw_document(self, pvt_name: str, document_id: str, analysis_id: str) -> PVT:
"""
Create a PVT object in the well group from a KW document
Parameters
----------
pvt_name: str
Name of the PVT object to create
document_id: str
Id of the document to use
analysis_id: str
Id of the analysis to use
Returns
-------
PVT
The PVT object created.
"""
dto = self.__dto_converter.well_dto_converter.get_command_pvt_from_kw_document_dto(pvt_name, self.__context, document_id, analysis_id)
pvt = self.__dto_converter.well_dto_converter.build_pvt(self.__context, self.__cluster_apis.automation_api.create_pvt_from_kw_document_wellgroup(self.id, dto))
self.pvts.append(pvt)
return pvt
