Coverage for src/CSET/operators/__init__.py: 100%
89 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-09 15:26 +0000
1# © Crown copyright, Met Office (2022-2026) and CSET contributors.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
15"""Subpackage contains all of CSET's operators."""
17import inspect
18import json
19import logging
20import os
21import zipfile
22from pathlib import Path
24from iris import FUTURE
26# Import operators here so they are exported for use by recipes.
27import CSET.operators
28from CSET.operators import (
29 ageofair,
30 aggregate,
31 aviation,
32 collapse,
33 constraints,
34 convection,
35 ensembles,
36 feature,
37 filters,
38 fluxes,
39 humidity,
40 imageprocessing,
41 mesoscale,
42 misc,
43 plot,
44 power_spectrum,
45 precipitation,
46 pressure,
47 read,
48 regrid,
49 scoreswrappers,
50 temperature,
51 transect,
52 wind,
53 write,
54)
56# Exported operators & functions to use elsewhere.
57__all__ = [
58 "ageofair",
59 "aggregate",
60 "aviation",
61 "collapse",
62 "constraints",
63 "convection",
64 "ensembles",
65 "execute_recipe",
66 "feature",
67 "filters",
68 "fluxes",
69 "humidity",
70 "get_operator",
71 "imageprocessing",
72 "mesoscale",
73 "misc",
74 "plot",
75 "power_spectrum",
76 "precipitation",
77 "pressure",
78 "read",
79 "regrid",
80 "temperature",
81 "scoreswrappers",
82 "transect",
83 "wind",
84 "write",
85]
87# Stop iris giving a warning whenever it loads something.
88FUTURE.datum_support = True
89# Stop iris giving a warning whenever it saves something.
90FUTURE.save_split_attrs = True
91# Accept microsecond precision in iris times.
92FUTURE.date_microseconds = True
95def get_operator(name: str):
96 """Get an operator by its name.
98 Parameters
99 ----------
100 name: str
101 The name of the desired operator.
103 Returns
104 -------
105 function
106 The named operator.
108 Raises
109 ------
110 ValueError
111 If name is not an operator.
113 Examples
114 --------
115 >>> CSET.operators.get_operator("read.read_cubes")
116 <function read_cubes at 0x7fcf9353c8b0>
117 """
118 logging.debug("get_operator(%s)", name)
119 try:
120 name_sections = name.split(".")
121 operator = CSET.operators
122 for section in name_sections:
123 operator = getattr(operator, section)
124 if callable(operator):
125 return operator
126 else:
127 raise AttributeError
128 except (AttributeError, TypeError) as err:
129 raise ValueError(f"Unknown operator: {name}") from err
132def _write_metadata(recipe: dict):
133 """Write a meta.json file in the CWD."""
134 metadata = recipe.copy()
135 # Remove steps, as not needed, and might contain non-serialisable types.
136 metadata.pop("steps", None)
137 # To remove long variable names with suffix
138 if "title" in metadata:
139 metadata["title"] = metadata["title"].replace("_for_climate_averaging", "")
140 metadata["title"] = metadata["title"].replace("_radiative_timestep", "")
141 metadata["title"] = metadata["title"].replace("_maximum_random_overlap", "")
142 with open("meta.json", "wt", encoding="UTF-8") as fp:
143 json.dump(metadata, fp, indent=2)
146def _step_parser(step: dict, step_input: any) -> str:
147 """Execute a recipe step, recursively executing any sub-steps."""
148 logging.debug("Executing step: %s", step)
149 kwargs = {}
150 for key in step.keys():
151 if key == "operator":
152 operator = get_operator(step["operator"])
153 logging.info("operator: %s", step["operator"])
154 elif isinstance(step[key], dict) and "operator" in step[key]:
155 logging.debug("Recursing into argument: %s", key)
156 kwargs[key] = _step_parser(step[key], step_input)
157 else:
158 kwargs[key] = step[key]
159 logging.debug("args: %s", kwargs)
160 logging.debug("step_input: %s", step_input)
161 # If first argument of operator is explicitly defined, use that rather
162 # than step_input. This is known through introspection of the operator.
163 first_arg = next(iter(inspect.signature(operator).parameters.keys()))
164 logging.debug("first_arg: %s", first_arg)
165 if first_arg not in kwargs:
166 logging.debug("first_arg not in kwargs, using step_input.")
167 return operator(step_input, **kwargs)
168 else:
169 logging.debug("first_arg in kwargs.")
170 return operator(**kwargs)
173def create_diagnostic_archive():
174 """Create archive for easy download of plots and data."""
175 output_directory: Path = Path.cwd()
176 archive_path = output_directory / "diagnostic.zip"
177 with zipfile.ZipFile(
178 archive_path, "w", compression=zipfile.ZIP_DEFLATED
179 ) as archive:
180 for file in output_directory.rglob("*"):
181 # Check the archive doesn't add itself.
182 if not file.samefile(archive_path):
183 archive.write(file, arcname=file.relative_to(output_directory))
186def execute_recipe(
187 recipe: dict,
188 output_directory: Path,
189 style_file: Path = None,
190 plot_resolution: int = None,
191 skip_write: bool = None,
192) -> None:
193 """Parse and executes the steps from a recipe file.
195 Parameters
196 ----------
197 recipe: dict
198 Parsed recipe.
199 output_directory: Path
200 Pathlike indicating desired location of output.
201 style_file: Path, optional
202 Path to a style file.
203 plot_resolution: int, optional
204 Resolution of plots in dpi.
205 skip_write: bool, optional
206 Skip saving processed output alongside plots.
208 Raises
209 ------
210 FileNotFoundError
211 The recipe or input file cannot be found.
212 FileExistsError
213 The output directory as actually a file.
214 ValueError
215 The recipe is not well formed.
216 TypeError
217 The provided recipe is not a stream or Path.
218 """
219 # Create output directory.
220 try:
221 output_directory.mkdir(parents=True, exist_ok=True)
222 except (FileExistsError, NotADirectoryError) as err:
223 logging.error("Output directory is a file. %s", output_directory)
224 raise err
225 steps = recipe["steps"]
227 # Execute the steps in a recipe.
228 original_working_directory = Path.cwd()
229 try:
230 os.chdir(output_directory)
231 logger = logging.getLogger(__name__)
232 diagnostic_log = logging.FileHandler(
233 filename="CSET.log", mode="w", encoding="UTF-8"
234 )
235 diagnostic_log.setFormatter(
236 logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")
237 )
238 logger.addHandler(diagnostic_log)
239 # Create metadata file used by some steps.
240 if style_file:
241 recipe["style_file_path"] = str(style_file)
242 if plot_resolution:
243 recipe["plot_resolution"] = plot_resolution
244 if skip_write:
245 recipe["skip_write"] = skip_write
246 _write_metadata(recipe)
248 # Execute the recipe.
249 step_input = None
250 for step in steps:
251 step_input = _step_parser(step, step_input)
252 logger.info("Recipe output:\n%s", step_input)
254 logger.info("Creating diagnostic archive.")
255 create_diagnostic_archive()
256 finally:
257 os.chdir(original_working_directory)