Add step plot for rate vectors

This adds a key util function is_rate which determines whether the key
represents a rate or not. Source resdata.

src/ert/gui/plottery/plots/ensemble.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Dict
+from typing import TYPE_CHECKING, Dict, Optional
 
 import numpy as np
 import pandas as pd
 
 from ert.gui.plottery.plots.history import plotHistory
 from ert.gui.tools.plot.plot_api import EnsembleObject
+from ert.shared.storage.key_utils import is_rate
 
 from .observations import plotObservations
 from .plot_tools import PlotTools
@@ -45,11 +46,13 @@ def plot(
                     plot_context.deactivateDateSupport()
                     plot_context.x_axis = plot_context.INDEX_AXIS
 
+                draw_style = "steps" if is_rate(plot_context.key) else None
                 self._plotLines(
                     axes,
                     config,
                     data,
                     f"{ensemble.experiment_name} : {ensemble.name}",
+                    draw_style,
                 )
                 config.nextColor()
 
@@ -71,6 +74,7 @@ def _plotLines(
         plot_config: PlotConfig,
         data: pd.DataFrame,
         ensemble_label: str,
+        draw_style: Optional[str] = None,
     ) -> None:
         style = plot_config.defaultStyle()
 
@@ -86,6 +90,7 @@ def _plotLines(
             linewidth=style.width,
             linestyle=style.line_style,
             markersize=style.size,
+            drawstyle=draw_style,
         )
 
         if len(lines) > 0:

src/ert/shared/storage/key_utils.py

@@ -0,0 +1,213 @@
+from enum import Enum, auto
+
+special_vars = [
+    "NAIMFRAC",
+    "NBAKFL",
+    "NBYTOT",
+    "NCPRLINS",
+    "NEWTFL",
+    "NEWTON",
+    "NLINEARP",
+    "NLINEARS",
+    "NLINSMAX",
+    "NLINSMIN",
+    "NLRESMAX",
+    "NLRESSUM",
+    "NMESSAGE",
+    "NNUMFL",
+    "NNUMST",
+    "NTS",
+    "NTSECL",
+    "NTSMCL",
+    "NTSPCL",
+    "ELAPSED",
+    "MAXDPR",
+    "MAXDSO",
+    "MAXDSG",
+    "MAXDSW",
+    "STEPTYPE",
+    "WNEWTON",
+]
+rate_vars = [
+    "OPR",
+    "OIR",
+    "OVPR",
+    "OVIR",
+    "OFR",
+    "OPP",
+    "OPI",
+    "OMR",
+    "GPR",
+    "GIR",
+    "GVPR",
+    "GVIR",
+    "GFR",
+    "GPP",
+    "GPI",
+    "GMR",
+    "WGPR",
+    "WGIR",
+    "WPR",
+    "WIR",
+    "WVPR",
+    "WVIR",
+    "WFR",
+    "WPP",
+    "WPI",
+    "WMR",
+    "LPR",
+    "LFR",
+    "VPR",
+    "VIR",
+    "VFR",
+    "GLIR",
+    "RGR",
+    "EGR",
+    "EXGR",
+    "SGR",
+    "GSR",
+    "FGR",
+    "GIMR",
+    "GCR",
+    "NPR",
+    "NIR",
+    "CPR",
+    "CIR",
+    "SIR",
+    "SPR",
+    "TIR",
+    "TPR",
+    "GOR",
+    "WCT",
+    "OGR",
+    "WGR",
+    "GLR",
+]
+
+seg_rate_vars = [
+    "OFR",
+    "GFR",
+    "WFR",
+    "CFR",
+    "SFR",
+    "TFR",
+    "CVPR",
+    "WCT",
+    "GOR",
+    "OGR",
+    "WGR",
+]
+
+
+class SummaryVarType(Enum):
+    RD_SMSPEC_INVALID_VAR = auto()
+    RD_SMSPEC_FIELD_VAR = auto()
+    RD_SMSPEC_REGION_VAR = auto()
+    RD_SMSPEC_GROUP_VAR = auto()
+    RD_SMSPEC_WELL_VAR = auto()
+    RD_SMSPEC_SEGMENT_VAR = auto()
+    RD_SMSPEC_BLOCK_VAR = auto()
+    RD_SMSPEC_AQUIFER_VAR = auto()
+    RD_SMSPEC_COMPLETION_VAR = auto()
+    RD_SMSPEC_NETWORK_VAR = auto()
+    RD_SMSPEC_REGION_2_REGION_VAR = auto()
+    RD_SMSPEC_LOCAL_BLOCK_VAR = auto()
+    RD_SMSPEC_LOCAL_COMPLETION_VAR = auto()
+    RD_SMSPEC_LOCAL_WELL_VAR = auto()
+    RD_SMSPEC_MISC_VAR = auto()
+
+    @staticmethod
+    def determine_var_type(var) -> "SummaryVarType":
+        # Default case for single characters or unknown vars
+        default_case = {
+            "A": SummaryVarType.RD_SMSPEC_AQUIFER_VAR,
+            "B": SummaryVarType.RD_SMSPEC_BLOCK_VAR,
+            "C": SummaryVarType.RD_SMSPEC_COMPLETION_VAR,
+            "F": SummaryVarType.RD_SMSPEC_FIELD_VAR,
+            "G": SummaryVarType.RD_SMSPEC_GROUP_VAR,
+            "N": SummaryVarType.RD_SMSPEC_NETWORK_VAR,
+            "S": SummaryVarType.RD_SMSPEC_SEGMENT_VAR,
+            "W": SummaryVarType.RD_SMSPEC_WELL_VAR,
+        }
+
+        # Special case for 'L' with nested logic
+        if var.startswith("L"):
+            secondary = var[1] if len(var) > 1 else None
+            return {
+                "B": SummaryVarType.RD_SMSPEC_LOCAL_BLOCK_VAR,
+                "C": SummaryVarType.RD_SMSPEC_LOCAL_COMPLETION_VAR,
+                "W": SummaryVarType.RD_SMSPEC_LOCAL_WELL_VAR,
+            }.get(secondary, SummaryVarType.RD_SMSPEC_MISC_VAR)
+
+        # Special case for 'R' with more complex nested logic
+        if var.startswith("R"):
+            if len(var) == 3 and var[2] == "F":
+                return SummaryVarType.RD_SMSPEC_REGION_2_REGION_VAR
+            if var == "RNLF":
+                return SummaryVarType.RD_SMSPEC_REGION_2_REGION_VAR
+            if var == "RORFR":
+                return SummaryVarType.RD_SMSPEC_REGION_VAR
+            if len(var) >= 4 and var[2] == "F" and var[3] in {"T", "R"}:
+                return SummaryVarType.RD_SMSPEC_REGION_2_REGION_VAR
+            if len(var) >= 5 and var[3] == "F" and var[4] in {"T", "R"}:
+                return SummaryVarType.RD_SMSPEC_REGION_2_REGION_VAR
+            return SummaryVarType.RD_SMSPEC_REGION_VAR
+
+        # Fallback to default cases or miscellaneous if not matched
+        return default_case.get(var[0], SummaryVarType.RD_SMSPEC_MISC_VAR)
+
+
+def identify_var_type(key: str) -> SummaryVarType:
+    if key not in special_vars:
+        return SummaryVarType.determine_var_type(key)
+    return SummaryVarType.RD_SMSPEC_MISC_VAR
+
+
+def match_keyword_vector(start, rate_vars, keyword) -> bool:
+    """
+    Check if 'keyword' matches any item in 'rate_vars' starting from
+    index 'start' onwards.
+    """
+    # Get the suffix of keyword starting from 'start' index (if not out of range)
+    suffix = keyword[start:] if len(keyword) > start else ""
+    return any(suffix.startswith(var) for var in rate_vars)
+
+
+def match_keyword_string(start, rate_string, keyword) -> bool:
+    """
+    Check if 'keyword' matches 'rate_string' starting from index 'start'.
+    """
+    return keyword[start:].startswith(rate_string)
+
+
+def is_rate(key: str) -> bool:
+    var_type = identify_var_type(key)
+    if var_type in {
+        SummaryVarType.RD_SMSPEC_WELL_VAR,
+        SummaryVarType.RD_SMSPEC_GROUP_VAR,
+        SummaryVarType.RD_SMSPEC_FIELD_VAR,
+        SummaryVarType.RD_SMSPEC_REGION_VAR,
+        SummaryVarType.RD_SMSPEC_COMPLETION_VAR,
+        SummaryVarType.RD_SMSPEC_LOCAL_WELL_VAR,
+        SummaryVarType.RD_SMSPEC_LOCAL_COMPLETION_VAR,
+        SummaryVarType.RD_SMSPEC_NETWORK_VAR,
+    }:
+        # Process local and network vars differently
+        if var_type in {
+            SummaryVarType.RD_SMSPEC_LOCAL_WELL_VAR,
+            SummaryVarType.RD_SMSPEC_LOCAL_COMPLETION_VAR,
+            SummaryVarType.RD_SMSPEC_NETWORK_VAR,
+        }:
+            return match_keyword_vector(2, rate_vars, key)
+        return match_keyword_vector(1, rate_vars, key)
+
+    if var_type == SummaryVarType.RD_SMSPEC_SEGMENT_VAR:
+        return match_keyword_vector(1, seg_rate_vars, key)
+
+    if var_type == SummaryVarType.RD_SMSPEC_REGION_2_REGION_VAR:
+        # Region to region rates are identified by R*FR or R**FR
+        if match_keyword_string(2, "FR", key):
+            return True
+        return match_keyword_string(3, "FR", key)
+
+    return False