13586595368

Committed 28 Feb 2025 10:50AM UTC coverage: 93.077% (+0.2%) from 92.875%

Build # 13586595368

Build Type

Pull #454

github

Committed by

web-flow

Commit Message

Merge a64f0ac04 into 80882e842

Pull Request Pull Request #454: List outputs

Run Details

182 of 188 new or added lines in 18 files covered. (96.81%)

1 existing line in 1 file now uncovered.

2689 of 2889 relevant lines covered (93.08%)

2.79 hits per line

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

92.59

/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 (
        Architectures,
        ASEReadArgs,
        Devices,
        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_paths_to_strs(value)
        elif isinstance(value, tuple):
            dictionary[key] = list(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, inputs: dict, 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.
    inputs
        Inputs to CLI command to save.
    output_files
        Output files with labels to be generated by CLI command.
    """
    output_files["summary"] = summary
    dict_paths_to_strs(output_files)

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

    with open(summary, "w", encoding="utf8") as outfile:
        yaml.dump(save_info, 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 save_struct_calc(
    *,
    inputs: dict,
    struct: MaybeSequence[Atoms],
    struct_path: Path,
    arch: Architectures,
    device: Devices,
    model_path: str,
    read_kwargs: ASEReadArgs,
    calc_kwargs: dict[str, Any],
    log: Path,
) -> None:
    """
    Add structure and calculator input information to a dictionary.

    Parameters
    ----------
    inputs
        Inputs dictionary to add information to.
    struct
        Structure to be simulated.
    struct_path
        Path of structure file.
    arch
        MLIP architecture.
    device
        Device to run calculations on.
    model_path
        Path to MLIP model.
    read_kwargs
        Keyword arguments to pass to ase.io.read.
    calc_kwargs
        Keyword arguments to pass to the calculator.
    log
        Path to log file.
    """
    from ase import Atoms

    # Clean up duplicate parameters
    for key in (
        "struct",
        "struct_path",
        "arch",
        "device",
        "model_path",
        "read_kwargs",
        "calc_kwargs",
        "log_kwargs",
    ):
        inputs.pop(key, None)

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

    inputs["calc"] = {
        "arch": arch,
        "device": device,
        "model_path": model_path,
        "read_kwargs": read_kwargs,
        "calc_kwargs": calc_kwargs,
    }

    inputs["log"] = log

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


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

stfc / janus-core / 13586595368

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