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 .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
[docs]
class WellGroup:
""" Well group object.
Presents a KAPPA Automate well group information object.
"""
[docs]
def __init__(self, well_group_id: str, field_id: str, name: str, cluster_apis: ClusterAPIS, uwi: Optional[str], file_folders: List[FileFolder], pvts: List[PVT], well_group_dto_converter: WellGroupDtoConverter, data_types_catalog: FieldDataTypesCatalog,
well_properties_catalog: FieldWellPropertiesCatalog):
self.__id: str = well_group_id
self.__field_id: str = field_id
self.__name: str = name
self.__cluster_apis: ClusterAPIS = cluster_apis
self.__uwi: Optional[str] = uwi
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
@property
def id(self) -> str:
""" Gets the id of the :class:`WellGroup` object.
"""
return self.__id
@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 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
[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.
"""
if file_folder_id is not None:
file_folder = find_file_folder_recursively_by_id(file_folder_id, self.__file_folders)
if file_folder is None:
raise ValueError(f"Missing File folder {file_folder_id} in well group {self.__name}")
file_dto = self.__cluster_apis.field_api.upload_file_to_file_folder_in_field(self.__field_id, file_folder.id, file_path, overwrite)
else:
file_dto = self.__cluster_apis.field_api.upload_file_to_well_group(self.__field_id, self.__id, file_path, overwrite)
file = self.__dto_converter.file_dto_converter.build_file_from_file_dto(self.__id, 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.__field_id, self.__id, self.__cluster_apis.field_api.create_well(self.__field_id, self.__id, payload), self.__data_type_catalog, self.__well_properties_catalog)
[docs]
def create_pvt_from_file(self, pvt_name: str, file_id: str, 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.
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.
"""
dto = self.__dto_converter.well_dto_converter.get_command_pvt_from_text_file_dto(pvt_name, self.__field_id, None, file_id, start_date, reservoir_pressure, reservoir_temperature, gas_oil_type, unit_system)
pvt = self.__dto_converter.well_dto_converter.build_pvt(self.__field_id, None, 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.__field_id, None, document_id, analysis_id)
pvt = self.__dto_converter.well_dto_converter.build_pvt(self.__field_id, None, self.__cluster_apis.automation_api.create_pvt_from_kw_document_wellgroup(self.id, dto))
self.pvts.append(pvt)
return pvt
