Coverage for src/CSET/__init__.py: 95%
104 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-08 11:00 +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"""CSET: Community Seamless Evaluation Toolkit."""
17import argparse
18import logging
19import os
20import sys
21from importlib.metadata import version
22from pathlib import Path
24from CSET._common import ArgumentError
26logger = logging.getLogger(__name__)
29def main(raw_cli_args: list[str] = sys.argv):
30 """CLI entrypoint.
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:])
38 setup_logging(args.verbose)
40 if args.subparser is None:
41 print("Please choose a command.", file=sys.stderr)
42 parser.print_usage()
43 sys.exit(127)
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)
63def setup_argument_parser() -> argparse.ArgumentParser:
64 """Create argument parser for CSET CLI."""
65 parser = argparse.ArgumentParser(
66 prog="cset", description="Community Seamless Evaluation Toolkit"
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 )
79 # https://docs.python.org/3/library/argparse.html#sub-commands
80 subparsers = parser.add_subparsers(title="subcommands", dest="subparser")
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)
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)
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)
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.add_argument(
173 "--restricted",
174 action="store_true",
175 help="install restricted site-specific files during extraction",
176 )
177 parser_extract_workflow.add_argument(
178 "--restricted-url",
179 type=str,
180 help=(
181 "Alternative Git URL to fetch the restricted files from. "
182 "If omitted, defaults to trying to clone first from "
183 "'localmirrors:', then from GitHub via SSH and HTTPS."
184 ),
185 )
186 parser_extract_workflow.set_defaults(func=_extract_workflow_command)
188 parser_install_restricted_files = subparsers.add_parser(
189 "install-restricted-files",
190 help="download and install restricted site-specific files for the CSET cylc workflow",
191 )
192 parser_install_restricted_files.add_argument(
193 "location", type=Path, help="directory containing workflow"
194 )
195 parser_install_restricted_files.add_argument(
196 "--restricted-url",
197 type=str,
198 help=(
199 "Alternative Git URL to fetch the restricted files from. "
200 "If omitted, defaults to trying to clone first from "
201 "'localmirrors:', then from GitHub via SSH and HTTPS."
202 ),
203 )
204 parser_install_restricted_files.set_defaults(func=_install_restricted_files_command)
206 return parser
209def setup_logging(verbosity: int):
210 """Configure logging level, format and output stream.
212 Level is based on verbose argument and the LOGLEVEL environment variable.
213 """
214 logging.captureWarnings(True)
216 # Calculate logging level.
217 # Level from CLI flags.
218 if verbosity >= 2:
219 cli_loglevel = logging.DEBUG
220 elif verbosity == 1:
221 cli_loglevel = logging.INFO
222 else:
223 cli_loglevel = logging.WARNING
225 # Level from $LOGLEVEL environment variable.
226 env_loglevel = logging.getLevelNamesMapping().get(
227 os.getenv("LOGLEVEL", ""), logging.ERROR
228 )
230 # Logging verbosity is the most verbose of CLI and environment setting.
231 loglevel = min(cli_loglevel, env_loglevel)
233 # Configure the root logger.
234 logger = logging.getLogger()
235 # Set logging level.
236 logger.setLevel(loglevel)
238 # Suppress matplotlib's verbose debug output.
239 logging.getLogger("matplotlib").setLevel(logging.WARNING)
241 stderr_log = logging.StreamHandler(stream=sys.stdout)
242 stderr_log.setFormatter(
243 logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")
244 )
245 logger.addHandler(stderr_log)
248def _bake_command(args, unparsed_args):
249 from CSET._common import parse_recipe, parse_variable_options
250 from CSET.operators import execute_recipe
252 recipe_variables = parse_variable_options(unparsed_args, args.input_dir)
253 recipe = parse_recipe(args.recipe, recipe_variables)
254 execute_recipe(
255 recipe,
256 args.output_dir,
257 args.style_file,
258 args.plot_resolution,
259 args.skip_write,
260 )
263def _graph_command(args, unparsed_args):
264 from CSET.graph import save_graph
266 save_graph(
267 args.recipe,
268 args.output_path,
269 auto_open=not args.output_path,
270 detailed=args.details,
271 )
274def _cookbook_command(args, unparsed_args):
275 from CSET.recipes import detail_recipe, list_available_recipes, unpack_recipe
277 if args.recipe:
278 if args.details:
279 detail_recipe(args.recipe)
280 else:
281 try:
282 unpack_recipe(args.output_dir, args.recipe)
283 except FileNotFoundError:
284 logger.error("Recipe %s does not exist.", args.recipe)
285 sys.exit(1)
286 else:
287 list_available_recipes()
290def _extract_workflow_command(args, unparsed_args):
291 from CSET.extract_workflow import install_restricted_files, install_workflow
293 workflow_dir = install_workflow(args.location)
294 if args.restricted: 294 ↛ 295line 294 didn't jump to line 295 because the condition on line 294 was never true
295 install_restricted_files(workflow_dir, args.restricted_url)
298def _install_restricted_files_command(args, unparsed_args):
299 from CSET.extract_workflow import install_restricted_files
301 install_restricted_files(args.location, args.restricted_url)