Coverage for src/CSET/recipes/__init__.py: 100%

1# © Crown copyright, Met Office (2022-2025) 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. 

14 

15"""Operations on recipes.""" 

16 

17import importlib.resources 

18import logging 

19import sys 

20from collections.abc import Iterable 

21from pathlib import Path 

22from typing import Any 

23 

24from ruamel.yaml import YAML 

25 

26from CSET._common import parse_recipe, slugify 

27from CSET.cset_workflow.lib.python.jinja_utils import get_models as get_models 

28 

29logger = logging.getLogger(__name__) 

30 

31 

32def _version_agnostic_importlib_resources_file() -> Path: 

33 """Transitional wrapper to importlib.resources.files(). 

34 

35 Importlib behaviour changed in 3.12 to avoid circular dependencies. 

36 """ 

37 if sys.version_info.minor >= 12: 

38 input_dir = importlib.resources.files() 

39 else: 

40 import CSET.recipes 

41 

42 input_dir = importlib.resources.files(CSET.recipes) 

43 return input_dir 

44 

45 

46def _recipe_files_in_tree( 

47 recipe_name: str | None = None, input_dir: Path | None = None 

48) -> Iterable[Path]: 

49 """Yield recipe file Paths matching the recipe name.""" 

50 if recipe_name is None: 

51 recipe_name = "" 

52 if input_dir is None: 

53 input_dir = _version_agnostic_importlib_resources_file() 

54 for file in input_dir.iterdir(): 

55 logger.debug("Testing %s", file) 

56 if recipe_name in file.name and file.is_file() and file.suffix == ".yaml": 

57 yield file 

58 elif file.is_dir() and file.name[0] != "_": # Excludes __pycache__ 

59 yield from _recipe_files_in_tree(recipe_name, file) 

60 

61 

62def _get_recipe_file(recipe_name: str, input_dir: Path | None = None) -> Path: 

63 """Return a Path to the recipe file.""" 

64 if input_dir is None: 

65 input_dir = _version_agnostic_importlib_resources_file() 

66 file = input_dir / recipe_name 

67 logger.debug("Getting recipe: %s", file) 

68 if not file.is_file(): 

69 raise FileNotFoundError("Recipe file does not exist.", recipe_name) 

70 return file 

71 

72 

73def unpack_recipe(recipe_dir: Path, recipe_name: str) -> None: 

74 """ 

75 Unpacks recipes files into a directory, creating it if it doesn't exist. 

76 

77 Parameters 

78 ---------- 

79 recipe_dir: Path 

80 Path to a directory into which to unpack the recipe files. 

81 recipe_name: str 

82 Name of recipe to unpack. 

83 

84 Raises 

85 ------ 

86 FileExistsError 

87 If recipe_dir already exists, and is not a directory. 

88 

89 OSError 

90 If recipe_dir cannot be created, such as insufficient permissions, or 

91 lack of space. 

92 """ 

93 recipe_dir.mkdir(parents=True, exist_ok=True) 

94 output_file = recipe_dir / recipe_name 

95 logger.debug("Saving recipe to %s", output_file) 

96 if output_file.exists(): 

97 logger.debug("%s already exists in target directory, skipping.", recipe_name) 

98 return 

99 logger.info("Unpacking %s to %s", recipe_name, output_file) 

100 file = _get_recipe_file(recipe_name) 

101 output_file.write_bytes(file.read_bytes()) 

102 

103 

104def list_available_recipes() -> None: 

105 """List available recipes to stdout.""" 

106 print("Available recipes:") 

107 for file in _recipe_files_in_tree(): 

108 print(f"\t{file.name}") 

109 

110 

111def detail_recipe(recipe_name: str) -> None: 

112 """Detail the recipe to stdout. 

113 

114 If multiple recipes match the given name they will all be displayed. 

115 

116 Parameters 

117 ---------- 

118 recipe_name: str 

119 Partial match for the recipe name. 

120 """ 

121 for file in _recipe_files_in_tree(recipe_name): 

122 with YAML(typ="safe", pure=True) as yaml: 

123 recipe = yaml.load(file) 

124 print(f"\n\t{file.name}\n\t{''.join('─' * len(file.name))}\n") 

125 print(recipe.get("description")) 

126 

127 

128class RawRecipe: 

129 """A recipe to be parbaked. 

130 

131 Parameters 

132 ---------- 

133 recipe: str 

134 Name of the recipe file. 

135 model_ids: int | list[int] 

136 Model IDs to set the input paths for. Matches the corresponding workflow 

137 model IDs. 

138 variables: dict[str, Any] aggregation: bool 

139 Recipe variables to be inserted into $VAR placeholders in the recipe. 

140 aggregation: bool 

141 Whether this is an aggregation recipe or just a single case. 

142 

143 Returns 

144 ------- 

145 RawRecipe 

146 """ 

147 

148 recipe: str 

149 model_ids: list[int] 

150 variables: dict[str, Any] 

151 aggregation: bool 

152 

153 def __init__( 

154 self, 

155 recipe: str, 

156 model_ids: int | list[int], 

157 variables: dict[str, Any], 

158 aggregation: bool, 

159 ) -> None: 

160 self.recipe = recipe 

161 self.model_ids = model_ids if isinstance(model_ids, list) else [model_ids] 

162 self.variables = variables 

163 self.aggregation = aggregation 

164 

165 def __str__(self) -> str: 

166 """Return str(self). 

167 

168 Examples 

169 -------- 

170 >>> print(raw_recipe) 

171 generic_surface_spatial_plot_sequence.yaml (model 1) 

172 VARNAME air_temperature 

173 MODEL_NAME Model A 

174 METHOD SEQ 

175 SUBAREA_TYPE None 

176 SUBAREA_EXTENT None 

177 """ 

178 recipe = self.recipe if self.recipe else "<unknown>" 

179 plural = "s" if len(self.model_ids) > 1 else "" 

180 ids = " ".join(str(m) for m in self.model_ids) 

181 aggregation = ", Aggregation" if self.aggregation else "" 

182 pad = max([0] + [len(k) for k in self.variables.keys()]) 

183 variables = "".join(f"\n\t{k:<{pad}} {v}" for k, v in self.variables.items()) 

184 return f"{recipe} (model{plural} {ids}{aggregation}){variables}" 

185 

186 def __eq__(self, value: object) -> bool: 

187 """Return self==value.""" 

188 if isinstance(value, self.__class__): 

189 return ( 

190 self.recipe == value.recipe 

191 and self.model_ids == value.model_ids 

192 and self.variables == value.variables 

193 and self.aggregation == value.aggregation 

194 ) 

195 return NotImplemented 

196 

197 def parbake(self, ROSE_DATAC: Path, SHARE_DIR: Path) -> None: 

198 """Pre-process recipe to bake in all variables. 

199 

200 Parameters 

201 ---------- 

202 ROSE_DATAC: Path 

203 Workflow shared per-cycle data location. 

204 SHARE_DIR: Path 

205 Workflow shared data location. 

206 """ 

207 # Ready recipe file to disk. 

208 unpack_recipe(Path.cwd(), self.recipe) 

209 

210 # Collect configuration from environment. 

211 if self.aggregation: 

212 # Construct the location for the recipe. 

213 recipe_dir = ROSE_DATAC / "aggregation_recipes" 

214 # Construct the input data directories for the cycle. 

215 data_dirs = [ 

216 SHARE_DIR / f"cycle/*/data/{model_id}" for model_id in self.model_ids 

217 ] 

218 else: 

219 recipe_dir = ROSE_DATAC / "recipes" 

220 data_dirs = [ROSE_DATAC / f"data/{model_id}" for model_id in self.model_ids] 

221 

222 # Ensure recipe dir exists. 

223 recipe_dir.mkdir(parents=True, exist_ok=True) 

224 

225 # Add input paths to recipe variables. 

226 self.variables["INPUT_PATHS"] = data_dirs 

227 

228 # Parbake this recipe, saving into recipe_dir. 

229 recipe = parse_recipe(Path(self.recipe), self.variables) 

230 output = recipe_dir / f"{slugify(recipe['title'])}.yaml" 

231 with open(output, "wt") as fp: 

232 with YAML(pure=True, output=fp) as yaml: 

233 yaml.dump(recipe) 

234 

235 

236class Config: 

237 """Namespace for easy access to configuration values. 

238 

239 A namespace for easy access to configuration values (via config.variable), 

240 where undefined attributes return an empty list. An empty list evaluates to 

241 False in boolean contexts and can be safely iterated over, so it acts as an 

242 effective unset value. 

243 

244 Parameters 

245 ---------- 

246 config: dict 

247 Configuration key-value pairs. 

248 

249 Example 

250 ------- 

251 >>> conf = Config({"key": "value"}) 

252 >>> conf.key 

253 'value' 

254 >>> conf.missing 

255 [] 

256 """ 

257 

258 d: dict 

259 

260 def __init__(self, config: dict) -> None: 

261 self.d = config 

262 

263 def __getattr__(self, name: str): 

264 """Return an empty list for missing names.""" 

265 return self.d.get(name, []) 

266 

267 def asdict(self) -> dict: 

268 """Return config as a dictionary.""" 

269 return self.d 

270 

271 

272def load_recipes(variables: dict[str, Any]) -> Iterable[RawRecipe]: 

273 """Load recipes enabled by configuration. 

274 

275 Recipes are loaded using all loaders (python modules) in CSET.loaders. Each 

276 of these loaders must define a function with the signature `load(conf: dict) 

277 -> Iterable[RawRecipe]`, which will be called with `variables`. 

278 

279 A minimal example can be found in `CSET.loaders.test`. 

280 

281 Parameters 

282 ---------- 

283 variables: dict[str, Any] 

284 Workflow configuration from ROSE_SUITE_VARIABLES. 

285 

286 Returns 

287 ------- 

288 Iterable[RawRecipe] 

289 Configured recipes. 

290 

291 Raises 

292 ------ 

293 AttributeError 

294 When a loader doesn't provide a `load` function. 

295 """ 

296 # Import here to avoid circular import. 

297 import CSET.loaders 

298 

299 config = Config(variables) 

300 for loader in CSET.loaders.__all__: 

301 logger.info("Loading recipes from %s", loader) 

302 module = getattr(CSET.loaders, loader) 

303 yield from module.load(config)