Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run_task(self, fw_spec):
# mendeleev params
a_mr = fw_spec['A']
b_mr = fw_spec['B']
c_mr = fw_spec['C']
# convert from mendeleev and score compound
a, b, c = mend_to_name(a_mr, b_mr, c_mr)
data = perovskites.loc[(perovskites['A'] == a) & (perovskites['B'] == b) & (perovskites['anion'] == c)]
score = float(data['complex_score'])
output = {'_y_opt': score}
return FWAction(update_spec=output)
task_fields_to_push = self.get("task_fields_to_push", None)
update_spec = {}
if task_fields_to_push:
if isinstance(task_fields_to_push, dict):
for key, path_in_task_doc in task_fields_to_push.items():
if has(task_doc, path_in_task_doc):
update_spec[key] = get(task_doc, path_in_task_doc)
else:
logger.warning("Could not find {} in task document. Unable to push to next firetask/firework".format(path_in_task_doc))
else:
raise RuntimeError("Inappropriate type {} for task_fields_to_push. It must be a "
"dictionary of format: {key: path} where key refers to a field "
"in the spec and path is a full mongo-style path to a "
"field in the task document".format(type(task_fields_to_push)))
return FWAction(stored_data={"task_id": task_doc.get("task_id", None)},
defuse_children=defuse_children, update_spec=update_spec)
def run_task(self, fw_spec):
pmr = PackmolRunner(self["molecules"], self["packing_config"],
tolerance=self.get("tolerance", 2.0),
filetype=self.get("filetype", "xyz"),
control_params=self.get("control_params", {"nloop": 1000}),
output_file=self.get("output_file", "packed_mol.xyz"),
bin=self["packmol_cmd"])
logger.info("Running {}".format(self["packmol_cmd"]))
packed_mol = pmr.run(self.get("copy_to_current_on_exit", False), site_property=self.get("site_property", None))
logger.info("Packmol finished running.")
return FWAction(mod_spec=[{'_set': {'packed_mol': packed_mol}}])
start_mol = self.get("molecule")
else:
raise KeyError(
"No molecule present, add as an optional param or check fw_spec"
)
babe_mol = BabelMolAdaptor(start_mol).openbabel_mol
babe_mol.SetTorsion(self["atom_indexes"][0], self["atom_indexes"][1],
self["atom_indexes"][2], self["atom_indexes"][3],
(self["angle"] * np.pi / 180.))
rotated_mol = BabelMolAdaptor(babe_mol).pymatgen_mol
# update the fw_spec with the rotated geometry
update_spec = {"prev_calc_molecule": rotated_mol}
return FWAction(update_spec=update_spec)
rescale_args = {"initial_pressure": pressure * 1000, "initial_temperature": 1, "beta": 0.0000005}
rescale_args = recursive_update(rescale_args, rescale_params)
# Spawn fw
fw = MDFW(structure, name=f'density_run_{density_spawn_count + 1}-{tag_id}',
previous_structure=False,
**run_specs, **md_params, **optional_params)
converge_params["density_spawn_count"] += 1
_spawner_args = {"converge_params": converge_params, "rescale_params": rescale_params,
"run_specs": run_specs, "md_params": md_params,
"optional_fw_params": optional_params, "tag_id": tag_id}
fw = powerups.add_rescale_volume(fw, **rescale_args)
fw = powerups.add_pass_pv(fw)
fw = powerups.add_converge_task(fw, **_spawner_args)
wf = Workflow([fw])
return FWAction(detours=wf, stored_data={'pressure': pressure, 'energy': mu})
else:
fw = MDFW(structure, name=f'energy_run_{energy_spawn_count + 1}_{tag_id}', previous_structure=False,
**run_specs, **md_params, **optional_params)
converge_params["energy_spawn_count"] += 1
_spawner_args = {"converge_params": converge_params, "rescale_params": rescale_params,
"run_specs": run_specs, "md_params": md_params,
"optional_fw_params": optional_params, "tag_id": tag_id}
fw = powerups.add_pass_pv(fw)
fw = powerups.add_converge_task(fw, **_spawner_args)
wf = Workflow([fw])
return FWAction(detours=wf, stored_data={'pressure': pressure, 'energy': mu})
else:
return FWAction(stored_data={'pressure': pressure,
'energy': mu,
'density_calculated': True})
def run_task(self, fw_spec):
default_list = ["INCAR", "POSCAR", "CONTCAR", "OUTCAR", "POTCAR", "vasprun.xml", "XDATCAR", "OSZICAR", "DOSCAR"]
files = self.get("files", default_list)
calc_home = self["calc_home"]
run_name = self["run_name"]
target_dir = os.path.join(calc_home, run_name)
if not os.path.exists(calc_home):
os.mkdir(calc_home)
if not os.path.exists(target_dir):
os.mkdir(target_dir)
for f in files:
try:
shutil.copy2(f, target_dir)
except:
pass
return FWAction()
def run_task(self, fw_spec):
structure_dict = fw_spec["structure"]
if self.get("rescale_volume", False):
spec_structure = Structure.from_dict(structure_dict)
scaling_volume = spec_structure.volume * self["rescale_volume"]
spec_structure.scale_lattice(scaling_volume)
structure_dict = spec_structure.as_dict()
_poscar = Poscar(structure_dict)
_poscar.write_file("POSCAR")
return FWAction()
def run_task(self, fw_spec):
mpr = MPRester(env_chk(self.get("MAPI_KEY"), fw_spec))
vasprun, outcar = get_vasprun_outcar(self.get("calc_dir", "."),
parse_dos=False,
parse_eigen=False)
my_entry = vasprun.get_computed_entry(inc_structure=False)
stored_data = mpr.get_stability([my_entry])[0]
if stored_data["e_above_hull"] > self.get("ehull_cutoff", 0.05):
logger.info("CheckStability: failed test!")
return FWAction(stored_data=stored_data, exit=True,
defuse_workflow=True)
else:
return FWAction(stored_data=stored_data)
def run_task(self, fw_spec):
x = fw_spec["_x"]
y = x[0] * x[1] / x[2]
return FWAction(update_spec={"_y": y})
target_pressure = self.get("target_pressure", 0.0)
alpha = self.get("alpha", 10e-6)
beta = self.get("beta", 10e-7)
corr_vol = RescaleVolume.of_poscar(poscar_path="./POSCAR", initial_temperature=initial_temperature,
initial_pressure=initial_pressure,
target_pressure=target_pressure,
target_temperature=target_temperature, alpha=alpha, beta=beta)
# Rescale volume based on temperature difference first. Const T will return no volume change:
corr_vol.by_thermo(scale='temperature')
# TO DB ("Rescaled volume due to delta T: ", corr_vol.structure.volume)
# Rescale volume based on pressure difference:
corr_vol.by_thermo(scale='pressure')
# TO DB ("Rescaled volume due to delta P: ", corr_vol.structure.volume)
corr_vol.poscar.write_file("./POSCAR")
# Pass the rescaled volume to Poscar
return FWAction(stored_data=corr_vol.structure.as_dict())