Add reader for casing and open hole;

PWPA-1933

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,

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 []
+        """ "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