Coverage for src/CSET/operators/__init__.py: 100%
89 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-06 11:34 +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 humidity,
39 imageprocessing,
40 mesoscale,
41 misc,
42 plot,
43 power_spectrum,
44 precipitation,
45 pressure,
46 read,
47 regrid,
48 scoreswrappers,
49 temperature,
50 transect,
51 wind,
52 write,
53)
55# Exported operators & functions to use elsewhere.
56__all__ = [
57 "ageofair",
58 "aggregate",
59 "aviation",
60 "collapse",
61 "constraints",
62 "convection",
63 "ensembles",
64 "execute_recipe",
65 "feature",
66 "filters",
67 "humidity",
68 "get_operator",
69 "imageprocessing",
70 "mesoscale",
71 "misc",
72 "plot",
73 "power_spectrum",
74 "precipitation",
75 "pressure",
76 "read",
77 "regrid",
78 "temperature",
79 "scoreswrappers",
80 "transect",
81 "wind",
82 "write",
83]
85# Stop iris giving a warning whenever it loads something.
86FUTURE.datum_support = True
87# Stop iris giving a warning whenever it saves something.
88FUTURE.save_split_attrs = True
89# Accept microsecond precision in iris times.
90FUTURE.date_microseconds = True
93def get_operator(name: str):
94 """Get an operator by its name.
96 Parameters
97 ----------
98 name: str
99 The name of the desired operator.
101 Returns
102 -------
103 function
104 The named operator.
106 Raises
107 ------
108 ValueError
109 If name is not an operator.
111 Examples
112 --------
113 >>> CSET.operators.get_operator("read.read_cubes")
114 <function read_cubes at 0x7fcf9353c8b0>
115 """
116 logging.debug("get_operator(%s)", name)
117 try:
118 name_sections = name.split(".")
119 operator = CSET.operators
120 for section in name_sections:
121 operator = getattr(operator, section)
122 if callable(operator):
123 return operator
124 else:
125 raise AttributeError
126 except (AttributeError, TypeError) as err:
127 raise ValueError(f"Unknown operator: {name}") from err
130def _write_metadata(recipe: dict):
131 """Write a meta.json file in the CWD."""
132 metadata = recipe.copy()
133 # Remove steps, as not needed, and might contain non-serialisable types.
134 metadata.pop("steps", None)
135 # To remove long variable names with suffix
136 if "title" in metadata:
137 metadata["title"] = metadata["title"].replace("_for_climate_averaging", "")
138 metadata["title"] = metadata["title"].replace("_radiative_timestep", "")
139 metadata["title"] = metadata["title"].replace("_maximum_random_overlap", "")
140 with open("meta.json", "wt", encoding="UTF-8") as fp:
141 json.dump(metadata, fp, indent=2)
144def _step_parser(step: dict, step_input: any) -> str:
145 """Execute a recipe step, recursively executing any sub-steps."""
146 logging.debug("Executing step: %s", step)
147 kwargs = {}
148 for key in step.keys():
149 if key == "operator":
150 operator = get_operator(step["operator"])
151 logging.info("operator: %s", step["operator"])
152 elif isinstance(step[key], dict) and "operator" in step[key]:
153 logging.debug("Recursing into argument: %s", key)
154 kwargs[key] = _step_parser(step[key], step_input)
155 else:
156 kwargs[key] = step[key]
157 logging.debug("args: %s", kwargs)
158 logging.debug("step_input: %s", step_input)
159 # If first argument of operator is explicitly defined, use that rather
160 # than step_input. This is known through introspection of the operator.
161 first_arg = next(iter(inspect.signature(operator).parameters.keys()))
162 logging.debug("first_arg: %s", first_arg)
163 if first_arg not in kwargs:
164 logging.debug("first_arg not in kwargs, using step_input.")
165 return operator(step_input, **kwargs)
166 else:
167 logging.debug("first_arg in kwargs.")
168 return operator(**kwargs)
171def create_diagnostic_archive():
172 """Create archive for easy download of plots and data."""
173 output_directory: Path = Path.cwd()
174 archive_path = output_directory / "diagnostic.zip"
175 with zipfile.ZipFile(
176 archive_path, "w", compression=zipfile.ZIP_DEFLATED
177 ) as archive:
178 for file in output_directory.rglob("*"):
179 # Check the archive doesn't add itself.
180 if not file.samefile(archive_path):
181 archive.write(file, arcname=file.relative_to(output_directory))
184def execute_recipe(
185 recipe: dict,
186 output_directory: Path,
187 style_file: Path = None,
188 plot_resolution: int = None,
189 skip_write: bool = None,
190) -> None:
191 """Parse and executes the steps from a recipe file.
193 Parameters
194 ----------
195 recipe: dict
196 Parsed recipe.
197 output_directory: Path
198 Pathlike indicating desired location of output.
199 style_file: Path, optional
200 Path to a style file.
201 plot_resolution: int, optional
202 Resolution of plots in dpi.
203 skip_write: bool, optional
204 Skip saving processed output alongside plots.
206 Raises
207 ------
208 FileNotFoundError
209 The recipe or input file cannot be found.
210 FileExistsError
211 The output directory as actually a file.
212 ValueError
213 The recipe is not well formed.
214 TypeError
215 The provided recipe is not a stream or Path.
216 """
217 # Create output directory.
218 try:
219 output_directory.mkdir(parents=True, exist_ok=True)
220 except (FileExistsError, NotADirectoryError) as err:
221 logging.error("Output directory is a file. %s", output_directory)
222 raise err
223 steps = recipe["steps"]
225 # Execute the steps in a recipe.
226 original_working_directory = Path.cwd()
227 try:
228 os.chdir(output_directory)
229 logger = logging.getLogger(__name__)
230 diagnostic_log = logging.FileHandler(
231 filename="CSET.log", mode="w", encoding="UTF-8"
232 )
233 diagnostic_log.setFormatter(
234 logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")
235 )
236 logger.addHandler(diagnostic_log)
237 # Create metadata file used by some steps.
238 if style_file:
239 recipe["style_file_path"] = str(style_file)
240 if plot_resolution:
241 recipe["plot_resolution"] = plot_resolution
242 if skip_write:
243 recipe["skip_write"] = skip_write
244 _write_metadata(recipe)
246 # Execute the recipe.
247 step_input = None
248 for step in steps:
249 step_input = _step_parser(step, step_input)
250 logger.info("Recipe output:\n%s", step_input)
252 logger.info("Creating diagnostic archive.")
253 create_diagnostic_archive()
254 finally:
255 os.chdir(original_working_directory)