from typing import List, Optional, Union, cast, TYPE_CHECKING
from .pseudo_pressures_results import PseudoPressuresResults
from .._private._cluster_apis import ClusterAPIS
from ..pvt_type_enum import PVTTypeEnum
from ..pvt_fluid_type_enum import PVTFluidTypeEnum
from .._private._tech_objects_dto import EosPvtQueryDto, BlackOilPvtQueryDto
from .black_oil_flash_results import BlackOilFlashResults
from .eos_flash_results import EosFlashResults
from .black_oil_pvt_property import BlackOilPvtProperty
from .black_oil_pvt_parameters import BlackOilPvtParameters
from .eos_flash_input import EosFlashInput
from .eos_pvt_parameters import EosPvtParameters
from .rate_conversion_results import RateConversionResults
from .._private._kw_dto import AutomaticRtmParametersDto
from ..ka_context import KAContext
if TYPE_CHECKING:
from .._private.dto_converters._pvt_dto_converter import PVTDtoConverter
[docs]
class PVT:
[docs]
def __init__(self, context: KAContext, id: str, name: str, technical_object_id: str, labels: Optional[List[str]], cluster_apis: ClusterAPIS, pvt_dto_converter: 'PVTDtoConverter') -> None:
self.__context = context
self.__id = id
self.__name = name
self.__technical_object_id = technical_object_id
self.__labels = labels
self.__cluster_apis = cluster_apis
self.__pvt_dto_converter = pvt_dto_converter
self.__pvt_dto: Optional[Union[BlackOilPvtQueryDto, EosPvtQueryDto]] = None
def __get_black_oil_dto(self) -> BlackOilPvtQueryDto:
if self.type is not PVTTypeEnum.black_oil:
raise ValueError("This property is only available for black oil PVT types")
return cast(BlackOilPvtQueryDto, self.__get_pvt_dto())
@property
def field_id(self) -> str:
"""Gets the id of the field that contains this :class:`PVT`."""
return self.__context.field_id
@property
def well_id(self) -> Optional[str]:
"""Gets the id of the well associated with this :class:`PVT`."""
return self.__context.well_id
@property
def id(self) -> str:
"""Gets the unique id of this :class:`PVT`."""
return self.__id
@property
def name(self) -> str:
"""Gets the name of this :class:`PVT`."""
return self.__name
@property
def technical_object_id(self) -> str:
"""Gets the technical object id associated with this :class:`PVT`."""
return self.__technical_object_id
@property
def labels(self) -> Optional[List[str]]:
"""Gets the labels associated with this :class:`PVT`."""
return self.__labels
@property
def type(self) -> PVTTypeEnum:
"""Gets the type of this :class:`PVT`."""
return self.__get_pvt_dto().pvtType
def __get_pvt_dto(self) -> Union[BlackOilPvtQueryDto, EosPvtQueryDto]:
if self.__pvt_dto is None:
self.__pvt_dto = self.__cluster_apis.tech_objects_api.get_pvt_dto(self.__technical_object_id)
return self.__pvt_dto
@property
def properties(self) -> List[BlackOilPvtProperty]:
"""
Retrieves the properties specific to the Black Oil PVT type.
This property checks if the current PVT type is `black_oil`. If so, it
builds and returns the relevant PVT properties using the appropriate
converter. Otherwise, an error is raised indicating that the properties
are only available for the `black_oil` type.
Returns
-------
List[BlackOilPvtProperty]
A list of BlackOilPvtProperty objects representing the PVT properties
for the `black_oil` PVT type.
Raises
------
ValueError
If the current PVT type is not `black_oil`.
"""
if self.type is PVTTypeEnum.black_oil:
return self.__pvt_dto_converter.build_pvt_properties(cast(BlackOilPvtQueryDto, self.__get_pvt_dto()).properties)
else:
raise ValueError("Properties are only available for black oil PVT types")
@property
def fluid_type(self) -> PVTFluidTypeEnum:
"""
Gets the fluid type value. Only available for black oil PVT.
Returns
-------
PVTFluidTypeEnum
The fluid type value.
"""
return self.__get_black_oil_dto().fluidType
@property
def reference_phase(self) -> Optional[str]:
"""
Gets the reference phase. Only available for black oil PVT.
Returns
-------
Optional[str]
The reference phase, or None if not set.
"""
return self.__get_black_oil_dto().referencePhase
@property
def parameters(self) -> List[BlackOilPvtParameters]:
"""
Gets the time-dependent parameter sets. Only available for black oil PVT.
Returns
-------
List[BlackOilPvtParameters]
One entry per configuration start date.
"""
return self.__pvt_dto_converter.build_black_oil_pvt_parameters_list(self.__get_black_oil_dto().allParameters)
@property
def eos_parameters(self) -> Optional[EosPvtParameters]:
"""
Gets the EOS model parameters. Only available for EOS PVT.
Returns
-------
Optional[EosPvtParameters]
The EOS parameters, or None if not set.
"""
if self.type is not PVTTypeEnum.eos:
raise ValueError("eos_parameters is only available for EOS PVT types")
dto = cast(EosPvtQueryDto, self.__get_pvt_dto()).parameters
return self.__pvt_dto_converter.build_eos_pvt_parameters(dto) if dto is not None else None
@property
def flash_inputs(self) -> List[EosFlashInput]:
"""
Gets the stored flash presets. Only available for EOS PVT.
Returns
-------
List[EosFlashInput]
The stored EOS flash presets.
"""
if self.type is not PVTTypeEnum.eos:
raise ValueError("flash_inputs is only available for EOS PVT types")
return self.__pvt_dto_converter.build_eos_flash_inputs(cast(EosPvtQueryDto, self.__get_pvt_dto()).flashInputs)
[docs]
def flash_black_oil_process(self, temperature: float, pressure: float, pb_imposed: bool, gor_or_pb: Optional[float] = None) -> BlackOilFlashResults:
"""Compute a flash on black oil PVT"""
if self.type is PVTTypeEnum.black_oil:
dto = self.__pvt_dto_converter.get_flash_black_oil_pvt_command_dto(temperature, pressure, pb_imposed, gor_or_pb)
response = self.__cluster_apis.tech_objects_api.flash_black_oil_pvt(self.__technical_object_id, dto)
return self.__pvt_dto_converter.build_black_oil_flash_results(response)
else:
raise ValueError("This is not a black oil PVT")
[docs]
def flash_eos_process(self, temperature: float, pressure: float, extended: bool, water_molar_fraction: Optional[float] = None, salinity: Optional[float] = None) -> EosFlashResults:
"""Compute a flash on EoS PVT"""
if self.type is PVTTypeEnum.eos:
if (water_molar_fraction is None and salinity is None) or (water_molar_fraction is not None and salinity is not None):
dto = self.__pvt_dto_converter.get_flash_eos_pvt_command_dto(temperature, pressure, extended, water_molar_fraction, salinity)
else:
raise ValueError("You must define water molar fraction and salinity")
response = self.__cluster_apis.tech_objects_api.flash_eos_pvt(self.__technical_object_id, dto)
return self.__pvt_dto_converter.build_eos_flash_results(response)
else:
raise ValueError("This is not a EoS PVT")
[docs]
def compute_pseudo_pressures(self, temperature: float, rock_compressibility: float, pressure_range_min: float, pressure_range_max: float, pressure_range_steps: int, sw: Optional[float] = None) -> PseudoPressuresResults:
dto = self.__pvt_dto_converter.get_compute_pseudo_pressures_command_dto(temperature, rock_compressibility, sw, pressure_range_min, pressure_range_max, pressure_range_steps=pressure_range_steps)
pseudo_pressure_query_dto = self.__cluster_apis.tech_objects_api.compute_pseudo_pressures(self.__technical_object_id, dto)
return self.__pvt_dto_converter.build_pseudo_pressures_results(pseudo_pressure_query_dto)
[docs]
def convert_rates(self, field_pressures: List[List[float]], field_temperatures: List[List[float]], surface_pressures: List[List[float]], surface_temperatures: List[List[float]],
standard_oil_rates: List[float], standard_gas_rates: List[float]) -> RateConversionResults:
dto = self.__pvt_dto_converter.get_rate_conversion_command_dto(field_pressures, field_temperatures, surface_pressures, surface_temperatures, standard_oil_rates, standard_gas_rates)
rate_conversion_query_dto = self.__cluster_apis.tech_objects_api.convert_rates(self.__technical_object_id, dto)
return self.__pvt_dto_converter.build_rate_conversion_results(rate_conversion_query_dto)
[docs]
def get_artm_parameters(self) -> AutomaticRtmParametersDto:
dto = self.__pvt_dto_converter.get_automatic_rtm_command_dto(self.fluid_type, self.__get_black_oil_dto().hasWater)
return self.__cluster_apis.kw_api.get_artm_parameters(dto)
