Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 61 additions & 42 deletions arc/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -98,20 +98,22 @@ class ARC(object):
ess_settings (dict, optional): A dictionary of available ESS (keys) and a corresponding server list (values).
bath_gas (str, optional): A bath gas. Currently used in OneDMin to calc L-J parameters.
Allowed values are He, Ne, Ar, Kr, H2, N2, O2.
adaptive_levels (dict, optional): A dictionary of levels of theory for ranges of the number of heavy atoms in
the molecule. Keys are tuples of (min_num_atoms, max_num_atoms), values are dictionaries. Keys of the
sub-dictionaries are tuples of job types, values are levels of theory (str, dict or Level).
adaptive_levels (list, optional): A list of levels of theory for ranges of the number of heavy atoms in the
molecule. Each entry is a dictionary with an ``atom_range`` 2-list (min_num_atoms, max_num_atoms; the upper
bound may be ``'inf'``) and a ``levels`` mapping from job type to level of theory (str, dict or Level).
Job types sharing a level may be given as a whitespace- or comma-separated key (e.g. ``'opt freq'``).
Job types not defined in adaptive levels will have non-adaptive (regular) levels.
Example::

adaptive_levels = {(1, 5): {('opt', 'freq'): 'wb97xd/6-311+g(2d,2p)',
'sp': 'ccsd(t)-f12/aug-cc-pvtz-f12'},
(6, 15): {('opt', 'freq'): 'b3lyp/cbsb7',
'sp': 'dlpno-ccsd(t)/def2-tzvp'},
(16, 30): {('opt', 'freq'): 'b3lyp/6-31g(d,p)',
'sp': 'wb97xd/6-311+g(2d,2p)'},
(31, 'inf'): {('opt', 'freq'): 'b3lyp/6-31g(d,p)',
'sp': 'b3lyp/6-311+g(d,p)'}}
adaptive_levels = [{'atom_range': [1, 5],
'levels': {'opt freq': 'wb97xd/6-311+g(2d,2p)',
'sp': 'ccsd(t)-f12/aug-cc-pvtz-f12'}},
{'atom_range': [6, 15],
'levels': {'opt freq': 'b3lyp/cbsb7',
'sp': 'dlpno-ccsd(t)/def2-tzvp'}},
{'atom_range': [16, 'inf'],
'levels': {'opt freq': 'b3lyp/6-31g(d,p)',
'sp': 'wb97xd/6-311+g(2d,2p)'}}]

freq_scale_factor (float, optional): The harmonic frequencies scaling factor. Could be automatically determined
if not available in Arkane and not provided by the user.
Expand Down Expand Up @@ -163,9 +165,8 @@ class ARC(object):
ts_guess_level (Level): Level of theory for comparisons of TS guesses between different methods.
irc_level (Level): The level of theory to use for IRC calculations.
orbitals_level (Level): Level of theory for molecular orbitals calculations.
adaptive_levels (dict): A dictionary of levels of theory for ranges of the number of heavy atoms in
the molecule. Keys are tuples of (min_num_atoms, max_num_atoms), values are dictionaries. Keys of the
sub-dictionaries are tuples of job types, values are levels of theory (str, dict or Level).
adaptive_levels (dict): The processed adaptive levels, keyed by (min_num_atoms, max_num_atoms) tuples, each
mapping job-type tuples to ``Level`` objects (built from the user-facing ``adaptive_levels`` list).
Job types not defined in adaptive levels will have non-adaptive (regular) levels.
output (dict): Output dictionary with status and final QM file paths for all species. Only used for restarting,
the actual object used is in the Scheduler class.
Expand Down Expand Up @@ -431,8 +432,10 @@ def as_dict(self) -> dict:
"""
restart_dict = dict()
if self.adaptive_levels is not None:
restart_dict['adaptive_levels'] = {atom_range: {job_type: level.as_dict() for job_type, level in levels_dict}
for atom_range, levels_dict in self.adaptive_levels.items()}
restart_dict['adaptive_levels'] = [
{'atom_range': [atom_range[0], atom_range[1]],
'levels': {' '.join(job_types): level.as_dict() for job_types, level in levels_dict.items()}}
for atom_range, levels_dict in self.adaptive_levels.items()]
if self.allow_nonisomorphic_2d:
restart_dict['allow_nonisomorphic_2d'] = self.allow_nonisomorphic_2d
if self.arkane_level_of_theory is not None:
Expand Down Expand Up @@ -1256,44 +1259,60 @@ def standardize_output_paths(self):
self.output = dict()


def process_adaptive_levels(adaptive_levels: dict | None) -> dict | None:
def process_adaptive_levels(adaptive_levels: list | None) -> dict | None:
"""
Process the ``adaptive_levels`` argument.

The user-facing form is a YAML-friendly list of entries, each a dictionary with an
``atom_range`` 2-list (the heavy-atom count range, the upper bound may be the string
``'inf'`` or ``float('inf')``) and a ``levels`` mapping of job types to levels of theory.
Job types that share a level may be given as a single whitespace- or comma-separated key
(e.g. ``'opt freq'``). A level value may be a string or a ``Level`` dictionary. For example::

adaptive_levels = [{'atom_range': [1, 6],
'levels': {'opt freq': 'wb97xd/def2tzvp',
'sp': 'ccsd(t)-f12/cc-pvtz-f12'}},
{'atom_range': [7, 'inf'],
'levels': {'opt freq': 'b3lyp/6-31g(d,p)',
'sp': 'dlpno-ccsd(t)/def2-tzvp'}}]

Args:
adaptive_levels (dict): The adaptive levels dictionary.
adaptive_levels (list): The adaptive levels specification (a list of entries).

Returns: dict
The processed adaptive levels dictionary.
Returns: dict | None
The processed adaptive levels keyed by ``(min_heavy_atoms, max_heavy_atoms)`` tuples,
each mapping job-type tuples to ``Level`` objects, or ``None`` if the input is ``None``.
"""
if adaptive_levels is None:
return None
processed = dict()
if not isinstance(adaptive_levels, dict):
raise InputError(f'The adaptive levels argument must be a dictionary, '
if not isinstance(adaptive_levels, list):
raise InputError(f'The adaptive levels argument must be a list of entries, '
f'got {adaptive_levels} which is a {type(adaptive_levels)}')
for atom_range, adaptive_level in adaptive_levels.items():
if not isinstance(atom_range, tuple) \
or not all([isinstance(a, int) or a == 'inf' for a in atom_range]) \
or len(atom_range) != 2:
raise InputError(f'Keys of the adaptive levels argument must be 2-length tuples of integers or an "inf" '
f'indicator, got {atom_range} which is a {type(atom_range)} in:\n{adaptive_levels}')
if not isinstance(adaptive_level, dict):
raise InputError(f'Each adaptive level in the adaptive levels argument must be a dictionary, '
f'got {adaptive_level} which is a {type(adaptive_level)} in:\n{adaptive_levels}')
processed = dict()
for entry in adaptive_levels:
if not isinstance(entry, dict) or 'atom_range' not in entry or 'levels' not in entry:
raise InputError(f'Each adaptive levels entry must be a dictionary with "atom_range" and "levels" '
f'keys, got {entry} which is a {type(entry)} in:\n{adaptive_levels}')
atom_range = entry['atom_range']
if not isinstance(atom_range, (list, tuple)) or len(atom_range) != 2 \
or not isinstance(atom_range[0], int) \
or not (isinstance(atom_range[1], int) or atom_range[1] in ('inf', float('inf'))):
raise InputError(f'The "atom_range" of each adaptive levels entry must be a 2-length list of an integer '
f'lower bound and an integer or "inf" upper bound, got {atom_range} '
f'in:\n{adaptive_levels}')
atom_range = (atom_range[0], 'inf' if atom_range[1] in ('inf', float('inf')) else atom_range[1])
levels = entry['levels']
if not isinstance(levels, dict):
raise InputError(f'The "levels" of each adaptive levels entry must be a dictionary, '
f'got {levels} which is a {type(levels)} in:\n{adaptive_levels}')
processed[atom_range] = dict()
for sub_key, level in adaptive_level.items():
new_sub_key = (sub_key,) if isinstance(sub_key, str) else sub_key
if not isinstance(new_sub_key, tuple):
raise InputError(f'Job types specifications in adaptive levels must be tuples, got {sub_key} '
f'which is a {type(sub_key)} in:\n{adaptive_levels}')
new_level = Level(repr=level)
processed[atom_range][new_sub_key] = new_level
atom_ranges = sorted(list(adaptive_levels.keys()), key=lambda x: x[0])
for job_types, level in levels.items():
job_type_tuple = tuple(job_types.replace(',', ' ').split())
processed[atom_range][job_type_tuple] = Level(repr=level)
atom_ranges = sorted(processed.keys(), key=lambda x: x[0])
for i, atom_range in enumerate(atom_ranges):
if i and atom_ranges[i-1][1] + 1 != atom_ranges[i][0]:
raise InputError(f'Atom ranges of adaptive levels must be consecutive. '
f'Got:\n{list(adaptive_levels.keys())}')
raise InputError(f'Atom ranges of adaptive levels must be consecutive. Got:\n{atom_ranges}')
if atom_ranges[-1][1] != 'inf':
raise InputError(f'The last atom range must be "inf", got {atom_ranges[-1][1]} in {atom_ranges}')
return processed
88 changes: 57 additions & 31 deletions arc/main_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -388,46 +388,72 @@ def test_add_hydrogen_for_bde(self):
self.assertIn('H', [spc.label for spc in arc1.species])

def test_process_adaptive_levels(self):
"""Test processing the adaptive levels"""
adaptive_levels_1 = {(1, 5): {('opt', 'freq'): 'wb97xd/6-311+g(2d,2p)',
('sp',): 'ccsd(t)-f12/aug-cc-pvtz-f12'},
(6, 15): {('opt', 'freq'): 'b3lyp/cbsb7',
'sp': 'dlpno-ccsd(t)/def2-tzvp'},
(16, 30): {('opt', 'freq'): 'b3lyp/6-31g(d,p)',
'sp': {'method': 'wb97xd', 'basis': '6-311+g(2d,2p)'}},
(31, 'inf'): {('opt', 'freq'): 'b3lyp/6-31g(d,p)',
'sp': 'b3lyp/6-311+g(d,p)'}}

"""Test processing the adaptive levels (YAML-friendly list-of-entries schema)"""
# None passes through.
self.assertIsNone(process_adaptive_levels(None))

# A normal, multi-range specification. Job types sharing a level are given as a
# whitespace- or comma-separated key; a level may be a string or a Level dict.
adaptive_levels_1 = [{'atom_range': [1, 5],
'levels': {'opt freq': 'wb97xd/6-311+g(2d,2p)',
'sp': 'ccsd(t)-f12/aug-cc-pvtz-f12'}},
{'atom_range': [6, 15],
'levels': {'opt, freq': 'b3lyp/cbsb7',
'sp': 'dlpno-ccsd(t)/def2-tzvp'}},
{'atom_range': [16, 30],
'levels': {'opt freq': 'b3lyp/6-31g(d,p)',
'sp': {'method': 'wb97xd', 'basis': '6-311+g(2d,2p)'}}},
{'atom_range': [31, 'inf'],
'levels': {'opt freq': 'b3lyp/6-31g(d,p)',
'sp': 'b3lyp/6-311+g(d,p)'}}]
processed_1 = process_adaptive_levels(adaptive_levels_1)
self.assertEqual(processed_1[(6, 15)][('sp',)].simple(), 'dlpno-ccsd(t)/def2-tzvp')
self.assertEqual(processed_1[(16, 30)][('sp',)].simple(), 'wb97xd/6-311+g(2d,2p)')

# test non dict
self.assertEqual(processed_1[(1, 5)][('opt', 'freq')].simple(), 'wb97xd/6-311+g(2d,2p)')

# A single range covering everything, and a float 'inf' is accepted as the upper bound.
processed_2 = process_adaptive_levels([{'atom_range': [1, float('inf')],
'levels': {'opt freq': 'b3lyp/6-31g(d,p)',
'sp': 'b3lyp/6-311+g(d,p)'}}])
self.assertEqual(processed_2[(1, 'inf')][('sp',)].simple(), 'b3lyp/6-311+g(d,p)')

# Restart round-trip: as_dict() must emit the list form and reproduce the same structure.
arc0 = ARC(project='adaptive_levels_test', adaptive_levels=adaptive_levels_1)
restart_levels = arc0.as_dict()['adaptive_levels']
self.assertIsInstance(restart_levels, list)
reprocessed = process_adaptive_levels(restart_levels)
self.assertEqual(reprocessed[(6, 15)][('sp',)].simple(), 'dlpno-ccsd(t)/def2-tzvp')
self.assertEqual(set(reprocessed.keys()), set(processed_1.keys()))

# Not a list (the legacy tuple-dict form is no longer accepted).
with self.assertRaises(InputError):
process_adaptive_levels(4)
# wrong atom range
with self.assertRaises(InputError):
process_adaptive_levels({5: {('opt', 'freq'): 'wb97xd/6-311+g(2d,2p)',
('sp',): 'ccsd(t)-f12/aug-cc-pvtz-f12'},
(6, 'inf'): {('opt', 'freq'): 'b3lyp/6-31g(d,p)',
'sp': 'b3lyp/6-311+g(d,p)'}})
# no 'inf
process_adaptive_levels({(1, 5): {('opt', 'freq'): 'wb97xd/6-311+g(2d,2p)'},
(6, 'inf'): {'sp': 'b3lyp/6-311+g(d,p)'}})
# atom_range is not a 2-length list.
with self.assertRaises(InputError):
process_adaptive_levels([{'atom_range': [5], 'levels': {'sp': 'b3lyp/6-311+g(d,p)'}}])
# 'inf' is only allowed as the upper bound.
with self.assertRaises(InputError):
process_adaptive_levels([{'atom_range': [float('inf'), 'inf'], 'levels': {'sp': 'b3lyp/6-311+g(d,p)'}}])
with self.assertRaises(InputError):
process_adaptive_levels([{'atom_range': ['inf', 10], 'levels': {'sp': 'b3lyp/6-311+g(d,p)'}}])
# The last range does not end with 'inf'.
with self.assertRaises(InputError):
process_adaptive_levels([{'atom_range': [1, 5], 'levels': {'sp': 'wb97xd/def2tzvp'}},
{'atom_range': [6, 75], 'levels': {'sp': 'b3lyp/6-311+g(d,p)'}}])
# 'levels' is not a dict.
with self.assertRaises(InputError):
process_adaptive_levels({(1, 5): {('opt', 'freq'): 'wb97xd/6-311+g(2d,2p)',
('sp',): 'ccsd(t)-f12/aug-cc-pvtz-f12'},
(6, 75): {('opt', 'freq'): 'b3lyp/6-31g(d,p)',
'sp': 'b3lyp/6-311+g(d,p)'}})
# adaptive level not a dict
process_adaptive_levels([{'atom_range': [1, 5], 'levels': {'sp': 'wb97xd/def2tzvp'}},
{'atom_range': [6, 'inf'], 'levels': 'b3lyp/6-31g(d,p)'}])
# Non-consecutive atom ranges.
with self.assertRaises(InputError):
process_adaptive_levels({(1, 5): {('opt', 'freq'): 'wb97xd/6-311+g(2d,2p)',
('sp',): 'ccsd(t)-f12/aug-cc-pvtz-f12'},
(6, 'inf'): 'b3lyp/6-31g(d,p)'})
# non-consecutive atom ranges
process_adaptive_levels([{'atom_range': [1, 5], 'levels': {'sp': 'wb97xd/def2tzvp'}},
{'atom_range': [15, 'inf'], 'levels': {'sp': 'b3lyp/6-311+g(d,p)'}}])
# An entry missing required keys.
with self.assertRaises(InputError):
process_adaptive_levels({(1, 5): {('opt', 'freq'): 'wb97xd/6-311+g(2d,2p)',
('sp',): 'ccsd(t)-f12/aug-cc-pvtz-f12'},
(15, 'inf'): {('opt', 'freq'): 'b3lyp/6-31g(d,p)',
'sp': 'b3lyp/6-311+g(d,p)'}})
process_adaptive_levels([{'levels': {'sp': 'wb97xd/def2tzvp'}}])

def test_process_level_of_theory(self):
"""
Expand Down
Loading