13967064830

Committed 20 Mar 2025 10:24AM UTC coverage: 92.434% (-0.2%) from 92.657%

Build # 13967064830

Build Type

push

github

Committed by

web-flow

Commit Message

List outputs (#454)

* Always return absolute filename

* Add output files property to calculations

* Save output files to CLI summary

* Save MD postprocessing filenames

* Add output files to missing CLI functions

* Fix Mixin tests

* Remove duplicate geomopt traj and fix output files

* Tidy building filenames

* Test output files

* Fix saving MD restart files

* Fix MD minimize filename

* Fix checking lists of output files

* Fix MD restart files list

* Fix inconsistent paths

* Fix local path

* Fix docstring

* Fix phonon output files

* Fix phonon output file tests

* Fix writing tuples in CLI summary

* Apply suggestions from code review

Co-authored-by: Jacob Wilkins <46597752+oerc0122@users.noreply.github.com>

* Update tests/utils.py

Co-authored-by: Jacob Wilkins <46597752+oerc0122@users.noreply.github.com>

* Use asserts for tests

* Fix when output dir is created

* Avoid rebuilding phonon files

---------

Co-authored-by: Jacob Wilkins <46597752+oerc0122@users.noreply.github.com>

Run Details

161 of 177 new or added lines in 18 files covered. (90.96%)

1 existing line in 1 file now uncovered.

2712 of 2934 relevant lines covered (92.43%)

2.77 hits per line

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

93.26

/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

from janus_core.helpers.utils import build_file_dir

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)

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

stfc / janus-core / 13967064830

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