13899239177

Committed 17 Mar 2025 12:18PM UTC coverage: 92.722% (+0.1%) from 92.617%

Build # 13899239177

Build Type

Pull #454

github

Committed by

web-flow

Commit Message

Merge 184dff930 into f8fd5c967

Pull Request Pull Request #454: List outputs

Run Details

148 of 154 new or added lines in 18 files covered. (96.1%)

1 existing line in 1 file now uncovered.

2688 of 2899 relevant lines covered (92.72%)

2.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

93.1

/janus_core/cli/utils.py

"""Utility functions for CLI."""

from __future__ import annotations

from collections.abc import Sequence
import datetime
import logging
from pathlib import Path
from typing import TYPE_CHECKING, Any

from typer_config import conf_callback_factory, yaml_loader
import yaml

if TYPE_CHECKING:
    from ase import Atoms
    from typer import Context

    from janus_core.cli.types import TyperDict
    from janus_core.helpers.janus_types import (
        MaybeSequence,
        PathLike,
    )


def dict_paths_to_strs(dictionary: dict) -> None:
    """
    Recursively iterate over dictionary, converting Path values to strings.

    Parameters
    ----------
    dictionary
        Dictionary to be converted.
    """
    for key, value in dictionary.items():
        if isinstance(value, dict):
            dict_paths_to_strs(value)
        elif isinstance(value, Sequence) and not isinstance(value, str):
            dictionary[key] = [
                str(path) if isinstance(path, Path) else path for path in value
            ]
        elif isinstance(value, Path):
            dictionary[key] = str(value)


def dict_tuples_to_lists(dictionary: dict) -> None:
    """
    Recursively iterate over dictionary, converting tuple values to lists.

    Parameters
    ----------
    dictionary
        Dictionary to be converted.
    """
    for key, value in dictionary.items():
        if isinstance(value, dict):
            dict_tuples_to_lists(value)
        elif isinstance(value, tuple):
            dictionary[key] = list(value)
        elif isinstance(value, list):
            dictionary[key] = [list(x) if isinstance(x, tuple) else x for x in value]


def dict_remove_hyphens(dictionary: dict) -> dict:
    """
    Recursively iterate over dictionary, replacing hyphens with underscores in keys.

    Parameters
    ----------
    dictionary
        Dictionary to be converted.

    Returns
    -------
    dict
        Dictionary with hyphens in keys replaced with underscores.
    """
    for key, value in dictionary.items():
        if isinstance(value, dict):
            dictionary[key] = dict_remove_hyphens(value)
    return {k.replace("-", "_"): v for k, v in dictionary.items()}


def set_read_kwargs_index(read_kwargs: dict[str, Any]) -> None:
    """
    Set default read_kwargs["index"] to final image and check its value is an integer.

    To ensure only a single Atoms object is read, slices such as ":" are forbidden.

    Parameters
    ----------
    read_kwargs
        Keyword arguments to be passed to ase.io.read. If specified,
        read_kwargs["index"] must be an integer, and if not, a default value
        of -1 is set.
    """
    read_kwargs.setdefault("index", -1)
    try:
        int(read_kwargs["index"])
    except ValueError as e:
        raise ValueError("`read_kwargs['index']` must be an integer") from e


def parse_typer_dicts(typer_dicts: list[TyperDict]) -> list[dict]:
    """
    Convert list of TyperDict objects to list of dictionaries.

    Parameters
    ----------
    typer_dicts
        List of TyperDict objects to convert.

    Returns
    -------
    list[dict]
        List of converted dictionaries.

    Raises
    ------
    ValueError
        If items in list are not converted to dicts.
    """
    for i, typer_dict in enumerate(typer_dicts):
        typer_dicts[i] = typer_dict.value if typer_dict else {}
        if not isinstance(typer_dicts[i], dict):
            raise ValueError(
                f"""{typer_dicts[i]} must be passed as a dictionary wrapped in quotes.\
 For example, "{{'key': value}}" """
            )
    return typer_dicts


def yaml_converter_loader(config_file: str) -> dict[str, Any]:
    """
    Load yaml configuration and replace hyphens with underscores.

    Parameters
    ----------
    config_file
        Yaml configuration file to read.

    Returns
    -------
    dict[str, Any]
        Dictionary with loaded configuration.
    """
    if not config_file:
        return {}

    config = yaml_loader(config_file)
    # Replace all "-"" with "_" in conf
    return dict_remove_hyphens(config)


yaml_converter_callback = conf_callback_factory(yaml_converter_loader)


def start_summary(
    *,
    command: str,
    summary: Path,
    config: dict[str, Any],
    info: dict[str, Any],
    output_files: dict[str, PathLike],
) -> None:
    """
    Write initial summary contents.

    Parameters
    ----------
    command
        Name of CLI command being used.
    summary
        Path to summary file being saved.
    config
        Inputs to CLI command to save.
    info
        Extra information to save.
    output_files
        Output files with labels to be generated by CLI command.
    """
    config.pop("config", None)
    output_files["summary"] = summary.absolute()

    summary_contents = {
        "command": f"janus {command}",
        "start_time": datetime.datetime.now().strftime("%d/%m/%Y, %H:%M:%S"),
        "config": config,
        "info": info,
        "output_files": output_files,
    }

    # Convert all paths to strings in inputs nested dictionary
    dict_paths_to_strs(summary_contents)
    dict_tuples_to_lists(summary_contents)

    with open(summary, "w", encoding="utf8") as outfile:
        yaml.dump(summary_contents, outfile, default_flow_style=False)


def carbon_summary(*, summary: Path, log: Path) -> None:
    """
    Calculate and write carbon tracking summary.

    Parameters
    ----------
    summary
        Path to summary file being saved.
    log
        Path to log file with carbon emissions saved.
    """
    with open(log, encoding="utf8") as file:
        logs = yaml.safe_load(file)

    emissions = sum(
        lg["message"]["emissions"]
        for lg in logs
        if isinstance(lg["message"], dict) and "emissions" in lg["message"]
    )

    with open(summary, "a", encoding="utf8") as outfile:
        yaml.dump({"emissions": emissions}, outfile, default_flow_style=False)


def end_summary(summary: Path) -> None:
    """
    Write final time to summary and close.

    Parameters
    ----------
    summary
        Path to summary file being saved.
    """
    with open(summary, "a", encoding="utf8") as outfile:
        yaml.dump(
            {"end_time": datetime.datetime.now().strftime("%d/%m/%Y, %H:%M:%S")},
            outfile,
            default_flow_style=False,
        )
    logging.shutdown()


def get_struct_info(
    *,
    struct: MaybeSequence[Atoms],
    struct_path: Path,
) -> dict[str, Any]:
    """
    Add structure information to a dictionary.

    Parameters
    ----------
    struct
        Structure to be simulated.
    struct_path
        Path of structure file.

    Returns
    -------
    dict[str, Any]
        Dictionary with structure information.
    """
    from ase import Atoms

    info = {}

    if isinstance(struct, Atoms):
        info["struct"] = {
            "n_atoms": len(struct),
            "struct_path": struct_path,
            "formula": struct.get_chemical_formula(),
        }
    elif isinstance(struct, Sequence):
        info["traj"] = {
            "length": len(struct),
            "struct_path": struct_path,
            "struct": {
                "n_atoms": len(struct[0]),
                "formula": struct[0].get_chemical_formula(),
            },
        }

    return info


def get_config(*, params: dict[str, Any], all_kwargs: dict[str, Any]) -> dict[str, Any]:
    """
    Get configuration and set kwargs dictionaries.

    Parameters
    ----------
    params
        CLI input parameters from ctx.
    all_kwargs
        Name and contents of all kwargs dictionaries.

    Returns
    -------
    dict[str, Any]
        Input parameters with parsed kwargs dictionaries substituted in.
    """
    for param in params:
        if param in all_kwargs:
            params[param] = all_kwargs[param]

    return params


def check_config(ctx: Context) -> None:
    """
    Check options in configuration file are valid options for CLI command.

    Parameters
    ----------
    ctx
        Typer (Click) Context within command.
    """
    # Compare options from config file (default_map) to function definition (params)
    for option in ctx.default_map:
        # Check options individually so can inform user of specific issue
        if option not in ctx.params:
            raise ValueError(f"'{option}' in configuration file is not a valid option")

1	"""Utility functions for CLI."""
2
3	from __future__ import annotations	3✔
4
5	from collections.abc import Sequence	3✔
6	import datetime	3✔
7	import logging	3✔
8	from pathlib import Path	3✔
9	from typing import TYPE_CHECKING, Any	3✔
10
11	from typer_config import conf_callback_factory, yaml_loader	3✔
12	import yaml	3✔
13
14	if TYPE_CHECKING:	3✔
15	from ase import Atoms	×
16	from typer import Context	×
17
18	from janus_core.cli.types import TyperDict	×
19	from janus_core.helpers.janus_types import (	×
20	MaybeSequence,
21	PathLike,
22	)
23
24
25	def dict_paths_to_strs(dictionary: dict) -> None:	3✔
26	"""
27	Recursively iterate over dictionary, converting Path values to strings.
28
29	Parameters
30	----------
31	dictionary
32	Dictionary to be converted.
33	"""
34	for key, value in dictionary.items():	3✔
35	if isinstance(value, dict):	3✔
36	dict_paths_to_strs(value)	3✔
37	elif isinstance(value, Sequence) and not isinstance(value, str):	3✔
38	dictionary[key] = [	3✔
39	str(path) if isinstance(path, Path) else path for path in value
40	]
41	elif isinstance(value, Path):	3✔
42	dictionary[key] = str(value)	3✔
43
44
45	def dict_tuples_to_lists(dictionary: dict) -> None:	3✔
46	"""
47	Recursively iterate over dictionary, converting tuple values to lists.
48
49	Parameters
50	----------
51	dictionary
52	Dictionary to be converted.
53	"""
54	for key, value in dictionary.items():	3✔
55	if isinstance(value, dict):	3✔
56	dict_tuples_to_lists(value)	3✔
57	elif isinstance(value, tuple):	3✔
UNCOV 58	dictionary[key] = list(value)	×
59	elif isinstance(value, list):	3✔
60	dictionary[key] = [list(x) if isinstance(x, tuple) else x for x in value]	3✔
61
62
63	def dict_remove_hyphens(dictionary: dict) -> dict:	3✔
64	"""
65	Recursively iterate over dictionary, replacing hyphens with underscores in keys.
66
67	Parameters
68	----------
69	dictionary
70	Dictionary to be converted.
71
72	Returns
73	-------
74	dict
75	Dictionary with hyphens in keys replaced with underscores.
76	"""
77	for key, value in dictionary.items():	3✔
78	if isinstance(value, dict):	3✔
79	dictionary[key] = dict_remove_hyphens(value)	3✔
80	return {k.replace("-", "_"): v for k, v in dictionary.items()}	3✔
81
82
83	def set_read_kwargs_index(read_kwargs: dict[str, Any]) -> None:	3✔
84	"""
85	Set default read_kwargs["index"] to final image and check its value is an integer.
86
87	To ensure only a single Atoms object is read, slices such as ":" are forbidden.
88
89	Parameters
90	----------
91	read_kwargs
92	Keyword arguments to be passed to ase.io.read. If specified,
93	read_kwargs["index"] must be an integer, and if not, a default value
94	of -1 is set.
95	"""
96	read_kwargs.setdefault("index", -1)	3✔
97	try:	3✔
98	int(read_kwargs["index"])	3✔
99	except ValueError as e:	3✔
100	raise ValueError("`read_kwargs['index']` must be an integer") from e	3✔
101
102
103	def parse_typer_dicts(typer_dicts: list[TyperDict]) -> list[dict]:	3✔
104	"""
105	Convert list of TyperDict objects to list of dictionaries.
106
107	Parameters
108	----------
109	typer_dicts
110	List of TyperDict objects to convert.
111
112	Returns
113	-------
114	list[dict]
115	List of converted dictionaries.
116
117	Raises
118	------
119	ValueError
120	If items in list are not converted to dicts.
121	"""
122	for i, typer_dict in enumerate(typer_dicts):	3✔
123	typer_dicts[i] = typer_dict.value if typer_dict else {}	3✔
124	if not isinstance(typer_dicts[i], dict):	3✔
125	raise ValueError(	×
126	f"""{typer_dicts[i]} must be passed as a dictionary wrapped in quotes.\
127	For example, "{{'key': value}}" """
128	)
129	return typer_dicts	3✔
130
131
132	def yaml_converter_loader(config_file: str) -> dict[str, Any]:	3✔
133	"""
134	Load yaml configuration and replace hyphens with underscores.
135
136	Parameters
137	----------
138	config_file
139	Yaml configuration file to read.
140
141	Returns
142	-------
143	dict[str, Any]
144	Dictionary with loaded configuration.
145	"""
146	if not config_file:	3✔
147	return {}	3✔
148
149	config = yaml_loader(config_file)	3✔
150	# Replace all "-"" with "_" in conf
151	return dict_remove_hyphens(config)	3✔
152
153
154	yaml_converter_callback = conf_callback_factory(yaml_converter_loader)	3✔
155
156
157	def start_summary(	3✔
158	*,
159	command: str,
160	summary: Path,
161	config: dict[str, Any],
162	info: dict[str, Any],
163	output_files: dict[str, PathLike],
164	) -> None:
165	"""
166	Write initial summary contents.
167
168	Parameters
169	----------
170	command
171	Name of CLI command being used.
172	summary
173	Path to summary file being saved.
174	config
175	Inputs to CLI command to save.
176	info
177	Extra information to save.
178	output_files
179	Output files with labels to be generated by CLI command.
180	"""
181	config.pop("config", None)	3✔
182	output_files["summary"] = summary.absolute()	3✔
183
184	summary_contents = {	3✔
185	"command": f"janus {command}",
186	"start_time": datetime.datetime.now().strftime("%d/%m/%Y, %H:%M:%S"),
187	"config": config,
188	"info": info,
189	"output_files": output_files,
190	}
191
192	# Convert all paths to strings in inputs nested dictionary
193	dict_paths_to_strs(summary_contents)	3✔
194	dict_tuples_to_lists(summary_contents)	3✔
195
196	with open(summary, "w", encoding="utf8") as outfile:	3✔
197	yaml.dump(summary_contents, outfile, default_flow_style=False)	3✔
198
199
200	def carbon_summary(*, summary: Path, log: Path) -> None:	3✔
201	"""
202	Calculate and write carbon tracking summary.
203
204	Parameters
205	----------
206	summary
207	Path to summary file being saved.
208	log
209	Path to log file with carbon emissions saved.
210	"""
211	with open(log, encoding="utf8") as file:	3✔
212	logs = yaml.safe_load(file)	3✔
213
214	emissions = sum(	3✔
215	lg["message"]["emissions"]
216	for lg in logs
217	if isinstance(lg["message"], dict) and "emissions" in lg["message"]
218	)
219
220	with open(summary, "a", encoding="utf8") as outfile:	3✔
221	yaml.dump({"emissions": emissions}, outfile, default_flow_style=False)	3✔
222
223
224	def end_summary(summary: Path) -> None:	3✔
225	"""
226	Write final time to summary and close.
227
228	Parameters
229	----------
230	summary
231	Path to summary file being saved.
232	"""
233	with open(summary, "a", encoding="utf8") as outfile:	3✔
234	yaml.dump(	3✔
235	{"end_time": datetime.datetime.now().strftime("%d/%m/%Y, %H:%M:%S")},
236	outfile,
237	default_flow_style=False,
238	)
239	logging.shutdown()	3✔
240
241
242	def get_struct_info(	3✔
243	*,
244	struct: MaybeSequence[Atoms],
245	struct_path: Path,
246	) -> dict[str, Any]:
247	"""
248	Add structure information to a dictionary.
249
250	Parameters
251	----------
252	struct
253	Structure to be simulated.
254	struct_path
255	Path of structure file.
256
257	Returns
258	-------
259	dict[str, Any]
260	Dictionary with structure information.
261	"""
262	from ase import Atoms	3✔
263
264	info = {}	3✔
265
266	if isinstance(struct, Atoms):	3✔
267	info["struct"] = {	3✔
268	"n_atoms": len(struct),
269	"struct_path": struct_path,
270	"formula": struct.get_chemical_formula(),
271	}
272	elif isinstance(struct, Sequence):	3✔
273	info["traj"] = {	3✔
274	"length": len(struct),
275	"struct_path": struct_path,
276	"struct": {
277	"n_atoms": len(struct[0]),
278	"formula": struct[0].get_chemical_formula(),
279	},
280	}
281
282	return info	3✔
283
284
285	def get_config(*, params: dict[str, Any], all_kwargs: dict[str, Any]) -> dict[str, Any]:	3✔
286	"""
287	Get configuration and set kwargs dictionaries.
288
289	Parameters
290	----------
291	params
292	CLI input parameters from ctx.
293	all_kwargs
294	Name and contents of all kwargs dictionaries.
295
296	Returns
297	-------
298	dict[str, Any]
299	Input parameters with parsed kwargs dictionaries substituted in.
300	"""
301	for param in params:	3✔
302	if param in all_kwargs:	3✔
303	params[param] = all_kwargs[param]	3✔
304
305	return params	3✔
306
307
308	def check_config(ctx: Context) -> None:	3✔
309	"""
310	Check options in configuration file are valid options for CLI command.
311
312	Parameters
313	----------
314	ctx
315	Typer (Click) Context within command.
316	"""
317	# Compare options from config file (default_map) to function definition (params)
318	for option in ctx.default_map:	3✔
319	# Check options individually so can inform user of specific issue
320	if option not in ctx.params:	3✔
321	raise ValueError(f"'{option}' in configuration file is not a valid option")	3✔

stfc / janus-core / 13899239177

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous