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

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. 

14 

15"""Operations on recipes.""" 

16 

17import importlib.resources 

18import logging 

19import sys 

20import warnings 

21from collections.abc import Iterable 

22from pathlib import Path 

23 

24import ruamel.yaml 

25 

26 

27class FileExistsWarning(UserWarning): 

28 """Warning a file already exists, and some unusual action shall be taken.""" 

29 

30 

31def _version_agnostic_importlib_resources_file() -> Path: 

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

33 

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 

40 

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

42 return input_dir 

43 

44 

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) 

59 

60 

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 

70 

71 

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. 

75 

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. 

82 

83 Raises 

84 ------ 

85 FileExistsError 

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

87 

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()) 

105 

106 

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}") 

112 

113 

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

115 """Detail the recipe to stdout. 

116 

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

118 

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"))