Coverage for src/CSET/recipes/__init__.py: 100%
53 statements
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-01 10:31 +0000
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-01 10:31 +0000
1# Copyright 2022-2023 Met Office and 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"""Operations on recipes."""
17import importlib.resources
18import logging
19import sys
20import warnings
21from collections.abc import Iterable
22from pathlib import Path
24import ruamel.yaml
27class FileExistsWarning(UserWarning):
28 """Warning a file already exists, and some unusual action shall be taken."""
31def _version_agnostic_importlib_resources_file() -> Path:
32 """Transitional wrapper to importlib.resources.files().
34 Importlib behaviour changed in 3.12 to avoid circular dependencies.
35 """
36 if sys.version_info.minor >= 12:
37 input_dir = importlib.resources.files()
38 else:
39 import CSET.recipes
41 input_dir = importlib.resources.files(CSET.recipes)
42 return input_dir
45def _recipe_files_in_tree(
46 recipe_name: str = None, input_dir: Path = None
47) -> Iterable[Path]:
48 """Yield recipe file Paths matching the recipe name."""
49 if recipe_name is None:
50 recipe_name = ""
51 if input_dir is None:
52 input_dir = _version_agnostic_importlib_resources_file()
53 for file in input_dir.iterdir():
54 logging.debug("Testing %s", file)
55 if recipe_name in file.name and file.is_file() and file.suffix == ".yaml":
56 yield file
57 elif file.is_dir() and file.name[0] != "_": # Excludes __pycache__
58 yield from _recipe_files_in_tree(recipe_name, file)
61def _get_recipe_file(recipe_name: str, input_dir: Path = None) -> Path:
62 """Return a Path to the recipe file."""
63 if input_dir is None:
64 input_dir = _version_agnostic_importlib_resources_file()
65 file = input_dir / recipe_name
66 logging.debug("Getting recipe: %s", file)
67 if not file.is_file():
68 raise FileNotFoundError("Recipe file does not exist.", recipe_name)
69 return file
72def unpack_recipe(recipe_dir: Path, recipe_name: str) -> None:
73 """
74 Unpacks recipes files into a directory, creating it if it doesn't exist.
76 Parameters
77 ----------
78 recipe_dir: Path
79 Path to a directory into which to unpack the recipe files.
80 recipe_name: str
81 Name of recipe to unpack.
83 Raises
84 ------
85 FileExistsError
86 If recipe_dir already exists, and is not a directory.
88 OSError
89 If recipe_dir cannot be created, such as insufficient permissions, or
90 lack of space.
91 """
92 file = _get_recipe_file(recipe_name)
93 recipe_dir.mkdir(parents=True, exist_ok=True)
94 output_file = recipe_dir / file.name
95 logging.debug("Saving recipe to %s", output_file)
96 if output_file.exists():
97 warnings.warn(
98 f"{file.name} already exists in target directory, skipping.",
99 FileExistsWarning,
100 stacklevel=2,
101 )
102 return
103 logging.info("Unpacking %s to %s", file.name, output_file)
104 output_file.write_bytes(file.read_bytes())
107def list_available_recipes() -> None:
108 """List available recipes to stdout."""
109 print("Available recipes:")
110 for file in _recipe_files_in_tree():
111 print(f"\t{file.name}")
114def detail_recipe(recipe_name: str) -> None:
115 """Detail the recipe to stdout.
117 If multiple recipes match the given name they will all be displayed.
119 Parameters
120 ----------
121 recipe_name: str
122 Partial match for the recipe name.
123 """
124 for file in _recipe_files_in_tree(recipe_name):
125 with ruamel.yaml.YAML(typ="safe", pure=True) as yaml:
126 recipe = yaml.load(file)
127 print(f"\n\t{file.name}\n\t{''.join('─' * len(file.name))}\n")
128 print(recipe.get("description"))