# -*- coding: utf-8 -*-
__copyright__ = """This file is part of SCINE Puffin.
This code is licensed under the 3-clause BSD license.
Copyright ETH Zurich, Laboratory for Physical Chemistry, Reiher Group.
See LICENSE.txt for details
"""
import os
from contextlib import redirect_stdout
from puffin.config import Configuration
from .job import ScineJob, calculation_context
[docs]class ScineIrcScan(ScineJob):
"""
A job scanning a single intrinsic reaction coordinate.
**Order Name**
``scine_irc_scan``
**Optional Settings**
Optional settings are read from the ``settings`` field, which is part of
any ``Calculation`` stored in a SCINE Database.
Possible settings for this job are:
All settings recognized by ReaDuct's IRC task. For a complete list see
the
`ReaDuct manual <https://scine.ethz.ch/static/download/readuct_manual.pdf>`_
Common examples are:
irc_allow_unconverged :: bool
If ``True``, the optimization does not need to fully converge but will
be accepted as a success even if it reaches the maximum amounts of
optimization cycles. Also the resulting structures ill be flaged as
``minimum_guess`` is this option is set ot be ``True``.
(Default: ``False``)
irc_mode :: int
The mode to follow during the IRC scan. By default the first mode (0).
(mode with the larges imaginary frequency will be followed).
All settings that are recognized by the SCF program chosen.
Common examples are:
max_scf_iterations :: int
The number of allowed SCF cycles until convergence.
**Required Packages**
- SCINE: Database (present by default)
- SCINE: Readuct (present by default)
- SCINE: Utils (present by default)
- A program implementing the SCINE Calculator interface, e.g. Sparrow
**Generated Data**
If successful the following data will be generated and added to the
database:
Structures
Both the forward and backward optimized structures will be added to the
database. They will be flaged as: ``minimum_optimized`` or
``minimum_guess`` if ``irc_allow_unconverged`` is set to ``True``.
Properties
The ``electronic_energy`` associated with both forward and backward
structures.
"""
def __init__(self):
super().__init__()
def run(self, manager, calculation, config: Configuration) -> bool:
import scine_database as db
import scine_readuct as readuct
import scine_utilities as utils
# Gather all required collections
structures = manager.get_collection('structures')
calculations = manager.get_collection('calculations')
properties = manager.get_collection('properties')
# Link calculation if needed
if not calculation.has_link():
calculation.link(calculations)
# Get structure
structure = db.Structure(calculation.get_structures()[0])
structure.link(structures)
# Get model
model = calculation.get_model()
# Structure based arguments
arguments = {}
arguments['molecular_charge'] = structure.get_charge()
arguments['spin_multiplicity'] = structure.get_multiplicity()
# Model based arguments
if model.method and (model.method != model.method_family):
arguments['method'] = model.method
if model.basis_set:
arguments['basis_set'] = model.basis_set
# arguments['spin_mode'] = model.spin_mode
if not model.program == 'any':
arguments['program'] = model.program
irc_allow_unconverged = False
with calculation_context(self._work_dir, 'output', 'errors'):
# Task based arguments
task_args = calculation.get_settings()
task_args['output'] = ['forward', 'backward']
if 'irc_allow_unconverged' in task_args:
irc_allow_unconverged = task_args['irc_allow_unconverged']
# Get the settings that are available for the chosen calculator
prog = model.program if (not model.program == "any") else ""
available_calculator_settings = utils.core.get_available_settings(model.method_family, prog)
# Transfer the settings that belong to the calculator to the other dictionary
keys_to_remove = []
for key, value in task_args.items():
if key in available_calculator_settings:
arguments[key] = value
keys_to_remove.append(key)
for key in keys_to_remove:
del task_args[key]
utils.IO.write('system.xyz', structure.get_atoms())
system = utils.core.load_system_into_calculator('system.xyz', model.method_family, **arguments)
systems = {'system': system}
systems, success = readuct.run_irc_task(systems, ['system'], **task_args)
# After the calculation, verify that the connection to the database still exists
self.verify_connection(manager)
# Capture raw output
stdout_path = os.path.join(self._work_dir, 'output')
stderr_path = os.path.join(self._work_dir, 'errors')
with open(stdout_path, 'r') as stdout, open(stderr_path, 'r') as stderr:
calculation.set_raw_output(stdout.read()+'\n'+stderr.read())
if not os.path.exists(os.path.join(self._work_dir, 'success')):
calculation.set_status(db.Status.FAILED)
return False
if not success:
calculation.set_status(db.Status.FAILED)
return False
# Basic check for a complete calculation
if not systems['forward'].has_results():
calculation.set_status(db.Status.FAILED)
return False
if not systems['backward'].has_results():
calculation.set_status(db.Status.FAILED)
return False
# Check successfull termination
for name in ['forward', 'backward']:
results = systems[name].get_results()
# Check successfull termination
if results.successful_calculation is None:
return False
if not results.successful_calculation:
calculation.set_status(db.Status.FAILED)
return False
# Update model
if results.program_name:
model.program = results.program_name.lower()
model.version = config.programs()[model.program]['version']
calculation.set_model(model)
# New structures
db_results = calculation.get_results()
label = db.Label.MINIMUM_OPTIMIZED
if irc_allow_unconverged:
label = db.Label.MINIMUM_GUESS
new_structure = db.Structure()
new_structure.link(structures)
new_structure.create(calculation.positions, arguments['molecular_charge'],
arguments['spin_multiplicity'], model, label)
db_results.add_structure(new_structure.id())
calculation.set_results(db_results)
# Store energy
if results.energy is None:
calculation.set_status(db.Status.FAILED)
return False
self.store_property(properties, 'electronic_energy', 'NumberProperty',
results.energy, model, calculation, new_structure)
calculation.set_executor(config['daemon']['uuid'])
calculation.set_status(db.Status.COMPLETE)
return True
def required_programs(self):
return ['database', 'utils', 'readuct']