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'
}
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()
def write_in_file(self, file_name: str) -> None:
self.__tree.write(file_name)
def export(self) -> str:
stream = io.StringIO()
self.__tree.write(stream, encoding='unicode', method='xml', xml_declaration=True)
return stream.getvalue()
def print_all(self) -> None:
self.__print_nodes(self.__root, "")
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 configure_datasets(self, wells: List[Keg5WellPropertyInputs]) -> None:
""" Will remove the existing datasets from this keg5 and create new ones taking
into account the parameters of each 'Keg5WellPropertyInputs', will only affect
permeability, porosity and thickness.
Will as well reset layers, horizons, zones and petrophysics (the analysis
will be single layer, one unique region).
Parameters
----------
wells: list of properties containing petrophysics data
"""
new_data_sets: VectorialReservoir = self.__replace_data_sets(wells)
self.__configure_geometry(new_data_sets)
def configure_time_stepping(self, start_date: datetime, end_date: datetime) -> None:
time_stepping = self.__find_element(self.__root, 'AA:Description/AA:TimeStepping')
date_format = "%Y-%m-%dT%H:%M:%S"
start_date_elm = self.__find_element(time_stepping, 'AA:RunStartDate')
start_date_elm.text = start_date.strftime(date_format)
end_date_elm = self.__find_element(time_stepping, 'AA:RunEndDate')
end_date_elm.text = end_date.strftime(date_format)
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)
