from . import fwio
import fireworks
from ase.db.row import AtomsRow
from ase.db import connect
from netrc import netrc
import numpy as np
[docs]class Laminar():
"""Simple submission script helper for CatFlow.
"""
def __init__(
self,
host,
username=None,
name=None,
password=None):
"""Initialize a fireworks instance."""
if username is None or name is None or password is None:
username, name, password = netrc().authenticators(host)
if username is None:
raise ValueError('username, name, and password required.')
launchpad = fireworks.LaunchPad(
host=host,
name=name,
username=username,
password=password)
self.launchpad = launchpad
[docs] def submit_relaxation(
self,
image,
workflow_name,
parameters=None,
spec=None):
"""Run a relaxation of a given DB entry or atoms object. If a
database object is used, the calculation will automatically store
the keys and data for later retrieval.
The entries uuid will also be stored and `data.calculator_parameters`
will be used as the calculation parameters.
Parameters
----------
images : Atoms object | AtomsRow object
ASE database entry or atoms object to relax.
workflow_name : str
Name of the fireworks calculation to be used.
parameters : dict
Calculation parameters to use. Will be pulled from
a database entry `data.calculator_parameters`.
spec : dict
Additional fireworks specifications to pass to the database.
"""
keys, data = {}, {}
if isinstance(image, AtomsRow):
atoms = image.toatoms()
keys.update(image.key_value_pairs)
keys.update({'uuid': image.unique_id})
data.update(image.data)
else:
atoms = image
if parameters is None:
if data.get('calculator_parameters'):
parameters = data.get('calculator_parameters')
del data['calculator_parameters']
elif atoms.info.get('calculator_parameters'):
parameters = atoms.info.get('calculator_parameters')
else:
raise ValueError('Calculation parameters missing.')
calculator = parameters.pop('calculator_name', None)
if calculator is None:
raise ValueError("'calculator_name' missing from parameters.")
atoms.info['calculator_parameters'] = parameters
for k, v in data.items():
if isinstance(v, np.ndarray):
fwio.array_to_list(v)
data[k] = v
encoding = fwio.atoms_to_encode(atoms)
t0 = fireworks.PyTask(
func='catkit.flow.fwio.encode_to_atoms',
args=[encoding])
t1 = fireworks.PyTask(
func='catkit.flow.fwase.get_potential_energy',
args=[calculator],
stored_data_varname='trajectory')
tasks = [t0, t1]
if spec is None:
spec = {'keys': keys, 'data': data}
else:
spec.update({'keys': keys, 'data': data})
firework = fireworks.Firework(tasks, spec=spec)
workflow = fireworks.Workflow([firework], name=workflow_name)
self.launchpad.add_wf(workflow)
[docs] def submit_relaxation_db(self, database, spec=None):
"""Submit each entry of an ASE database for relaxation.
This requires that the calculation parameters be stored in
the data under `data.calculator_parameters`.
Parameters
----------
database : str
Path to ASE database to be looped over for relaxation.
spec : dict
Additional specification to be passed to Fireworks.
"""
filename = database.split('/')[-1]
db = connect(database)
for d in db.select():
self.submit_relaxation(
d, workflow_name=filename, spec=spec)
[docs] def bulk_relaxation(
self,
atoms,
parameters,
spec=None):
"""Run a relaxation of a given DB entry or atoms object. If a
database object is used, the calculation will automatically store
the keys and data for later retrieval.
The entries uuid will also be stored and `data.calculator_parameters`
will be used as the calculation parameters.
Parameters
----------
images : Atoms object
Initial atoms to perform workflow on.
parameters : dict
Calculation parameters to use.
workflow_name : str
Name of the fireworks calculation to be used.
spec : dict
Additional fireworks specifications to pass to the database.
"""
atoms.info['calculator_parameters'] = parameters
encoding = fwio.atoms_to_encode(atoms)
t0 = fireworks.PyTask(
func='catkit.flow.fwio.encode_to_atoms',
args=[encoding])
t1 = fireworks.PyTask(
func='catkit.flow.fwase.catflow_relaxation',
stored_data_varname='trajectory')
tasks = [t0, t1]
if spec is None:
spec = {}
firework = fireworks.Firework(tasks, spec=spec)
workflow = fireworks.Workflow([firework], name='bulk_relaxation')
workflow_id = self.launchpad.add_wf(workflow)[-1]
return workflow_id