Source code for kappa_sdk.workflow_settings

from dataclasses import dataclass
from enum import Enum
from typing import Optional, Dict, Any, List, Union


[docs] class IptaParameter(Enum): C = "Wellbore storage" Skin = "Skin" k = "Permeability" h = "Thickness" phi = "Porosity" S = "South" E = "East" N = "North" W = "West" L = "Distance"
class IrtaParameterEnum(Enum): Skin = "-Skin" Xmf = "-Xmf" Pi = '-Pi' k = "-k" h = "-h" phi = "-ɸ" zone_1_d = "Zone 1-d" zone_1_ri = "Zone 1-Ri" zone_2_d = "Zone 2-d" @dataclass class ImproveTarget: """An improve target for an IPTA or IRTA workflow.""" target_type: str is_selected: bool target_well_id: str on_log: Optional[bool] = None global_weight: Optional[float] = None def dump(self) -> Dict[str, Any]: return { "targetType": self.target_type, "isSelected": self.is_selected, "targetWellId": self.target_well_id, "onLog": self.on_log, "globalWeight": self.global_weight, } @dataclass class ImproveParameter: """An improve parameter for an IPTA or IRTA workflow.""" name: str id: str is_included: bool value: float max_value: float min_value: float measure: str category: Optional[str] = None def dump(self) -> Dict[str, Any]: result: Dict[str, Any] = { "id": self.id, "isIncluded": self.is_included, "value": self.value, "maxValue": self.max_value, "minValue": self.min_value, } return result
[docs] class WorkflowImproveSettings:
[docs] def __init__(self, improve_parameters: List[ImproveParameter], use_all_parameters: bool = False, use_wide_search: bool = False, update_values: bool = False, update_bounds: bool = False, min_elapsed_time: Optional[float] = None, locate_and_ignore_wellbore_effect: bool = False, improve_targets: Optional[List[ImproveTarget]] = None): self.__parameters: List[ImproveParameter] = list(improve_parameters) if use_all_parameters: for parameter in self.__parameters: parameter.is_included = True self.__wide_search_settings: Dict[str, Any] = {"useWideSearch": use_wide_search, "initialPopulation": 50, "boundsStretch": 10} self.__update_values: bool = update_values self.__update_bounds: bool = update_bounds self.__min_elapsed_time: Optional[float] = min_elapsed_time self.__locate_and_ignore_wellbore_effect: bool = locate_and_ignore_wellbore_effect self.__improve_targets: Optional[List[ImproveTarget]] = improve_targets
@property def parameters(self) -> List[ImproveParameter]: """The improve parameters.""" return self.__parameters @property def improve_targets(self) -> Optional[List[ImproveTarget]]: """The improve targets, if any.""" return self.__improve_targets @property def min_elapsed_time(self) -> Optional[float]: return self.__min_elapsed_time @min_elapsed_time.setter def min_elapsed_time(self, value: Optional[float]) -> None: self.__min_elapsed_time = value
[docs] def select_parameter(self, parameter: Union[IptaParameter, IrtaParameterEnum], min_value: Optional[float] = None, max_value: Optional[float] = None) -> None: try: selected_parameter = next(x for x in self.__parameters if x.id == parameter.value) except StopIteration: raise ValueError(f"Cannot find {parameter.value} in the improve parameter list, make sure you are using the right type of parameter, could be IptaParameter or IrtaParameter") if min_value is not None: selected_parameter.min_value = min_value if max_value is not None: selected_parameter.max_value = max_value selected_parameter.is_included = True
[docs] def remove_parameter(self, parameter: Union[IptaParameter, IrtaParameterEnum]) -> None: try: selected_parameter = next(x for x in self.__parameters if x.id == parameter.value) except StopIteration: raise ValueError(f"Cannot find {parameter.value} in the improve parameter list, make sure you are using the right type of parameter, could be IptaParameter or IrtaParameter") selected_parameter.is_included = False
[docs] def update_wide_search_settings(self, initial_population: float = 50, bounds_stretch: float = 10) -> None: self.__wide_search_settings["initialPopulation"] = initial_population self.__wide_search_settings["boundsStretch"] = bounds_stretch
[docs] def dump(self) -> Dict[str, Any]: result: Dict[str, Any] = { "updateValues": self.__update_values, "updateBounds": self.__update_bounds, "nlrSettings": { "locateAndIgnoreWellboreEffect": self.__locate_and_ignore_wellbore_effect, "minElapsedTime": self.__min_elapsed_time, "wideSearchSettings": self.__wide_search_settings, }, "parameters": [p.dump() for p in self.__parameters], } if self.__improve_targets is not None: result["improveTargets"] = [t.dump() for t in self.__improve_targets] return result