Make streams configurable at dataset level

CMEW/app/configure_for/bin/test_update_recipe_file.py

@@ -38,6 +38,15 @@ def mock_env_vars(monkeypatch):
         "DATASETS_LIST_DIR",
         str(Path(__file__).parent.parent.parent / "unittest" / "mock_data"),
     )
+    monkeypatch.setenv(
+        "MODEL_RUNS_CONFIG",
+        str(
+            Path(__file__).parent.parent.parent
+            / "unittest"
+            / "mock_data"
+            / "model_runs.yml"
+        ),
+    )
 
 
 @pytest.fixture

CMEW/app/configure_for/bin/update_recipe_file.py

@@ -72,7 +72,12 @@ def add_extra_datasets(recipe, yaml_filepath):
     }
 
     # Some attributes are neither needed nor wanted by ESMValTool
-    unwanted_keys = ["calendar", "suite_id"]
+    unwanted_keys = [
+        "calendar",
+        "path_to_custom_streams_config",
+        "suite_id",
+        "use_custom_data_streams",
+    ]
 
     # Convert the variable names in the extra datasets
     for dataset, inner_dict in extra_datasets.items():
@@ -179,7 +184,7 @@ def main():
     amended_recipe = remove_additional_datasets(blank_recipe)
 
     # Add the model runs into the datasets section of the recipe
-    model_runs_fp = f"{os.environ['DATASETS_LIST_DIR']}/model_runs.yml"
+    model_runs_fp = os.environ["MODEL_RUNS_CONFIG"]
     logger.info("Adding model runs to recipe")
     updated_recipe = add_extra_datasets(amended_recipe, model_runs_fp)
 

CMEW/app/configure_standardise/bin/create_request_file.py

@@ -10,6 +10,7 @@
 from pathlib import Path
 import yaml
 import logging
+from determine_streams_config import determine_stream_config_fp
 
 logging.basicConfig(level=logging.INFO, stream=sys.stdout)
 filename = os.path.basename(__file__)
@@ -49,7 +50,7 @@ def list_streams():
         Space separated list of all streams.
     """
     # Get path to stream mappings
-    streams_config = os.environ["STREAM_CONFIG_PATH"]
+    streams_config = determine_stream_config_fp()
 
     # Read the stream mappings
     with open(streams_config, "r") as f:
@@ -90,7 +91,7 @@ def create_request(model_run):
     mip_table_dir = os.environ["MIP_TABLE_DIR"]
 
     # Read the model run information from the model_runs.yml file
-    model_runs_yaml = Path(os.environ["DATASETS_LIST_DIR"]) / "model_runs.yml"
+    model_runs_yaml = os.environ["MODEL_RUNS_CONFIG"]
     with open(model_runs_yaml, "r") as f:
         dataset_dict = yaml.safe_load(f)[model_run]
     logger.debug(

CMEW/app/configure_standardise/bin/create_variables_file.py

@@ -8,6 +8,7 @@
 import yaml
 import sys
 import logging
+from determine_streams_config import determine_stream_config_fp
 
 
 logging.basicConfig(level=logging.INFO, stream=sys.stdout)
@@ -46,15 +47,17 @@ def combine_variable_lists(directory):
 
 def load_stream_dict():
     """
-    Loads stream information from the ../etc/streams.yml file.
+    Loads stream information for the dataset.
+
+    This will either be a custom filepath or the ../etc/streams.yml file.
 
     Returns
     -------
     dict
         A mapping of pre-defined streams to their associated variables
     """
     # Get path to stream mappings
-    streams_config = os.environ["STREAM_CONFIG_PATH"]
+    streams_config = determine_stream_config_fp()
     logger.debug("Reading streams from %s", streams_config)
 
     # Read the stream mappings

CMEW/app/configure_standardise/bin/determine_streams_config.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python
+# (C) Crown Copyright 2026, Met Office.
+# The LICENSE.md file contains full licensing details.
+"""
+Determines the stream config used to create the request and variables files.
+"""
+import os
+import sys
+import yaml
+import logging
+
+logging.basicConfig(level=logging.INFO, stream=sys.stdout)
+filename = os.path.basename(__file__)
+logger = logging.getLogger(filename)
+
+
+def determine_stream_config_fp():
+    """
+    Provide a path to pp file stream configuration data.
+
+    Uses the value of key `use_custom_data_streams` in the dataset information,
+    and returns either `path_to_custom_streams_config` or a default value.
+    The dataset is determined by variable `CYLC_TASK_PARAM_dataset`.
+
+    Returns
+    -------
+    str
+        Filepath to stream information for the dataset.
+    """
+    # Load the dataset information from model_runs YAML
+    dataset = os.environ["CYLC_TASK_PARAM_dataset"].strip()
+    with open(os.environ["MODEL_RUNS_CONFIG"]) as f:
+        content = yaml.safe_load(f)
+    dataset_dict = content[dataset]
+    logger.debug("Dataset information:\n%s", dataset_dict)
+
+    # Check whether a custom stream map is set to be used
+    if dataset_dict.get("use_custom_data_streams"):
+        streams_config = dataset_dict["path_to_custom_streams_config"]
+    else:
+        streams_config = os.environ["DEFAULT_STREAM_CONFIG_PATH"]
+    logger.info(
+        "Stream info for dataset % will use: %s", dataset, streams_config
+    )
+
+    return streams_config

CMEW/app/configure_standardise/bin/test_create_request_file.py

@@ -14,14 +14,25 @@
 
 def test_create_request(monkeypatch):
     monkeypatch.setenv(
-        "DATASETS_LIST_DIR",
-        str(Path(__file__).parent.parent.parent / "unittest" / "mock_data"),
+        "MODEL_RUNS_CONFIG",
+        str(
+            Path(__file__).parent.parent.parent
+            / "unittest"
+            / "mock_data"
+            / "model_runs.yml"
+        ),
     )
+    monkeypatch.setenv("CYLC_TASK_PARAM_dataset", "u-bv526")
 
     request_defaults_path = (
         Path(__file__).parent.parent / "etc" / "request_defaults.yml"
     )
-    stream_config_path = Path(__file__).parent.parent / "etc" / "streams.yml"
+    stream_config_path = (
+        Path(__file__).parent.parent.parent
+        / "unittest"
+        / "mock_data"
+        / "streams.yml"
+    )
     root_proc_dir = "/path/to/proc/dir/"
     root_data_dir = "/path/to/data/dir/"
     variables_path = "/path/to/variables.txt"
@@ -30,7 +41,7 @@ def test_create_request(monkeypatch):
 
     monkeypatch.setenv("RAW_DATA_DIR_MODE", "use_saved")
     monkeypatch.setenv("REQUEST_DEFAULTS_PATH", str(request_defaults_path))
-    monkeypatch.setenv("STREAM_CONFIG_PATH", str(stream_config_path))
+    monkeypatch.setenv("DEFAULT_STREAM_CONFIG_PATH", str(stream_config_path))
     monkeypatch.setenv("ROOT_PROC_DIR", root_proc_dir)
     monkeypatch.setenv("ROOT_DATA_DIR", root_data_dir)
     monkeypatch.setenv("VARIABLES_PATH", variables_path)

CMEW/app/configure_standardise/bin/test_create_variables_file.py

@@ -58,8 +58,25 @@ def test_combine_variable_lists():
 
 
 def test_add_stream_to_variables(monkeypatch, path_to_combined_variables):
-    stream_config_path = Path(__file__).parent.parent / "etc" / "streams.yml"
-    monkeypatch.setenv("STREAM_CONFIG_PATH", str(stream_config_path))
+    default_stream_config_path = (
+        Path(__file__).parent.parent.parent
+        / "unittest"
+        / "mock_data"
+        / "streams.yml"
+    )
+    monkeypatch.setenv(
+        "MODEL_RUNS_CONFIG",
+        str(
+            Path(__file__).parent.parent.parent
+            / "unittest"
+            / "mock_data"
+            / "model_runs.yml"
+        ),
+    )
+    monkeypatch.setenv("CYLC_TASK_PARAM_dataset", "u-bv526")
+    monkeypatch.setenv(
+        "DEFAULT_STREAM_CONFIG_PATH", str(default_stream_config_path)
+    )
     input = [
         "Amon/hfls",
         "Amon/hfss",

CMEW/app/configure_standardise/bin/test_determine_streams_config.py

@@ -0,0 +1,42 @@
+# (C) Crown Copyright 2026, Met Office.
+# The LICENSE.md file contains full licensing details.
+"""
+Unit tests for determine_streams_config.py
+
+Test data files:
+/app/unittest/mock_data/model_run_stream_config.yml
+    input for test_create_request
+"""
+import pytest
+from pathlib import Path
+from determine_streams_config import determine_stream_config_fp
+
+
+@pytest.fixture
+def mock_env_vars(monkeypatch):
+    monkeypatch.setenv(
+        "MODEL_RUNS_CONFIG",
+        str(
+            Path(__file__).parent.parent.parent
+            / "unittest"
+            / "mock_data"
+            / "model_run_stream_config.yml"
+        ),
+    )
+    monkeypatch.setenv(
+        "DEFAULT_STREAM_CONFIG_PATH", "/path/to/default_streams.yml"
+    )
+
+
+def test_determine_stream_config_fp_custom(mock_env_vars, monkeypatch):
+    monkeypatch.setenv("CYLC_TASK_PARAM_dataset", "run-1")
+    expected = "/path/to/custom_streams.yml"
+    actual = determine_stream_config_fp()
+    assert actual == expected
+
+
+def test_determine_stream_config_fp_default(mock_env_vars, monkeypatch):
+    monkeypatch.setenv("CYLC_TASK_PARAM_dataset", "run-2")
+    expected = "/path/to/default_streams.yml"
+    actual = determine_stream_config_fp()
+    assert actual == expected

CMEW/app/configure_standardise/etc/streams.yml

@@ -1,17 +1,21 @@
 # (C) Crown Copyright 2026, Met Office.
 # The LICENSE.md file contains full licensing details.
+apa:
+  - day/pr
+ape:
+  - day/ua
 apm:
-- Amon/hfls
-- Amon/hfss
-- Amon/rlds
-- Amon/rlut
-- Amon/rlutcs
-- Amon/rsds
-- Amon/rsdt
-- Amon/rsut
-- Amon/rsutcs
-- Amon/tas
-- Emon/rls
-- Emon/rss
+  - Amon/hfls
+  - Amon/hfss
+  - Amon/rlds
+  - Amon/rlut
+  - Amon/rlutcs
+  - Amon/rsds
+  - Amon/rsdt
+  - Amon/rsut
+  - Amon/rsutcs
+  - Amon/tas
+  - Emon/rls
+  - Emon/rss
 inm:
-- SImon/siconc
+  - SImon/siconc

CMEW/app/copy_datasets/bin/add_datasets_to_share.py

@@ -311,6 +311,32 @@ def add_reference_key(filepath):
         yaml.dump(dataset_dict, f)
 
 
+def format_as_bools(filepath):
+    """
+    Format second level key true or false strings as booleans.
+
+    Parameters
+    ----------
+    filepath: str
+        The location of the YAML file to be edited.
+    """
+    # Read the yaml as a dictionary without the extra key
+    with open(filepath, "r") as f:
+        dataset_dict = yaml.safe_load(f)
+
+    # Reformat true and false keys
+    for dataset in dataset_dict:
+        for key in dataset_dict[dataset]:
+            if str(dataset_dict[dataset][key]).lower() == "true":
+                dataset_dict[dataset][key] = True
+            elif str(dataset_dict[dataset][key]).lower() == "false":
+                dataset_dict[dataset][key] = False
+
+    # Re-save the file
+    with open(filepath, "w") as f:
+        yaml.dump(dataset_dict, f)
+
+
 def main():
     """Copy dataset information from configuration to the share directory."""
     # Read the target (shared) directory from the environment
@@ -346,14 +372,17 @@ def main():
 
     # Reformat the YAML files to use unique identifiers as keys
     logger.info("Reformatting YAML files to use suite IDs as keys")
-    model_runs_yaml = f"{target_dir}/model_runs.yml"
+    model_runs_yaml = os.environ["MODEL_RUNS_CONFIG"]
     use_facet_as_key(model_runs_yaml, "suite_id")
     use_facet_as_key(f"{target_dir}/cmip6_datasets.yml", "model_id")
 
     # Add the reference identifier
     logger.info("Adding benchmarking key to model runs YAML")
     add_reference_key(model_runs_yaml)
 
+    # Reformat Cylc or Rose booleans as python booleans
+    format_as_bools(model_runs_yaml)
+
 
 if __name__ == "__main__":
     main()

CMEW/app/unittest/mock_data/model_run_stream_config.yml

@@ -0,0 +1,9 @@
+# (C) Crown Copyright 2026, Met Office.
+# The LICENSE.md file contains full licensing details.
+run-1:
+  path_to_custom_streams_config: /path/to/custom_streams.yml
+  use_custom_data_streams: true
+  other_key: something
+run-2:
+  use_custom_data_streams: false
+  other_key: something else

CMEW/app/unittest/mock_data/streams.yml

@@ -0,0 +1,17 @@
+# (C) Crown Copyright 2026, Met Office.
+# The LICENSE.md file contains full licensing details.
+apm:
+  - Amon/hfls
+  - Amon/hfss
+  - Amon/rlds
+  - Amon/rlut
+  - Amon/rlutcs
+  - Amon/rsds
+  - Amon/rsdt
+  - Amon/rsut
+  - Amon/rsutcs
+  - Amon/tas
+  - Emon/rls
+  - Emon/rss
+inm:
+  - SImon/siconc

CMEW/flow.cylc

@@ -85,7 +85,8 @@
             RAW_DATA_DIR = {{ RAW_DATA_DIR | default("") }}
             RAW_DATA_DIR_MODE = {{ RAW_DATA_DIR_MODE }}
             REQUEST_DEFAULTS_PATH = ${CYLC_WORKFLOW_RUN_DIR}/app/configure_standardise/etc/request_defaults.yml
-            STREAM_CONFIG_PATH = ${CYLC_WORKFLOW_RUN_DIR}/app/configure_standardise/etc/streams.yml
+            ROOT_RESTRUCTURED_DIR=${CYLC_WORKFLOW_SHARE_DIR}/work/
+            DEFAULT_STREAM_CONFIG_PATH = ${CYLC_WORKFLOW_RUN_DIR}/app/configure_standardise/etc/streams.yml
             # Workaround for bug in CDDS: ROOT_SOFTWARE_DIR: unbound variable.
             ROOT_SOFTWARE_DIR = ${CDDS_SOFTWARE_DIR}
             CDDS_VERSION = {{ CDDS_VERSION }}
@@ -99,7 +100,9 @@
     [[MODEL_RUNS]]
         [[[environment]]]
             # Directory containing dataset lists as YAML files
-            DATASETS_LIST_DIR = "${CYLC_WORKFLOW_SHARE_DIR}/dataset_lists"
+            DATASETS_LIST_DIR = ${CYLC_WORKFLOW_SHARE_DIR}/dataset_lists
+            # This particular file controls behaviours throughout the workflow
+            MODEL_RUNS_CONFIG = ${DATASETS_LIST_DIR}/model_runs.yml
             START_YEAR = {{ START_YEAR }}
             NUMBER_OF_YEARS = {{ NUMBER_OF_YEARS }}