Support glob pattern in seismic observations

Author: achaikou

src/ert/config/_observations.py

@@ -1076,7 +1076,7 @@ def from_obs_dict(
             shape_registry: ShapeRegistry for storing geometry.
         """
         name = ""
-        filepath: str | Path | None = None
+        filepath: str | None = None
         boundary_filepath: str | Path | None = None
 
         if "CSV" in observation_dict and "OBS_FILE" in observation_dict:
@@ -1105,22 +1105,6 @@ def from_obs_dict(
                 case _:
                     raise _unknown_key_error(str(key), observation_dict.context)
 
-        if filepath is None:
-            raise _missing_value_error(observation_dict.context, "OBS_FILE")
-
-        filepath = Path(directory) / filepath
-        if not filepath.exists():
-            raise ObservationConfigError.with_context(
-                f"The seismic observations file ({filepath.absolute()}) "
-                "does not exist or is not accessible.",
-                filepath,
-            )
-
-        if not name:
-            name = filepath.stem
-
-        df = cls._load_observations(filepath)
-
         boundary_id = None
         if boundary_filepath is not None:
             boundary_filepath = Path(directory) / boundary_filepath
@@ -1133,6 +1117,40 @@ def from_obs_dict(
             boundary = PolygonShapeConfig.from_file(str(boundary_filepath))
             boundary_id = shape_registry.register(boundary)
 
+        if filepath is None:
+            raise _missing_value_error(observation_dict.context, "OBS_FILE")
+
+        matching_filepaths = SeismicData.resolve_pattern_filepaths(
+            directory,
+            filepath,
+            on_error=lambda msg: ObservationConfigError.with_context(msg, filepath),
+        )
+        seismic_observations_for_filepath: list[Self] = []
+        for matching_filepath in matching_filepaths:
+            seismic_observations_for_filepath.extend(
+                cls._from_filepath(
+                    name=name,
+                    filepath=matching_filepath,
+                    boundary_id=boundary_id,
+                    shape_registry=shape_registry,
+                )
+            )
+        return seismic_observations_for_filepath
+
+    @classmethod
+    def _from_filepath(
+        cls,
+        name: str,
+        filepath: Path,
+        boundary_id: int | None,
+        shape_registry: ShapeRegistry,
+    ) -> list[Self]:
+
+        if not name:
+            name = filepath.stem
+
+        df = cls._load_observations(filepath)
+
         seismic_observations = []
         for row in df.iter_rows(named=True):
             east = validate_float(str(row["X_UTME"]), "X_UTME")

src/ert/config/_reservoir_data_utils.py

@@ -1,4 +1,7 @@
-from collections.abc import Sequence
+import fnmatch
+import re
+from collections.abc import Callable, Sequence
+from pathlib import Path
 from typing import ClassVar
 
 import numpy as np
@@ -83,3 +86,54 @@ def use_observation_locations_in_respective_responses(
             )
             .drop(["east_obs", "north_obs"])
         )
+
+    @staticmethod
+    def resolve_pattern_filepaths(
+        basedir: str,
+        pattern: str,
+        on_error: Callable[[str], Exception],
+    ) -> list[Path]:
+        """Resolve all file paths matching a pattern against filenames only.
+
+        The pattern can include a directory path (e.g., "../subdir/obs-*.csv"),
+        but the pattern itself is only matched against the filename component.
+        Literal glob metacharacters (* ?) should be wrapped in brackets: "test[*].csv".
+
+        Args:
+            basedir: Base directory.
+            pattern: Pattern with optional directory path and possible glob-style
+            wildcards in the filename.
+            on_error: Callable to raise a custom exception.
+
+        """
+        pattern_path = Path(pattern)
+        search_dir = Path(basedir) / pattern_path.parent
+        if not search_dir.exists():
+            raise on_error(
+                f"Directory '{search_dir.absolute()}' does not exist "
+                "or is not accessible."
+            )
+
+        filename_pattern = pattern_path.name
+        has_glob = any(char in filename_pattern for char in "*?[")
+        if not has_glob:
+            path = search_dir / filename_pattern
+            if not path.exists():
+                raise on_error(
+                    f"File '{path.absolute()}' does not exist or is not accessible."
+                )
+            return [path]
+
+        compiled_pattern = re.compile(fnmatch.translate(filename_pattern))
+        matching_paths = [
+            path
+            for path in search_dir.iterdir()
+            if path.is_file() and compiled_pattern.fullmatch(path.name)
+        ]
+
+        if not matching_paths:
+            raise on_error(
+                f"No files matching pattern '{filename_pattern}' found "
+                f"in '{search_dir}'"
+            )
+        return sorted(matching_paths)

tests/ert/unit_tests/config/test_observation_declaration.py

@@ -4,6 +4,7 @@
 from datetime import datetime
 from pathlib import Path
 from textwrap import dedent
+from typing import cast
 
 import hypothesis.extra.lark as stlark
 import polars as pl
@@ -16,6 +17,7 @@
     BreakthroughObservation,
     GeneralObservation,
     RFTObservation,
+    SeismicObservation,
     SummaryObservation,
     make_observations,
 )
@@ -862,10 +864,13 @@ def test_that_seismic_observation_instantiates(file_context_token):
     ]
 
 
+@pytest.mark.usefixtures("use_tmpdir")
 def test_that_non_existent_seismic_observation_file_raises_error(file_context_token):
+    directory = "dir"
+    Path(directory).mkdir()
     with pytest.raises(ObservationConfigError) as err:
         make_observations(
-            "dir",
+            directory,
             [
                 ObservationDict(
                     {
@@ -879,7 +884,7 @@ def test_that_non_existent_seismic_observation_file_raises_error(file_context_to
             shape_registry=ShapeRegistry(),
         )
 
-    assert "/dir/seismic_observations.csv) does not exist or is not accessible." in str(
+    assert "seismic_observations.csv' does not exist or is not accessible" in str(
         err.value
     )
 
@@ -1180,17 +1185,27 @@ def test_that_seismic_observation_coordinate_distance_below_tolerance_raises(
     )
 
 
-@pytest.mark.usefixtures("use_tmpdir")
-def test_that_seismic_observation_reads_boundary_file(file_context_token):
-    Path("obs.csv").write_text(
-        dedent(
-            """
-            X_UTME,Y_UTMN,OBS,OBS_ERROR,REGION
-            1.0,1.0,1.0,0.005,1.0
-            """
-        ),
+def default_seismic_file_content() -> str:
+    return dedent(
+        """
+        X_UTME,Y_UTMN,OBS,OBS_ERROR,REGION
+        1.0,1.0,1.0,0.005,1.0
+        """
+    )
+
+
+def write_default_seismic_file_content(
+    filename="horizon--amplitude_full_min_depth--20250101_20240101.csv",
+):
+    Path(filename).write_text(
+        default_seismic_file_content(),
         encoding="utf8",
     )
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_seismic_observation_reads_boundary_file(file_context_token):
+    write_default_seismic_file_content("obs.csv")
     Path("boundary.pol").write_text(
         dedent(
             """
@@ -1249,15 +1264,9 @@ def test_that_non_existent_boundary_seismic_observation_file_raises_error(
 ):
     os.makedirs("directory/right/path", exist_ok=True)
     os.makedirs("directory/wrong/path", exist_ok=True)
-    Path("directory/obs.csv").write_text(
-        dedent(
-            """
-            X_UTME,Y_UTMN,OBS,OBS_ERROR,REGION
-            1.0,1.0,1.0,0.005,1.0
-            """
-        ),
-        encoding="utf8",
-    )
+
+    write_default_seismic_file_content("directory/obs.csv")
+
     Path("directory/right/path/bound.pol").write_text(
         "Unexpected file location",
         encoding="utf8",
@@ -1283,3 +1292,97 @@ def test_that_non_existent_boundary_seismic_observation_file_raises_error(
         "/directory/wrong/path/bound.pol) does not exist or is not accessible."
         in str(err.value)
     )
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_seismic_observation_filenames_can_be_glob_pattern(file_context_token):
+    filename0 = "surface--amplitude_far_mean_depth--20190701_20180101.csv"
+    filename1 = "surface--amplitude_full_min_depth--20190901_20180101.csv"
+    filename2 = "surface--amplitude_full_min_depth--20180701_20180101.csv"
+    filename3 = ".surface--amplitude_full_mean_depth--20190701_20180101.csv.yml"
+
+    directory = "dir1/dir2/.."
+    os.makedirs(directory, exist_ok=True)
+
+    for filename in [filename0, filename1, filename2, filename3]:
+        write_default_seismic_file_content(f"{directory}/{filename}")
+
+    def make_observations_with_pattern(
+        pattern: str, directory: str = directory
+    ) -> list[SeismicObservation]:
+        shape_registry = ShapeRegistry()
+        obs = make_observations(
+            "",
+            [
+                ObservationDict(
+                    {
+                        "type": ObservationType.SEISMIC,
+                        "OBS_FILE": f"{directory}/{pattern}",
+                    },
+                    context=file_context_token(obs_type="SEISMIC_OBSERVATION"),
+                )
+            ],
+            shape_registry=shape_registry,
+        )
+        return [cast(SeismicObservation, o) for o in obs]
+
+    p1 = "surface--amplitude_*_*_depth--20190[1-9]01_20180101.csv"
+    obs = make_observations_with_pattern(p1)
+    assert len(obs) == 2
+    assert sorted([o.filepath for o in obs]) == sorted(
+        [
+            Path(f"{directory}/{filename0}"),
+            Path(f"{directory}/{filename1}"),
+        ]
+    )
+
+    p2 = "surface*"
+    obs = make_observations_with_pattern(p2)
+    assert len(obs) == 3
+    assert sorted([o.filepath for o in obs]) == sorted(
+        [
+            Path(f"{directory}/{filename0}"),
+            Path(f"{directory}/{filename1}"),
+            Path(f"{directory}/{filename2}"),
+        ]
+    )
+
+    p3 = p1[:-4]
+    with pytest.raises(ObservationConfigError) as err:
+        make_observations_with_pattern(p3)
+    assert f"No files matching pattern '{p3}' found in '{directory}'" in str(err.value)
+
+    with pytest.raises(ObservationConfigError) as err:
+        make_observations_with_pattern(p1, directory="ufo")
+    assert "ufo' does not exist or is not accessible" in str(err.value)
+
+
+@pytest.mark.usefixtures("use_tmpdir")
+def test_that_filepath_can_have_literal_metacharacters(file_context_token):
+    def make_observations_with_pattern(pattern: str) -> list[SeismicObservation]:
+        shape_registry = ShapeRegistry()
+        obs = make_observations(
+            "",
+            [
+                ObservationDict(
+                    {
+                        "type": ObservationType.SEISMIC,
+                        "OBS_FILE": pattern,
+                    },
+                    context=file_context_token(obs_type="SEISMIC_OBSERVATION"),
+                )
+            ],
+            shape_registry=shape_registry,
+        )
+        return [cast(SeismicObservation, o) for o in obs]
+
+    path_with_literal_wildcard = r"test*.csv"
+    path_fitting_to_wildcard = "test123.csv"
+    write_default_seismic_file_content(path_with_literal_wildcard)
+    write_default_seismic_file_content(path_fitting_to_wildcard)
+
+    wildcard_pattern = "test**"
+    literal_pattern = "test[*]*"
+
+    assert len(make_observations_with_pattern(pattern=wildcard_pattern)) == 2
+    assert len(make_observations_with_pattern(pattern=literal_pattern)) == 1

tests/ert/unit_tests/config/test_seismic_config.py

@@ -1,4 +1,5 @@
 from io import BytesIO, StringIO
+from pathlib import Path
 from typing import cast
 
 import polars as pl
@@ -31,13 +32,14 @@ def _mock_seismic_response(
         mocked_files[path] = buf.getvalue()
 
 
+@pytest.mark.usefixtures("use_tmpdir")
 @pytest.mark.parametrize("suffix", [".csv", ".parquet"])
 def test_that_seismic_observation_response_key_matches_simulated_response_key(
     mocked_files, suffix
 ):
     expected_response_key = "horizon--amplitude_full_min_depth--20250101_20240101"
     name = f"{expected_response_key}{suffix}"
-    runpath = "/runpath"
+    runpath = "runpath"
     obs_path = "share/preprocessed/tables/" + name
     simulated_path_relative_to_runpath = "share/results/tables/" + name
     simulated_path = runpath + "/" + simulated_path_relative_to_runpath
@@ -65,6 +67,16 @@ def test_that_seismic_observation_response_key_matches_simulated_response_key(
     _mock_seismic_response(mocked_files, obs_path, obs_frame, suffix)
     _mock_seismic_response(mocked_files, simulated_path, simulated_frame, suffix)
 
+    Path(obs_path).parent.mkdir(parents=True, exist_ok=True)
+    Path(simulated_path).parent.mkdir(parents=True, exist_ok=True)
+
+    if suffix.endswith("parquet"):
+        Path(obs_path).write_bytes(mocked_files[obs_path])
+        Path(simulated_path).write_bytes(mocked_files[simulated_path])
+    else:
+        Path(obs_path).write_text(mocked_files[obs_path], encoding="utf8")
+        Path(simulated_path).write_text(mocked_files[simulated_path], encoding="utf8")
+
     config = ErtConfig.from_dict(
         {
             "SEISMIC": [simulated_path_relative_to_runpath],
@@ -119,7 +131,7 @@ def test_that_seismic_config_raises_when_reading_from_non_existing_file(tmp_path
         keys=["key"],
     )
     with pytest.raises(InvalidResponseFile):
-        seismic_config.read_from_file(tmp_path / "non-existent-file.csv", 1, 1)
+        seismic_config.read_from_file(tmp_path, 1, 1)
 
 
 @pytest.mark.parametrize("suffix", [".csv", ".parquet"])