from typing import List, Optional, Dict, Any, Literal
from ._private._field_dto import FieldDtoWithHierarchy
from .well_production_enum import WellProductionTypeEnum
from .well import Well
from .well_group import WellGroup
from .file import File
from .document import Document
from ._private._cluster_apis import ClusterAPIS
from datetime import datetime
from .field_data_types_catalog import FieldDataTypesCatalog
from .field_well_properties_catalog import FieldWellPropertiesCatalog
from ._private.dto_converters._field_dto_converter import FieldDtoConverter
from .file_folder import FileFolder
from .file_folders_extensions import find_file_folder_recursively_by_id
from .pvt.pvt import PVT
from .gas_oil_type_enum import GasOilTypeEnum
from .unit_system_pvt_enum import UnitSystemPvtEnum
from .artm_model_enum import ARTMModelEnum
from .artm_improve_target_enum import ARTMImproveTargetEnum
from .artm_regression_algorithm_enum import ARTMRegressionAlgorithmEnum
from .artm_forecast_type_enum import ARTMForecastTypeEnum
[docs]
class Field:
""" Field object.
Presents a KAPPA Automate field object that can be queried for contained wells.
Returned as a result of the :meth:`Connection.get_fields` query.
.. note:: Should not be instantiated directly.
.. note:: :py:obj:`Field.wells` property is populated on-demand and is cached for the duration of the :class:`Connection`.
"""
def __init__(self, field_id: str, name: str, asset: str, field_dto: Optional[FieldDtoWithHierarchy], cluster_apis: ClusterAPIS, dto_converter: FieldDtoConverter) -> None:
self.__id: str = field_id
self.__name: str = name
self.__asset: str = asset
self.__reference_date: Optional[datetime] = None
self.__field_dto: Optional[FieldDtoWithHierarchy] = field_dto
self.__wells: Optional[List[Well]] = None
self.__well_groups: Optional[List[WellGroup]] = None
self.__files: Optional[List[File]] = None
self.__documents: Optional[List[Document]] = None
self.__cluster_apis: ClusterAPIS = cluster_apis
self.__dto_converter: FieldDtoConverter = dto_converter
self.__data_types_catalog: Optional[FieldDataTypesCatalog] = None
self.__well_properties_catalog: Optional[FieldWellPropertiesCatalog] = None
self.__file_folders: Optional[List[FileFolder]] = None
self.__pvts: Optional[List[PVT]] = None
def __get_field_dto(self) -> FieldDtoWithHierarchy:
if self.__field_dto is None:
self.__field_dto = self.__cluster_apis.field_api.get_field_dto(self.__id)
return self.__field_dto
@property
def id(self) -> str:
""" Gets the id of the :class:`Field` object.
"""
return self.__id
@property
def name(self) -> str:
""" Gets the name of the :class:`Field`.
"""
return self.__name
@property
def asset(self) -> str:
""" Gets the asset of the :class:`Field`.
"""
return self.__asset
@property
def data_types_catalog(self) -> FieldDataTypesCatalog:
"""Get the Field Data Types catalog object of the :class:`Field`"""
if self.__data_types_catalog is None:
self.__data_types_catalog = self.__dto_converter.get_field_data_types_catalog_from_data_type_catalog_dto(self.__id, self.__get_field_dto().dataTypeCatalog)
return self.__data_types_catalog
@property
def well_properties_catalog(self) -> FieldWellPropertiesCatalog:
"""Get the Field Well Properties catalog object of the :class:`Field`"""
if self.__well_properties_catalog is None:
self.__well_properties_catalog = self.__dto_converter.get_field_well_properties_catalog_from_well_property_catalog_dto(self.__id, self.__get_field_dto().wellPropertyCatalog)
return self.__well_properties_catalog
@property
def reference_date(self) -> datetime:
""" Gets the reference date of the :class:`Field`.
"""
if self.__reference_date is None:
self.__reference_date = self.__get_field_dto().referenceDate
return self.__reference_date
@property
def wells(self) -> List[Well]:
""" Gets the list of all wells contained in the :class:`Field`, including contained in the well groups.
.. note:: This property is populated on-demand and is cached for the duration of the :class:`Connection`.
"""
if self.__wells is None:
self.__wells = self.__dto_converter.get_wells_by_field_from_field_dto(self.id, self.__get_field_dto(), self.data_types_catalog, self.well_properties_catalog)
return self.__wells
@property
def well_groups(self) -> List[WellGroup]:
""" Gets the list of all well groups contained in the :class:`Field`.
.. note:: This property is populated on-demand and is cached for the duration of the :class:`Connection`.
"""
if self.__well_groups is None:
self.__well_groups = self.__dto_converter.get_well_groups_recursively(self.id, self.__get_field_dto(), self.data_types_catalog, self.well_properties_catalog)
return self.__well_groups if self.__well_groups is not None else list()
@property
def files(self) -> List[File]:
""" Gets the list of files contained in this :class:`Field` and its well groups.
.. note:: This property is populated on-demand and is cached for the duration of the :class:`Connection`.
"""
if self.__files is None:
self.__files = self.__dto_converter.well_dto_converter.get_files_recursively(self.__id, None, self.__get_field_dto())
return self.__files
@property
def file_folders(self) -> List[FileFolder]:
""" Gets the list of file folders contained in this :class:`Well`.
.. note:: This property is populated on-demand and is cached for the duration of the :class:`Connection`.
"""
if self.__file_folders is None:
self.__file_folders = self.__dto_converter.well_dto_converter.file_dto_converter.get_file_folders_from_file_folder_dto_recursively(self.__id, None, None, None, None, self.__get_field_dto().fileFolders)
return self.__file_folders
@property
def documents(self) -> List[Document]:
""" Gets the list of KW documents contained in this :class:`Field` and its well groups.
.. note:: This property is populated on-demand and is cached for the duration of the :class:`Connection`.
"""
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
def upload_file(self, file_path: str, file_folder_id: Optional[str] = None, overwrite: bool = False) -> File:
""" Uploads a file to this :class:`Field`.
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 field.
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 field {self.__name}")
file_dto = self.__cluster_apis.field_api.upload_file_to_file_folder_in_field(self.__id, file_folder.id, file_path, overwrite)
else:
file_dto = self.__cluster_apis.field_api.upload_file_to_field(self.__id, file_path, overwrite)
file = self.__dto_converter.well_dto_converter.file_dto_converter.build_file_from_file_dto(None, file_dto)
self.files.append(file)
return file
def refresh_data(self) -> None:
"""
Clean all attributes and dto, to grab updated attributes.
"""
self.__field_dto = None
self.__data_types_catalog = None
self.__well_properties_catalog = None
self.__reference_date = None
self.__wells = None
self.__well_groups = None
self.__files = None
self.__pvts = None
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:`Field`
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, unknown 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.well_group_dto_converter.get_create_well_payload(name, uwi, comment, production_type, labels, well_properties_values)
well = self.__dto_converter.well_group_dto_converter.build_well(self.id, None, self.__cluster_apis.field_api.create_well(self.__id, None, payload), self.data_types_catalog, self.well_properties_catalog)
if self.__wells is not None:
self.__wells.append(well)
return well
def rename(self, new_name: str) -> None:
""" Rename the field
Parameters
----------
new_name:
New name of the field
"""
self.__cluster_apis.field_api.rename_field(self.__id, {"name": new_name})
self.__name = new_name
def delete(self) -> None:
""" Delete the field"""
self.__cluster_apis.field_api.delete_field(self.__id)
def __str__(self) -> str:
"""String representation of the Field."""
return f"Field(name='{self.name}', id='{self.id}')"
def __repr__(self) -> str:
"""Detailed representation of the Field."""
return f"Field(id='{self.id}', name='{self.name}', asset='{self.asset}')"
@property
def pvts(self) -> List[PVT]:
""" Gets the list of PVTs contained in this :class:`Field`.
.. note:: This property is populated on-demand and is cached for the duration of the :class:`Connection`.
"""
if self.__pvts is None:
self.__pvts = self.__dto_converter.well_dto_converter.get_pvts_from_pvts_dto(self.__id, None, self.__get_field_dto().pvts)
return self.__pvts
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.__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.__id, None, self.__cluster_apis.automation_api.create_pvt_from_text_file_field(self.id, dto))
self.pvts.append(pvt)
return pvt
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.__id, None, document_id, analysis_id)
pvt = self.__dto_converter.well_dto_converter.build_pvt(self.__id, None, self.__cluster_apis.automation_api.create_pvt_from_kw_document_field(self.id, dto))
self.pvts.append(pvt)
return pvt
def create_automatic_rtm_aggregator(self, pvt_document: Document, name: str = "Automatic RTM Agg #1", labels: Optional[List[str]] = None, use_black_oil_pvt: bool = False, pressure_data_type: str = "BHP", pressure_label: Optional[List[str]] = None, use_oil_rate: Optional[bool] = None, use_gas_rate: Optional[bool] = None, use_water_rate: Optional[bool] = None,
wells: Optional[List[Well]] = None, power_law_segments: Literal[1, 2] = 1,
artm_model: ARTMModelEnum = ARTMModelEnum.mzfd, improve_target: ARTMImproveTargetEnum = ARTMImproveTargetEnum.cumulative, regression_algorithm: ARTMRegressionAlgorithmEnum = ARTMRegressionAlgorithmEnum.genetic_algorithm, replicate_results_to_master_container: bool = False, use_equivalent_rate: bool = False,
forecast_duration_hours: float = 2160, forecast_type: ARTMForecastTypeEnum = ARTMForecastTypeEnum.constant_pressure, forecast_value: Optional[float] = None, mandatory_well_properties: Optional[List[str]] = None) -> None:
if wells is None:
wells = self.wells
if len(wells) == 0:
raise ValueError("No well in the field to create the RTM aggregator")
artm_parameters = pvt_document.get_artm_parameters()
artm_creation_dto = self.__dto_converter.get_artm_aggretor_creation_dto(self.__id, name, [x.id for x in wells], pvt_document.file_id, artm_parameters, use_black_oil_pvt, pressure_data_type, pressure_label, use_oil_rate, use_gas_rate, use_water_rate, power_law_segments, artm_model, improve_target, regression_algorithm,
replicate_results_to_master_container, use_equivalent_rate, forecast_duration_hours, forecast_type, forecast_value, mandatory_well_properties, labels)
self.__cluster_apis.automation_api.create_automatic_rtm_aggregator_field(self.__id, artm_creation_dto)
