Skip to content

Commit

Permalink
Read casing and open hole data;
Browse files Browse the repository at this point in the history 
Fill casing and open hole data to create description;
Read casing materials;
Add new constants for missing valus;
Add new unit for roughness;

PWPA-1933
  • Loading branch information
Gabriel Antão committed Jun 19, 2024
1 parent acc3283 commit 0b5b557
Show file tree
Hide file tree
Showing 4 changed files with 127 additions and 50 deletions.
8 changes: 8 additions & 0 deletions src/alfasim_score/constants.py
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,13 @@
from barril.units import Scalar

from alfasim_score.units import ROUGHNESS_UNIT

WELLBORE_TOP_NODE = "WELBORE_TOP_NODE"
WELLBORE_BOTTOM_NODE = "WELBORE_BOTTOM_NODE"
ANNULUS_TOP_NODE_NAME = "WELLBORE_ANNULUS_TOP_NODE"

CEMENT_NAME = "cement"

ROCK_DEFAULT_ROUGHNESS = Scalar(0.1, ROUGHNESS_UNIT)
CASING_DEFAULT_ROUGHNESS = Scalar(0.05, ROUGHNESS_UNIT)
TUBING_DEFAULT_ROUGHNESS = Scalar(0.05, ROUGHNESS_UNIT)
69 changes: 50 additions & 19 deletions src/alfasim_score/converter/alfacase/convert_alfacase.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,24 @@
from barril.units import Scalar

from alfasim_score.constants import ANNULUS_TOP_NODE_NAME
from alfasim_score.constants import CASING_DEFAULT_ROUGHNESS
from alfasim_score.constants import CEMENT_NAME
from alfasim_score.constants import ROCK_DEFAULT_ROUGHNESS
from alfasim_score.constants import TUBING_DEFAULT_ROUGHNESS
from alfasim_score.constants import WELLBORE_BOTTOM_NODE
from alfasim_score.constants import WELLBORE_TOP_NODE
from alfasim_score.converter.alfacase.score_input_reader import ScoreInputReader
from alfasim_score.units import LENGTH_UNIT


def filter_duplicated_materials(
material_list: List[MaterialDescription],
) -> List[MaterialDescription]:
"""Remove the duplicated materials parsed by the reader"""
# TODO: implement it to filter the duplicated materials
return material_list


class ScoreAlfacaseConverter:
def __init__(self, score_reader: ScoreInputReader):
self.score_input = score_reader
Expand All @@ -39,11 +50,13 @@ def _convert_well_trajectory(self) -> ProfileDescription:
def convert_materials(self) -> List[MaterialDescription]:
"""Convert list of materials from SCORE file"""
material_descriptions = []
material_list = (
material_list = filter_duplicated_materials(
self.score_input.read_cement_material()
+ self.score_input.read_casing_materials()
+ self.score_input.read_tubing_materials()
+ self.score_input.read_lithology_materials()
)

for data in material_list:
material_descriptions.append(
MaterialDescription(
Expand All @@ -66,27 +79,29 @@ def _convert_formation(self) -> AnnulusDescription:
return FormationDescription(reference_y_coordinate=Scalar(0.0, "m", "length"))

def _convert_casing_list(self) -> List[CasingSectionDescription]:
"""Create the description for the casings."""
casing_sections = []
for i, data in enumerate(self.score_input.read_casings(), start=1):
casing_sections.append(
CasingSectionDescription(
name=f"CASING_{i}",
hanger_depth=data["hanger_md"],
settings_depth=data["final_md"],
hole_diameter=data["hole_diameter"],
outer_diameter=data["outer_diameter"],
# TODO: missing values
# inner_diameter: Scalar
# inner_roughness: Scalar
# material: Optional[str] = None
top_of_filler=data["top_of_cement"],
filler_material=CEMENT_NAME,
# material_above_filler: Optional[str] = None
for section in data["sections"]:
casing_sections.append(
CasingSectionDescription(
name=f"{data['function']}_{data['type']}_{i}",
hanger_depth=section["top_md"],
settings_depth=section["base_md"],
hole_diameter=data["hole_diameter"],
outer_diameter=section["outer_diameter"],
inner_diameter=section["inner_diameter"],
inner_roughness=CASING_DEFAULT_ROUGHNESS,
material=section["material"],
top_of_filler=data["top_of_cement"],
filler_material=CEMENT_NAME,
material_above_filler=data["material_above"],
)
)
)
return casing_sections

def _convert_tubing_list(self) -> List[TubingDescription]:
"""Create the description for the tubing list."""
tubing_sections = []
for i, data in enumerate(self.score_input.read_tubing(), start=1):
tubing_sections.append(
Expand All @@ -95,8 +110,7 @@ def _convert_tubing_list(self) -> List[TubingDescription]:
length=data["base_md"] - data["top_md"],
outer_diameter=data["outer_diameter"],
inner_diameter=data["inner_diameter"],
# TODO: set right the value for roughness...
inner_roughness=Scalar(0.0, "mm"),
inner_roughness=TUBING_DEFAULT_ROUGHNESS,
material=data["material"],
)
)
Expand All @@ -116,7 +130,24 @@ def _convert_packer_list(self) -> List[PackerDescription]:
return packers

def _convert_open_hole_list(self) -> List[OpenHoleDescription]:
return []
"""Create the description for the open hole."""
open_hole = []
start_position = Scalar(
max([data["shoe_md"] for data in self.score_input.read_casings()]),
LENGTH_UNIT,
"length",
)
for i, data in enumerate(self.score_input.read_tubing(), start=1):
open_hole.append(
OpenHoleDescription(
name=f"OPEN_HOLE_{i}",
length=data["final_md"] - start_position,
diameter=data["hole_diameter"],
inner_roughness=ROCK_DEFAULT_ROUGHNESS,
)
)
start_position = data["final_md"]
return open_hole

def _convert_casings(self) -> WellDescription:
"""Create the description for the casings."""
Expand Down
96 changes: 67 additions & 29 deletions src/alfasim_score/converter/alfacase/score_input_reader.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,10 @@
class WellItemType(StrEnum):
DRILLING = "DRILLING"
CASING = "CASING"
NONE = "NONE"


class WellItemInterval(StrEnum):
class WellItemFunction(StrEnum):
CONDUCTOR = "CONDUCTOR"
SURFACE = "SURFACE"
PRODUCTION = "PRODUCTION"
Expand Down Expand Up @@ -67,15 +68,37 @@ def read_tubing_materials(self) -> List[Dict[str, Union[Scalar, str]]]:
)
return tubing_data

def read_casing_materials(self) -> List[Dict[str, Union[Scalar, str]]]:
"""Read the data for the casing from SCORE input file."""
casing_data = []
for item in self.input_content["operation"]["thermal_simulation"]["well_strings"]:
for section in item["string_sections"]:
properties = section["pipe"]["grade"]["thermomechanical_property"]
casing_data.append(
{
"name": section["pipe"]["grade"]["name"],
"density": Scalar(properties["density"], DENSITY_UNIT),
"thermal_conductivity": Scalar(
properties["thermal_conductivity"], THERMAL_CONDUCTIVITY_UNIT
),
"specific_heat": Scalar(properties["specific_heat"], SPECIFIC_HEAT_UNIT),
"thermal_expansion": Scalar(
properties["thermal_expansion_coefficient"], THERMAL_EXPANSION_UNIT
),
"young_modulus": Scalar(properties["e"], YOUNG_MODULUS_UNIT),
"poisson_ratio": Scalar(properties["nu"], FRACTION_UNIT),
}
)
return casing_data

def read_cement_material(self) -> List[Dict[str, Union[Scalar, str]]]:
"""
Read the data for the cement from SCORE input file.
This method assumes all configured cement properties are the same and that
the first_slurry and second_slurry have the same properties.
"""
properties = self.input_content["well_strings"][0]["cementing"]["first_slurry"][
"thermomechanical_property"
]
well_strings = self.input_content["operation"]["thermal_simulation"]["well_strings"]
properties = well_strings[0]["cementing"]["first_slurry"]["thermomechanical_property"]
return [
{
"name": CEMENT_NAME,
Expand Down Expand Up @@ -113,27 +136,47 @@ def read_lithology_materials(self) -> List[Dict[str, Union[Scalar, str]]]:
)
return lithology_data

def read_casings(self) -> List[Dict[str, Scalar | str]]:
""" "Read the data for the casing from SCORE input file"""
def read_casings(self) -> List[Dict[str, Any]]:
"""Read the data for the casing from SCORE input file"""
casing_data = []
for section in self.input_content["well_strings"]:
if section["interval"] != WellItemInterval.OPEN.value:
for item in self.input_content["operation"]["thermal_simulation"]["well_strings"]:
if item["interval"] != WellItemFunction.OPEN.value:
casing_data.append(
{
"hanger_md": Scalar(section["hanger_md"], LENGTH_UNIT, "length"),
"shoe_md": Scalar(section["shoe_md"], LENGTH_UNIT, "length"),
"final_md": Scalar(section["final_md"], LENGTH_UNIT, "length"),
"top_of_cement": Scalar(section["toc_md"], LENGTH_UNIT, "length"),
"hole_diameter": Scalar(section["hole_size"], DIAMETER_UNIT, "diameter"),
# TODO: check missing inner diameter here
# "inner_diameter": Scalar(inner_diameter, DIAMETER_UNIT, "diameter"),
"outer_diameter": Scalar(section["od"], DIAMETER_UNIT, "diameter"),
"type": WellItemType(item["type"]) if "type" in item else WellItemType.NONE,
"function": WellItemFunction(item["interval"]),
"hanger_md": Scalar(item["hanger_md"], LENGTH_UNIT, "length"),
"shoe_md": Scalar(item["shoe_md"], LENGTH_UNIT, "length"),
"final_md": Scalar(item["final_md"], LENGTH_UNIT, "length"),
"top_of_cement": Scalar(item["toc_md"], LENGTH_UNIT, "length"),
# TODO: check how to get this material obove filler
"material_above_filler": None,
"hole_diameter": Scalar(item["hole_size"], DIAMETER_UNIT, "diameter"),
# TODO: check if these diameters should be used here
# "inner_diameter":,
# "outer_diameter": Scalar(item["od"], DIAMETER_UNIT, "diameter"),
"sections": [
{
"material": section["pipe"]["grade"]["name"],
"top_md": Scalar(section["top_md"], LENGTH_UNIT, "length"),
"base_md": Scalar(section["base_md"], LENGTH_UNIT, "length"),
"inner_diameter": Scalar(
section["pipe"]["od"] - 2.0 * section["pipe"]["wt"],
DIAMETER_UNIT,
"diameter",
),
"outer_diameter": Scalar(
section["pipe"]["od"], DIAMETER_UNIT, "diameter"
),
}
for section in item["string_sections"]
],
}
)
return casing_data

def read_tubing(self) -> List[Dict[str, Any]]:
""" "Read the data for the tubing from SCORE input file"""
"""Read the data for the tubing from SCORE input file"""
tubing_data = []
for section in self.input_content["operation"]["tubing_string"]["string_sections"]:
outer_radius = section["pipe"]["od"] / 2.0
Expand All @@ -150,8 +193,8 @@ def read_tubing(self) -> List[Dict[str, Any]]:
)
return tubing_data

def read_packers(self) -> List[Dict[str, Scalar | str]]:
""" "Read the data for the packers from SCORE input file"""
def read_packers(self) -> List[Dict[str, Any]]:
"""Read the data for the packers from SCORE input file"""
packer_data = []
for component in self.input_content["operation"]["tubing_string"]["components"]:
if component["component"]["type"] == "PACKER":
Expand All @@ -160,26 +203,21 @@ def read_packers(self) -> List[Dict[str, Scalar | str]]:
"name": component["name"],
"position": Scalar(component["depth"], LENGTH_UNIT, "length"),
# TODO: get material above from somewhere else
"material_above": "",
# well_string items have the annular_fluids, but the fluids don't have properties in file
"material_above": None,
}
)
return packer_data

def read_open_hole(self) -> List[Dict[str, Scalar | str]]:
""" "Read the data for the open hole from SCORE input file"""
"""Read the data for the open hole from SCORE input file"""
casing_data = []
for section in self.input_content["well_strings"]:
if section["interval"] == WellItemInterval.OPEN.value:
for section in self.input_content["operation"]["thermal_simulation"]["well_strings"]:
if section["interval"] == WellItemFunction.OPEN.value:
casing_data.append(
{
"hanger_md": Scalar(section["hanger_md"], LENGTH_UNIT, "length"),
"shoe_md": Scalar(section["shoe_md"], LENGTH_UNIT, "length"),
"final_md": Scalar(section["final_md"], LENGTH_UNIT, "length"),
"top_of_cement": Scalar(section["toc_md"], LENGTH_UNIT, "length"),
"hole_diameter": Scalar(section["hole_size"], DIAMETER_UNIT, "diameter"),
# TODO: check missing inner diameter here
# "inner_diameter": Scalar(inner_diameter, DIAMETER_UNIT, "diameter"),
"outer_diameter": Scalar(section["od"], DIAMETER_UNIT, "diameter"),
}
)
return casing_data
4 changes: 2 additions & 2 deletions src/alfasim_score/units.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
THERMAL_EXPANSION_UNIT = "1/K"
YOUNG_MODULUS_UNIT = "psi"
FRACTION_UNIT = "-"

ROUGHNESS_UNIT = "mm"
DENSITY_UNIT = ""
THERMAL_CONDUCTIVITY_UNIT = "W/m.K"
SPECIFIT_HEAT_UNIT = "J/kg.K"
SPECIFIT_HEAT_UNIT = "J/kg.K"

0 comments on commit 0b5b557

Please sign in to comment.