Coverage for src/CSET/__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"""CSET: Convective and turbulence Scale Evaluation Tool.""" 

16 

17import argparse 

18import logging 

19import os 

20import sys 

21from importlib.metadata import version 

22from pathlib import Path 

23 

24from CSET._common import ArgumentError 

25 

26logger = logging.getLogger(__name__) 

27 

28 

29def main(raw_cli_args: list[str] = sys.argv): 

30 """CLI entrypoint. 

31 

32 Handles argument parsing, setting up logging, top level error capturing, 

33 and execution of the desired subcommand. 

34 """ 

35 parser = setup_argument_parser() 

36 args, unparsed_args = parser.parse_known_args(raw_cli_args[1:]) 

37 

38 setup_logging(args.verbose) 

39 

40 if args.subparser is None: 

41 print("Please choose a command.", file=sys.stderr) 

42 parser.print_usage() 

43 sys.exit(127) 

44 

45 try: 

46 # Execute the specified subcommand. 

47 args.func(args, unparsed_args) 

48 except ArgumentError as err: 

49 # Error message for when needed template variables are missing. 

50 print(err, file=sys.stderr) 

51 parser.print_usage() 

52 sys.exit(127) 

53 except Exception as err: 

54 # Provide slightly nicer error messages for unhandled exceptions. 

55 print(err, file=sys.stderr) 

56 # Display the time and full traceback when debug logging. 

57 logger.debug("An unhandled exception occurred.") 

58 if logger.isEnabledFor(logging.DEBUG): 

59 raise 

60 sys.exit(1) 

61 

62 

63def setup_argument_parser() -> argparse.ArgumentParser: 

64 """Create argument parser for CSET CLI.""" 

65 parser = argparse.ArgumentParser( 

66 prog="cset", description="Convective Scale Evaluation Tool" 

67 ) 

68 parser.add_argument( 

69 "-v", 

70 "--verbose", 

71 action="count", 

72 default=0, 

73 help="increase output verbosity, may be specified multiple times", 

74 ) 

75 parser.add_argument( 

76 "--version", action="version", version=f"CSET v{version('CSET')}" 

77 ) 

78 

79 # https://docs.python.org/3/library/argparse.html#sub-commands 

80 subparsers = parser.add_subparsers(title="subcommands", dest="subparser") 

81 

82 # Run operator chain 

83 parser_bake = subparsers.add_parser("bake", help="run steps from a recipe file") 

84 parser_bake.add_argument( 

85 "-i", 

86 "--input-dir", 

87 type=str, 

88 action="extend", 

89 nargs="+", 

90 help="Alternate way to set the INPUT_PATHS recipe variable", 

91 ) 

92 parser_bake.add_argument( 

93 "-o", 

94 "--output-dir", 

95 type=Path, 

96 required=True, 

97 help="directory to write output into", 

98 ) 

99 parser_bake.add_argument( 

100 "-r", 

101 "--recipe", 

102 type=Path, 

103 required=True, 

104 help="recipe file to read", 

105 ) 

106 parser_bake.add_argument( 

107 "-s", "--style-file", type=Path, help="colour bar definition to use" 

108 ) 

109 parser_bake.add_argument( 

110 "--plot-resolution", type=int, help="plotting resolution in dpi" 

111 ) 

112 parser_bake.add_argument( 

113 "--skip-write", action="store_true", help="Skip saving processed output" 

114 ) 

115 parser_bake.set_defaults(func=_bake_command) 

116 

117 parser_graph = subparsers.add_parser("graph", help="visualise a recipe file") 

118 parser_graph.add_argument( 

119 "-d", 

120 "--details", 

121 action="store_true", 

122 help="include operator arguments in output", 

123 ) 

124 parser_graph.add_argument( 

125 "-o", 

126 "--output-path", 

127 type=Path, 

128 nargs="?", 

129 help="persistent file to save the graph. Otherwise the file is opened", 

130 default=None, 

131 ) 

132 parser_graph.add_argument( 

133 "-r", 

134 "--recipe", 

135 type=Path, 

136 required=True, 

137 help="recipe file to read", 

138 ) 

139 parser_graph.set_defaults(func=_graph_command) 

140 

141 parser_cookbook = subparsers.add_parser( 

142 "cookbook", help="unpack included recipes to a folder" 

143 ) 

144 parser_cookbook.add_argument( 

145 "-d", 

146 "--details", 

147 action="store_true", 

148 help="list available recipes. Supplied recipes are detailed", 

149 ) 

150 parser_cookbook.add_argument( 

151 "-o", 

152 "--output-dir", 

153 type=Path, 

154 help="directory to save recipes. If omitted uses $PWD", 

155 default=Path.cwd(), 

156 ) 

157 parser_cookbook.add_argument( 

158 "recipe", 

159 type=str, 

160 nargs="?", 

161 help="recipe to output or detail", 

162 default="", 

163 ) 

164 parser_cookbook.set_defaults(func=_cookbook_command) 

165 

166 parser_extract_workflow = subparsers.add_parser( 

167 "extract-workflow", help="extract the CSET cylc workflow" 

168 ) 

169 parser_extract_workflow.add_argument( 

170 "location", type=Path, help="directory to save workflow into" 

171 ) 

172 parser_extract_workflow.set_defaults(func=_extract_workflow_command) 

173 

174 return parser 

175 

176 

177def setup_logging(verbosity: int): 

178 """Configure logging level, format and output stream. 

179 

180 Level is based on verbose argument and the LOGLEVEL environment variable. 

181 """ 

182 logging.captureWarnings(True) 

183 

184 # Calculate logging level. 

185 # Level from CLI flags. 

186 if verbosity >= 2: 

187 cli_loglevel = logging.DEBUG 

188 elif verbosity == 1: 

189 cli_loglevel = logging.INFO 

190 else: 

191 cli_loglevel = logging.WARNING 

192 

193 # Level from $LOGLEVEL environment variable. 

194 env_loglevel = logging.getLevelNamesMapping().get( 

195 os.getenv("LOGLEVEL"), logging.ERROR 

196 ) 

197 

198 # Logging verbosity is the most verbose of CLI and environment setting. 

199 loglevel = min(cli_loglevel, env_loglevel) 

200 

201 # Configure the root logger. 

202 logger = logging.getLogger() 

203 # Set logging level. 

204 logger.setLevel(loglevel) 

205 

206 # Hide matplotlib's many font messages. 

207 class NoFontMessageFilter(logging.Filter): 

208 def filter(self, record): 

209 return not record.getMessage().startswith("findfont:") 

210 

211 logging.getLogger("matplotlib.font_manager").addFilter(NoFontMessageFilter()) 

212 

213 stderr_log = logging.StreamHandler() 

214 stderr_log.setFormatter( 

215 logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s") 

216 ) 

217 logger.addHandler(stderr_log) 

218 

219 

220def _bake_command(args, unparsed_args): 

221 from CSET._common import parse_recipe, parse_variable_options 

222 from CSET.operators import execute_recipe 

223 

224 recipe_variables = parse_variable_options(unparsed_args, args.input_dir) 

225 recipe = parse_recipe(args.recipe, recipe_variables) 

226 execute_recipe( 

227 recipe, 

228 args.output_dir, 

229 args.style_file, 

230 args.plot_resolution, 

231 args.skip_write, 

232 ) 

233 

234 

235def _graph_command(args, unparsed_args): 

236 from CSET.graph import save_graph 

237 

238 save_graph( 

239 args.recipe, 

240 args.output_path, 

241 auto_open=not args.output_path, 

242 detailed=args.details, 

243 ) 

244 

245 

246def _cookbook_command(args, unparsed_args): 

247 from CSET.recipes import detail_recipe, list_available_recipes, unpack_recipe 

248 

249 if args.recipe: 

250 if args.details: 

251 detail_recipe(args.recipe) 

252 else: 

253 try: 

254 unpack_recipe(args.output_dir, args.recipe) 

255 except FileNotFoundError: 

256 logger.error("Recipe %s does not exist.", args.recipe) 

257 sys.exit(1) 

258 else: 

259 list_available_recipes() 

260 

261 

262def _extract_workflow_command(args, unparsed_args): 

263 from CSET.extract_workflow import install_workflow 

264 

265 install_workflow(args.location)