18732618520

Committed 22 Oct 2025 11:21PM UTC coverage: 25.34% (-70.0%) from 95.34%

Build # 18732618520

Build Type

Pull #70

github

Committed by

miketynes

Commit Message

fix init, logging init, waiting

Pull Request Pull Request #70: Academy proto

Run Details

261 of 1030 relevant lines covered (25.34%)

0.25 hits per line

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

30.0

/cascade/calculator.py

"""Utilities for employing ASE calculators"""
from typing import List
from pathlib import Path
from string import Template
from hashlib import sha256
import numpy as np
import yaml
import os

from ase.calculators.calculator import Calculator, all_changes, all_properties
from ase.calculators.cp2k import CP2K
from ase import units, Atoms

_file_dir = Path(__file__).parent / 'files'


def create_run_hash(atoms: Atoms, **kwargs) -> str:
    """Generate a unique has for a certain simulation

    Args:
        atoms: Atoms describing the start of the dynamics
        kwargs: Any other keyword arguments used to describe the run
    Returns:
        A hash describing the run
    """

    # Update using the structure
    hasher = sha256()
    hasher.update(atoms.get_atomic_numbers().tobytes())
    hasher.update(atoms.positions.tobytes())

    # Add the optional arguments
    options = sorted(kwargs.items())
    for key, value in options:
        hasher.update(key.encode())
        hasher.update(str(value).encode())

    return hasher.hexdigest()[-8:]


def make_calculator(
        method: str,
        multiplicity: int = 0,
        command: str | None = None,
        directory: str = 'run',
        template_dir: str | Path | None = None,
        set_pos_file: bool = False,
        debug: bool = False
) -> CP2K:
    """Make a calculator ready to run with different configurations

    Supported methods:
        - `pm6`: A force-matched PM6 shown by `Welborn et al <https://onlinelibrary.wiley.com/doi/10.1002/jcc.23887>`_
           to agree better for properties of liquid water than the original formulation
        - `blyp`: The BLYP GGA potential with D3 vdW corrections, suggested by `Lin et al. <https://pubs.acs.org/doi/10.1021/ct3001848>`_
            to give the best properties of liquid water
        - `b97m`: The `B97M-rV <http://xlink.rsc.org/?DOI=C6SC04711D>`_ metaGGA functional

    Args:
        method: Which method to run
        multiplicity: Multiplicity of the system
        command: Command used to launch CP2K. Defaults to whatever ASE autodetermines or ``cp2k_shell``
        directory: Path in which to write run file
        template_dir: Path to the directory containing templates.
            Default is to use the value of CASCADE_CP2K_TEMPLATE environment variable
            or the template directory provided with cascade if the environment variable
            has not been set.
        set_pos_file: whether cp2k and ase communicate positions via a file on disk
        debug: Whether to run the ase cp2k calculator in debug mode
    Returns:
        Calculator configured for target method
    """
    # Default to the environment variable
    if template_dir is None:
        template_dir = Path(os.environ.get('CASCADE_CP2K_TEMPLATE', _file_dir))

    # Load the presets file
    with (template_dir / 'presets.yml').open() as fp:
        presets = yaml.safe_load(fp)
    if method not in presets:
        raise ValueError(f'"{method}" not in presets file')
    kwargs = presets[method]
    if kwargs.get('cutoff') is not None:
        kwargs['cutoff'] *= units.Ry
    kwargs.pop('description')

    # Get the input file and replace any templated arguments
    input_file = template_dir / f'cp2k-{method}-template.inp'
    inp = Template(input_file.read_text()).substitute(mult=multiplicity)

    cp2k_opts = dict(
        xc=None,
        inp=inp,
        poisson_solver=None,
        **kwargs
    )
    if command is not None:
        cp2k_opts['command'] = command
    return CP2K(directory=directory,
                stress_tensor=True,
                potential_file=None,
                set_pos_file=set_pos_file,
                **cp2k_opts)


class EnsembleCalculator(Calculator):
    """A single calculator which combines the results of many


    Stores the mean of all calculators as the standard property names,
    and stores the values for each calculator in the :attr:`results`
    as the name of the property with "_ens" appended (e.g., "forces_ens")

    Args:
        calculators: the calculators to ensemble over
    """

    def __init__(self,
                 calculators: list[Calculator],
                 **kwargs):
        Calculator.__init__(self, **kwargs)
        self.calculators = calculators
        self.num_calculators = len(calculators)

    @property
    def implemented_properties(self) -> List[str]:
        joint = set(self.calculators[0].implemented_properties)
        for calc in self.calculators[1:]:
            joint.intersection_update(calc.implemented_properties)

        return list(joint)

    def calculate(self,
                  atoms: Atoms = None,
                  properties=all_properties,
                  system_changes=all_changes):
        # Run each of the subcalculators
        for calc in self.calculators:
            calc.calculate(atoms, properties=properties, system_changes=system_changes)

        # Determine the intersection of the properties computed from all calculators
        all_results = set(self.calculators[0].results.keys())
        for calc in self.calculators[1:]:
            all_results.intersection_update(calc.results.keys())

        # Merge their results
        results = {}
        for key in all_results:
            all_results = np.concatenate([np.expand_dims(calc.results[key], axis=0) for calc in self.calculators], axis=0)
            results[f'{key}_ens'] = all_results
            results[key] = all_results.mean(axis=0)

        self.results = results

1	"""Utilities for employing ASE calculators"""
2	from typing import List	1✔
3	from pathlib import Path	1✔
4	from string import Template	1✔
5	from hashlib import sha256	1✔
6	import numpy as np	1✔
7	import yaml	1✔
8	import os	1✔
9
10	from ase.calculators.calculator import Calculator, all_changes, all_properties	1✔
11	from ase.calculators.cp2k import CP2K	1✔
12	from ase import units, Atoms	1✔
13
14	_file_dir = Path(__file__).parent / 'files'	1✔
15
16
17	def create_run_hash(atoms: Atoms, **kwargs) -> str:	1✔
18	"""Generate a unique has for a certain simulation
19
20	Args:
21	atoms: Atoms describing the start of the dynamics
22	kwargs: Any other keyword arguments used to describe the run
23	Returns:
24	A hash describing the run
25	"""
26
27	# Update using the structure
28	hasher = sha256()	×
29	hasher.update(atoms.get_atomic_numbers().tobytes())	×
30	hasher.update(atoms.positions.tobytes())	×
31
32	# Add the optional arguments
33	options = sorted(kwargs.items())	×
34	for key, value in options:	×
35	hasher.update(key.encode())	×
36	hasher.update(str(value).encode())	×
37
38	return hasher.hexdigest()[-8:]	×
39
40
41	def make_calculator(	1✔
42	method: str,
43	multiplicity: int = 0,
44	command: str \| None = None,
45	directory: str = 'run',
46	template_dir: str \| Path \| None = None,
47	set_pos_file: bool = False,
48	debug: bool = False
49	) -> CP2K:
50	"""Make a calculator ready to run with different configurations
51
52	Supported methods:
53	- `pm6`: A force-matched PM6 shown by `Welborn et al <https://onlinelibrary.wiley.com/doi/10.1002/jcc.23887>`_
54	to agree better for properties of liquid water than the original formulation
55	- `blyp`: The BLYP GGA potential with D3 vdW corrections, suggested by `Lin et al. <https://pubs.acs.org/doi/10.1021/ct3001848>`_
56	to give the best properties of liquid water
57	- `b97m`: The `B97M-rV <http://xlink.rsc.org/?DOI=C6SC04711D>`_ metaGGA functional
58
59	Args:
60	method: Which method to run
61	multiplicity: Multiplicity of the system
62	command: Command used to launch CP2K. Defaults to whatever ASE autodetermines or ``cp2k_shell``
63	directory: Path in which to write run file
64	template_dir: Path to the directory containing templates.
65	Default is to use the value of CASCADE_CP2K_TEMPLATE environment variable
66	or the template directory provided with cascade if the environment variable
67	has not been set.
68	set_pos_file: whether cp2k and ase communicate positions via a file on disk
69	debug: Whether to run the ase cp2k calculator in debug mode
70	Returns:
71	Calculator configured for target method
72	"""
73	# Default to the environment variable
74	if template_dir is None:	×
75	template_dir = Path(os.environ.get('CASCADE_CP2K_TEMPLATE', _file_dir))	×
76
77	# Load the presets file
78	with (template_dir / 'presets.yml').open() as fp:	×
79	presets = yaml.safe_load(fp)	×
80	if method not in presets:	×
81	raise ValueError(f'"{method}" not in presets file')	×
82	kwargs = presets[method]	×
83	if kwargs.get('cutoff') is not None:	×
84	kwargs['cutoff'] *= units.Ry	×
85	kwargs.pop('description')	×
86
87	# Get the input file and replace any templated arguments
88	input_file = template_dir / f'cp2k-{method}-template.inp'	×
89	inp = Template(input_file.read_text()).substitute(mult=multiplicity)	×
90
91	cp2k_opts = dict(	×
92	xc=None,
93	inp=inp,
94	poisson_solver=None,
95	**kwargs
96	)
97	if command is not None:	×
98	cp2k_opts['command'] = command	×
99	return CP2K(directory=directory,	×
100	stress_tensor=True,
101	potential_file=None,
102	set_pos_file=set_pos_file,
103	**cp2k_opts)
104
105
106	class EnsembleCalculator(Calculator):	1✔
107	"""A single calculator which combines the results of many
108
109
110	Stores the mean of all calculators as the standard property names,
111	and stores the values for each calculator in the :attr:`results`
112	as the name of the property with "_ens" appended (e.g., "forces_ens")
113
114	Args:
115	calculators: the calculators to ensemble over
116	"""
117
118	def __init__(self,	1✔
119	calculators: list[Calculator],
120	**kwargs):
121	Calculator.__init__(self, **kwargs)	×
122	self.calculators = calculators	×
123	self.num_calculators = len(calculators)	×
124
125	@property	1✔
126	def implemented_properties(self) -> List[str]:	1✔
127	joint = set(self.calculators[0].implemented_properties)	×
128	for calc in self.calculators[1:]:	×
129	joint.intersection_update(calc.implemented_properties)	×
130
131	return list(joint)	×
132
133	def calculate(self,	1✔
134	atoms: Atoms = None,
135	properties=all_properties,
136	system_changes=all_changes):
137	# Run each of the subcalculators
138	for calc in self.calculators:	×
139	calc.calculate(atoms, properties=properties, system_changes=system_changes)	×
140
141	# Determine the intersection of the properties computed from all calculators
142	all_results = set(self.calculators[0].results.keys())	×
143	for calc in self.calculators[1:]:	×
144	all_results.intersection_update(calc.results.keys())	×
145
146	# Merge their results
147	results = {}	×
148	for key in all_results:	×
149	all_results = np.concatenate([np.expand_dims(calc.results[key], axis=0) for calc in self.calculators], axis=0)	×
150	results[f'{key}_ens'] = all_results	×
151	results[key] = all_results.mean(axis=0)	×
152
153	self.results = results	×

globus-labs / cascade / 18732618520

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