import io
from xml.etree.ElementTree import ElementTree, Element, fromstring, register_namespace
from ..parser_exception import ParserException
from typing import Optional, List, Union, Tuple, cast
from .intake_properties_enum import IntakePropertiesEnum
from .trajectory_enum import TrajectoryEnum
from .geometry_properties_enum import GeometryPropertiesEnum
from .temperature_properties_enum import TemperaturePropertiesEnum
from .gas_lift_properties_enum import GasLiftPropertiesEnum
from .flow_path_enum import FlowPathEnum
from ...datetime_utils import datetime_to_str, str_to_datetime
from datetime import datetime
from uuid import UUID
[docs]
class WellIntakeParser:
"""
Parses XML Well Intake from KW document's well intake 5.50
"""
__ns = {'Export': 'KWKA_XML_Export', 'Intake': 'KWKA_XML_Intake', 'Base': 'KW_XML_Base', 'xsi': 'http://www.w3.org/2001/XMLSchema-instance'}
def __init__(self, xml_string: str) -> None:
register_namespace('AA', 'KWKA_XML_Export')
register_namespace('KWKAModel', 'KWKA_XML_Model')
register_namespace('ns', 'KWKA_XML_Intake')
register_namespace('Export', 'KWKA_XML_Export')
register_namespace('Intake', 'KWKA_XML_Intake')
register_namespace('Base', 'KW_XML_Base')
register_namespace('xsi', 'http://www.w3.org/2001/XMLSchema-instance')
self.__tree = ElementTree(fromstring(xml_string))
self.__root = self.__tree.getroot()
self.__model = self.__get_intake_element("Model")
self.__version = cast(str, self.__find_parameter_in_element(self.__root, "Export", "WellIntake").get('WellIntakeVersion'))
@property
def version(self) -> str:
return self.__version
def __get_intake_element(self, name: str) -> Element:
well_intake_element = self.__find_parameter_in_element(self.__root, "Export", "WellIntake")
return self.__find_parameter_in_element(well_intake_element, "Intake", name)
def __find_parameter_in_element(self, element: Element, namespace_prefix: str, parameter_name: str) -> Element:
parameter = element.find(f"{namespace_prefix}:{parameter_name}", self.__ns)
if parameter is None:
raise ParserException(f"Cannot find {self.__ns[namespace_prefix]}:{parameter_name} in the xml document")
return parameter
def __get_intake_info_element(self, intake_element_name: str) -> Element:
well_intake_element = self.__find_parameter_in_element(self.__root, "Export", "WellIntake")
return self.__find_parameter_in_element(well_intake_element, "Intake", intake_element_name)
def __set_well_log_value(self, element: Element, value: Union[Tuple[List[float], List[float]], float, int], is_only_non_constant_well_log: bool = False) -> None:
element.clear()
if isinstance(value, tuple):
x_values, y_values = value
non_constant_value_element = Element("Base:NonConstantValue")
array = self.__build_array(x_values, y_values)
non_constant_value_element.append(array)
element.append(non_constant_value_element)
else:
if not is_only_non_constant_well_log:
constant_value_element = Element("Base:ConstantValue")
constant_value_element.text = str(value)
element.append(constant_value_element)
else:
raise ValueError(f"{element.tag} can only be a tuple of 2 lists of floats")
def __build_array(self, x_values: List[float], y_values: List[float]) -> Element:
array = Element("Base:Array")
for x, y in zip(x_values, y_values):
well_log_element = Element("Base:WellLogElement")
x_element = Element("Base:X")
x_element.text = str(x)
y_element = Element("Base:Y")
y_element.text = str(y)
well_log_element.append(x_element)
well_log_element.append(y_element)
array.append(well_log_element)
return array
def get_properties_value(self, parameter: IntakePropertiesEnum) -> Optional[Union[str, datetime]]:
if parameter == IntakePropertiesEnum.start_date:
return cast(datetime, str_to_datetime(self.__find_parameter_in_element(self.__model, "Intake", parameter.value).text))
elif parameter == IntakePropertiesEnum.gauge_depth:
geometry = self.__find_parameter_in_element(self.__model, "Intake", "Geometry")
return self.__find_parameter_in_element(geometry, "Intake", "BottomholeDepth").text
else:
try:
value = cast(str, self.__get_intake_info_element(str(parameter.value)).text)
except ParserException:
return None
return value
def set_properties_value(self, parameter: IntakePropertiesEnum, value: Union[str, float, datetime]) -> None:
if parameter == IntakePropertiesEnum.start_date:
self.__find_parameter_in_element(self.__model, "Intake", parameter.value).text = datetime_to_str(cast(datetime, value))
elif parameter == IntakePropertiesEnum.gauge_depth:
geometry = self.__find_parameter_in_element(self.__model, "Intake", "Geometry")
self.__find_parameter_in_element(geometry, "Intake", "BottomholeDepth").text = str(value)
else:
self.__get_intake_info_element(str(parameter.value)).text = str(value)
def set_trajectory(self, trajectory_type: TrajectoryEnum, value: Union[Tuple[List[float], List[float]], float]) -> None:
if trajectory_type == TrajectoryEnum.md_tvd and isinstance(value, float):
raise ValueError(f"{trajectory_type} accept only 2 lists of floats")
trajectory = self.__get_intake_element("Trajectory")
try:
element = self.__find_parameter_in_element(trajectory, "Intake", trajectory_type.value)
except ParserException:
trajectory.clear()
element = Element(f"Intake:{trajectory_type.value}")
trajectory.append(element)
if trajectory_type == TrajectoryEnum.md_tvd:
element.clear()
x_values, y_values = cast(Tuple[List[float], List[float]], value)
array = self.__build_array(x_values, y_values)
element.append(array)
else:
self.__set_well_log_value(element, value)
def set_geometry(self, geometry_property: GeometryPropertiesEnum, value: Union[Tuple[List[float], List[float]], float]) -> None:
geometry = self.__find_parameter_in_element(self.__model, "Intake", "Geometry")
if geometry_property == GeometryPropertiesEnum.tubing_id or geometry_property == GeometryPropertiesEnum.tubing_absolute_roughness:
tubing_geometry = self.__find_parameter_in_element(geometry, "Intake", "TubingGeometry")
element = self.__find_parameter_in_element(tubing_geometry, "Intake", geometry_property.value)
self.__set_well_log_value(element, value)
if geometry_property is not GeometryPropertiesEnum.tubing_id:
tubing_geometry = self.__find_parameter_in_element(geometry, "Intake", "CasingGeometry")
element = self.__find_parameter_in_element(tubing_geometry, "Intake", geometry_property.value)
self.__set_well_log_value(element, value)
def set_temperature(self, temperature_property: TemperaturePropertiesEnum, value: Union[Tuple[List[float], List[float]], float, int]) -> None:
temperature = self.__find_parameter_in_element(self.__model, "Intake", "Temperature")
linear = self.__find_parameter_in_element(temperature, "Intake", "Linear")
self.__find_parameter_in_element(linear, "Intake", temperature_property.value).text = str(value)
def set_gas_lift_properties(self, gas_lift_property: GasLiftPropertiesEnum, value: Union[str, float, int, UUID]) -> None:
gas_lift_element = self.__find_parameter_in_element(self.__model, "Intake", "GasLift")
if gas_lift_property in [GasLiftPropertiesEnum.specific_gas_gravity, GasLiftPropertiesEnum.z_correlation_choice, GasLiftPropertiesEnum.cg_correlation_choice, GasLiftPropertiesEnum.mug_correlation_choice]:
injected_gas_pvt_description_element = self.__find_parameter_in_element(gas_lift_element, "Intake", "InjectedGasPvtDescription")
parameter_element = self.__find_parameter_in_element(injected_gas_pvt_description_element, "Intake", str(gas_lift_property.value))
parameter_element.text = str(value)
else:
element = self.__find_parameter_in_element(gas_lift_element, "Intake", str(gas_lift_property.value))
if gas_lift_property == GasLiftPropertiesEnum.injection_pressure:
element.clear()
if type(value) is str:
value = UUID(value)
if type(value) is float or type(value) is int:
injection_pressure_value_element = Element("Intake:ConstantPressure")
elif type(value) is UUID:
injection_pressure_value_element = Element("Intake:InjectedPressureGaugeID")
else:
raise ValueError(f"To set the value of {gas_lift_property.value} you must pass a float or a UUID")
injection_pressure_value_element.text = str(value)
element.append(injection_pressure_value_element)
elif gas_lift_property == GasLiftPropertiesEnum.injection_gas_rate:
element.clear()
if type(value) is str:
value = UUID(value)
if type(value) is float or type(value) is int:
injection_gas_rate_value_element = Element("Intake:ConstantGasRate")
elif type(value) is UUID:
injection_gas_rate_value_element = Element("Intake:InjectedGasRateGaugeID")
else:
raise ValueError(f"To set the value of {gas_lift_property.value} you must pass a float or a UUID")
injection_gas_rate_value_element.text = str(value)
element.append(injection_gas_rate_value_element)
else:
element.text = str(value)
def is_gas_lift(self) -> bool:
gas_lift_parameter = self.__model.find("Intake:GasLift", self.__ns)
if gas_lift_parameter is None:
return False
else:
return True
def set_flow_path(self, flow_path: FlowPathEnum) -> None:
geometry = self.__find_parameter_in_element(self.__model, "Intake", "Geometry")
self.__find_parameter_in_element(geometry, "Intake", "FlowPath").text = str(flow_path.value)
def export(self) -> str:
self.__root.set('xmlns:ns', self.__ns['Intake'])
stream = io.StringIO()
self.__tree.write(stream, encoding='unicode', method='xml', xml_declaration=True)
return stream.getvalue()
