diff --git a/src/alfasim_score/converter/alfacase/convert_alfacase.py b/src/alfasim_score/converter/alfacase/convert_alfacase.py index 275fe5e..3daae99 100644 --- a/src/alfasim_score/converter/alfacase/convert_alfacase.py +++ b/src/alfasim_score/converter/alfacase/convert_alfacase.py @@ -1,16 +1,21 @@ from typing import List from alfasim_sdk import AnnulusDescription +from alfasim_sdk import CasingDescription +from alfasim_sdk import CasingSectionDescription from alfasim_sdk import FormationDescription from alfasim_sdk import MaterialDescription from alfasim_sdk import MaterialType +from alfasim_sdk import OpenHoleDescription +from alfasim_sdk import PackerDescription from alfasim_sdk import ProfileDescription +from alfasim_sdk import TubingDescription from alfasim_sdk import WellDescription from alfasim_sdk import XAndYDescription -from alfasim_sdk import CasingDescription, PackerDescription, TubingDescription, CasingSectionDescription, OpenHoleDescription from barril.units import Scalar from alfasim_score.constants import ANNULUS_TOP_NODE_NAME +from alfasim_score.constants import CEMENT_NAME 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 @@ -61,24 +66,42 @@ def _convert_formation(self) -> AnnulusDescription: return FormationDescription(reference_y_coordinate=Scalar(0.0, "m", "length")) def _convert_casing_list(self) -> List[CasingSectionDescription]: - return [] - + 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 + ) + ) + return casing_sections + def _convert_tubing_list(self) -> List[TubingDescription]: tubing_sections = [] - for i, data in enumerate(self.score_input.read_casings(), start=1): + for i, data in enumerate(self.score_input.read_tubing(), start=1): tubing_sections.append( TubingDescription( name=f"TUBING_{i}", - length=Scalar(data["base_md"] - data["top_md"], LENGTH_UNIT, "length"), + 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"), - material=data["material"] + material=data["material"], ) ) return tubing_sections - + def _convert_packer_list(self) -> List[PackerDescription]: """Create the description for the packers.""" packers = [] @@ -92,24 +115,24 @@ def _convert_packer_list(self) -> List[PackerDescription]: ) return packers - def _convert_open_hole(self) -> List[OpenHoleDescription]: + def _convert_open_hole_list(self) -> List[OpenHoleDescription]: return [] - def _convert_casing_list(self) -> WellDescription: + def _convert_casings(self) -> WellDescription: """Create the description for the casings.""" return CasingDescription( casing_sections=self._convert_casing_list(), tubings=self._convert_tubing_list(), packers=self._convert_packer_list(), - open_holes=self._convert_open_hole(), + open_holes=self._convert_open_hole_list(), ) - + def build_well(self) -> WellDescription: """Create the description for the well.""" return WellDescription( name=self.well_name, profile=self._convert_well_trajectory(), - casing=self._convert_casing(), + casing=self._convert_casings(), annulus=self._convert_annulus(), formation=self._convert_formation(), top_node=WELLBORE_TOP_NODE, diff --git a/src/alfasim_score/converter/alfacase/score_input_reader.py b/src/alfasim_score/converter/alfacase/score_input_reader.py index eca0f1b..3e2cdd9 100644 --- a/src/alfasim_score/converter/alfacase/score_input_reader.py +++ b/src/alfasim_score/converter/alfacase/score_input_reader.py @@ -1,15 +1,18 @@ +from typing import Any from typing import Dict from typing import List from typing import Tuple from typing import Union import json -from barril.units import Array, Scalar +from barril.units import Array +from barril.units import Scalar +from enum import StrEnum from pathlib import Path -from typing import Tuple, List, Dict from alfasim_score.constants import CEMENT_NAME from alfasim_score.units import DENSITY_UNIT +from alfasim_score.units import DIAMETER_UNIT from alfasim_score.units import FRACTION_UNIT from alfasim_score.units import LENGTH_UNIT from alfasim_score.units import SPECIFIC_HEAT_UNIT @@ -18,6 +21,18 @@ from alfasim_score.units import YOUNG_MODULUS_UNIT +class WellItemType(StrEnum): + DRILLING = "DRILLING" + CASING = "CASING" + + +class WellItemInterval(StrEnum): + CONDUCTOR = "CONDUCTOR" + SURFACE = "SURFACE" + PRODUCTION = "PRODUCTION" + OPEN = "OPEN" + + class ScoreInputReader: def __init__(self, score_filepath: Path): self.score_filepath = score_filepath @@ -98,30 +113,45 @@ def read_lithology_materials(self) -> List[Dict[str, Union[Scalar, str]]]: ) return lithology_data - # TODO: implement the casing parser - def read_casings() -> List[Dict[str, Scalar | str]]: - return [] - - def read_tubing(self) -> List[Dict[str, Scalar | str]]: - """"Read the data for the tubing from SCORE input file""" + def read_casings(self) -> List[Dict[str, Scalar | str]]: + """ "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: + 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 + + def read_tubing(self) -> List[Dict[str, Any]]: + """ "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 thickness = section["pipe"]["wt"] inner_diameter = 2.0 * (outer_radius - thickness) - casing_data.append( + tubing_data.append( { "top_md": Scalar(section["top_md"], LENGTH_UNIT, "length"), "base_md": Scalar(section["base_md"], LENGTH_UNIT, "length"), "inner_diameter": Scalar(inner_diameter, DIAMETER_UNIT, "diameter"), - "outer_diameter": Scalar(section["od"], DIAMETER_UNIT, "diameter"), - "material": section["pipe"]["grade"]["name"] + "outer_diameter": Scalar(section["pipe"]["od"], DIAMETER_UNIT, "diameter"), + "material": section["pipe"]["grade"]["name"], } ) - return casing_data + return tubing_data def read_packers(self) -> List[Dict[str, Scalar | str]]: - """"Read the data for the packers from SCORE input file""" + """ "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": @@ -135,6 +165,21 @@ def read_packers(self) -> List[Dict[str, Scalar | str]]: ) return packer_data - # TODO: implement the open holes parser def read_open_hole(self) -> List[Dict[str, Scalar | str]]: - return [] \ No newline at end of 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: + 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