import io
from datetime import datetime
from typing import List, Any
from xml.etree.ElementTree import ElementTree, Element, fromstring, register_namespace, SubElement
from ..kw.parser_exception import ParserException
from .vectorial_reservoir import VectorialReservoir
from .data_set_data import DataSet
from .keg5_well_property_inputs import Keg5WellPropertyInputs
keg5_AA_alias = "AA"
keg5_wellgeometry_alias = "keg5_wellgeometry"
keg5_reservoirgeometry_alias = "keg5_reservoirgeometry"
keg5_baseobjects_alias = "keg5_baseobjects"
keg5_well_alias = "keg5_well"
keg5_reservoir_alias = "keg5_reservoir"
keg5_petrophysics_alias = "keg5_petrophysics"
keg5_identifiers_alias = "keg5_identifiers"
keg5_wellcontrol_alias = "keg5_wellcontrol"
keg5_wellintake_alias = "keg5_wellintake"
[docs]
class ModelParser:
"""
Parses XML from KEG5 KW document's model
"""
# TODO default values to be changed?
__default_z_top: float = 1828.8000000000002
__default_layer_name = "Layer #1"
__default_name = "Default"
__ns = {
keg5_AA_alias: 'KEG5',
keg5_identifiers_alias: 'KEG5_Identifiers',
keg5_baseobjects_alias: 'KEG5_BaseObjects',
keg5_reservoir_alias: 'KEG5_Reservoir',
keg5_well_alias: 'KEG5_Well',
keg5_wellcontrol_alias: 'KEG5_WellControl',
keg5_wellgeometry_alias: 'KEG5_WellGeometry',
keg5_petrophysics_alias: 'KEG5_Petrophysics',
keg5_reservoirgeometry_alias: 'KEG5_ReservoirGeometry',
'keg5_pvt': 'KEG5_PVT',
'keg5_compaction': 'KEG5_Compaction',
'KEG5_desorption': 'KEG5_Desorption',
'keg5_initialstate': 'KEG5_InitialState',
'keg5_krpc': 'KEG5_KrPc',
'keg5_aquifer': 'KEG5_Aquifer',
'keg5_fault': 'KEG5_Fault',
'keg5_welldata': 'KEG5_WellData',
keg5_wellintake_alias: 'KEG5_WellIntake',
'keg5_wellgroup': 'KEG5_WellGroup',
'keg5_grid': 'KEG5_Grids',
'keg5_results': 'KEG5_Results'
}
[docs]
def __init__(self, xml_string: str) -> None:
for xmlns, value in self.__ns.items():
register_namespace(xmlns, value)
self.__tree = ElementTree(fromstring(xml_string))
self.__root = self.__tree.getroot()
[docs]
def write_in_file(self, file_name: str) -> None:
self.__tree.write(file_name)
[docs]
def export(self) -> str:
stream = io.StringIO()
self.__tree.write(stream, encoding='unicode', method='xml', xml_declaration=True)
return stream.getvalue()
[docs]
def print_all(self) -> None:
self.__print_nodes(self.__root, "")
[docs]
def add_vectorial_objects(self, wells: List[Keg5WellPropertyInputs]) -> None:
""" Will remove the existing wells from this keg5 and create new ones taking
into account the parameters of each input
Parameters
----------
wells: list of all needed properties to create new wells
"""
common_field: Element = self.__find_element(self.__root, 'AA:CommonFields/AA:CommonField')
common_wells_container: Element = self.__get_or_add_element(common_field, keg5_AA_alias, "CommonWells")
self.__remove_children(common_wells_container, f"{keg5_AA_alias}:Well")
vectorial_field: Element = self.__find_element(self.__root, 'AA:VectorialFields/AA:VectorialField')
vectorial_wells_container: Element = self.__get_or_add_element(vectorial_field, keg5_AA_alias, "VectorialWells")
self.__remove_children(vectorial_wells_container, f"{keg5_AA_alias}:VectorialWell")
for well in wells:
self.__add_well(common_wells_container, vectorial_wells_container, well)
def __print_nodes(self, node: Element, indent: str) -> None:
print(f"{indent}{node.tag}")
for sub_node in node:
self.__print_nodes(sub_node, indent + "\t")
def __add_well(self, common_wells_container: Element, vectorial_wells_container: Element, data: Keg5WellPropertyInputs) -> None:
self.__add_common_well(common_wells_container, data)
self.__add_vectorial_well(vectorial_wells_container, data)
def __add_common_well(self, common_wells_container: Element, data: Keg5WellPropertyInputs) -> None:
new_well = self.__add_sub_element(common_wells_container, keg5_AA_alias, "Well")
self.__set_ab_type_and_value(new_well, f"{keg5_well_alias}:KEG5_CommonWell")
self.__add_sub_element(new_well, keg5_wellcontrol_alias, "Controls")
self.__add_sub_element(new_well, keg5_well_alias, "Name", data.name)
self.__add_sub_element(new_well, keg5_well_alias, "WellID", data.well_id)
well_general_info_element = self.__add_sub_element(new_well, keg5_well_alias, "WellGeneralInformation")
self.__fill_well_general_info(well_general_info_element, data)
common_perforations = self.__add_sub_element(new_well, keg5_well_alias, "CommonPerforations")
self.__fill_common_perforations(common_perforations, data)
intake = self.__add_sub_element(new_well, keg5_well_alias, "Intake")
self.__add_sub_element(intake, keg5_wellintake_alias, "Name", f"{data.name} - intake")
self.__add_sub_element(intake, keg5_wellintake_alias, "PressureDrops")
def __fill_well_general_info(self, general_info_element: Element, data: Keg5WellPropertyInputs) -> None:
self.__add_sub_element(general_info_element, keg5_well_alias, "UniqueWellIndentifier", data.uwi)
self.__add_sub_element(general_info_element, keg5_well_alias, "WellType", data.get_well_type())
self.__add_sub_element(general_info_element, keg5_well_alias, "BottomholeDepth", str(self.__default_z_top))
self.__add_sub_element(general_info_element, keg5_well_alias, "DrillFloorElevation", "0")
well_head_coord = self.__add_sub_element(general_info_element, keg5_well_alias, "WellHeadCoordinates")
self.__add_sub_element(well_head_coord, keg5_well_alias, "X", str(data.x))
self.__add_sub_element(well_head_coord, keg5_well_alias, "Y", str(data.y))
def __fill_common_perforations(self, common_perforations: Element, data: Keg5WellPropertyInputs) -> None:
perforation = self.__add_sub_element(common_perforations, keg5_well_alias, "CommonPerforation")
self.__add_sub_element(perforation, keg5_wellgeometry_alias, "ID", data.perforation_id)
skins = self.__add_sub_element(perforation, keg5_wellgeometry_alias, "PerforationSkins")
default_skin = self.__add_sub_element(skins, keg5_wellgeometry_alias, "DefaultSkin")
self.__add_sub_element(default_skin, keg5_wellgeometry_alias, "Skin", "0" if data.skin is None else str(data.skin))
self.__add_sub_element(default_skin, keg5_wellgeometry_alias, "TurbulenceSkinFactor", "0")
def __add_vectorial_well(self, vectorial_wells_container: Element, data: Keg5WellPropertyInputs) -> None:
new_well = self.__add_sub_element(vectorial_wells_container, keg5_AA_alias, "VectorialWell")
trajectory = self.__add_sub_element(new_well, keg5_well_alias, "VectorialTrajectory")
self.__fill_well_trajectory(trajectory, data)
perforations = self.__add_sub_element(new_well, keg5_well_alias, "VectorialPerforations")
self.__fill_vectorial_perforations(perforations, data)
self.__add_sub_element(new_well, keg5_well_alias, "VectorialWellProperties", data.well_id)
self.__add_sub_element(new_well, keg5_well_alias, "HydraulicFractures")
def __fill_well_trajectory(self, trajectory: Element, data: Keg5WellPropertyInputs) -> None:
mainbore = self.__add_sub_element(trajectory, keg5_wellgeometry_alias, "VectorialMainbore")
self.__add_sub_element(mainbore, keg5_wellgeometry_alias, "BoreholeID", data.borehole_id)
self.__add_sub_element(mainbore, keg5_wellgeometry_alias, "Radius", str(data.radius))
path = self.__add_sub_element(mainbore, keg5_wellgeometry_alias, "Path")
for point in data.get_well_trajectory(self.__default_z_top):
self.__add_point_to_geometry(path, point.x, point.y, point.z)
def __fill_vectorial_perforations(self, perforations: Element, data: Keg5WellPropertyInputs) -> None:
perforation = self.__add_sub_element(perforations, keg5_well_alias, "Perforation")
self.__add_sub_element(perforation, keg5_wellgeometry_alias, "RefBoreholeID", data.borehole_id)
self.__add_sub_element(perforation, keg5_wellgeometry_alias, "PerforationID", data.perforation_id)
self.__add_sub_element(perforation, keg5_wellgeometry_alias, "MDStart", str(data.get_perforation_md_start(self.__default_z_top)))
self.__add_sub_element(perforation, keg5_wellgeometry_alias, "MDEnd", str(data.get_perforation_md_end(self.__default_z_top)))
def __add_point_to_geometry(self, path: Element, x: float, y: float, z: float) -> None:
point = self.__add_sub_element(path, keg5_wellgeometry_alias, "Point")
self.__add_sub_element(point, keg5_baseobjects_alias, "X", str(x))
self.__add_sub_element(point, keg5_baseobjects_alias, "Y", str(y))
self.__add_sub_element(point, keg5_baseobjects_alias, "Z", str(z))
def __add_sub_sub_element(self, parent: Element, namespace: str, name: str, sub_namespace: str, sub_name: str, text: str) -> Element:
element = self.__add_sub_element(parent, namespace, name)
self.__add_sub_element(element, sub_namespace, sub_name, text)
return element
def __add_sub_element(self, parent: Element, namespace: str, name: str, text: str = "") -> Element:
path = '{' + f'{self.__ns[namespace]}' + '}' + f'{name}'
sub_element = SubElement(parent, path)
if text is not None and text != "":
sub_element.text = text
return sub_element
def __get_or_add_element(self, parent: Element, namespace: str, name: str) -> Element:
path = '{' + f'{self.__ns[namespace]}' + '}' + f'{name}'
child_element = parent.find(path, self.__ns)
if child_element is None:
return self.__add_sub_element(parent, namespace, name)
return child_element
def __replace_data_sets(self, well_data: List[Keg5WellPropertyInputs]) -> VectorialReservoir:
x: List[float] = [well.x for well in well_data]
y: List[float] = [well.y for well in well_data]
data_sets: List[DataSet] = [
DataSet("PermDataSet", x, y, [well.permeability for well in well_data]),
DataSet("PorosityDataSet", x, y, [well.porosity for well in well_data]),
DataSet("ThicknessDataSet", x, y, [well.thickness for well in well_data])
]
vectorial_reservoir = self.__get_vectorial_reservoir()
self.__remove_children(vectorial_reservoir, f"{keg5_reservoir_alias}:DataSets")
for data_set in data_sets:
self.__add_data_set(vectorial_reservoir, data_set)
return VectorialReservoir(data_sets[0].id, data_sets[1].id, data_sets[2].id)
def __add_data_set(self, vectorial_reservoir: Element, data_set_data: DataSet) -> None:
data_set: Element = self.__add_sub_element(vectorial_reservoir, keg5_reservoir_alias, "DataSets")
self.__add_sub_element(data_set, keg5_baseobjects_alias, "ID", data_set_data.id)
self.__add_sub_element(data_set, keg5_baseobjects_alias, "Name", data_set_data.name)
varying_values: Element = self.__add_sub_element(data_set, keg5_baseobjects_alias, "VaryingValues")
self.__add_vector_double(varying_values, "X", data_set_data.x)
self.__add_vector_double(varying_values, "Y", data_set_data.y)
self.__add_vector_double(varying_values, "Value", data_set_data.values)
def __configure_geometry(self, data_sets: VectorialReservoir) -> None:
vectorial_reservoir = self.__get_vectorial_reservoir()
vectorial_geometry = self.__find_element(vectorial_reservoir, f"{keg5_reservoir_alias}:VectorialGeometry")
self.__configure_horizons(vectorial_geometry, data_sets)
self.__configure_layers(vectorial_geometry, data_sets)
self.__configure_property_zones(vectorial_geometry, data_sets)
self.__configure_regions(vectorial_geometry)
self.__configure_petrophysics(vectorial_reservoir, data_sets)
def __configure_horizons(self, geometry: Element, data_sets: VectorialReservoir) -> None:
horizons: Element = self.__find_element(geometry, f"{keg5_reservoirgeometry_alias}:Horizons")
self.__remove_children(horizons, f"{keg5_reservoirgeometry_alias}:Horizon")
top_horizon = self.__add_sub_element(horizons, keg5_reservoirgeometry_alias, "Horizon")
self.__add_sub_element(top_horizon, keg5_reservoirgeometry_alias, "ID", data_sets.top_horizon_id)
self.__add_sub_sub_element(top_horizon,
keg5_reservoirgeometry_alias, "HorizonGeometry",
keg5_baseobjects_alias,
"ConstantValue",
str(self.__default_z_top))
bottom_horizon = self.__add_sub_element(horizons, keg5_reservoirgeometry_alias, "Horizon")
self.__add_sub_element(bottom_horizon, keg5_reservoirgeometry_alias, "ID", data_sets.bottom_horizon_id)
self.__add_sub_sub_element(bottom_horizon,
keg5_reservoirgeometry_alias, "HorizonGeometry",
keg5_baseobjects_alias,
"DataSetID",
data_sets.thickness_id)
def __configure_layers(self, geometry: Element, data_sets: VectorialReservoir) -> None:
layers: Element = self.__find_element(geometry, f"{keg5_reservoirgeometry_alias}:Layers")
self.__remove_children(layers, f"{keg5_reservoirgeometry_alias}:Layer")
layer = self.__add_sub_element(layers, keg5_reservoirgeometry_alias, "Layer")
self.__add_sub_element(layer, keg5_reservoirgeometry_alias, "Name", self.__default_layer_name)
self.__add_sub_element(layer, keg5_reservoirgeometry_alias, "ID", data_sets.layer_id)
self.__add_sub_element(layer, keg5_reservoirgeometry_alias, "TopHorizon", data_sets.top_horizon_id)
self.__add_sub_element(layer, keg5_reservoirgeometry_alias, "BottomHorizon", data_sets.bottom_horizon_id)
def __configure_property_zones(self, geometry: Element, data_sets: VectorialReservoir) -> None:
zones: Element = self.__find_element(geometry, f"{keg5_reservoirgeometry_alias}:PropertyZones")
self.__remove_children(zones, f"{keg5_reservoirgeometry_alias}:PropertyZone")
zone = self.__add_sub_element(zones, keg5_reservoirgeometry_alias, "PropertyZone")
self.__set_ab_type_and_value(zone, f"{keg5_reservoirgeometry_alias}:KEG5_VectorialPropertyZone")
self.__add_sub_element(zone, keg5_identifiers_alias, "ZoneID", data_sets.zone_id)
self.__add_sub_element(zone, keg5_reservoirgeometry_alias, "Name", self.__default_name)
zone_geometry = self.__add_sub_element(zone, keg5_reservoirgeometry_alias, "PropertyZoneGeometry")
layer_region_geo = self.__add_sub_element(zone_geometry, keg5_reservoirgeometry_alias, "LayerAndRegionIDs")
self.__add_sub_element(layer_region_geo, keg5_identifiers_alias, "LayerID", data_sets.layer_id)
self.__add_sub_element(layer_region_geo, keg5_identifiers_alias, "RegionID", data_sets.region_id)
def __configure_regions(self, geometry: Element) -> None:
regions: Element = self.__find_element(geometry, f"{keg5_reservoirgeometry_alias}:Regions")
self.__remove_children(regions, f"{keg5_reservoirgeometry_alias}:Region")
def __configure_petrophysics(self, vectorial_reservoir: Element, data_sets: VectorialReservoir) -> None:
vectorial_petro: Element = self.__find_element(vectorial_reservoir, f"{keg5_reservoir_alias}:VectorialPetrophysics")
self.__remove_children(vectorial_petro, f"{keg5_reservoir_alias}:Petro")
petro = self.__add_sub_element(vectorial_petro, keg5_reservoir_alias, "Petro")
self.__set_ab_type_and_value(petro, f"{keg5_petrophysics_alias}:KEG5_VectorialPetrophysics")
self.__add_sub_element(petro, keg5_identifiers_alias, "ZoneID", data_sets.zone_id)
self.__add_sub_element(petro, keg5_petrophysics_alias, "Name")
data_set_id_tag = "DataSetID"
perm_map = self.__add_sub_element(petro, keg5_petrophysics_alias, "PermeabilityMap")
self.__add_sub_sub_element(perm_map, keg5_petrophysics_alias, "PermeabilityXY", keg5_baseobjects_alias, data_set_id_tag, data_sets.perm_id)
self.__add_sub_sub_element(perm_map, keg5_petrophysics_alias, "PermeabilityZ", keg5_baseobjects_alias, data_set_id_tag, data_sets.perm_id)
self.__add_sub_sub_element(petro, keg5_petrophysics_alias, "PorosityMap", keg5_baseobjects_alias, data_set_id_tag, data_sets.porosity_id)
self.__add_sub_sub_element(petro, keg5_petrophysics_alias, "NetToGrossMap", keg5_baseobjects_alias, "ConstantValue", "1")
self.__add_sub_element(petro, keg5_petrophysics_alias, "Type", "REFERENCE")
def __get_vectorial_reservoir(self) -> Element:
return self.__find_element(self.__root, 'AA:VectorialFields/AA:VectorialField/AA:VectorialReservoir')
def __add_vector_double(self, parent: Element, name: str, values: List[float]) -> None:
self.__add_vector(parent, "Double", name, values)
def __add_vector_int(self, parent: Element, name: str, values: List[int]) -> None:
self.__add_vector(parent, "Integer", name, values)
def __add_vector(self, parent: Element, vector_type: str, name: str, values: List[Any]) -> None:
values_container: Element = self.__add_sub_element(parent, keg5_baseobjects_alias, name)
self.__set_ab_type_and_value(values_container, f"{keg5_baseobjects_alias}:KEG5_Vector{vector_type}")
direct_doubles_container = self.__add_sub_element(values_container, keg5_baseobjects_alias, f"Direct{vector_type}")
value_type_tag = f"{vector_type}Value"
for value in values:
self.__add_sub_element(direct_doubles_container, keg5_baseobjects_alias, value_type_tag, str(value))
def __remove_children(self, parent: Element, nodes: str) -> None:
children = parent.findall(nodes, self.__ns)
for child in children:
parent.remove(child)
def __find_element(self, parent: Element, child: str) -> Element:
child_element = parent.find(child, self.__ns)
if child_element is None:
raise ParserException(f"Document does not contain {child}")
return child_element
@staticmethod
def __set_ab_type_and_value(element: Element, value: str) -> None:
element.set("xmlns:AB", "http://www.w3.org/2001/XMLSchema-instance")
element.set("AB:type", value)
