feat: Injection valve

completor/completion.py

@@ -20,7 +20,7 @@
     pass
 
 # Use more precise type information, if possible
-DeviceType: TypeAlias = 'Literal["AICD", "ICD", "DAR", "VALVE", "AICV", "ICV"]'
+DeviceType: TypeAlias = 'Literal["AICD", "ICD", "DAR", "VALVE", "AICV", "ICV", "INJV"]'
 
 
 class Information:
@@ -612,7 +612,7 @@ def get_device(df_well: pd.DataFrame, df_device: pd.DataFrame, device_type: Devi
     Args:
         df_well: Must contain device type, device number, and the scaling factor.
         df_device: Device table.
-        device_type: Device type. `AICD`, `ICD`, `DAR`, `VALVE`, `AICV`, `ICV`.
+        device_type: Device type. `AICD`, `ICD`, `DAR`, `VALVE`, `AICV`, `ICV`, `INJV`.
 
     Returns:
         Updated well information with device characteristics.
@@ -635,6 +635,10 @@ def get_device(df_well: pd.DataFrame, df_device: pd.DataFrame, device_type: Devi
         # rescale the Cv
         # because no scaling factor in WSEGVALV
         df_well[Headers.CV_DAR] = -df_well[Headers.CV_DAR] / df_well[Headers.SCALING_FACTOR]
+    elif device_type == "INJV":
+        # rescale the Cv
+        # because no scaling factor in WSEGVALV
+        df_well[Headers.CV_INJV] = -df_well[Headers.CV_INJV] / df_well[Headers.SCALING_FACTOR]
     return df_well
 
 

completor/constants.py

@@ -126,6 +126,12 @@ class Headers:
     GHF_LCF_DAR = "GHF_LCF_DAR"
     GHF_HCF_DAR = "GHF_HCF_DAR"
 
+    CV_INJV = "CV_INJV"
+    AC_PRIMARY = "AC_PRIMARY"
+    AC_SECONDARY = "AC_SECONDARY"
+    WR_CF_INJV = "WR_CF_INJV"
+    PRD_CF_INJV = "PRD_CF_INJV"
+
     ALPHA = "ALPHA"
     X = "X"
     Y = "Y"
@@ -207,6 +213,7 @@ class _Keywords:
     WSEGICV = "WSEGICV"
     WSEGSICD = "WSEGSICD"
     WSEGDAR = "WSEGDAR"
+    WSEGINJV = "WSEGINJV"
 
     SCHFILE = "SCHFILE"
     OUTFILE = "OUTFILE"

completor/create_output.py

@@ -29,7 +29,7 @@ class CreateOutput:
         schedule: ReadSchedule object.
         wells: CreateWells object.
         well_name: Well name.
-        iwell: Well number used in creating WSEGAICV and WSEGDAR output.
+        iwell: Well number used in creating WSEGAICV, WSEGDAR, and WSEGINJV output.
         version: Completor version information.
         show_figure: Flag for pdf export of well completion schematic.
         figure_no: Figure number.
@@ -132,6 +132,17 @@ def __init__(
 {"-" * 100}{self.newline1}"""
 
         self.print_wsegdarinit = self.print_wsegdar
+        self.print_wseginjv = f"""\
+{'-' * 100}
+-- This is how we model Injection Valve technology using sets of ACTIONX keywords.
+-- The segment dP curves changes according to the segment water-
+-- rate and segment pressure drop at downhole condition.
+-- The value of Cv is adjusted according to the segment length and the number of
+-- devices per joint. The constriction area varies according to values of
+-- volume fractions.
+{"-" * 100}{self.newline1}"""
+
+        self.print_wseginjvinit = self.print_wseginjv
         self.print_wsegaicv = f"""\
 {"-" * 100}
 -- This is how we model AICV technology using sets of ACTIONX keyword
@@ -186,6 +197,7 @@ def __init__(
             self.df_wsegsicd = po.prepare_wsegsicd(self.well_name, lateral, self.df_well, self.df_device)
             self.df_wsegaicd = po.prepare_wsegaicd(self.well_name, lateral, self.df_well, self.df_device)
             self.df_wsegdar = po.prepare_wsegdar(self.well_name, lateral, self.df_well, self.df_device)
+            self.df_wseginjv = po.prepare_wseginjv(self.well_name, lateral, self.df_well, self.df_device)
             self.df_wsegaicv = po.prepare_wsegaicv(self.well_name, lateral, self.df_well, self.df_device)
             self.df_wsegicv = po.prepare_wsegicv(
                 self.well_name,
@@ -205,6 +217,7 @@ def __init__(
             self.make_wsegaicd(lateral)
             self.make_wsegicv(lateral)
             self.make_wsegdar()
+            self.make_wseginjv()
             self.make_wsegaicv()
 
             if show_figure and figure_name is not None:
@@ -365,6 +378,11 @@ def make_wsegdar(self) -> None:
         if self.df_wsegdar.shape[0] > 0:
             self.print_wsegdar += po.print_wsegdar(self.df_wsegdar, self.iwell + 1) + "\n"
 
+    def make_wseginjv(self) -> None:
+        """Print WSEGINJV to file."""
+        if self.df_wseginjv.shape[0] > 0:
+            self.print_wseginjv += po.print_wseginjv(self.df_wseginjv, self.iwell + 1) + "\n"
+
     def make_wsegaicv(self) -> None:
         """Print WSEGAICV to file."""
         if self.df_wsegaicv.shape[0] > 0:
@@ -412,7 +430,12 @@ def fix_printing(self) -> None:
             self.print_wsegdar = ""
         else:
             self.print_wsegdar += self.newline1
-        # if no DAR then dont print
+        # if no Injection Valve then dont print
+        if self.print_wseginjv == self.print_wseginjvinit:
+            self.print_wseginjv = ""
+        else:
+            self.print_wseginjv += self.newline1
+        # if no AICV then dont print
         if self.print_wsegaicv == self.print_wsegaicvinit:
             self.print_wsegaicv = ""
         else:
@@ -440,6 +463,7 @@ def print_per_well(self) -> None:
             + self.print_wsegsicd
             + self.print_wsegaicd
             + self.print_wsegdar
+            + self.print_wseginjv
             + self.print_wsegaicv
             + self.print_wsegicv
         )

completor/create_wells.py

@@ -103,7 +103,7 @@ def _active_wells(self) -> npt.NDArray[np.unicode_]:
                 Headers.DEVICE_TYPE
             ]
             gp_check = not ann_series.isin(["OA"]).any()
-            perf_check = not type_series.isin(["AICD", "AICV", "DAR", "ICD", "VALVE", "ICV"]).any()
+            perf_check = not type_series.isin(["AICD", "AICV", "DAR", "ICD", "VALVE", "ICV", "INJV"]).any()
             if gp_check and perf_check and not self.case.gp_perf_devicelayer:
                 # De-activate wells with GP_PERF if instructed to do so:
                 active_wells.remove(well_name)
@@ -266,6 +266,8 @@ def get_devices(self) -> None:
             self.df_well = completion.get_device(self.df_well, self.case.wsegaicd_table, "AICD")
         if "DAR" in active_devices:
             self.df_well = completion.get_device(self.df_well, self.case.wsegdar_table, "DAR")
+        if "INJV" in active_devices:
+            self.df_well = completion.get_device(self.df_well, self.case.wseginjv_table, "INJV")
         if "AICV" in active_devices:
             self.df_well = completion.get_device(self.df_well, self.case.wsegaicv_table, "AICV")
         if "ICV" in active_devices:

completor/input_validation.py

@@ -162,10 +162,10 @@ def _check_for_errors(df_comp: pd.DataFrame, well_name: str, idx: int) -> None:
                 f"t{df_comp[Headers.END_MEASURED_DEPTH].iloc[idx - 1]} "
                 f"to depth {(df_comp[Headers.START_MEASURED_DEPTH].iloc[idx])}"
             )
-    if df_comp[Headers.DEVICE_TYPE].iloc[idx] not in ["PERF", "AICD", "ICD", "VALVE", "DAR", "AICV", "ICV"]:
+    if df_comp[Headers.DEVICE_TYPE].iloc[idx] not in ["PERF", "AICD", "ICD", "VALVE", "DAR", "INJV", "AICV", "ICV"]:
         raise CompletorError(
             f"{df_comp[Headers.DEVICE_TYPE].iloc[idx]} not a valid device type. "
-            "Valid types are PERF, AICD, ICD, VALVE, DAR, AICV, and ICV."
+            "Valid types are PERF, AICD, ICD, VALVE, DAR, INJV, AICV, and ICV."
         )
     if df_comp[Headers.ANNULUS].iloc[idx] not in ["GP", "OA", "PA"]:
         raise CompletorError(
@@ -250,6 +250,24 @@ def set_format_wsegdar(df_temp: pd.DataFrame) -> pd.DataFrame:
     return df_temp
 
 
+def set_format_wseginjv(df_temp: pd.DataFrame) -> pd.DataFrame:
+    """Format the well segments Injection Valve (WSEGINJV) data.
+
+    Args:
+        df_temp: Well segments Injection Valve device data.
+
+    Returns:
+        Updated data.
+    """
+    df_temp[Headers.DEVICE_NUMBER] = df_temp[Headers.DEVICE_NUMBER].astype(np.int64)
+    # left out devicenumber because it has been formatted as integer
+    columns = df_temp.columns.to_numpy()[1:]
+    df_temp[columns] = df_temp[columns].astype(np.float64)
+    # Create ID device column
+    df_temp.insert(0, Headers.DEVICE_TYPE, np.full(df_temp.shape[0], "INJV"))
+    return df_temp
+
+
 def set_format_wsegaicv(df_temp: pd.DataFrame) -> pd.DataFrame:
     """Format the well segments automatic inflow control valve (WSEGAICV) table.
 

completor/prepare_outputs.py

@@ -112,16 +112,21 @@ def dataframe_tostring(
                 Headers.END_MEASURED_DEPTH: "{:.3f}".format,
                 Headers.CV_DAR: "{:.10g}".format,
                 Headers.CV: "{:.10g}".format,
+                Headers.CV_INJV: "{:.10g}".format,
                 Headers.AC: "{:.3e}".format,
                 Headers.AC_OIL: "{:.3e}".format,
                 Headers.AC_GAS: "{:.3e}".format,
                 Headers.AC_WATER: "{:.3e}".format,
+                Headers.AC_PRIMARY: "{:.3e}".format,
+                Headers.AC_SECONDARY: "{:.3e}".format,
                 Headers.AC_MAX: "{:.3e}".format,
                 Headers.DEFAULTS: "{:.10s}".format,
                 Headers.WHF_LCF_DAR: "{:.10g}".format,
                 Headers.WHF_HCF_DAR: "{:.10g}".format,
                 Headers.GHF_LCF_DAR: "{:.10g}".format,
                 Headers.GHF_HCF_DAR: "{:.10g}".format,
+                Headers.WR_CF_INJV: "{:.10g}".format,
+                Headers.PRD_CF_INJV: "{:.10g}".format,
                 Headers.ALPHA_MAIN: "{:.10g}".format,
                 Headers.ALPHA_PILOT: "{:.10g}".format,
             }
@@ -438,9 +443,13 @@ def prepare_device_layer(
                         df_well[Headers.DEVICE_TYPE] == "DAR",
                         "/ -- DAR types",
                         np.where(
-                            df_well[Headers.DEVICE_TYPE] == "AICV",
-                            "/ -- AICV types",
-                            np.where(df_well[Headers.DEVICE_TYPE] == "ICV", "/ -- ICV types", ""),
+                            df_well[Headers.DEVICE_TYPE] == "INJV",
+                            "/ -- Injection Valve types",
+                            np.where(
+                                df_well[Headers.DEVICE_TYPE] == "AICV",
+                                "/ -- AICV types",
+                                np.where(df_well[Headers.DEVICE_TYPE] == "ICV", "/ -- ICV types", ""),
+                            ),
                         ),
                     ),
                 ),
@@ -1190,6 +1199,42 @@ def prepare_wsegdar(well_name: str, lateral: int, df_well: pd.DataFrame, df_devi
     return wsegdar
 
 
+def prepare_wseginjv(well_name: str, lateral: int, df_well: pd.DataFrame, df_device: pd.DataFrame) -> pd.DataFrame:
+    """Prepare data frame for Injection Valve.
+
+    Args:
+        well_name: Well name.
+        lateral: Lateral number.
+        df_well: df_well from class CreateWells.
+        df_device: From function prepare_device_layer for this well and this lateral.
+
+    Returns:
+        DataFrame for Injection Valve.
+    """
+    df_well = df_well[df_well[Headers.LATERAL] == lateral]
+    df_well = df_well[(df_well[Headers.DEVICE_TYPE] == "PERF") | (df_well[Headers.NUMBER_OF_DEVICES] > 0)]
+    if df_well.shape[0] == 0:
+        return pd.DataFrame()
+    df_merge = pd.merge_asof(
+        left=df_device, right=df_well, left_on=[Headers.MD], right_on=[Headers.TUB_MD], direction="nearest"
+    )
+    df_merge = df_merge[df_merge[Headers.DEVICE_TYPE] == "INJV"]
+    wseginjv = pd.DataFrame()
+    if df_merge.shape[0] > 0:
+        wseginjv[Headers.WELL] = [well_name] * df_merge.shape[0]
+        wseginjv[Headers.SEG] = df_merge[Headers.SEG].to_numpy()
+        # the Cv is already corrected by the scaling factor
+        wseginjv[Headers.CV_INJV] = df_merge[Headers.CV_INJV].to_numpy()
+        wseginjv[Headers.AC_PRIMARY] = df_merge[Headers.AC_PRIMARY].to_numpy()
+        wseginjv[Headers.AC_SECONDARY] = df_merge[Headers.AC_SECONDARY].to_numpy()
+        wseginjv[Headers.WR_CF_INJV] = df_merge[Headers.WR_CF_INJV].to_numpy()
+        wseginjv[Headers.PRD_CF_INJV] = df_merge[Headers.PRD_CF_INJV].to_numpy()
+        wseginjv[Headers.DEFAULTS] = "5*"
+        wseginjv[Headers.AC_MAX] = wseginjv[Headers.AC_PRIMARY].to_numpy()
+        wseginjv[Headers.EMPTY] = "/"
+    return wseginjv
+
+
 def prepare_wsegaicv(well_name: str, lateral: int, df_well: pd.DataFrame, df_device: pd.DataFrame) -> pd.DataFrame:
     """Prepare data frame for AICV.
 
@@ -1368,6 +1413,110 @@ def print_wsegdar(df_wsegdar: pd.DataFrame, well_number: int) -> str:
     return action
 
 
+def print_wseginjv(df_wseginjv: pd.DataFrame, well_number: int) -> str:
+    """Print Injection Valves devices.
+
+    Args:
+        df_wseginjv: Output from function prepare_wseginjv.
+        well_number: Well number.
+
+    Returns:
+        Formatted actions to be included in the output file.
+
+    Raises:
+        CompletorError: If there are to many wells and/or segments with Injection Valve.
+    """
+
+    """Important Information
+
+    -- the valves will react on both pressure drop and rate limit ----
+    -- it will move to smaller nozzle, and back again if certain criteria is fulfilled ---
+    -- the SUVTRIG is a marker flag to check if it is on big nozzle or small nozzle --
+    -- primary nozzle is 0 and secondary nozzle is 1 --
+
+    """
+
+    header = [
+        [Headers.WELL, Headers.SEG, Headers.CV_INJV, Headers.AC_PRIMARY, Headers.DEFAULTS, Headers.AC_MAX],
+        [Headers.WELL, Headers.SEG, Headers.CV_INJV, Headers.AC_SECONDARY, Headers.DEFAULTS, Headers.AC_MAX],
+    ]
+    sign_rate = ["<"]
+    sign_pressure_drop = [">"]
+    suvtrig = ["0", "1"]
+    action = "UDQ\n"
+    for idx in range(df_wseginjv.shape[0]):
+        segment_number = df_wseginjv[Headers.SEG].iloc[idx]
+        well_name = df_wseginjv[Headers.WELL].iloc[idx]
+        action += f"  ASSIGN SUVTRIG {well_name} {segment_number} 0 /\n"
+    action += "/\n\n"
+    iaction = 1
+    action += Keywords.WSEGVALV + "\n"
+    header_string = "--"
+    for itm in header[iaction]:
+        header_string += "  " + itm
+    action += header_string.rstrip() + "\n"
+    for idx in range(df_wseginjv.shape[0]):
+        segment_number = df_wseginjv[Headers.SEG].iloc[idx]
+        print_df = df_wseginjv[df_wseginjv[Headers.SEG] == segment_number]
+        print_df = print_df[header[iaction]]
+        print_df = dataframe_tostring(print_df, True, False, False) + "\n"
+        action += print_df
+    action += "/\n\n"
+    for idx in range(df_wseginjv.shape[0]):
+        segment_number = df_wseginjv[Headers.SEG].iloc[idx]
+        well_name = df_wseginjv[Headers.WELL].iloc[idx]
+        water_segment_rate_cutoff = df_wseginjv[Headers.WR_CF_INJV].iloc[idx]
+        pressure_drop_cutoff = df_wseginjv[Headers.PRD_CF_INJV].iloc[idx]
+
+        iaction = 0
+        act_number = iaction + 1
+        act_name = f"INJVOP{well_number:03d}{segment_number:03d}{act_number:1d}"
+        if len(act_name) > 13:
+            raise CompletorError("Too many wells and/or too many segments with Injection Valve")
+        action += (
+            f"ACTIONX\n{act_name} 1000000 /\n"
+            f"SWFR '{well_name}' {segment_number} "
+            f"{sign_rate[iaction]} {water_segment_rate_cutoff} AND /\n"
+            f"SUVTRIG '{well_name}' {segment_number} "
+            f"= {suvtrig[iaction]} /\n/\n\n"
+        )
+        print_df = df_wseginjv[df_wseginjv[Headers.SEG] == segment_number]
+        print_df = print_df[header[iaction]]  # type: ignore
+        header_string = Keywords.WSEGVALV + "\n--"
+        for item in header[iaction]:
+            header_string += "  " + item
+        header_string = header_string.rstrip() + "\n"
+        print_df = header_string + dataframe_tostring(print_df, True, False, False)  # type: ignore
+        print_df += "\n/\n"
+        print_df += f"\nUDQ\n  ASSIGN SUVTRIG {well_name} {segment_number} 0 /\n/\n"
+        action += print_df + "\nENDACTIO\n\n"
+
+        iaction = 0
+        act_number = iaction + 1
+        act_name = f"INJVCL{well_number:03d}{segment_number:03d}{act_number:1d}"
+        if len(act_name) > 13:
+            raise CompletorError("Too many wells and/or too many segments with Injection Valve")
+        action += (
+            f"ACTIONX\n{act_name} 1000000 /\n"
+            f"SPRD '{well_name}' {segment_number} "
+            f"{sign_pressure_drop[iaction]} {pressure_drop_cutoff} AND /\n"
+            f"SUVTRIG '{well_name}' {segment_number} "
+            f"= {suvtrig[iaction]} /\n/\n\n"
+        )
+        print_df = df_wseginjv[df_wseginjv[Headers.SEG] == segment_number]
+        print_df = print_df[header[iaction]]  # type: ignore
+        header_string = Keywords.WSEGVALV + "\n--"
+        for item in header[iaction]:
+            header_string += "  " + item
+        header_string = header_string.rstrip() + "\n"
+        print_df = header_string + dataframe_tostring(print_df, True, False, False)  # type: ignore
+        print_df += "\n/\n"
+        print_df += f"\nUDQ\n  ASSIGN SUVTRIG {well_name} {segment_number} 1 /\n/\n"
+        action += print_df + "\nENDACTIO\n\n"
+
+    return action
+
+
 def print_wsegaicv(df_wsegaicv: pd.DataFrame, well_number: int) -> str:
     """Print for AICV devices.