Type code in src/ert/resources

.mypy.ini

@@ -14,8 +14,10 @@ warn_redundant_casts = True
 strict_equality = True
 extra_checks = True
 
-
-exclude = (?x)(src/ert/resources | src/_ert/forward_model_runner)
+exclude = (?x)(
+    src/ert/resources/forward_models/res/script/ecl.*py
+    | src/ert/resources/.*/flow.py
+ )
 
 [mypy-scipy.*]
 ignore_missing_imports = True
@@ -47,6 +49,9 @@ ignore_missing_imports = True
 [mypy-colorama.*]
 ignore_missing_imports = True
 
+[mypy-yaml.*]
+ignore_missing_imports = True
+
 [mypy-ruamel.*]
 ignore_missing_imports = True
 

src/ert/resources/forward_models/template_render.py

@@ -2,14 +2,15 @@
 import argparse
 import json
 import os
+from typing import Any, Sequence
 
 import jinja2
 import yaml
 
 DEFAULT_GEN_KW_EXPORT_NAME = "parameters"
 
 
-def load_data(filename):
+def load_data(filename: str) -> dict[str, Any]:
     """Will try to load data from @filename first as yaml, and if that fails,
     as json. If both fail, a ValueError with both of the error messages will be
     raised.
@@ -32,18 +33,18 @@ def load_data(filename):
     )
 
 
-def _load_template(template_path):
+def _load_template(template_path: str) -> jinja2.Template:
     path, filename = os.path.split(template_path)
     return jinja2.Environment(
         loader=jinja2.FileSystemLoader(path or "./")
     ).get_template(filename)
 
 
-def _generate_file_namespace(filename):
+def _generate_file_namespace(filename: str) -> str:
     return os.path.splitext(os.path.basename(filename))[0]
 
 
-def _load_input(input_files):
+def _load_input(input_files: Sequence[str]) -> dict[str, Any]:
     """
     Loads input files (JSON or YAML) and returns the content as dict.
     """
@@ -55,7 +56,9 @@ def _load_input(input_files):
     return data
 
 
-def _assert_input(input_files, template_file, output_file):
+def _assert_input(
+    input_files: Sequence[str], template_file: str, output_file: str
+) -> None:
     """
     validates input for template rendering.
     Throws ValueError if input files or template file is not found.
@@ -72,7 +75,9 @@ def _assert_input(input_files, template_file, output_file):
         raise TypeError("Expected output path to be a string")
 
 
-def render_template(input_files, template_file, output_file):
+def render_template(
+    input_files: Sequence[str], template_file: str, output_file: str
+) -> None:
     """
     Will render a jinja2 template file with the parameters given
     :param input_files: parameters as a list of JSON or YAML files
@@ -82,7 +87,7 @@ def render_template(input_files, template_file, output_file):
     if isinstance(input_files, str) and input_files:
         input_files = (input_files,)
 
-    all_input_files = ()
+    all_input_files: tuple[str, ...] = ()
 
     gen_kw_export_path = DEFAULT_GEN_KW_EXPORT_NAME + ".json"
     if os.path.isfile(gen_kw_export_path):
@@ -99,7 +104,7 @@ def render_template(input_files, template_file, output_file):
         fout.write(template.render(**data))
 
 
-def _build_argument_parser():
+def _build_argument_parser() -> argparse.ArgumentParser:
     description = """
 Loads the data from each file ("some/path/filename.xxx") in INPUT_FILES
 and exposes it as the variable "filename". It then loads the Jinja2

src/ert/resources/shell_scripts/careful_copy_file.py

@@ -4,7 +4,7 @@
 import sys
 
 
-def careful_copy_file(src, target=None):
+def careful_copy_file(src: str, target: str | None = None) -> None:
     if target is None:
         target = os.path.basename(src)
     if os.path.exists(target):

src/ert/resources/shell_scripts/copy_directory.py

@@ -3,10 +3,10 @@
 import shutil
 import sys
 
-from make_directory import mkdir
+from make_directory import mkdir  # type: ignore
 
 
-def copy_directory(src_path, target_path):
+def copy_directory(src_path: str, target_path: str) -> None:
     if os.path.isdir(src_path):
         src_basename = os.path.basename(src_path)
         target_root, _ = os.path.split(target_path)

src/ert/resources/shell_scripts/copy_file.py

@@ -4,7 +4,7 @@
 import sys
 
 
-def copy_file(src, target=None):
+def copy_file(src: str, target: str | None = None) -> None:
     if os.path.isfile(src):
         if target is None:
             target = os.path.basename(src)

src/ert/resources/shell_scripts/delete_directory.py

@@ -3,7 +3,7 @@
 import sys
 
 
-def delete_file(filename):
+def delete_file(filename: str) -> None:
     stat_info = os.lstat(filename)
     uid = stat_info.st_uid
     if uid == os.getuid():
@@ -13,7 +13,7 @@ def delete_file(filename):
         sys.stderr.write(f"Sorry you are not owner of file:{filename} - not deleted\n")
 
 
-def delete_empty_directory(dirname):
+def delete_empty_directory(dirname: str) -> None:
     stat_info = os.stat(dirname)
     uid = stat_info.st_uid
     if uid == os.getuid():
@@ -37,7 +37,7 @@ def delete_empty_directory(dirname):
         )
 
 
-def delete_directory(path):
+def delete_directory(path: str) -> None:
     """
     Will ignore if you are not owner.
     """

src/ert/resources/shell_scripts/delete_file.py

@@ -3,7 +3,7 @@
 import sys
 
 
-def delete_file(filename):
+def delete_file(filename: str) -> None:
     if os.path.exists(filename):
         if os.path.isfile(filename):
             stat_info = os.stat(filename)

src/ert/resources/shell_scripts/make_directory.py

@@ -3,7 +3,7 @@
 import sys
 
 
-def mkdir(path):
+def mkdir(path: str) -> None:
     if os.path.isdir(path):
         print(f"OK - directory: '{path}' already exists")
     else:

src/ert/resources/shell_scripts/move_directory.py

@@ -4,7 +4,7 @@
 import sys
 
 
-def move_directory(src_dir, target):
+def move_directory(src_dir: str, target: str) -> None:
     """
     Will raise IOError if src_dir is not a folder.
 

src/ert/resources/shell_scripts/move_file.py

@@ -4,7 +4,7 @@
 import sys
 
 
-def move_file(src_file, target):
+def move_file(src_file: str, target: str) -> None:
     """
     Will raise IOError if src_file is not a file.
 

src/ert/resources/shell_scripts/symlink.py

@@ -3,7 +3,7 @@
 import sys
 
 
-def symlink(target, link_name):
+def symlink(target: str, link_name: str) -> None:
     """Will create a symbol link 'link_name -> target'.
 
     If the @link_name already exists as a symbolic link it will be

src/ert/resources/workflows/jobs/internal-gui/scripts/csv_export.py

@@ -1,12 +1,15 @@
 import json
 import os
+from typing import Sequence
 
 import pandas
 
 from ert import ErtScript, LibresFacade
+from ert.config import ErtConfig
+from ert.storage import Storage
 
 
-def loadDesignMatrix(filename) -> pandas.DataFrame:
+def loadDesignMatrix(filename: str) -> pandas.DataFrame:
     dm = pandas.read_csv(filename, delim_whitespace=True)
     dm = dm.rename(columns={dm.columns[0]: "Realization"})
     dm = dm.set_index(["Realization"])
@@ -37,22 +40,22 @@ class CSVExportJob(ErtScript):
     """
 
     @staticmethod
-    def getName():
+    def getName() -> str:
         return "CSV Export"
 
     @staticmethod
-    def getDescription():
+    def getDescription() -> str:
         return (
             "Export GenKW, design matrix, misfit data "
             "and summary data into a single CSV file."
         )
 
     def run(
         self,
-        ert_config,
-        storage,
-        workflow_args,
-    ):
+        ert_config: ErtConfig,
+        storage: Storage,
+        workflow_args: Sequence[str],
+    ) -> str:
         output_file = workflow_args[0]
         ensemble_data_as_json = None if len(workflow_args) < 2 else workflow_args[1]
         design_matrix_path = None if len(workflow_args) < 3 else workflow_args[2]
@@ -67,6 +70,7 @@ def run(
         # Use the keys (UUIDs as strings) to get ensembles
         ensembles = []
         for ensemble_id in ensemble_data_as_dict:
+            assert self.storage is not None
             ensemble = self.storage.get_ensemble(ensemble_id)
             ensembles.append(ensemble)
 

src/ert/resources/workflows/jobs/internal-gui/scripts/gen_data_rft_export.py

@@ -1,16 +1,18 @@
 import contextlib
 import json
 import os
+from typing import Any, Sequence
 
 import numpy
 import pandas as pd
 import polars
-from qtpy.QtWidgets import QCheckBox
+from qtpy.QtWidgets import QCheckBox, QWidget
 
 from ert.config import CancelPluginException, ErtPlugin
+from ert.storage import Storage
 
 
-def load_args(filename, column_names=None):
+def load_args(filename: str, column_names: list[str] | None = None) -> pd.DataFrame:
     rows = 0
     columns = 0
     with open(filename, "r", encoding="utf-8") as fileH:
@@ -67,18 +69,18 @@ class GenDataRFTCSVExportJob(ErtPlugin):
     """
 
     @staticmethod
-    def getName():
+    def getName() -> str:
         return "GEN_DATA RFT CSV Export"
 
     @staticmethod
-    def getDescription():
+    def getDescription() -> str:
         return "Export gen_data RFT results into a single CSV file."
 
     def run(
         self,
-        storage,
-        workflow_args,
-    ):
+        storage: Storage,
+        workflow_args: Sequence[str],
+    ) -> str:
         """The run method will export the RFT's for all wells and all ensembles."""
 
         output_file = workflow_args[0]
@@ -202,7 +204,7 @@ def run(
         )
         return export_info
 
-    def getArguments(self, parent, storage):
+    def getArguments(self, parent: QWidget, storage: Storage) -> list[Any]:  # type: ignore
         # Importing ert.gui on-demand saves ~0.5 seconds off `from ert import __main__`
         from ert.gui.ertwidgets import (  # noqa: PLC0415
             CustomDialog,