From 1ca461af0e24d948c58f59d98c27de1b7a52d13a Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 29 Sep 2025 14:58:03 +0200 Subject: [PATCH 01/18] Maker to run QE scf calculations --- .../auto/GenMLFF/QuantumEspressoSCF.py | 414 ++++++++++++++++++ 1 file changed, 414 insertions(+) create mode 100644 src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py diff --git a/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py b/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py new file mode 100644 index 000000000..a92df4a29 --- /dev/null +++ b/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py @@ -0,0 +1,414 @@ +"""Jobs to create training data for ML potentials.""" + +import os +import logging +import subprocess +import numpy as np +from glob import glob +from dataclasses import dataclass, field + +from ase import Atoms +from ase.io import read +from jobflow import job, Flow, Maker, Response + +logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s") + +def qe_params_from_config(config: dict): + """ + Return QEstaticLabelling params from a configuration dictionary. + + Args: + config (dict): Keys should match __init__ parameters. For example: + { + "qe_run_cmd": qe_run_cmd, + "num_qe_workers": num_qe_workers, + "fname_pwi_template": fname_pwi_template, + "kspace_resolution" : Kspace_resolution, + "koffset": Koffset, + "fname_structures": fname_structures, + } + + Returns: + params: dict + Dictionary with parameters for QEstaticLabelling. + """ + #Get default parameters + params = { + "qe_run_cmd": "pw.x", + "num_qe_workers": 1, + "fname_pwi_template": None, + "kspace_resolution" : None, + "koffset": [False, False, False], + "fname_structures": None, + } + + # Update parameters with values from the config file + if config is None: raise ValueError("Configuration file is empty or not properly formatted.") + params.update(config) + + #Check a valid reference pwi path is provided + if not os.path.exists(params["fname_pwi_template"]): raise ValueError(f"Reference QE input file '{params['fname_pwi_template']}' not found.") + + return params + +@dataclass +class QEstaticLabelling(Maker): + """ + Maker to set up and run Quantum Espresso static calculations for input structures, including bulk, isolated atoms, and dimers. + Parameters + ---------- + name: str + Name of the flow. + qe_run_cmd: str + String with the command to run QE (including its executable path/or application name). + fname_pwi_template: str + Path to file containing the template computational parameters. + fname_structures: str + Path to ASE-readible file containing the structures to be computed. + num_qe_workers: int | None + Number of workers to use for the calculations. If None, defaults to the number of structures. + """ + + name: str = "do_qe_labelling" + qe_run_cmd: str | None = None #String with the command to run QE (including its executable path/or application name) + fname_pwi_template: str | None = None #Path to file containing the template computational parameters + fname_structures: str | list[str] | None = None #Path or list[Path] to ASE-readible file containing the structures to be computed + num_qe_workers: int | None = None #Number of workers to use for the calculations. + kspace_resolution: float | None = None #K-space resolution in Angstrom^-1, used to set the K-points in the pwi file + koffset: list[bool] = field(default_factory=lambda: [False, False, False]) #K-points offset in the pwi file + + def make(self): + #Define jobs + joblist = [] + + # Load structures + structures = self.load_structures(fname_structures=self.fname_structures) + if len(structures) == 0: + logging.info("No structures found to compute with DFT. Exiting.") + return Response(replace=None, output=[]) + + # Check pwi template + pwi_reference_lines = self.check_pwi_template(self.fname_pwi_template) + + # Write pwi input files for each structure + work_dir = os.getcwd() + path_to_qe_workdir = os.path.join(work_dir, "scf_files") + os.makedirs(path_to_qe_workdir, exist_ok=True) + for i, structure in enumerate(structures): + #Get fname of the next pwi file + fname_new_pwi = os.path.join(path_to_qe_workdir, f"structure_{i}.pwi") + + #Write pwi file for the structure + self.write_pwi( + fname_pwi_output=fname_new_pwi, + structure=structure, + pwi_reference=pwi_reference_lines, + ) + + # Set number of QE workers + if self.num_qe_workers is None: # 1 worker per structure (all DFT jobs in parallel) + num_qe_workers = len(glob(os.path.join(path_to_qe_workdir, "*.pwi"))) + else: + num_qe_workers = self.num_qe_workers + + # Launch QE workers + outputs = [] + for id_qe_worker in range(num_qe_workers): + qe_worker = self.run_qe_worker( + id=id_qe_worker, + command=self.qe_run_cmd, + work_dir=path_to_qe_workdir + ) + + qe_worker.name = f"run_qe_worker_{id_qe_worker}" + joblist.append(qe_worker) + outputs.append(qe_worker.output) #Contains list of dict{'successes', 'pwo_files', 'outdirs'} for each worker + + qe_wrk_flow = Flow(jobs=joblist, output=outputs, name="qe_workers") + + # Output is a list of success status, one for each worker + # The success status is a dictionary with the pwo file name as key and the calculation success status as value (True/False) + return Response(replace=qe_wrk_flow, output=qe_wrk_flow.output) + + def load_structures(self, + fname_structures: str | list[str] | None = None, + ): + """ + Load structures from a file or a list of files. + Parameters + ---------- + fname_structures : str | list[str] | None + Path or list of paths to ASE-readable files containing the structures to be loaded. + If None, no structures will be loaded. + Returns + ------- + list[Atoms] + List of ASE Atoms objects representing the loaded structures. + """ + #Convert fname_structures to a list if it is a string + if isinstance(fname_structures, str): + fname_structures = [fname_structures] + elif fname_structures is None: + return [] + elif not isinstance(fname_structures, list): + raise ValueError("fname_structures must be a string or a list of strings.") + + #Loop over provided files and load structures + structures = [] + for fname in fname_structures: + #Check if all files exist + if not os.path.exists(fname): raise FileNotFoundError(f"File {fname} does not exist.") + + #Read structures from file + try: + structures += read(fname, index=":") + except Exception as e: + logging.error(f"Error reading file {fname}: {e}") + + return structures + + def check_pwi_template(self, fname_template): + """ + Check the pwi template file for the required parameters. + """ + # Read template file + tmp_pwi_lines = [] + with open(fname_template, 'r') as f: + tmp_pwi_lines = f.readlines() + + # Modify lines with structure information: + # Assume ntyp, atom_types and pseudoptentials are already defined in the template and consistent with the structures + # Assume ibrav=0 and Kspacing is already defined in the template + idx_nat_line, idx_pos_line, idx_cell_line = 0, 0, 0 + for i, line in enumerate(tmp_pwi_lines): + if 'nat' in line: idx_nat_line = i + + elif 'ATOMIC_POSITIONS' in line: idx_pos_line = i + + elif 'CELL_PARAMETERS' in line: idx_cell_line = i + + # Set nat line + if idx_nat_line == 0: # nat not defined, assume nat = 0 + raise ValueError("Number of atoms line not defined in the template file. Please define \'nat =\' in the template file.") + else: + tmp_pwi_lines[idx_nat_line] = f'nat = \n' + + # Cancel lines with ATOMIC_POSITIONS and CELL_PARAMETERS + if idx_pos_line == 0 and idx_cell_line > 0: + idx_to_delete = idx_pos_line + del(tmp_pwi_lines[idx_to_delete:]) + + elif idx_pos_line > 0 and idx_cell_line == 0: + idx_to_delete = idx_pos_line + del(tmp_pwi_lines[idx_to_delete:]) + + elif idx_pos_line > 0 and idx_cell_line > 0: + idx_to_delete = min([idx_pos_line, idx_cell_line]) + del(tmp_pwi_lines[idx_to_delete:]) + + return tmp_pwi_lines + + def write_pwi( + self, + fname_pwi_output: str, + structure: Atoms, + pwi_reference: list[str], + ): + """ + Write the pwi input file for the given structure. + """ + #Duplicate the pwi template to avoid overwriting the reference lines + pwi_template = pwi_reference.copy() + + # Check pwi lines + idx_diskio, idx_outdir, idx_nat_line, idx_kpoints_line, nat = 0, 0, 0, 0, len(structure) + for idx, line in enumerate(pwi_template): + if 'nat =' in line: idx_nat_line = idx + elif 'disk_io' in line: idx_diskio = idx + elif 'outdir' in line: idx_outdir = idx + elif 'K_POINTS' in line: idx_kpoints_line = idx + + #Update number of atoms + pwi_template[idx_nat_line] = f'nat = {nat}\n' + + #Get identifier for this structure + structure_id = fname_pwi_output.split('/')[-1].replace('.pwi', '') + + #Update outdir based on disk_io + if idx_diskio == 0 or 'none' not in pwi_template[idx_diskio]: #disk_io is not 'none' (QE default is low for scf) + if idx_outdir == 0: # outdir not defined, define it + pwi_template.insert(idx_diskio + 1, f"outdir = {structure_id}\n") + else: # outdir is defined, update it + pwi_template[idx_outdir] = f"outdir = '{structure_id}'\n" + else: #disk_io is 'none', remove outdir line + if idx_outdir == 0: + pwi_template.insert(idx_diskio + 1, f"outdir = 'OUT'\n") + + kpoints_lines = self._set_Kpoints( + tmp_pwi_lines=pwi_template, + idx_kpoints_line=idx_kpoints_line, + atoms=structure, + Kspace_resolution=self.kspace_resolution, + Koffset=self.koffset, + ) + + #Write cell lines + cell_lines = ["\nCELL_PARAMETERS (angstrom)\n"] + cell_lines += [f"{structure.cell[i, 0]:.10f} {structure.cell[i, 1]:.10f} {structure.cell[i, 2]:.10f}\n" for i in range(3)] + + #Write positions lines + pos_lines = ["\nATOMIC_POSITIONS (angstrom)\n"] + for i, atom in enumerate(structure): + pos_lines.append(f"{atom.symbol} {structure.positions[i, 0]:.10f} {structure.positions[i, 1]:.10f} {structure.positions[i, 2]:.10f}\n") + + # Write the modified lines to the new pwi file + with open(fname_pwi_output, 'w') as f: + for line in pwi_template: #Write reference pwi lines (computational parameters) + f.write(line) + for line in kpoints_lines: #Write K-points lines + f.write(line) + for line in cell_lines: #Write cell lines + f.write(line) + for line in pos_lines: #Write positions lines + f.write(line) + + def _set_Kpoints(self, + tmp_pwi_lines: list[str], + idx_kpoints_line: int, + atoms: Atoms, + Kspace_resolution: float | None = None, + Koffset: list[bool] = [False, False, False], + ): + """ + Set the K-points in the pwi file based on user definition or K-space resolution. + """ + # Define K-points lines + kpoints_lines = [] + + # K_POINTS line not found + if idx_kpoints_line == 0: + if Kspace_resolution is None: # K_POINTS line not found and Kspace_resolution is not defined + raise ValueError("K_POINTS line not found in the template file. Please define K_POINTS in the template file or provide Kspace_resolution.") + else: # Find k-points grid using Monkorst-Pack method based on K-space resolution + #Get real space cell + cell = atoms.cell + + #Find Kpoints grid + #TODO: use structure_type info to generalize to non-periodic systems (3d, 2d, 1d, 0d) + MP_mesh = self._compute_kpoints_grid(cell, Kspace_resolution) + + #Format k-points lines + kpoints_lines.append(f"\nK_POINTS automatic\n") #Header for MP-grid + Kpoint_line = f"{MP_mesh[0]} {MP_mesh[1]} {MP_mesh[2]}" #K-points grid line + for offset in Koffset: #Add offset + if offset: Kpoint_line += " 1" + else: Kpoint_line += " 0" + Kpoint_line += "\n" + kpoints_lines.append(Kpoint_line) + + # K_POINTS is defined by user in reference pwi file, keep the line/s + elif idx_kpoints_line > 0: + if 'gamma' in tmp_pwi_lines[idx_kpoints_line] or 'Gamma' in tmp_pwi_lines[idx_kpoints_line]: # KPOINT is 1 line + kpoints_lines = tmp_pwi_lines[idx_kpoints_line:idx_kpoints_line+1] + del tmp_pwi_lines[idx_kpoints_line:] + elif 'automatic' in tmp_pwi_lines[idx_kpoints_line]: # KPOINTS is 2 lines + kpoints_lines = tmp_pwi_lines[idx_kpoints_line:idx_kpoints_line+2] + del tmp_pwi_lines[idx_kpoints_line:] + elif 'tpiba' in tmp_pwi_lines[idx_kpoints_line] or 'crystal' in tmp_pwi_lines[idx_kpoints_line]: #KPOINTS is multiple lines + num_ks = int(tmp_pwi_lines[idx_kpoints_line+1].split()[0]) #Get number of k-points + kpoints_lines = tmp_pwi_lines[idx_kpoints_line:idx_kpoints_line+num_ks+2] #Get k-points lines + del tmp_pwi_lines[idx_kpoints_line:] + else: + raise ValueError(f"K_POINTS format: {tmp_pwi_lines[idx_kpoints_line]} is unknown in pwi template file") + + return kpoints_lines + + def _compute_kpoints_grid(self, cell, Kspace_resolution): + """ + Compute the k-points grid using Monkhorst-Pack method based on the cell and K-space resolution. + """ + #Compute the reciprocal cell vectors: b_i = 2π * (a_j x a_k) / (a_i . (a_j x a_k)) + rec_cell = 2.0 * np.pi * np.linalg.inv(cell).T + + #Compute reciprocal lattice vecotors' lenghts + lengths = np.linalg.norm(rec_cell, axis=1) + + #Compute mesh size + mesh = [int(np.ceil(L / Kspace_resolution)) for L in lengths] + print(f"Computed k-points mesh: {mesh} for K-space resolution: {Kspace_resolution} Angstrom^-1") #DEBUG + + return mesh + + @job + def run_qe_worker( + self, + id, + command, + work_dir, + ): + """ + Run the QE command in a subprocess. + """ + #Get pwi files + pwi_files = glob(os.path.join(work_dir, "*.pwi")) + + #Check pwo does not exist + worker_output = {'success' : [], 'output' : [], 'outdir' : []} + for pwi in pwi_files: + #Try locking the pwi file + lock_pwi, pwo_fname = self.lock_input(pwi_fname=pwi, worker_id=id) + + if lock_pwi == "": continue #Skip to next pwi if lock failed + + #Get output directory of this calculation + with open(lock_pwi, 'r') as f: + pwi_lines = f.readlines() + outdir_line = [line.split('=')[1] for line in pwi_lines if 'outdir' in line][0] + outdir_line = outdir_line.strip().replace("'", "").replace('"', '') # Remove quotes + outdir = os.getcwd() + f"/{outdir_line}" + + #Launch QE calculation + success = self.run_qe(command=command, fname_pwi=lock_pwi, fname_pwo=pwo_fname) + + #Update output + worker_output['success'].append(success) + worker_output['output'].append(pwo_fname) + worker_output['outdir'].append(outdir) + + return worker_output + + def run_qe(self, command, fname_pwi, fname_pwo): + """ + Run the QE command in a subprocess. Execute one QuantumEspresso calculation on the current input file. + """ + #Assemble QE command + run_cmd = f"{command} < {fname_pwi} >> {fname_pwo}" + + success = False + try: + # Launch QE and wait till ending + subprocess.run(run_cmd, shell=True, check=True, executable="/bin/bash") + + success = True + + except subprocess.CalledProcessError as e: + + success = False + + return success + + def lock_input(self, pwi_fname, worker_id): + + pwi_lock_fname = "" + #Check if pwo exists + pwo_fname = pwi_fname.replace('.pwi', '.pwo') + if os.path.exists(pwo_fname): return pwi_lock_fname, pwo_fname #If exists, skip to next pwi + + # Try to lock the pwi file by renaming it + pwi_lock_fname = f'{pwi_fname}.lock_{worker_id}' + try: + os.rename(f'{pwi_fname}', f'{pwi_lock_fname}') + except Exception as e: + pwi_lock_fname = "" + + return pwi_lock_fname, pwo_fname \ No newline at end of file From 6823fef9d1b380bdd95e4ffba926abbfdc06ae19 Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 29 Sep 2025 15:16:24 +0200 Subject: [PATCH 02/18] Template pwi and QE test --- tests/auto/GenMLFF/reference.pwi | 47 ++++++++++++++++++++++++++++++++ tests/auto/GenMLFF/test_qe.py | 44 ++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+) create mode 100644 tests/auto/GenMLFF/reference.pwi create mode 100644 tests/auto/GenMLFF/test_qe.py diff --git a/tests/auto/GenMLFF/reference.pwi b/tests/auto/GenMLFF/reference.pwi new file mode 100644 index 000000000..22927b5fb --- /dev/null +++ b/tests/auto/GenMLFF/reference.pwi @@ -0,0 +1,47 @@ +&CONTROL + calculation = 'scf' + restart_mode = 'from_scratch' + prefix = 'FeCOH' + tprnfor = .true. + tstress = .true. + outdir = './OUT/' + disk_io = 'none' + pseudo_dir = '/path/to/directory/PSEUDO' + max_seconds = 86000 +/ + +&SYSTEM + ibrav = 0 + nat = 506 + ntyp = 4 + ecutwfc = 60 + ecutrho = 480 + occupations = 'smearing' + smearing = 'gaussian' + degauss = 0.00735 + nosym=.true. + vdw_corr = 'dft-d3' + nspin = 2 + starting_magnetization(1) = 0.4 +/ + +&ELECTRONS + diagonalization = 'david' + mixing_beta = 0.15 + electron_maxstep = 150 + mixing_mode = 'local-TF' + mixing_ndim = 16 + conv_thr = 1.0d-6 +/ + +&IONS +/ + +ATOMIC_SPECIES +Fe 55.845 Fe.pbe-sp-van.UPF +C 12.011 C.pbe-n-kjpaw_psl.1.0.0.UPF +O 15.999 O.pbe-n-kjpaw_psl.1.0.0.UPF +H 2.016 H.pbe-kjpaw_psl.1.0.0.UPF + +K_POINTS automatic +4 4 4 1 1 1 diff --git a/tests/auto/GenMLFF/test_qe.py b/tests/auto/GenMLFF/test_qe.py new file mode 100644 index 000000000..faa60562c --- /dev/null +++ b/tests/auto/GenMLFF/test_qe.py @@ -0,0 +1,44 @@ +from jobflow import Flow, job +from jobflow_remote import submit_flow, set_run_config +from autoplex.auto.GenMLFF.QuantumEspressoSCF import qe_params_from_config, QEstaticLabelling + +#Define resources +parallel_gpu_resources = { + "account": "IscrB_CNT-HARV", + "partition": "boost_usr_prod", + "qos": "boost_qos_dbg", + "time": "00:30:00", + "nodes": 1, + "ntasks_per_node": 2, + "cpus_per_task": 8, + "gres": "gpu:2", + "mem": "240000", + "job_name": "qe_auto", + "qerr_path": "JOB.err", + "qout_path": "JOB.out", + } + + +#Define QE test parameters +qe_test_params = { + "qe_run_cmd": "mpirun -np 1 pw.x", + "num_qe_workers": 2, #Number of workers to use for the calculations. If None setp up 1 worker per scf + "kspace_resolution": 0.25, #k-point spacing in 1/Angstrom + "koffset": [False, False, True], #k-point offset + "fname_pwi_template": "/leonardo_work/EUHPC_A04_113/Alberto/GenMLFF-progect/autoplex/tests/auto/GenMLFF/reference.pwi", #Path to file containing the template QE input +} + +# Update QE default parameters +qe_params = qe_params_from_config(qe_test_params) + +# Define QEscf job +fname_structures_to_be_computed = "" +qe_scf_maker = QEstaticLabelling(**qe_params, fname_structures=fname_structures_to_be_computed) +qe_scf_job = qe_scf_maker.make() + +# Define flow +flow = Flow([qe_scf_job]) +set_run_config(flow, name_filter="run_qe_worker", exec_config="qe_config", worker="QuantumEspresso", resources=parallel_gpu_resources) + +# Submit flow +submit_flow(flow, worker="local_worker", resources={}, project="GenMLFF") \ No newline at end of file From 6f381fb8098c371031db586fa6bb266e0f626e45 Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 29 Sep 2025 15:30:14 +0200 Subject: [PATCH 03/18] Updated configuration for QE test --- tests/auto/GenMLFF/test_qe.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/auto/GenMLFF/test_qe.py b/tests/auto/GenMLFF/test_qe.py index faa60562c..2c9ad603d 100644 --- a/tests/auto/GenMLFF/test_qe.py +++ b/tests/auto/GenMLFF/test_qe.py @@ -38,7 +38,7 @@ # Define flow flow = Flow([qe_scf_job]) -set_run_config(flow, name_filter="run_qe_worker", exec_config="qe_config", worker="QuantumEspresso", resources=parallel_gpu_resources) +set_run_config(flow, name_filter="run_qe_worker", exec_config="qe_config", worker="schedule_worker", resources=parallel_gpu_resources) # Submit flow submit_flow(flow, worker="local_worker", resources={}, project="GenMLFF") \ No newline at end of file From 37c595dd34e41695fae707dfe9b31c33a034bed7 Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 29 Sep 2025 17:23:18 +0200 Subject: [PATCH 04/18] Reordering of QEscf maker to be used as stand-alone module --- .../auto/GenMLFF/QuantumEspressoSCF.py | 171 +++++++++--------- 1 file changed, 86 insertions(+), 85 deletions(-) diff --git a/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py b/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py index a92df4a29..69a42e51d 100644 --- a/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py +++ b/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py @@ -51,6 +51,81 @@ def qe_params_from_config(config: dict): return params + +def run_qe(command, fname_pwi, fname_pwo): + """ + Run the QE command in a subprocess. Execute one QuantumEspresso calculation on the current input file. + """ + #Assemble QE command + run_cmd = f"{command} < {fname_pwi} >> {fname_pwo}" + + success = False + try: + # Launch QE and wait till ending + subprocess.run(run_cmd, shell=True, check=True, executable="/bin/bash") + + success = True + + except subprocess.CalledProcessError as e: + + success = False + + return success + +def lock_input(pwi_fname, worker_id): + + pwi_lock_fname = "" + #Check if pwo exists + pwo_fname = pwi_fname.replace('.pwi', '.pwo') + if os.path.exists(pwo_fname): return pwi_lock_fname, pwo_fname #If exists, skip to next pwi + + # Try to lock the pwi file by renaming it + pwi_lock_fname = f'{pwi_fname}.lock_{worker_id}' + try: + os.rename(f'{pwi_fname}', f'{pwi_lock_fname}') + except Exception as e: + pwi_lock_fname = "" + + return pwi_lock_fname, pwo_fname + +@job +def run_qe_worker( + id, + command, + work_dir, + ): + """ + Run the QE command in a subprocess. + """ + #Get pwi files + pwi_files = glob(os.path.join(work_dir, "*.pwi")) + + #Check pwo does not exist + worker_output = {'success' : [], 'output' : [], 'outdir' : []} + for pwi in pwi_files: + #Try locking the pwi file + lock_pwi, pwo_fname = lock_input(pwi_fname=pwi, worker_id=id) + + if lock_pwi == "": continue #Skip to next pwi if lock failed + + #Get output directory of this calculation + with open(lock_pwi, 'r') as f: + pwi_lines = f.readlines() + outdir_line = [line.split('=')[1] for line in pwi_lines if 'outdir' in line][0] + outdir_line = outdir_line.strip().replace("'", "").replace('"', '') # Remove quotes + outdir = os.getcwd() + f"/{outdir_line}" + + #Launch QE calculation + success = run_qe(command=command, fname_pwi=lock_pwi, fname_pwo=pwo_fname) + + #Update output + worker_output['success'].append(success) + worker_output['output'].append(pwo_fname) + worker_output['outdir'].append(outdir) + + return worker_output + + @dataclass class QEstaticLabelling(Maker): """ @@ -114,21 +189,21 @@ def make(self): # Launch QE workers outputs = [] for id_qe_worker in range(num_qe_workers): - qe_worker = self.run_qe_worker( - id=id_qe_worker, - command=self.qe_run_cmd, - work_dir=path_to_qe_workdir - ) - - qe_worker.name = f"run_qe_worker_{id_qe_worker}" - joblist.append(qe_worker) - outputs.append(qe_worker.output) #Contains list of dict{'successes', 'pwo_files', 'outdirs'} for each worker + worker_job = run_qe_worker( + id=id_qe_worker, + command=self.qe_run_cmd, + work_dir=path_to_qe_workdir, + ) + worker_job.name = f"run_qe_worker_{id_qe_worker}" + + joblist.append(worker_job) + outputs.append(worker_job.output) qe_wrk_flow = Flow(jobs=joblist, output=outputs, name="qe_workers") # Output is a list of success status, one for each worker # The success status is a dictionary with the pwo file name as key and the calculation success status as value (True/False) - return Response(replace=qe_wrk_flow, output=qe_wrk_flow.output) + return qe_wrk_flow def load_structures(self, fname_structures: str | list[str] | None = None, @@ -337,78 +412,4 @@ def _compute_kpoints_grid(self, cell, Kspace_resolution): mesh = [int(np.ceil(L / Kspace_resolution)) for L in lengths] print(f"Computed k-points mesh: {mesh} for K-space resolution: {Kspace_resolution} Angstrom^-1") #DEBUG - return mesh - - @job - def run_qe_worker( - self, - id, - command, - work_dir, - ): - """ - Run the QE command in a subprocess. - """ - #Get pwi files - pwi_files = glob(os.path.join(work_dir, "*.pwi")) - - #Check pwo does not exist - worker_output = {'success' : [], 'output' : [], 'outdir' : []} - for pwi in pwi_files: - #Try locking the pwi file - lock_pwi, pwo_fname = self.lock_input(pwi_fname=pwi, worker_id=id) - - if lock_pwi == "": continue #Skip to next pwi if lock failed - - #Get output directory of this calculation - with open(lock_pwi, 'r') as f: - pwi_lines = f.readlines() - outdir_line = [line.split('=')[1] for line in pwi_lines if 'outdir' in line][0] - outdir_line = outdir_line.strip().replace("'", "").replace('"', '') # Remove quotes - outdir = os.getcwd() + f"/{outdir_line}" - - #Launch QE calculation - success = self.run_qe(command=command, fname_pwi=lock_pwi, fname_pwo=pwo_fname) - - #Update output - worker_output['success'].append(success) - worker_output['output'].append(pwo_fname) - worker_output['outdir'].append(outdir) - - return worker_output - - def run_qe(self, command, fname_pwi, fname_pwo): - """ - Run the QE command in a subprocess. Execute one QuantumEspresso calculation on the current input file. - """ - #Assemble QE command - run_cmd = f"{command} < {fname_pwi} >> {fname_pwo}" - - success = False - try: - # Launch QE and wait till ending - subprocess.run(run_cmd, shell=True, check=True, executable="/bin/bash") - - success = True - - except subprocess.CalledProcessError as e: - - success = False - - return success - - def lock_input(self, pwi_fname, worker_id): - - pwi_lock_fname = "" - #Check if pwo exists - pwo_fname = pwi_fname.replace('.pwi', '.pwo') - if os.path.exists(pwo_fname): return pwi_lock_fname, pwo_fname #If exists, skip to next pwi - - # Try to lock the pwi file by renaming it - pwi_lock_fname = f'{pwi_fname}.lock_{worker_id}' - try: - os.rename(f'{pwi_fname}', f'{pwi_lock_fname}') - except Exception as e: - pwi_lock_fname = "" - - return pwi_lock_fname, pwo_fname \ No newline at end of file + return mesh \ No newline at end of file From 429a646dd15b0ee24d8f696db76d59983328952f Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 29 Sep 2025 17:23:46 +0200 Subject: [PATCH 05/18] Test and reference PWI file --- tests/auto/GenMLFF/reference.pwi | 4 +- tests/auto/GenMLFF/test_qe.py | 67 +++++++++++++++----------------- 2 files changed, 33 insertions(+), 38 deletions(-) diff --git a/tests/auto/GenMLFF/reference.pwi b/tests/auto/GenMLFF/reference.pwi index 22927b5fb..3954f485f 100644 --- a/tests/auto/GenMLFF/reference.pwi +++ b/tests/auto/GenMLFF/reference.pwi @@ -6,7 +6,7 @@ tstress = .true. outdir = './OUT/' disk_io = 'none' - pseudo_dir = '/path/to/directory/PSEUDO' + pseudo_dir = '/leonardo/home/userexternal/apacini0/PSEUDO' max_seconds = 86000 / @@ -43,5 +43,3 @@ C 12.011 C.pbe-n-kjpaw_psl.1.0.0.UPF O 15.999 O.pbe-n-kjpaw_psl.1.0.0.UPF H 2.016 H.pbe-kjpaw_psl.1.0.0.UPF -K_POINTS automatic -4 4 4 1 1 1 diff --git a/tests/auto/GenMLFF/test_qe.py b/tests/auto/GenMLFF/test_qe.py index 2c9ad603d..a3bddf7c6 100644 --- a/tests/auto/GenMLFF/test_qe.py +++ b/tests/auto/GenMLFF/test_qe.py @@ -1,44 +1,41 @@ -from jobflow import Flow, job from jobflow_remote import submit_flow, set_run_config from autoplex.auto.GenMLFF.QuantumEspressoSCF import qe_params_from_config, QEstaticLabelling -#Define resources -parallel_gpu_resources = { - "account": "IscrB_CNT-HARV", - "partition": "boost_usr_prod", - "qos": "boost_qos_dbg", - "time": "00:30:00", - "nodes": 1, - "ntasks_per_node": 2, - "cpus_per_task": 8, - "gres": "gpu:2", - "mem": "240000", - "job_name": "qe_auto", - "qerr_path": "JOB.err", - "qout_path": "JOB.out", - } +if __name__ == "__main__": + # Resources for QE + parallel_gpu_resources = { + "account": "IscrB_CNT-HARV", + "partition": "boost_usr_prod", + "qos": "boost_qos_dbg", + "time": "00:30:00", + "nodes": 1, + "ntasks_per_node": 2, + "cpus_per_task": 8, + "gres": "gpu:2", + "mem": "240000", + "job_name": "qe_auto", + "qerr_path": "JOB.err", + "qout_path": "JOB.out", + } -#Define QE test parameters -qe_test_params = { - "qe_run_cmd": "mpirun -np 1 pw.x", - "num_qe_workers": 2, #Number of workers to use for the calculations. If None setp up 1 worker per scf - "kspace_resolution": 0.25, #k-point spacing in 1/Angstrom - "koffset": [False, False, True], #k-point offset - "fname_pwi_template": "/leonardo_work/EUHPC_A04_113/Alberto/GenMLFF-progect/autoplex/tests/auto/GenMLFF/reference.pwi", #Path to file containing the template QE input -} -# Update QE default parameters -qe_params = qe_params_from_config(qe_test_params) + # QE job parameters (test) + qe_test_params = { + "qe_run_cmd": "mpirun -np 2 pw.x -nk 2 ", #Command to run QE scf calculation + "num_qe_workers": 2, #Number of workers to use for the calculations. If None setp up 1 worker per scf + "kspace_resolution": 0.25, #k-point spacing in 1/Angstrom + "koffset": [False, False, True], #k-point offset + "fname_pwi_template": "/leonardo_work/EUHPC_A04_113/Alberto/GenMLFF-progect/autoplex/tests/auto/GenMLFF/reference.pwi", #Path to file containing the template QE input + "fname_structures": "/leonardo_work/EUHPC_A04_113/Alberto/GenMLFF-progect/Test-QE/initial_dataset.extxyz", #Path to file containing the structures to be computed + } -# Define QEscf job -fname_structures_to_be_computed = "" -qe_scf_maker = QEstaticLabelling(**qe_params, fname_structures=fname_structures_to_be_computed) -qe_scf_job = qe_scf_maker.make() + # Define QE scf workflow + qe_params = qe_params_from_config(qe_test_params) + qe_workflow = QEstaticLabelling(**qe_params).make() -# Define flow -flow = Flow([qe_scf_job]) -set_run_config(flow, name_filter="run_qe_worker", exec_config="qe_config", worker="schedule_worker", resources=parallel_gpu_resources) + # Update flow config + set_run_config(qe_workflow, name_filter="run_qe_worker", exec_config="qe_config", worker="schedule_worker", resources=parallel_gpu_resources) -# Submit flow -submit_flow(flow, worker="local_worker", resources={}, project="GenMLFF") \ No newline at end of file + # Submit flow + submit_flow(qe_workflow, worker="local_worker", resources={}, project="GenMLFF") \ No newline at end of file From b7eaf49bffdef50b66548b9043fa52f34f77817f Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 29 Sep 2025 15:39:38 +0000 Subject: [PATCH 06/18] pre-commit auto-fixes --- .../auto/GenMLFF/QuantumEspressoSCF.py | 441 ++++++++++-------- 1 file changed, 257 insertions(+), 184 deletions(-) diff --git a/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py b/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py index 69a42e51d..d53d6da96 100644 --- a/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py +++ b/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py @@ -1,18 +1,19 @@ """Jobs to create training data for ML potentials.""" -import os import logging +import os import subprocess -import numpy as np -from glob import glob from dataclasses import dataclass, field +from glob import glob +import numpy as np from ase import Atoms from ase.io import read -from jobflow import job, Flow, Maker, Response +from jobflow import Flow, Maker, Response, job logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s") + def qe_params_from_config(config: dict): """ Return QEstaticLabelling params from a configuration dictionary. @@ -28,26 +29,31 @@ def qe_params_from_config(config: dict): "fname_structures": fname_structures, } - Returns: + Returns + ------- params: dict Dictionary with parameters for QEstaticLabelling. """ - #Get default parameters + # Get default parameters params = { "qe_run_cmd": "pw.x", "num_qe_workers": 1, "fname_pwi_template": None, - "kspace_resolution" : None, + "kspace_resolution": None, "koffset": [False, False, False], "fname_structures": None, - } + } # Update parameters with values from the config file - if config is None: raise ValueError("Configuration file is empty or not properly formatted.") + if config is None: + raise ValueError("Configuration file is empty or not properly formatted.") params.update(config) - #Check a valid reference pwi path is provided - if not os.path.exists(params["fname_pwi_template"]): raise ValueError(f"Reference QE input file '{params['fname_pwi_template']}' not found.") + # Check a valid reference pwi path is provided + if not os.path.exists(params["fname_pwi_template"]): + raise ValueError( + f"Reference QE input file '{params['fname_pwi_template']}' not found." + ) return params @@ -56,72 +62,78 @@ def run_qe(command, fname_pwi, fname_pwo): """ Run the QE command in a subprocess. Execute one QuantumEspresso calculation on the current input file. """ - #Assemble QE command + # Assemble QE command run_cmd = f"{command} < {fname_pwi} >> {fname_pwo}" success = False - try: + try: # Launch QE and wait till ending subprocess.run(run_cmd, shell=True, check=True, executable="/bin/bash") - + success = True - - except subprocess.CalledProcessError as e: - + + except subprocess.CalledProcessError: + success = False return success + def lock_input(pwi_fname, worker_id): - + pwi_lock_fname = "" - #Check if pwo exists - pwo_fname = pwi_fname.replace('.pwi', '.pwo') - if os.path.exists(pwo_fname): return pwi_lock_fname, pwo_fname #If exists, skip to next pwi + # Check if pwo exists + pwo_fname = pwi_fname.replace(".pwi", ".pwo") + if os.path.exists(pwo_fname): + return pwi_lock_fname, pwo_fname # If exists, skip to next pwi # Try to lock the pwi file by renaming it - pwi_lock_fname = f'{pwi_fname}.lock_{worker_id}' + pwi_lock_fname = f"{pwi_fname}.lock_{worker_id}" try: - os.rename(f'{pwi_fname}', f'{pwi_lock_fname}') - except Exception as e: - pwi_lock_fname = "" - + os.rename(f"{pwi_fname}", f"{pwi_lock_fname}") + except Exception: + pwi_lock_fname = "" + return pwi_lock_fname, pwo_fname + @job -def run_qe_worker( - id, - command, - work_dir, - ): +def run_qe_worker( + id, + command, + work_dir, +): """ Run the QE command in a subprocess. """ - #Get pwi files + # Get pwi files pwi_files = glob(os.path.join(work_dir, "*.pwi")) - #Check pwo does not exist - worker_output = {'success' : [], 'output' : [], 'outdir' : []} + # Check pwo does not exist + worker_output = {"success": [], "output": [], "outdir": []} for pwi in pwi_files: - #Try locking the pwi file + # Try locking the pwi file lock_pwi, pwo_fname = lock_input(pwi_fname=pwi, worker_id=id) - if lock_pwi == "": continue #Skip to next pwi if lock failed + if lock_pwi == "": + continue # Skip to next pwi if lock failed - #Get output directory of this calculation - with open(lock_pwi, 'r') as f: + # Get output directory of this calculation + with open(lock_pwi) as f: pwi_lines = f.readlines() - outdir_line = [line.split('=')[1] for line in pwi_lines if 'outdir' in line][0] - outdir_line = outdir_line.strip().replace("'", "").replace('"', '') # Remove quotes + outdir_line = [line.split("=")[1] for line in pwi_lines if "outdir" in line][0] + outdir_line = ( + outdir_line.strip().replace("'", "").replace('"', "") + ) # Remove quotes outdir = os.getcwd() + f"/{outdir_line}" - #Launch QE calculation + # Launch QE calculation success = run_qe(command=command, fname_pwi=lock_pwi, fname_pwo=pwo_fname) - #Update output - worker_output['success'].append(success) - worker_output['output'].append(pwo_fname) - worker_output['outdir'].append(outdir) + # Update output + worker_output["success"].append(success) + worker_output["output"].append(pwo_fname) + worker_output["outdir"].append(outdir) return worker_output @@ -130,6 +142,7 @@ def run_qe_worker( class QEstaticLabelling(Maker): """ Maker to set up and run Quantum Espresso static calculations for input structures, including bulk, isolated atoms, and dimers. + Parameters ---------- name: str @@ -145,15 +158,25 @@ class QEstaticLabelling(Maker): """ name: str = "do_qe_labelling" - qe_run_cmd: str | None = None #String with the command to run QE (including its executable path/or application name) - fname_pwi_template: str | None = None #Path to file containing the template computational parameters - fname_structures: str | list[str] | None = None #Path or list[Path] to ASE-readible file containing the structures to be computed - num_qe_workers: int | None = None #Number of workers to use for the calculations. - kspace_resolution: float | None = None #K-space resolution in Angstrom^-1, used to set the K-points in the pwi file - koffset: list[bool] = field(default_factory=lambda: [False, False, False]) #K-points offset in the pwi file - + qe_run_cmd: str | None = ( + None # String with the command to run QE (including its executable path/or application name) + ) + fname_pwi_template: str | None = ( + None # Path to file containing the template computational parameters + ) + fname_structures: str | list[str] | None = ( + None # Path or list[Path] to ASE-readible file containing the structures to be computed + ) + num_qe_workers: int | None = None # Number of workers to use for the calculations. + kspace_resolution: float | None = ( + None # K-space resolution in Angstrom^-1, used to set the K-points in the pwi file + ) + koffset: list[bool] = field( + default_factory=lambda: [False, False, False] + ) # K-points offset in the pwi file + def make(self): - #Define jobs + # Define jobs joblist = [] # Load structures @@ -170,23 +193,25 @@ def make(self): path_to_qe_workdir = os.path.join(work_dir, "scf_files") os.makedirs(path_to_qe_workdir, exist_ok=True) for i, structure in enumerate(structures): - #Get fname of the next pwi file + # Get fname of the next pwi file fname_new_pwi = os.path.join(path_to_qe_workdir, f"structure_{i}.pwi") - #Write pwi file for the structure + # Write pwi file for the structure self.write_pwi( fname_pwi_output=fname_new_pwi, - structure=structure, - pwi_reference=pwi_reference_lines, - ) + structure=structure, + pwi_reference=pwi_reference_lines, + ) # Set number of QE workers - if self.num_qe_workers is None: # 1 worker per structure (all DFT jobs in parallel) + if ( + self.num_qe_workers is None + ): # 1 worker per structure (all DFT jobs in parallel) num_qe_workers = len(glob(os.path.join(path_to_qe_workdir, "*.pwi"))) - else: + else: num_qe_workers = self.num_qe_workers - # Launch QE workers + # Launch QE workers outputs = [] for id_qe_worker in range(num_qe_workers): worker_job = run_qe_worker( @@ -205,36 +230,40 @@ def make(self): # The success status is a dictionary with the pwo file name as key and the calculation success status as value (True/False) return qe_wrk_flow - def load_structures(self, - fname_structures: str | list[str] | None = None, - ): + def load_structures( + self, + fname_structures: str | list[str] | None = None, + ): """ Load structures from a file or a list of files. + Parameters ---------- fname_structures : str | list[str] | None Path or list of paths to ASE-readable files containing the structures to be loaded. If None, no structures will be loaded. + Returns ------- list[Atoms] List of ASE Atoms objects representing the loaded structures. """ - #Convert fname_structures to a list if it is a string + # Convert fname_structures to a list if it is a string if isinstance(fname_structures, str): fname_structures = [fname_structures] elif fname_structures is None: return [] elif not isinstance(fname_structures, list): raise ValueError("fname_structures must be a string or a list of strings.") - - #Loop over provided files and load structures + + # Loop over provided files and load structures structures = [] for fname in fname_structures: - #Check if all files exist - if not os.path.exists(fname): raise FileNotFoundError(f"File {fname} does not exist.") - - #Read structures from file + # Check if all files exist + if not os.path.exists(fname): + raise FileNotFoundError(f"File {fname} does not exist.") + + # Read structures from file try: structures += read(fname, index=":") except Exception as e: @@ -248,168 +277,212 @@ def check_pwi_template(self, fname_template): """ # Read template file tmp_pwi_lines = [] - with open(fname_template, 'r') as f: + with open(fname_template) as f: tmp_pwi_lines = f.readlines() - # Modify lines with structure information: + # Modify lines with structure information: # Assume ntyp, atom_types and pseudoptentials are already defined in the template and consistent with the structures # Assume ibrav=0 and Kspacing is already defined in the template idx_nat_line, idx_pos_line, idx_cell_line = 0, 0, 0 for i, line in enumerate(tmp_pwi_lines): - if 'nat' in line: idx_nat_line = i + if "nat" in line: + idx_nat_line = i + + elif "ATOMIC_POSITIONS" in line: + idx_pos_line = i - elif 'ATOMIC_POSITIONS' in line: idx_pos_line = i + elif "CELL_PARAMETERS" in line: + idx_cell_line = i - elif 'CELL_PARAMETERS' in line: idx_cell_line = i - # Set nat line - if idx_nat_line == 0: # nat not defined, assume nat = 0 - raise ValueError("Number of atoms line not defined in the template file. Please define \'nat =\' in the template file.") - else: - tmp_pwi_lines[idx_nat_line] = f'nat = \n' + if idx_nat_line == 0: # nat not defined, assume nat = 0 + raise ValueError( + "Number of atoms line not defined in the template file. Please define 'nat =' in the template file." + ) + tmp_pwi_lines[idx_nat_line] = "nat = \n" # Cancel lines with ATOMIC_POSITIONS and CELL_PARAMETERS - if idx_pos_line == 0 and idx_cell_line > 0: - idx_to_delete = idx_pos_line - del(tmp_pwi_lines[idx_to_delete:]) - - elif idx_pos_line > 0 and idx_cell_line == 0: + if (idx_pos_line == 0 and idx_cell_line > 0) or ( + idx_pos_line > 0 and idx_cell_line == 0 + ): idx_to_delete = idx_pos_line - del(tmp_pwi_lines[idx_to_delete:]) - + del tmp_pwi_lines[idx_to_delete:] + elif idx_pos_line > 0 and idx_cell_line > 0: idx_to_delete = min([idx_pos_line, idx_cell_line]) - del(tmp_pwi_lines[idx_to_delete:]) + del tmp_pwi_lines[idx_to_delete:] return tmp_pwi_lines def write_pwi( - self, - fname_pwi_output: str, - structure: Atoms, - pwi_reference: list[str], - ): + self, + fname_pwi_output: str, + structure: Atoms, + pwi_reference: list[str], + ): """ Write the pwi input file for the given structure. """ - #Duplicate the pwi template to avoid overwriting the reference lines + # Duplicate the pwi template to avoid overwriting the reference lines pwi_template = pwi_reference.copy() # Check pwi lines - idx_diskio, idx_outdir, idx_nat_line, idx_kpoints_line, nat = 0, 0, 0, 0, len(structure) + idx_diskio, idx_outdir, idx_nat_line, idx_kpoints_line, nat = ( + 0, + 0, + 0, + 0, + len(structure), + ) for idx, line in enumerate(pwi_template): - if 'nat =' in line: idx_nat_line = idx - elif 'disk_io' in line: idx_diskio = idx - elif 'outdir' in line: idx_outdir = idx - elif 'K_POINTS' in line: idx_kpoints_line = idx - - #Update number of atoms - pwi_template[idx_nat_line] = f'nat = {nat}\n' - - #Get identifier for this structure - structure_id = fname_pwi_output.split('/')[-1].replace('.pwi', '') - - #Update outdir based on disk_io - if idx_diskio == 0 or 'none' not in pwi_template[idx_diskio]: #disk_io is not 'none' (QE default is low for scf) - if idx_outdir == 0: # outdir not defined, define it + if "nat =" in line: + idx_nat_line = idx + elif "disk_io" in line: + idx_diskio = idx + elif "outdir" in line: + idx_outdir = idx + elif "K_POINTS" in line: + idx_kpoints_line = idx + + # Update number of atoms + pwi_template[idx_nat_line] = f"nat = {nat}\n" + + # Get identifier for this structure + structure_id = fname_pwi_output.split("/")[-1].replace(".pwi", "") + + # Update outdir based on disk_io + if ( + idx_diskio == 0 or "none" not in pwi_template[idx_diskio] + ): # disk_io is not 'none' (QE default is low for scf) + if idx_outdir == 0: # outdir not defined, define it pwi_template.insert(idx_diskio + 1, f"outdir = {structure_id}\n") - else: # outdir is defined, update it + else: # outdir is defined, update it pwi_template[idx_outdir] = f"outdir = '{structure_id}'\n" - else: #disk_io is 'none', remove outdir line + else: # disk_io is 'none', remove outdir line if idx_outdir == 0: - pwi_template.insert(idx_diskio + 1, f"outdir = 'OUT'\n") + pwi_template.insert(idx_diskio + 1, "outdir = 'OUT'\n") kpoints_lines = self._set_Kpoints( - tmp_pwi_lines=pwi_template, - idx_kpoints_line=idx_kpoints_line, + tmp_pwi_lines=pwi_template, + idx_kpoints_line=idx_kpoints_line, atoms=structure, Kspace_resolution=self.kspace_resolution, Koffset=self.koffset, - ) + ) - #Write cell lines + # Write cell lines cell_lines = ["\nCELL_PARAMETERS (angstrom)\n"] - cell_lines += [f"{structure.cell[i, 0]:.10f} {structure.cell[i, 1]:.10f} {structure.cell[i, 2]:.10f}\n" for i in range(3)] - - #Write positions lines + cell_lines += [ + f"{structure.cell[i, 0]:.10f} {structure.cell[i, 1]:.10f} {structure.cell[i, 2]:.10f}\n" + for i in range(3) + ] + + # Write positions lines pos_lines = ["\nATOMIC_POSITIONS (angstrom)\n"] for i, atom in enumerate(structure): - pos_lines.append(f"{atom.symbol} {structure.positions[i, 0]:.10f} {structure.positions[i, 1]:.10f} {structure.positions[i, 2]:.10f}\n") + pos_lines.append( + f"{atom.symbol} {structure.positions[i, 0]:.10f} {structure.positions[i, 1]:.10f} {structure.positions[i, 2]:.10f}\n" + ) # Write the modified lines to the new pwi file - with open(fname_pwi_output, 'w') as f: - for line in pwi_template: #Write reference pwi lines (computational parameters) + with open(fname_pwi_output, "w") as f: + for ( + line + ) in pwi_template: # Write reference pwi lines (computational parameters) f.write(line) - for line in kpoints_lines: #Write K-points lines + for line in kpoints_lines: # Write K-points lines f.write(line) - for line in cell_lines: #Write cell lines + for line in cell_lines: # Write cell lines f.write(line) - for line in pos_lines: #Write positions lines + for line in pos_lines: # Write positions lines f.write(line) - def _set_Kpoints(self, - tmp_pwi_lines: list[str], - idx_kpoints_line: int, - atoms: Atoms, - Kspace_resolution: float | None = None, - Koffset: list[bool] = [False, False, False], - ): - """ - Set the K-points in the pwi file based on user definition or K-space resolution. - """ - # Define K-points lines - kpoints_lines = [] - - # K_POINTS line not found - if idx_kpoints_line == 0: - if Kspace_resolution is None: # K_POINTS line not found and Kspace_resolution is not defined - raise ValueError("K_POINTS line not found in the template file. Please define K_POINTS in the template file or provide Kspace_resolution.") - else: # Find k-points grid using Monkorst-Pack method based on K-space resolution - #Get real space cell - cell = atoms.cell - - #Find Kpoints grid - #TODO: use structure_type info to generalize to non-periodic systems (3d, 2d, 1d, 0d) - MP_mesh = self._compute_kpoints_grid(cell, Kspace_resolution) - - #Format k-points lines - kpoints_lines.append(f"\nK_POINTS automatic\n") #Header for MP-grid - Kpoint_line = f"{MP_mesh[0]} {MP_mesh[1]} {MP_mesh[2]}" #K-points grid line - for offset in Koffset: #Add offset - if offset: Kpoint_line += " 1" - else: Kpoint_line += " 0" - Kpoint_line += "\n" - kpoints_lines.append(Kpoint_line) - - # K_POINTS is defined by user in reference pwi file, keep the line/s - elif idx_kpoints_line > 0: - if 'gamma' in tmp_pwi_lines[idx_kpoints_line] or 'Gamma' in tmp_pwi_lines[idx_kpoints_line]: # KPOINT is 1 line - kpoints_lines = tmp_pwi_lines[idx_kpoints_line:idx_kpoints_line+1] - del tmp_pwi_lines[idx_kpoints_line:] - elif 'automatic' in tmp_pwi_lines[idx_kpoints_line]: # KPOINTS is 2 lines - kpoints_lines = tmp_pwi_lines[idx_kpoints_line:idx_kpoints_line+2] - del tmp_pwi_lines[idx_kpoints_line:] - elif 'tpiba' in tmp_pwi_lines[idx_kpoints_line] or 'crystal' in tmp_pwi_lines[idx_kpoints_line]: #KPOINTS is multiple lines - num_ks = int(tmp_pwi_lines[idx_kpoints_line+1].split()[0]) #Get number of k-points - kpoints_lines = tmp_pwi_lines[idx_kpoints_line:idx_kpoints_line+num_ks+2] #Get k-points lines - del tmp_pwi_lines[idx_kpoints_line:] + def _set_Kpoints( + self, + tmp_pwi_lines: list[str], + idx_kpoints_line: int, + atoms: Atoms, + Kspace_resolution: float | None = None, + Koffset: list[bool] = [False, False, False], + ): + """ + Set the K-points in the pwi file based on user definition or K-space resolution. + """ + # Define K-points lines + kpoints_lines = [] + + # K_POINTS line not found + if idx_kpoints_line == 0: + if ( + Kspace_resolution is None + ): # K_POINTS line not found and Kspace_resolution is not defined + raise ValueError( + "K_POINTS line not found in the template file. Please define K_POINTS in the template file or provide Kspace_resolution." + ) + # Find k-points grid using Monkorst-Pack method based on K-space resolution + # Get real space cell + cell = atoms.cell + + # Find Kpoints grid + # TODO: use structure_type info to generalize to non-periodic systems (3d, 2d, 1d, 0d) + MP_mesh = self._compute_kpoints_grid(cell, Kspace_resolution) + + # Format k-points lines + kpoints_lines.append("\nK_POINTS automatic\n") # Header for MP-grid + Kpoint_line = ( + f"{MP_mesh[0]} {MP_mesh[1]} {MP_mesh[2]}" # K-points grid line + ) + for offset in Koffset: # Add offset + if offset: + Kpoint_line += " 1" else: - raise ValueError(f"K_POINTS format: {tmp_pwi_lines[idx_kpoints_line]} is unknown in pwi template file") - - return kpoints_lines - + Kpoint_line += " 0" + Kpoint_line += "\n" + kpoints_lines.append(Kpoint_line) + + # K_POINTS is defined by user in reference pwi file, keep the line/s + elif idx_kpoints_line > 0: + if ( + "gamma" in tmp_pwi_lines[idx_kpoints_line] + or "Gamma" in tmp_pwi_lines[idx_kpoints_line] + ): # KPOINT is 1 line + kpoints_lines = tmp_pwi_lines[idx_kpoints_line : idx_kpoints_line + 1] + del tmp_pwi_lines[idx_kpoints_line:] + elif "automatic" in tmp_pwi_lines[idx_kpoints_line]: # KPOINTS is 2 lines + kpoints_lines = tmp_pwi_lines[idx_kpoints_line : idx_kpoints_line + 2] + del tmp_pwi_lines[idx_kpoints_line:] + elif ( + "tpiba" in tmp_pwi_lines[idx_kpoints_line] + or "crystal" in tmp_pwi_lines[idx_kpoints_line] + ): # KPOINTS is multiple lines + num_ks = int( + tmp_pwi_lines[idx_kpoints_line + 1].split()[0] + ) # Get number of k-points + kpoints_lines = tmp_pwi_lines[ + idx_kpoints_line : idx_kpoints_line + num_ks + 2 + ] # Get k-points lines + del tmp_pwi_lines[idx_kpoints_line:] + else: + raise ValueError( + f"K_POINTS format: {tmp_pwi_lines[idx_kpoints_line]} is unknown in pwi template file" + ) + + return kpoints_lines + def _compute_kpoints_grid(self, cell, Kspace_resolution): """ Compute the k-points grid using Monkhorst-Pack method based on the cell and K-space resolution. """ - #Compute the reciprocal cell vectors: b_i = 2π * (a_j x a_k) / (a_i . (a_j x a_k)) - rec_cell = 2.0 * np.pi * np.linalg.inv(cell).T + # Compute the reciprocal cell vectors: b_i = 2π * (a_j x a_k) / (a_i . (a_j x a_k)) + rec_cell = 2.0 * np.pi * np.linalg.inv(cell).T - #Compute reciprocal lattice vecotors' lenghts - lengths = np.linalg.norm(rec_cell, axis=1) + # Compute reciprocal lattice vecotors' lenghts + lengths = np.linalg.norm(rec_cell, axis=1) - #Compute mesh size - mesh = [int(np.ceil(L / Kspace_resolution)) for L in lengths] - print(f"Computed k-points mesh: {mesh} for K-space resolution: {Kspace_resolution} Angstrom^-1") #DEBUG + # Compute mesh size + mesh = [int(np.ceil(L / Kspace_resolution)) for L in lengths] + print( + f"Computed k-points mesh: {mesh} for K-space resolution: {Kspace_resolution} Angstrom^-1" + ) # DEBUG - return mesh \ No newline at end of file + return mesh From 14989e9a6648179223c0d08aeefbb573880efd15 Mon Sep 17 00:00:00 2001 From: 41bY Date: Fri, 3 Oct 2025 20:22:24 +0200 Subject: [PATCH 07/18] First version for QEStaticMaker --- src/autoplex/misc/qe/__init__.py | 19 +++ src/autoplex/misc/qe/jobs.py | 75 +++++++++ src/autoplex/misc/qe/run.py | 75 +++++++++ src/autoplex/misc/qe/schema.py | 57 +++++++ src/autoplex/misc/qe/utils.py | 258 +++++++++++++++++++++++++++++++ tests/auto/QE/test_qe.py | 102 ++++++++++++ 6 files changed, 586 insertions(+) create mode 100644 src/autoplex/misc/qe/__init__.py create mode 100644 src/autoplex/misc/qe/jobs.py create mode 100644 src/autoplex/misc/qe/run.py create mode 100644 src/autoplex/misc/qe/schema.py create mode 100644 src/autoplex/misc/qe/utils.py create mode 100644 tests/auto/QE/test_qe.py diff --git a/src/autoplex/misc/qe/__init__.py b/src/autoplex/misc/qe/__init__.py new file mode 100644 index 000000000..851f02afc --- /dev/null +++ b/src/autoplex/misc/qe/__init__.py @@ -0,0 +1,19 @@ +from .schema import ( + QeRunSettings, + QeKpointsSettings, + QeInputSet, + QeRunResult, +) +from .utils import QeStaticInputGenerator +from .run import run_qe_static +from .jobs import QEStaticMaker + +__all__ = [ + "QeRunSettings", + "QeKpointsSettings", + "QeInputSet", + "QeRunResult", + "QeStaticInputGenerator", + "run_qe_static", + "QEStaticMaker", +] diff --git a/src/autoplex/misc/qe/jobs.py b/src/autoplex/misc/qe/jobs.py new file mode 100644 index 000000000..53798c86e --- /dev/null +++ b/src/autoplex/misc/qe/jobs.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import List, Optional + +from jobflow import Flow, Maker + +from .run import run_qe_static +from .schema import QeKpointsSettings, QeRunSettings +from .utils import QeStaticInputGenerator + + +@dataclass +class QEStaticMaker(Maker): + """ + StaticMaker for Quantum ESPRESSO: + - assemble and write one .pwi per structure using InputGenerator; + - create a `run_qe_static` job for each input; + - assemble flow with all jobs. + + Parameters + -------------------- + name : str + Name of the Flow. + command : str + Command to execute QE (e.g. "pw.x" or "mpirun -np 4 pw.x -nk 2"). + template_pwi : str | None + Path to template `.pwi` with namelists: &control, &system, &electrons. + structures : str | list[str] | None + ASE-readables file (or list of files) containing the structures. + workdir : str | None + Directory used to write input/output files. Default: "/qe_static". + run_settings : QeRunSettings | None + Update namelists (&control, &system, &electrons). + kpoints : QeKpointsSettings | None + Set up for K_POINTS if it is not contained in the template. + pseudo : dict[str, str] | None + Dictionary of atomic symbols and corresponding pseudopotential files. + """ + + name: str = "qe_static" + command: str = "pw.x" + template_pwi: Optional[str] = None + structures: Optional[str | List[str]] = None + workdir: Optional[str] = None + run_settings: Optional[QeRunSettings] = None + kpoints: Optional[QeKpointsSettings] = None + pseudo: Optional[dict[str, str]] = None + + def make(self) -> Flow: + workdir = self.workdir or os.path.join(os.getcwd(), "qe_static") + os.makedirs(workdir, exist_ok=True) + + # Generate one input per structure + generator = QeStaticInputGenerator( + template_pwi=self.template_pwi, + run_settings=self.run_settings or QeRunSettings(), + kpoints=self.kpoints or QeKpointsSettings(), + pseudo=self.pseudo, + ) + input_sets = generator.generate_for_structures( + structures=self.structures, workdir=workdir, seed_prefix="structure" + ) + + # Create one SCF job per structure + jobs = [] + outputs = [] + for i, inp in enumerate(input_sets): + j = run_qe_static(pwi_path=inp.pwi_path, command=self.command) + j.name = f"{self.name}_{i}" + jobs.append(j) + outputs.append(j.output) + + return Flow(jobs=jobs, output=outputs, name=self.name) \ No newline at end of file diff --git a/src/autoplex/misc/qe/run.py b/src/autoplex/misc/qe/run.py new file mode 100644 index 000000000..b34772e6b --- /dev/null +++ b/src/autoplex/misc/qe/run.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import logging +import os +import re +import subprocess +from typing import Optional + +from jobflow import job + +from .schema import QeRunResult + +logger = logging.getLogger(__name__) + + +_ENERGY_RE = re.compile(r"!\s+total energy\s+=\s+([-\d\.Ee+]+)\s+Ry") + + +def _parse_total_energy_ev(pwo_path: str) -> Optional[float]: + """ + Extract and return total energy (eV) if found in QE output (.pwo) + """ + if not os.path.exists(pwo_path): + return None + + try: + with open(pwo_path, "r", errors="ignore") as fh: + for line in fh: + m = _ENERGY_RE.search(line) + if m: + # convert energy in eV + ry = float(m.group(1)) + return ry * 13.605693009 + except Exception: + return None + return None + + +@job +def run_qe_static(pwi_path: str, command: str) -> QeRunResult: + """ + Execute single QE SCF static calculation from .pwi file. + """ + pwo_path = pwi_path.replace(".pwi", ".pwo") + # Assemble pwscf run command e.g. "pw.x < input.pwi >> input.pwo" + run_cmd = f"{command} < {pwi_path} >> {pwo_path}" + + success = False + try: + subprocess.run(run_cmd, shell=True, check=True, executable="/bin/bash") + success = True + except subprocess.CalledProcessError as exc: + logger.error("QE failed for %s: %s", pwi_path, exc) + success = False + + # Return outdir read from .pwi + outdir = "" + try: + with open(pwi_path, "r") as fh: + for line in fh: + if "outdir" in line: + outdir = line.split("=")[1].strip().strip("'").strip('"') + break + except Exception: + pass + + energy_ev = _parse_total_energy_ev(pwo_path) + + return QeRunResult( + success=success, + pwi=os.path.abspath(pwi_path), + pwo=os.path.abspath(pwo_path), + outdir=os.path.abspath(outdir) if outdir else "", + total_energy_ev=energy_ev, + ) \ No newline at end of file diff --git a/src/autoplex/misc/qe/schema.py b/src/autoplex/misc/qe/schema.py new file mode 100644 index 000000000..d22a90666 --- /dev/null +++ b/src/autoplex/misc/qe/schema.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from typing import Dict, List, Optional + +from pydantic import BaseModel, Field, field_validator + + +class QeRunSettings(BaseModel): + """ + Set QE namelists for static calculation SCF + Standard namelists: CONTROL, SYSTEM, ELECTRONS + """ + control: Dict[str, object] = Field(default_factory=dict) + system: Dict[str, object] = Field(default_factory=dict) + electrons: Dict[str, object] = Field(default_factory=dict) + + @field_validator("control", "system", "electrons") + @classmethod + def _lowercase_keys(cls, v: Dict[str, object]) -> Dict[str, object]: + # default lowercase keywords + return {str(k).lower(): v[k] for k in v} + + +class QeKpointsSettings(BaseModel): + """ + K-points use k-space resoultion with automatic Monkhorst-Pack + if K_POINTS are not defined in reference 'template.pwi' + """ + kspace_resolution: Optional[float] = None # angstrom^-1 + koffset: List[bool] = Field(default_factory=lambda: [False, False, False]) + + @field_validator("koffset") + @classmethod + def _len3(cls, v: List[bool]) -> List[bool]: + if len(v) != 3: + raise ValueError("koffset must be a list of 3 booleans.") + return v + + +class QeInputSet(BaseModel): + """ + Files and contexts for static SCF job + """ + workdir: str + pwi_path: str + seed: str + + +class QeRunResult(BaseModel): + """ + Results of the SCF job + """ + success: bool + pwi: str + pwo: str + outdir: str + total_energy_ev: Optional[float] = None \ No newline at end of file diff --git a/src/autoplex/misc/qe/utils.py b/src/autoplex/misc/qe/utils.py new file mode 100644 index 000000000..78c140278 --- /dev/null +++ b/src/autoplex/misc/qe/utils.py @@ -0,0 +1,258 @@ +from __future__ import annotations + +import logging +import os +from typing import List, Optional + +import numpy as np +from ase import Atoms +from ase.data import atomic_numbers, atomic_masses +from ase.io import read + +from .schema import QeKpointsSettings, QeInputSet, QeRunSettings + +logger = logging.getLogger(__name__) + + +class QeStaticInputGenerator: + """ + Input generator to run static SCF calculations with Quantum Espresso (.pwi), using: + - .pwi template containing computational parameters (&control, &system, &electrons); + - ASE structures; + - k-points optional settings if K_POINTS is not defined within template + + Used to write one .pwi for each structure inside `workdir`. + """ + + def __init__( + self, + template_pwi: Optional[str] = None, + run_settings: Optional[QeRunSettings] = None, + kpoints: Optional[QeKpointsSettings] = None, + pseudo: Optional[dict[str, str]] = None, + ) -> None: + self.template_pwi = template_pwi + self.run_settings = run_settings or QeRunSettings() + self.kpoints = kpoints or QeKpointsSettings() + self.pseudo = pseudo + + def generate_for_structures( + self, + structures: str | List[str] | None, + workdir: str, + seed_prefix: str = "structure", + ) -> List[QeInputSet]: + """ + Generate one .pwi for each structure read from ASE-readable files. + """ + os.makedirs(workdir, exist_ok=True) + atoms_list = _load_structures(structures) + input_sets: List[QeInputSet] = [] + template_lines = _read_template(self.template_pwi) if self.template_pwi else None + + for i, atoms in enumerate(atoms_list): + seed = f"{seed_prefix}_{i}" + pwi_path = os.path.join(workdir, f"{seed}.pwi") + self._write_pwi( + pwi_output=pwi_path, + atoms=atoms, + template_lines=list(template_lines) if template_lines else None, + kpoints=self.kpoints, + pseudo=self.pseudo, + ) + input_sets.append(QeInputSet(workdir=workdir, pwi_path=pwi_path, seed=seed)) + return input_sets + + def _write_pwi( + self, + *, + pwi_output: str, + atoms: Atoms, + template_lines: Optional[List[str]], + kpoints: QeKpointsSettings, + pseudo: Optional[dict[str, str]] = None, + ) -> None: + """ + Write .pwi file using template (if present) + K_POINTS/cell/positions + Updating `nat`, `outdir` + """ + if template_lines is None: + # Minimal template with main QE namelists + template_lines = _render_minimal_namelists(self.run_settings) + + pwi = list(template_lines) + + idx_diskio = None + idx_outdir = None + idx_nat = None + idx_ntyp = None + idx_kpts = None + for idx, line in enumerate(pwi): + ls = line.lower() + if "nat" in ls and "!" not in ls: + idx_nat = idx + elif "ntyp" in ls and "!" not in ls: + idx_ntyp = idx + elif "disk_io" in ls: + idx_diskio = idx + elif "outdir" in ls: + idx_outdir = idx + elif "k_points" in ls: + idx_kpts = idx + + # Number of atoms + nat = len(atoms) + if idx_nat is None: + raise ValueError("Template must define a 'nat = ' line (in &system).") + pwi[idx_nat] = f" nat = {nat}\n" + + # Output directory + structure_id = os.path.basename(pwi_output).replace(".pwi", "") + if idx_diskio is None or "none" not in pwi[idx_diskio].lower(): + if idx_outdir is None: + insert_at = (idx_diskio + 1) if idx_diskio is not None else len(pwi) + pwi.insert(insert_at, f" outdir = '{structure_id}'\n") + else: + pwi[idx_outdir] = f" outdir = '{structure_id}'\n" + else: + if idx_outdir is None: + insert_at = (idx_diskio + 1) if idx_diskio is not None else len(pwi) + pwi.insert(insert_at, " outdir = 'OUT'\n") + + # ATOMIC_SPECIES + if pseudo is None: + raise ValueError("Pseudo dictionary must be provided to write ATOMIC_SPECIES.") + species = set(atoms.get_chemical_symbols()) + ntyp = len(species) + pwi[idx_ntyp] = f" ntyp = {ntyp}\n" + species_lines = ["\nATOMIC_SPECIES\n"] + for s in sorted(species): + if s not in pseudo: + raise ValueError(f"Missing pseudo for atomic symbol '{s}' in pseudo dictionary.") + mass = atomic_masses[atomic_numbers[s]] + species_lines.append(f"{s} {mass:.4f} {pseudo[s]}\n") + species_lines.append("\n") + + # K-POINTS + kpt_lines = _render_kpoints( + template_lines=pwi, idx_kpts=idx_kpts, atoms=atoms, kpoints=kpoints + ) + kpt_lines.append("\n") + + # CELL_PARAMETERS + cell = atoms.cell + cell_lines = ["\nCELL_PARAMETERS (angstrom)\n"] + for i in range(3): + cell_lines.append(f"{cell[i, 0]:.10f} {cell[i, 1]:.10f} {cell[i, 2]:.10f}\n") + cell_lines.append("\n") + + # ATOMIC_POSITIONS + pos_lines = ["\nATOMIC_POSITIONS (angstrom)\n"] + for i, atom in enumerate(atoms): + x, y, z = atoms.positions[i] + pos_lines.append(f"{atom.symbol} {x:.10f} {y:.10f} {z:.10f}\n") + pos_lines.append("\n") + + # Assemble and write + with open(pwi_output, "w") as fh: + fh.writelines(pwi) + fh.writelines(species_lines) + fh.writelines(kpt_lines) + fh.writelines(cell_lines) + fh.writelines(pos_lines) + + +# --------- Utilities --------- + +def _load_structures(paths: str | List[str] | None) -> List[Atoms]: + if paths is None: + return [] + if isinstance(paths, str): + paths = [paths] + atoms_list: List[Atoms] = [] + for fname in paths: + if not os.path.exists(fname): + raise FileNotFoundError(f"Structure file not found: {fname}") + atoms_list += read(fname, index=":") + return atoms_list + + +def _read_template(path: Optional[str]) -> List[str]: + if path is None: + return [] + if not os.path.exists(path): + raise FileNotFoundError(f"template_pwi not found: {path}") + with open(path, "r") as fh: + return fh.readlines() + + +def _render_minimal_namelists(settings: QeRunSettings) -> List[str]: + """Create draft lines for namelists &control, &system, &electrons.""" + def _render_block(name: str, kv: dict) -> List[str]: + lines = [f"&{name}\n"] + for k, v in kv.items(): + if isinstance(v, bool): + vv = ".true." if v else ".false." + elif isinstance(v, (int, float)): + vv = v + else: + vv = f"'{v}'" + lines.append(f" {k} = {vv}\n") + lines.append("/\n\n") + return lines + + out: List[str] = [] + out += _render_block("control", settings.control) + out += _render_block("system", settings.system) + out += _render_block("electrons", settings.electrons) + return out + + +def _render_kpoints( + *, + template_lines: List[str], + idx_kpts: int | None, + atoms: Atoms, + kpoints: QeKpointsSettings, +) -> List[str]: + """Create lines for K_POINTS section.""" + + # if K_POINTS in template keep it + if idx_kpts is not None and idx_kpts >= 0: + line = template_lines[idx_kpts].lower() + if "gamma" in line: + kpts = template_lines[idx_kpts : idx_kpts + 1] + elif "automatic" in line: + kpts = template_lines[idx_kpts : idx_kpts + 2] + elif "tpiba" in line or "crystal" in line: + n = int(template_lines[idx_kpts + 1].split()[0]) + kpts = template_lines[idx_kpts : idx_kpts + n + 2] + else: + raise ValueError( + f"Unknown K_POINTS format: {template_lines[idx_kpts].strip()}" + ) + del template_lines[idx_kpts:] + return ["\n"] + kpts + + # Otherwise create K_POINTS using kspace_resolution + if kpoints.kspace_resolution is None: + raise ValueError( + "K_POINTS not found in template and kspace_resolution is None. " + "Either add K_POINTS to template or set kspace_resolution." + ) + + # Generate Monkhorst–Pack mesh + mp_mesh = _compute_kpoints_grid(atoms.cell, kpoints.kspace_resolution) + line = f"{mp_mesh[0]} {mp_mesh[1]} {mp_mesh[2]}" + for off in kpoints.koffset: + line += " 1" if off else " 0" + return ["\nK_POINTS automatic\n", f"{line}\n"] + + +def _compute_kpoints_grid(cell: np.ndarray, kspace_resolution: float) -> List[int]: + """Compute Monkhorst-Pack mesh from cell and kspace_resolution (in angstrom^-1).""" + rec = 2.0 * np.pi * np.linalg.inv(cell).T + lengths = np.linalg.norm(rec, axis=1) + mesh = [max(1, int(np.ceil(L / kspace_resolution))) for L in lengths] + logger.debug("QE MP mesh %s using k-resolution %s angstrom^-1", mesh, kspace_resolution) + return mesh \ No newline at end of file diff --git a/tests/auto/QE/test_qe.py b/tests/auto/QE/test_qe.py new file mode 100644 index 000000000..6ad90d5a0 --- /dev/null +++ b/tests/auto/QE/test_qe.py @@ -0,0 +1,102 @@ +from jobflow_remote import submit_flow, set_run_config +from autoplex.misc.qe import QEStaticMaker, QeRunSettings, QeKpointsSettings + +# --------- QE namelists dictionaries --------- +control_dict = { + "calculation" : "scf", + "restart_mode" : "from_scratch", + "prefix" : "FeCOH", + "tprnfor" : True, + "tstress" : True, + "outdir" : "./OUT/", + "disk_io" : 'none', + "pseudo_dir" : '/leonardo/home/userexternal/apacini0/PSEUDO', + "max_seconds" : 86000 +} + +system_dict = { + "ibrav" : 0, + "nat" : 506, + "ntyp" : 4, + "ecutwfc" : 60, + "ecutrho" : 480, + "occupations" : "smearing", + "smearing" : "gaussian", + "degauss" : 0.00735, + "nosym" : True, + "vdw_corr" : "dft-d3", + "nspin" : 2, + "starting_magnetization(1)" : 0.4 +} + +electrons_dict = { + "diagonalization" : "david", + "mixing_beta" : 0.15, + "electron_maxstep" : 150, + "mixing_mode" : "local-TF", + "mixing_ndim" : 16, + "conv_thr" : 1.0e-6, +} + +pseudo_dict = { + "Fe": "Fe.pbe-sp-van.UPF", + "C": "C.pbe-n-kjpaw_psl.1.0.0.UPF", + "O": "O.pbe-n-kjpaw_psl.1.0.0.UPF", + "H": "H.pbe-kjpaw_psl.1.0.0.UPF", +} + +# --------- Resources --------- +parallel_gpu_resources = { + "account": "IscrB_CNT-HARV", + "partition": "boost_usr_prod", + "qos": "boost_qos_dbg", + "time": "00:30:00", + "nodes": 1, + "ntasks_per_node": 4, + "cpus_per_task": 8, + "gres": "gpu:4", + "mem": "480000", + "job_name": "qe_auto", + "qerr_path": "JOB.err", + "qout_path": "JOB.out", + } + + + +if __name__ == "__main__": + # QE command + qe_command = "mpirun -np 4 pw.x -nk 2" + + # QE run seetings (computational parameters namelist) + qe_run_settings = QeRunSettings( + control=control_dict, + system=system_dict, + electrons=electrons_dict, + ) + + # QE KPOINTS settings + k_points_settings = QeKpointsSettings( + kspace_resolution=0.25, # angstrom^-1 + koffset=[False, False, True], + ) + + # Instance of QEStaticMaker + qe_maker = QEStaticMaker( + name="static_qe", + command=qe_command, + template_pwi=None, # Optional if run_settings + structures="/leonardo_work/EUHPC_A04_113/Alberto/GenMLFF-progect/Test-QE/initial_dataset.extxyz", + workdir=None, # default /qe_static + run_settings=qe_run_settings, + kpoints=k_points_settings, + pseudo=pseudo_dict, + ) + + # Define QE scf workflow + qe_workflow = qe_maker.make() + + # Update flow config + set_run_config(qe_workflow, name_filter="static_qe", exec_config="qe_config", worker="schedule_worker", resources=parallel_gpu_resources) + + # Submit flow + submit_flow(qe_workflow, worker="local_worker", resources={}, project="GenMLFF") \ No newline at end of file From 7ddce5f0f32df682260e31f3241b541024d540f2 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 6 Oct 2025 06:41:58 +0000 Subject: [PATCH 08/18] pre-commit auto-fixes --- src/autoplex/misc/qe/__init__.py | 14 +++---- src/autoplex/misc/qe/jobs.py | 16 ++++---- src/autoplex/misc/qe/run.py | 8 ++-- src/autoplex/misc/qe/schema.py | 20 +++++---- src/autoplex/misc/qe/utils.py | 69 ++++++++++++++++++-------------- 5 files changed, 71 insertions(+), 56 deletions(-) diff --git a/src/autoplex/misc/qe/__init__.py b/src/autoplex/misc/qe/__init__.py index 851f02afc..ccdbbcbc0 100644 --- a/src/autoplex/misc/qe/__init__.py +++ b/src/autoplex/misc/qe/__init__.py @@ -1,19 +1,19 @@ +from .jobs import QEStaticMaker +from .run import run_qe_static from .schema import ( - QeRunSettings, - QeKpointsSettings, QeInputSet, + QeKpointsSettings, QeRunResult, + QeRunSettings, ) from .utils import QeStaticInputGenerator -from .run import run_qe_static -from .jobs import QEStaticMaker __all__ = [ - "QeRunSettings", - "QeKpointsSettings", + "QEStaticMaker", "QeInputSet", + "QeKpointsSettings", "QeRunResult", + "QeRunSettings", "QeStaticInputGenerator", "run_qe_static", - "QEStaticMaker", ] diff --git a/src/autoplex/misc/qe/jobs.py b/src/autoplex/misc/qe/jobs.py index 53798c86e..b3014b1b8 100644 --- a/src/autoplex/misc/qe/jobs.py +++ b/src/autoplex/misc/qe/jobs.py @@ -20,7 +20,7 @@ class QEStaticMaker(Maker): - assemble flow with all jobs. Parameters - -------------------- + ---------- name : str Name of the Flow. command : str @@ -41,12 +41,12 @@ class QEStaticMaker(Maker): name: str = "qe_static" command: str = "pw.x" - template_pwi: Optional[str] = None - structures: Optional[str | List[str]] = None - workdir: Optional[str] = None - run_settings: Optional[QeRunSettings] = None - kpoints: Optional[QeKpointsSettings] = None - pseudo: Optional[dict[str, str]] = None + template_pwi: str | None = None + structures: str | list[str] | None = None + workdir: str | None = None + run_settings: QeRunSettings | None = None + kpoints: QeKpointsSettings | None = None + pseudo: dict[str, str] | None = None def make(self) -> Flow: workdir = self.workdir or os.path.join(os.getcwd(), "qe_static") @@ -72,4 +72,4 @@ def make(self) -> Flow: jobs.append(j) outputs.append(j.output) - return Flow(jobs=jobs, output=outputs, name=self.name) \ No newline at end of file + return Flow(jobs=jobs, output=outputs, name=self.name) diff --git a/src/autoplex/misc/qe/run.py b/src/autoplex/misc/qe/run.py index b34772e6b..f8d6e1146 100644 --- a/src/autoplex/misc/qe/run.py +++ b/src/autoplex/misc/qe/run.py @@ -16,7 +16,7 @@ _ENERGY_RE = re.compile(r"!\s+total energy\s+=\s+([-\d\.Ee+]+)\s+Ry") -def _parse_total_energy_ev(pwo_path: str) -> Optional[float]: +def _parse_total_energy_ev(pwo_path: str) -> float | None: """ Extract and return total energy (eV) if found in QE output (.pwo) """ @@ -24,7 +24,7 @@ def _parse_total_energy_ev(pwo_path: str) -> Optional[float]: return None try: - with open(pwo_path, "r", errors="ignore") as fh: + with open(pwo_path, errors="ignore") as fh: for line in fh: m = _ENERGY_RE.search(line) if m: @@ -56,7 +56,7 @@ def run_qe_static(pwi_path: str, command: str) -> QeRunResult: # Return outdir read from .pwi outdir = "" try: - with open(pwi_path, "r") as fh: + with open(pwi_path) as fh: for line in fh: if "outdir" in line: outdir = line.split("=")[1].strip().strip("'").strip('"') @@ -72,4 +72,4 @@ def run_qe_static(pwi_path: str, command: str) -> QeRunResult: pwo=os.path.abspath(pwo_path), outdir=os.path.abspath(outdir) if outdir else "", total_energy_ev=energy_ev, - ) \ No newline at end of file + ) diff --git a/src/autoplex/misc/qe/schema.py b/src/autoplex/misc/qe/schema.py index d22a90666..4b3204ad4 100644 --- a/src/autoplex/misc/qe/schema.py +++ b/src/autoplex/misc/qe/schema.py @@ -10,13 +10,14 @@ class QeRunSettings(BaseModel): Set QE namelists for static calculation SCF Standard namelists: CONTROL, SYSTEM, ELECTRONS """ - control: Dict[str, object] = Field(default_factory=dict) - system: Dict[str, object] = Field(default_factory=dict) - electrons: Dict[str, object] = Field(default_factory=dict) + + control: dict[str, object] = Field(default_factory=dict) + system: dict[str, object] = Field(default_factory=dict) + electrons: dict[str, object] = Field(default_factory=dict) @field_validator("control", "system", "electrons") @classmethod - def _lowercase_keys(cls, v: Dict[str, object]) -> Dict[str, object]: + def _lowercase_keys(cls, v: dict[str, object]) -> dict[str, object]: # default lowercase keywords return {str(k).lower(): v[k] for k in v} @@ -26,12 +27,13 @@ class QeKpointsSettings(BaseModel): K-points use k-space resoultion with automatic Monkhorst-Pack if K_POINTS are not defined in reference 'template.pwi' """ - kspace_resolution: Optional[float] = None # angstrom^-1 - koffset: List[bool] = Field(default_factory=lambda: [False, False, False]) + + kspace_resolution: float | None = None # angstrom^-1 + koffset: list[bool] = Field(default_factory=lambda: [False, False, False]) @field_validator("koffset") @classmethod - def _len3(cls, v: List[bool]) -> List[bool]: + def _len3(cls, v: list[bool]) -> list[bool]: if len(v) != 3: raise ValueError("koffset must be a list of 3 booleans.") return v @@ -41,6 +43,7 @@ class QeInputSet(BaseModel): """ Files and contexts for static SCF job """ + workdir: str pwi_path: str seed: str @@ -50,8 +53,9 @@ class QeRunResult(BaseModel): """ Results of the SCF job """ + success: bool pwi: str pwo: str outdir: str - total_energy_ev: Optional[float] = None \ No newline at end of file + total_energy_ev: float | None = None diff --git a/src/autoplex/misc/qe/utils.py b/src/autoplex/misc/qe/utils.py index 78c140278..50c5837ab 100644 --- a/src/autoplex/misc/qe/utils.py +++ b/src/autoplex/misc/qe/utils.py @@ -6,10 +6,10 @@ import numpy as np from ase import Atoms -from ase.data import atomic_numbers, atomic_masses +from ase.data import atomic_masses, atomic_numbers from ase.io import read -from .schema import QeKpointsSettings, QeInputSet, QeRunSettings +from .schema import QeInputSet, QeKpointsSettings, QeRunSettings logger = logging.getLogger(__name__) @@ -26,10 +26,10 @@ class QeStaticInputGenerator: def __init__( self, - template_pwi: Optional[str] = None, - run_settings: Optional[QeRunSettings] = None, - kpoints: Optional[QeKpointsSettings] = None, - pseudo: Optional[dict[str, str]] = None, + template_pwi: str | None = None, + run_settings: QeRunSettings | None = None, + kpoints: QeKpointsSettings | None = None, + pseudo: dict[str, str] | None = None, ) -> None: self.template_pwi = template_pwi self.run_settings = run_settings or QeRunSettings() @@ -38,17 +38,19 @@ def __init__( def generate_for_structures( self, - structures: str | List[str] | None, + structures: str | list[str] | None, workdir: str, seed_prefix: str = "structure", - ) -> List[QeInputSet]: + ) -> list[QeInputSet]: """ Generate one .pwi for each structure read from ASE-readable files. """ os.makedirs(workdir, exist_ok=True) atoms_list = _load_structures(structures) - input_sets: List[QeInputSet] = [] - template_lines = _read_template(self.template_pwi) if self.template_pwi else None + input_sets: list[QeInputSet] = [] + template_lines = ( + _read_template(self.template_pwi) if self.template_pwi else None + ) for i, atoms in enumerate(atoms_list): seed = f"{seed_prefix}_{i}" @@ -68,9 +70,9 @@ def _write_pwi( *, pwi_output: str, atoms: Atoms, - template_lines: Optional[List[str]], + template_lines: list[str] | None, kpoints: QeKpointsSettings, - pseudo: Optional[dict[str, str]] = None, + pseudo: dict[str, str] | None = None, ) -> None: """ Write .pwi file using template (if present) + K_POINTS/cell/positions @@ -118,17 +120,21 @@ def _write_pwi( if idx_outdir is None: insert_at = (idx_diskio + 1) if idx_diskio is not None else len(pwi) pwi.insert(insert_at, " outdir = 'OUT'\n") - + # ATOMIC_SPECIES if pseudo is None: - raise ValueError("Pseudo dictionary must be provided to write ATOMIC_SPECIES.") + raise ValueError( + "Pseudo dictionary must be provided to write ATOMIC_SPECIES." + ) species = set(atoms.get_chemical_symbols()) ntyp = len(species) pwi[idx_ntyp] = f" ntyp = {ntyp}\n" species_lines = ["\nATOMIC_SPECIES\n"] for s in sorted(species): if s not in pseudo: - raise ValueError(f"Missing pseudo for atomic symbol '{s}' in pseudo dictionary.") + raise ValueError( + f"Missing pseudo for atomic symbol '{s}' in pseudo dictionary." + ) mass = atomic_masses[atomic_numbers[s]] species_lines.append(f"{s} {mass:.4f} {pseudo[s]}\n") species_lines.append("\n") @@ -143,7 +149,9 @@ def _write_pwi( cell = atoms.cell cell_lines = ["\nCELL_PARAMETERS (angstrom)\n"] for i in range(3): - cell_lines.append(f"{cell[i, 0]:.10f} {cell[i, 1]:.10f} {cell[i, 2]:.10f}\n") + cell_lines.append( + f"{cell[i, 0]:.10f} {cell[i, 1]:.10f} {cell[i, 2]:.10f}\n" + ) cell_lines.append("\n") # ATOMIC_POSITIONS @@ -164,12 +172,13 @@ def _write_pwi( # --------- Utilities --------- -def _load_structures(paths: str | List[str] | None) -> List[Atoms]: + +def _load_structures(paths: str | list[str] | None) -> list[Atoms]: if paths is None: return [] if isinstance(paths, str): paths = [paths] - atoms_list: List[Atoms] = [] + atoms_list: list[Atoms] = [] for fname in paths: if not os.path.exists(fname): raise FileNotFoundError(f"Structure file not found: {fname}") @@ -177,18 +186,19 @@ def _load_structures(paths: str | List[str] | None) -> List[Atoms]: return atoms_list -def _read_template(path: Optional[str]) -> List[str]: +def _read_template(path: str | None) -> list[str]: if path is None: return [] if not os.path.exists(path): raise FileNotFoundError(f"template_pwi not found: {path}") - with open(path, "r") as fh: + with open(path) as fh: return fh.readlines() -def _render_minimal_namelists(settings: QeRunSettings) -> List[str]: +def _render_minimal_namelists(settings: QeRunSettings) -> list[str]: """Create draft lines for namelists &control, &system, &electrons.""" - def _render_block(name: str, kv: dict) -> List[str]: + + def _render_block(name: str, kv: dict) -> list[str]: lines = [f"&{name}\n"] for k, v in kv.items(): if isinstance(v, bool): @@ -201,7 +211,7 @@ def _render_block(name: str, kv: dict) -> List[str]: lines.append("/\n\n") return lines - out: List[str] = [] + out: list[str] = [] out += _render_block("control", settings.control) out += _render_block("system", settings.system) out += _render_block("electrons", settings.electrons) @@ -210,13 +220,12 @@ def _render_block(name: str, kv: dict) -> List[str]: def _render_kpoints( *, - template_lines: List[str], + template_lines: list[str], idx_kpts: int | None, atoms: Atoms, kpoints: QeKpointsSettings, -) -> List[str]: +) -> list[str]: """Create lines for K_POINTS section.""" - # if K_POINTS in template keep it if idx_kpts is not None and idx_kpts >= 0: line = template_lines[idx_kpts].lower() @@ -249,10 +258,12 @@ def _render_kpoints( return ["\nK_POINTS automatic\n", f"{line}\n"] -def _compute_kpoints_grid(cell: np.ndarray, kspace_resolution: float) -> List[int]: +def _compute_kpoints_grid(cell: np.ndarray, kspace_resolution: float) -> list[int]: """Compute Monkhorst-Pack mesh from cell and kspace_resolution (in angstrom^-1).""" rec = 2.0 * np.pi * np.linalg.inv(cell).T lengths = np.linalg.norm(rec, axis=1) mesh = [max(1, int(np.ceil(L / kspace_resolution))) for L in lengths] - logger.debug("QE MP mesh %s using k-resolution %s angstrom^-1", mesh, kspace_resolution) - return mesh \ No newline at end of file + logger.debug( + "QE MP mesh %s using k-resolution %s angstrom^-1", mesh, kspace_resolution + ) + return mesh From a0e61590844ef0bd6fc14edc5d42ee867dc5f6dd Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 6 Oct 2025 08:52:39 +0200 Subject: [PATCH 09/18] Consistent format for QE schema --- src/autoplex/misc/qe/schema.py | 62 +++++++++++++++++++++++++++++----- 1 file changed, 53 insertions(+), 9 deletions(-) diff --git a/src/autoplex/misc/qe/schema.py b/src/autoplex/misc/qe/schema.py index d22a90666..47bb10af2 100644 --- a/src/autoplex/misc/qe/schema.py +++ b/src/autoplex/misc/qe/schema.py @@ -2,6 +2,10 @@ from typing import Dict, List, Optional +from pymatgen.core import Structure +from emmet.core.structure import StructureMetadata +from emmet.core.math import Matrix3D, Vector3D + from pydantic import BaseModel, Field, field_validator @@ -37,21 +41,61 @@ def _len3(cls, v: List[bool]) -> List[bool]: return v -class QeInputSet(BaseModel): +class InputDoc(BaseModel): """ - Files and contexts for static SCF job + Inputs and contexts used to run the static SCF job """ workdir: str pwi_path: str seed: str + run_settings: QeRunSettings = Field(None, description="QE namelist section with: &control, &system, &electrons") + kpoints: QeKpointsSettings = Field(None, description="QE K_POINTS settings") -class QeRunResult(BaseModel): +class OutputDoc(BaseModel): """ - Results of the SCF job + The outputs of this jobs """ - success: bool - pwi: str - pwo: str - outdir: str - total_energy_ev: Optional[float] = None \ No newline at end of file + energy: float | None = Field(None, description="Total energy in units of eV.") + + energy_per_atom: float | None = Field( + None, + description="Energy per atom of the final molecule or structure " + "in units of eV/atom.", + ) + + forces: list[Vector3D] | None = Field( + None, + description=( + "The force on each atom in units of eV/A for the final molecule " + "or structure." + ), + ) + + # NOTE: units for stresses were converted to kbar (* -10 from standard output) + # to comply with MP convention + stress: Matrix3D | None = Field( + None, description="The stress on the cell in units of kbar." + ) + +class TaskDoc(StructureMetadata): + """Document containing information on structure manipulation using Quantum ESPRESSO.""" + + structure: Structure = Field( + None, description="Final output structure from the task" + ) + + input: InputDoc = Field( + None, description="The input information used to run this job." + ) + + output: OutputDoc = Field(None, description="The output information from this job.") + + task_label: str = Field( + None, + description="Description of the CASTEP task (e.g., static, relax)", + ) + + dir_name: str | None = Field( + None, description="Directory where the QE calculations are performed." + ) \ No newline at end of file From 3dcdb74f97bc91bc5ca906234141bcdc653e765a Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 6 Oct 2025 13:26:15 +0200 Subject: [PATCH 10/18] Build pwi input using settings dicts (run_settings, kpoints and pseuso). make method accecepts structures in different formats: Ase, pymatgen, ASE-readable files --- src/autoplex/misc/qe/jobs.py | 61 +++++++++++++++++++++++------------- 1 file changed, 39 insertions(+), 22 deletions(-) diff --git a/src/autoplex/misc/qe/jobs.py b/src/autoplex/misc/qe/jobs.py index 53798c86e..2304446da 100644 --- a/src/autoplex/misc/qe/jobs.py +++ b/src/autoplex/misc/qe/jobs.py @@ -2,7 +2,9 @@ import os from dataclasses import dataclass -from typing import List, Optional + +from ase import Atoms +from pymatgen.core import Structure from jobflow import Flow, Maker @@ -12,7 +14,7 @@ @dataclass -class QEStaticMaker(Maker): +class QeStaticMaker(Maker): """ StaticMaker for Quantum ESPRESSO: - assemble and write one .pwi per structure using InputGenerator; @@ -20,15 +22,11 @@ class QEStaticMaker(Maker): - assemble flow with all jobs. Parameters - -------------------- + ---------- name : str Name of the Flow. command : str Command to execute QE (e.g. "pw.x" or "mpirun -np 4 pw.x -nk 2"). - template_pwi : str | None - Path to template `.pwi` with namelists: &control, &system, &electrons. - structures : str | list[str] | None - ASE-readables file (or list of files) containing the structures. workdir : str | None Directory used to write input/output files. Default: "/qe_static". run_settings : QeRunSettings | None @@ -41,35 +39,54 @@ class QEStaticMaker(Maker): name: str = "qe_static" command: str = "pw.x" - template_pwi: Optional[str] = None - structures: Optional[str | List[str]] = None - workdir: Optional[str] = None - run_settings: Optional[QeRunSettings] = None - kpoints: Optional[QeKpointsSettings] = None - pseudo: Optional[dict[str, str]] = None + workdir: str | None = None + run_settings: QeRunSettings | None = None + kpoints: QeKpointsSettings | None = None + pseudo: dict[str, str] | None = None + + def make( + self, + structures: Atoms | list[Atoms] | Structure | list[Structure] | str | list[str] + ) -> Flow: + """ + Create a Flow to run static SCF calculations with QE for given structures. - def make(self) -> Flow: + Parameters + ---------- + structures : Atoms | list[Atoms] | Structure | list[Structure] | str | list[str] + Single or list of ASE Atoms, pymatgen Structures, or ASE-readable files. + + Returns + ------- + Flow + A jobflow Flow with one `run_qe_static` job per structure. + """ workdir = self.workdir or os.path.join(os.getcwd(), "qe_static") os.makedirs(workdir, exist_ok=True) # Generate one input per structure generator = QeStaticInputGenerator( - template_pwi=self.template_pwi, run_settings=self.run_settings or QeRunSettings(), kpoints=self.kpoints or QeKpointsSettings(), - pseudo=self.pseudo, + pseudo=self.pseudo or {}, ) input_sets = generator.generate_for_structures( - structures=self.structures, workdir=workdir, seed_prefix="structure" + structures=structures, workdir=workdir, seed_prefix="structure" ) - # Create one SCF job per structure + # If single structure, generate one job + if len(input_sets) == 1: + job = run_qe_static(input_sets[0], command=self.command) + job.name = self.name + return job + + # Else create one SCF job per structure and assemble flow jobs = [] - outputs = [] + tasks = [] for i, inp in enumerate(input_sets): - j = run_qe_static(pwi_path=inp.pwi_path, command=self.command) + j = run_qe_static(inp, command=self.command) j.name = f"{self.name}_{i}" jobs.append(j) - outputs.append(j.output) + tasks.append(j.output) - return Flow(jobs=jobs, output=outputs, name=self.name) \ No newline at end of file + return Flow(jobs=jobs, output=tasks, name=self.name) \ No newline at end of file From e95de7acbb9ac4e294661d604c1a6c020ab8a693 Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 6 Oct 2025 13:27:34 +0200 Subject: [PATCH 11/18] Use of ase.io.read to extract output quantities. The @job 'run_qe_static' return TaskDoc with input, output and metadata --- src/autoplex/misc/qe/run.py | 75 +++++++++++++++++++++++++------------ 1 file changed, 52 insertions(+), 23 deletions(-) diff --git a/src/autoplex/misc/qe/run.py b/src/autoplex/misc/qe/run.py index f8d6e1146..3650b1a32 100644 --- a/src/autoplex/misc/qe/run.py +++ b/src/autoplex/misc/qe/run.py @@ -4,11 +4,14 @@ import os import re import subprocess -from typing import Optional + +from ase.io import read +from ase.units import GPa +from pymatgen.io.ase import AseAtomsAdaptor from jobflow import job -from .schema import QeRunResult +from .schema import InputDoc, OutputDoc, TaskDoc logger = logging.getLogger(__name__) @@ -19,6 +22,16 @@ def _parse_total_energy_ev(pwo_path: str) -> float | None: """ Extract and return total energy (eV) if found in QE output (.pwo) + + Parameters + ---------- + pwo_path : str + Path to QE output file (.pwo) + + Returns + ------- + float | None + Total energy in eV if found, else None """ if not os.path.exists(pwo_path): return None @@ -37,10 +50,24 @@ def _parse_total_energy_ev(pwo_path: str) -> float | None: @job -def run_qe_static(pwi_path: str, command: str) -> QeRunResult: +def run_qe_static(input: InputDoc, command: str) -> TaskDoc: """ Execute single QE SCF static calculation from .pwi file. + Parse output .pwo file to extract total energy, forces, stress, and final structure. + + Parameters + ---------- + input : InputDoc + Input document containing paths and settings for the QE run. + command : str + Command to execute QE (e.g. "pw.x" or "mpirun -np 4 pw.x -nk 2"). + + Returns + ------- + TaskDoc + Document containing input, output, and metadata of the QE run. """ + pwi_path = input.pwi_path pwo_path = pwi_path.replace(".pwi", ".pwo") # Assemble pwscf run command e.g. "pw.x < input.pwi >> input.pwo" run_cmd = f"{command} < {pwi_path} >> {pwo_path}" @@ -48,28 +75,30 @@ def run_qe_static(pwi_path: str, command: str) -> QeRunResult: success = False try: subprocess.run(run_cmd, shell=True, check=True, executable="/bin/bash") - success = True except subprocess.CalledProcessError as exc: logger.error("QE failed for %s: %s", pwi_path, exc) - success = False - - # Return outdir read from .pwi - outdir = "" - try: - with open(pwi_path) as fh: - for line in fh: - if "outdir" in line: - outdir = line.split("=")[1].strip().strip("'").strip('"') - break - except Exception: - pass - energy_ev = _parse_total_energy_ev(pwo_path) + # # Manual parse of total energy in eV from .pwo + # energy_ev = _parse_total_energy_ev(pwo_path) + + # Parse with ASE + atoms = read(pwo_path) + energy_ev = atoms.get_total_energy() + forces_evA = atoms.get_forces() + stress_kbar = atoms.get_stress(voigt=False)*(-10/GPa) + final_structure = AseAtomsAdaptor().get_structure(atoms) - return QeRunResult( - success=success, - pwi=os.path.abspath(pwi_path), - pwo=os.path.abspath(pwo_path), - outdir=os.path.abspath(outdir) if outdir else "", - total_energy_ev=energy_ev, + output = OutputDoc( + energy=energy_ev, + forces=forces_evA.tolist(), + stress=stress_kbar.tolist(), + energy_per_atom=energy_ev / len(atoms) if energy_ev is not None else None, ) + + return TaskDoc( + structure=final_structure, + dir_name=os.path.dirname(pwi_path), + task_label="qe_scf", + input=input, + output=output, + ) \ No newline at end of file From 1ce724e98e653e7ea9101636aecd675406a50625 Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 6 Oct 2025 13:28:05 +0200 Subject: [PATCH 12/18] Added InputDoc, OutputDoc and TaskDoc --- src/autoplex/misc/qe/schema.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/autoplex/misc/qe/schema.py b/src/autoplex/misc/qe/schema.py index a677a4d33..531d3aecc 100644 --- a/src/autoplex/misc/qe/schema.py +++ b/src/autoplex/misc/qe/schema.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Dict, List, Optional +from typing import Dict from pymatgen.core import Structure from emmet.core.structure import StructureMetadata @@ -28,8 +28,8 @@ def _lowercase_keys(cls, v: dict[str, object]) -> dict[str, object]: class QeKpointsSettings(BaseModel): """ - K-points use k-space resoultion with automatic Monkhorst-Pack - if K_POINTS are not defined in reference 'template.pwi' + K-points: use k-space resoultion with automatic Monkhorst-Pack grid + k-points offset as in Quantum ESPRESSO manual: 0: False, 1: True """ kspace_resolution: float | None = None # angstrom^-1 @@ -52,6 +52,7 @@ class InputDoc(BaseModel): pwi_path: str seed: str run_settings: QeRunSettings = Field(None, description="QE namelist section with: &control, &system, &electrons") + pseudo: Dict[str, str] = Field(None, description="Dictionary of atomic symbols and corresponding pseudopotential files.") kpoints: QeKpointsSettings = Field(None, description="QE K_POINTS settings") @@ -96,7 +97,7 @@ class TaskDoc(StructureMetadata): task_label: str = Field( None, - description="Description of the CASTEP task (e.g., static, relax)", + description="Description of the QE task (e.g., static, relax)", ) dir_name: str | None = Field( From 7dd361189afcffaa50341eb647bcfcce6c1c2aa8 Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 6 Oct 2025 13:29:06 +0200 Subject: [PATCH 13/18] Use QeStaticInputGenerator to generate one InputDoc for each scf calculation, containing QE input settings --- src/autoplex/misc/qe/utils.py | 270 +++++++++++++++++++++++++--------- 1 file changed, 201 insertions(+), 69 deletions(-) diff --git a/src/autoplex/misc/qe/utils.py b/src/autoplex/misc/qe/utils.py index 78c140278..fb6eabca4 100644 --- a/src/autoplex/misc/qe/utils.py +++ b/src/autoplex/misc/qe/utils.py @@ -1,54 +1,74 @@ from __future__ import annotations -import logging import os -from typing import List, Optional +import logging +from dataclasses import dataclass import numpy as np -from ase import Atoms -from ase.data import atomic_numbers, atomic_masses + +from pymatgen.core import Structure +from pymatgen.io.ase import AseAtomsAdaptor + +from ase import Atoms, Atom +from ase.data import atomic_masses, atomic_numbers from ase.io import read -from .schema import QeKpointsSettings, QeInputSet, QeRunSettings +from .schema import InputDoc, QeKpointsSettings, QeRunSettings logger = logging.getLogger(__name__) - +@dataclass class QeStaticInputGenerator: """ - Input generator to run static SCF calculations with Quantum Espresso (.pwi), using: - - .pwi template containing computational parameters (&control, &system, &electrons); - - ASE structures; - - k-points optional settings if K_POINTS is not defined within template - + Input generator to run static SCF calculations with Quantum Espresso (.pwi). Used to write one .pwi for each structure inside `workdir`. + + Parameters + ---------- + run_settings : QeRunSettings | None + Update namelists (&control, &system, &electrons). + kpoints : QeKpointsSettings | None + Set up for K_POINTS. + pseudo : dict[str, str] | None + Dictionary of atomic symbols and corresponding pseudopotential files. """ def __init__( self, - template_pwi: Optional[str] = None, - run_settings: Optional[QeRunSettings] = None, - kpoints: Optional[QeKpointsSettings] = None, - pseudo: Optional[dict[str, str]] = None, + run_settings: QeRunSettings | None = None, + kpoints: QeKpointsSettings | None = None, + pseudo: dict[str, str] | None = None, ) -> None: - self.template_pwi = template_pwi self.run_settings = run_settings or QeRunSettings() self.kpoints = kpoints or QeKpointsSettings() - self.pseudo = pseudo + self.pseudo = pseudo or {} def generate_for_structures( self, - structures: str | List[str] | None, + structures: Atoms | list[Atoms] | Structure | list[Structure] | str | list[str], workdir: str, seed_prefix: str = "structure", - ) -> List[QeInputSet]: + ) -> list[InputDoc]: """ Generate one .pwi for each structure read from ASE-readable files. + + Parameters + ---------- + structures : Atoms | list[Atoms] | Structure | list[Structure] | str | list[str] + Single or list of ASE Atoms, pymatgen Structures, or ASE-readable files. + workdir : str + Directory used to write input/output files. + seed_prefix : str + Prefix for naming input/output QE files. + + Returns + ------- + list[InputDoc] + List of InputDoc containing path to generated .pwi files and metadata. """ os.makedirs(workdir, exist_ok=True) atoms_list = _load_structures(structures) - input_sets: List[QeInputSet] = [] - template_lines = _read_template(self.template_pwi) if self.template_pwi else None + input_sets: list[InputDoc] = [] for i, atoms in enumerate(atoms_list): seed = f"{seed_prefix}_{i}" @@ -56,11 +76,16 @@ def generate_for_structures( self._write_pwi( pwi_output=pwi_path, atoms=atoms, - template_lines=list(template_lines) if template_lines else None, - kpoints=self.kpoints, - pseudo=self.pseudo, ) - input_sets.append(QeInputSet(workdir=workdir, pwi_path=pwi_path, seed=seed)) + input_sets.append( + InputDoc( + workdir=workdir, + pwi_path=pwi_path, + seed=seed, + run_settings=self.run_settings, + kpoints=self.kpoints, + pseudo=self.pseudo, + )) return input_sets def _write_pwi( @@ -68,20 +93,27 @@ def _write_pwi( *, pwi_output: str, atoms: Atoms, - template_lines: Optional[List[str]], - kpoints: QeKpointsSettings, - pseudo: Optional[dict[str, str]] = None, ) -> None: """ - Write .pwi file using template (if present) + K_POINTS/cell/positions - Updating `nat`, `outdir` + Write .pwi file for the current Atoms. + + Parameters + ---------- + pwi_output : str + Path to output .pwi file. + atoms : Atoms + ASE Atoms object. + + Raises + ------ + ValueError + If required information is missing to write the .pwi file. """ - if template_lines is None: - # Minimal template with main QE namelists - template_lines = _render_minimal_namelists(self.run_settings) - + # Minimal template with main QE namelists + template_lines = _render_minimal_namelists(self.run_settings) pwi = list(template_lines) + # Find indices of key lines in template idx_diskio = None idx_outdir = None idx_nat = None @@ -118,24 +150,28 @@ def _write_pwi( if idx_outdir is None: insert_at = (idx_diskio + 1) if idx_diskio is not None else len(pwi) pwi.insert(insert_at, " outdir = 'OUT'\n") - + # ATOMIC_SPECIES - if pseudo is None: - raise ValueError("Pseudo dictionary must be provided to write ATOMIC_SPECIES.") + if self.pseudo is None: + raise ValueError( + "Pseudo dictionary must be provided to write ATOMIC_SPECIES." + ) species = set(atoms.get_chemical_symbols()) ntyp = len(species) pwi[idx_ntyp] = f" ntyp = {ntyp}\n" species_lines = ["\nATOMIC_SPECIES\n"] for s in sorted(species): - if s not in pseudo: - raise ValueError(f"Missing pseudo for atomic symbol '{s}' in pseudo dictionary.") + if s not in self.pseudo: + raise ValueError( + f"Missing pseudo for atomic symbol '{s}' in pseudo dictionary." + ) mass = atomic_masses[atomic_numbers[s]] - species_lines.append(f"{s} {mass:.4f} {pseudo[s]}\n") + species_lines.append(f"{s} {mass:.4f} {self.pseudo[s]}\n") species_lines.append("\n") # K-POINTS kpt_lines = _render_kpoints( - template_lines=pwi, idx_kpts=idx_kpts, atoms=atoms, kpoints=kpoints + template_lines=pwi, idx_kpts=idx_kpts, atoms=atoms, kpoints=self.kpoints ) kpt_lines.append("\n") @@ -143,7 +179,9 @@ def _write_pwi( cell = atoms.cell cell_lines = ["\nCELL_PARAMETERS (angstrom)\n"] for i in range(3): - cell_lines.append(f"{cell[i, 0]:.10f} {cell[i, 1]:.10f} {cell[i, 2]:.10f}\n") + cell_lines.append( + f"{cell[i, 0]:.10f} {cell[i, 1]:.10f} {cell[i, 2]:.10f}\n" + ) cell_lines.append("\n") # ATOMIC_POSITIONS @@ -161,34 +199,95 @@ def _write_pwi( fh.writelines(cell_lines) fh.writelines(pos_lines) - -# --------- Utilities --------- - -def _load_structures(paths: str | List[str] | None) -> List[Atoms]: - if paths is None: - return [] - if isinstance(paths, str): - paths = [paths] - atoms_list: List[Atoms] = [] - for fname in paths: - if not os.path.exists(fname): - raise FileNotFoundError(f"Structure file not found: {fname}") - atoms_list += read(fname, index=":") - return atoms_list +# --------- Utils --------- -def _read_template(path: Optional[str]) -> List[str]: +def _read_template(path: str | None) -> list[str]: + """Read template .pwi file if provided, else return empty list.""" if path is None: return [] if not os.path.exists(path): raise FileNotFoundError(f"template_pwi not found: {path}") - with open(path, "r") as fh: + with open(path) as fh: return fh.readlines() -def _render_minimal_namelists(settings: QeRunSettings) -> List[str]: - """Create draft lines for namelists &control, &system, &electrons.""" - def _render_block(name: str, kv: dict) -> List[str]: +def _load_structures( + structures: Atoms | list[Atoms] | Structure | list[Structure] | str | list[str] +) -> list[Atoms]: + """ + Load structures from various input types and return a list of ASE Atoms objects. + + Parameters + ---------- + structures : Atoms | list[Atoms] | Structure | list[Structure] | str | list[str] + Single or list of ASE Atoms, pymatgen Structures, or ASE-readable files. + + Returns + ------- + list[Atoms] + List of ASE Atoms objects. + """ + + # ASE readable files + # Sinlge ASE-readable file + if isinstance(structures, str): + atoms_list = read(structures, index=":") + # List of ASE readable files + elif isinstance(structures, list) and all(isinstance(s, str) for s in structures): + atom_list = [] + for fname in structures: + atoms_list += read(fname, index=":") + + + # ASE Atoms + elif isinstance(structures, Atoms): + # List of ASE Atoms objects + if isinstance(structures[0], Atoms): + atoms_list = structures + # Single ASE Atoms + elif isinstance(structures, Atom): + atoms_list = [structures] + else: + raise ValueError("Unsupported ASE Atoms input.") + # List of ASE Atoms objects + elif isinstance(structures, list) and all(isinstance(s, Atoms) for s in structures): + atoms_list = structures + + + # Pymatgen Structures + # Single pymatgen structure + elif isinstance(structures, Structure): + atoms_list = [AseAtomsAdaptor().get_atoms(structures)] + # List of pymatgen structures + elif isinstance(structures, list) and all(isinstance(s, Structure) for s in structures): + adaptor = AseAtomsAdaptor() + atoms_list = [adaptor.get_atoms(s) for s in structures] + + + else: + raise ValueError("Unsupported type for structures input.") + + return atoms_list + + +def _render_minimal_namelists(settings: QeRunSettings) -> list[str]: + """ + Create draft lines for namelists &control, &system, &electrons. + If some namelist is missing in `settings`, it will be created with minimal entries. + + Parameters + ---------- + settings : QeRunSettings + QE namelist settings. + + Returns + ------- + list[str] + List of lines for the namelists to be written in the .pwi file. + """ + + def _render_block(name: str, kv: dict) -> list[str]: lines = [f"&{name}\n"] for k, v in kv.items(): if isinstance(v, bool): @@ -201,7 +300,7 @@ def _render_block(name: str, kv: dict) -> List[str]: lines.append("/\n\n") return lines - out: List[str] = [] + out: list[str] = [] out += _render_block("control", settings.control) out += _render_block("system", settings.system) out += _render_block("electrons", settings.electrons) @@ -210,13 +309,30 @@ def _render_block(name: str, kv: dict) -> List[str]: def _render_kpoints( *, - template_lines: List[str], + template_lines: list[str], idx_kpts: int | None, atoms: Atoms, kpoints: QeKpointsSettings, -) -> List[str]: - """Create lines for K_POINTS section.""" - +) -> list[str]: + """ + Create lines for K_POINTS section. + + Parameters + ---------- + template_lines : list[str] + Lines of the template .pwi file. + idx_kpts : int | None + Index of K_POINTS line in template_lines, or None if not present. + atoms : Atoms + ASE Atoms object. + kpoints : QeKpointsSettings + K-points settings. + + Returns + ------- + list[str] + Lines for K_POINTS section to be written in the .pwi file. + """ # if K_POINTS in template keep it if idx_kpts is not None and idx_kpts >= 0: line = template_lines[idx_kpts].lower() @@ -249,10 +365,26 @@ def _render_kpoints( return ["\nK_POINTS automatic\n", f"{line}\n"] -def _compute_kpoints_grid(cell: np.ndarray, kspace_resolution: float) -> List[int]: - """Compute Monkhorst-Pack mesh from cell and kspace_resolution (in angstrom^-1).""" +def _compute_kpoints_grid(cell: np.ndarray, kspace_resolution: float) -> list[int]: + """ + Compute Monkhorst-Pack mesh from cell and kspace_resolution (in angstrom^-1). + + Parameters + ---------- + cell : np.ndarray + 3x3 array with cell vectors in angstrom. + kspace_resolution : float + Desired k-space resolution in angstrom^-1. + + Returns + ------- + list[int] + List of 3 integers defining the Monkhorst-Pack mesh. + """ rec = 2.0 * np.pi * np.linalg.inv(cell).T lengths = np.linalg.norm(rec, axis=1) mesh = [max(1, int(np.ceil(L / kspace_resolution))) for L in lengths] - logger.debug("QE MP mesh %s using k-resolution %s angstrom^-1", mesh, kspace_resolution) + logger.debug( + "QE MP mesh %s using k-resolution %s angstrom^-1", mesh, kspace_resolution + ) return mesh \ No newline at end of file From ab91986805938d39f4bf7b4533956bdd69834c62 Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 6 Oct 2025 13:29:39 +0200 Subject: [PATCH 14/18] Added new methods and classes to init --- src/autoplex/misc/qe/__init__.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/autoplex/misc/qe/__init__.py b/src/autoplex/misc/qe/__init__.py index ccdbbcbc0..6e61b2ff0 100644 --- a/src/autoplex/misc/qe/__init__.py +++ b/src/autoplex/misc/qe/__init__.py @@ -1,9 +1,10 @@ -from .jobs import QEStaticMaker +from .jobs import QeStaticMaker from .run import run_qe_static from .schema import ( - QeInputSet, + InputDoc, + OutputDoc, + TaskDoc, QeKpointsSettings, - QeRunResult, QeRunSettings, ) from .utils import QeStaticInputGenerator From b2f684cdb6006bfd65eacf426f302b7ea4c7a904 Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 6 Oct 2025 13:30:16 +0200 Subject: [PATCH 15/18] Basic files to reproduce and test QeStaticMaker --- tests/auto/QE/initial_dataset.extxyz | 23 +++++++++++++++++++++++ tests/auto/QE/test_qe.py | 17 +++++++++-------- 2 files changed, 32 insertions(+), 8 deletions(-) create mode 100644 tests/auto/QE/initial_dataset.extxyz diff --git a/tests/auto/QE/initial_dataset.extxyz b/tests/auto/QE/initial_dataset.extxyz new file mode 100644 index 000000000..617a82597 --- /dev/null +++ b/tests/auto/QE/initial_dataset.extxyz @@ -0,0 +1,23 @@ +5 +Lattice="4.305297245283376 0.0 1.1625217649309496 0.8909299950637738 4.280333028546382 1.2880608262677014 0.0 0.0 4.72729418447685" Properties=species:S:1:pos:R:3:forces:R:3:magmoms:R:1 energy=-2514.1464323047767 free_energy=-2514.1464323047767 stress="0.007359951443438806 0.00299778461362621 0.0052683883960144545 0.00299778461362621 0.0026837747092279974 0.0037396559667073666 0.0052683883960144545 0.0037396559667073666 0.0063628322733322" pbc="T T T" +O 4.50768846 1.33619342 2.92160180 -0.17530373 -0.10837200 -0.14322382 -0.20730000 +O 1.82101029 2.69224629 4.07320896 0.10518717 0.08744862 0.14952482 -0.00050000 +O 3.74062566 0.79659710 2.12389339 0.17508853 0.10835117 0.14262192 -0.20730000 +C 2.41609781 3.20583755 4.94720316 -0.00019309 0.00017149 -0.00040289 0.00000000 +O 3.01116349 3.71950374 5.82116302 -0.10477888 -0.08759928 -0.14852003 -0.00060000 +8 +Lattice="4.6056051052883795 0.0 2.029353509877331 1.811973285238557 4.241415722640877 2.024043819726518 0.0 0.0 6.196473833460105" Properties=species:S:1:pos:R:3:forces:R:3:magmoms:R:1 energy=-3894.545303604514 free_energy=-3894.545303604514 stress="0.02617208282126037 -0.0035165436662139 0.01209121763456158 -0.0035165436662139 0.02923413846795054 0.007873201550335298 0.01209121763456158 0.007873201550335298 0.004317177106667822" pbc="T T T" +C 3.52254256 3.39084463 3.80256102 0.02988496 -0.02485177 -0.00790871 -0.00000000 +C 4.92641658 1.26664697 6.90007421 -0.02382616 0.01373226 0.01277735 0.00000000 +O 5.38937929 2.34182500 7.41231336 -0.36160612 -0.84502596 -0.40126976 -0.00040000 +O 4.68875284 3.39517442 4.32552071 -0.92541382 -0.00182266 -0.41182953 0.00000000 +O 3.75714279 1.25867136 6.38448820 0.92343690 0.00715384 0.40076119 0.00030000 +O 3.06303694 2.31729625 3.28347615 0.35900468 0.84653006 0.40832589 0.00010000 +O 1.02422967 0.20052108 4.87576818 -0.54057470 0.82168523 -0.00692809 0.00010000 +O 1.00749081 0.21551202 1.77354824 0.53909426 -0.81740075 0.00607192 -0.00020000 +4 +Lattice="3.5430015722830794 0.0 1.3334437401239645 -0.19990356019914338 3.755621037089823 1.4851449742235037 0.0 0.0 4.927789704843149" Properties=species:S:1:pos:R:3:forces:R:3:magmoms:R:1 energy=-1947.2726323461177 free_energy=-1947.2726323461177 stress="0.018799277171208773 -0.009603010497371065 -0.01168631012625862 -0.009603010497371065 0.02233969293541786 -0.010452306291430558 -0.01168631012625862 -0.010452306291430558 0.022246040858667517" pbc="T T T" +O 2.62466719 0.38683228 3.18016181 -0.08298467 0.86162243 -0.71473377 0.00010000 +O -0.10657170 1.28243135 4.94944308 -0.81817131 0.06298020 0.80407420 0.00020000 +C 2.57548477 1.32559350 2.30699142 0.40906354 -0.33314401 -0.03192410 0.00000000 +O 1.72125422 2.26331329 2.38182181 0.49209243 -0.59145863 -0.05741608 -0.00030000 diff --git a/tests/auto/QE/test_qe.py b/tests/auto/QE/test_qe.py index 6ad90d5a0..444dbf58e 100644 --- a/tests/auto/QE/test_qe.py +++ b/tests/auto/QE/test_qe.py @@ -1,5 +1,7 @@ +from ase.io import read +from pymatgen.io.ase import AseAtomsAdaptor from jobflow_remote import submit_flow, set_run_config -from autoplex.misc.qe import QEStaticMaker, QeRunSettings, QeKpointsSettings +from autoplex.misc.qe import QeStaticMaker, QeRunSettings, QeKpointsSettings # --------- QE namelists dictionaries --------- control_dict = { @@ -65,7 +67,7 @@ if __name__ == "__main__": # QE command - qe_command = "mpirun -np 4 pw.x -nk 2" + qe_command = "mpirun -np 4 pw.x -nk 4" # QE run seetings (computational parameters namelist) qe_run_settings = QeRunSettings( @@ -81,11 +83,9 @@ ) # Instance of QEStaticMaker - qe_maker = QEStaticMaker( + qe_maker = QeStaticMaker( name="static_qe", command=qe_command, - template_pwi=None, # Optional if run_settings - structures="/leonardo_work/EUHPC_A04_113/Alberto/GenMLFF-progect/Test-QE/initial_dataset.extxyz", workdir=None, # default /qe_static run_settings=qe_run_settings, kpoints=k_points_settings, @@ -93,10 +93,11 @@ ) # Define QE scf workflow - qe_workflow = qe_maker.make() + structures_path="/leonardo_work/EUHPC_A04_113/Alberto/GenMLFF-progect/autoplex/tests/auto/QE/initial_dataset.extxyz" + qe_scf_workflow = qe_maker.make(structures_path) # Update flow config - set_run_config(qe_workflow, name_filter="static_qe", exec_config="qe_config", worker="schedule_worker", resources=parallel_gpu_resources) + set_run_config(qe_scf_workflow, name_filter="static_qe", exec_config="qe_config", worker="schedule_worker", resources=parallel_gpu_resources) # Submit flow - submit_flow(qe_workflow, worker="local_worker", resources={}, project="GenMLFF") \ No newline at end of file + submit_flow(qe_scf_workflow, worker="local_worker", resources={}, project="GenMLFF") \ No newline at end of file From e844830859945aa57ef07c5e2dc53509d4410fab Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 6 Oct 2025 11:30:44 +0000 Subject: [PATCH 16/18] pre-commit auto-fixes --- src/autoplex/misc/qe/__init__.py | 2 +- src/autoplex/misc/qe/jobs.py | 11 ++++---- src/autoplex/misc/qe/run.py | 13 +++++----- src/autoplex/misc/qe/schema.py | 18 ++++++++----- src/autoplex/misc/qe/utils.py | 43 ++++++++++++++++---------------- 5 files changed, 45 insertions(+), 42 deletions(-) diff --git a/src/autoplex/misc/qe/__init__.py b/src/autoplex/misc/qe/__init__.py index 6e61b2ff0..4ce69eee7 100644 --- a/src/autoplex/misc/qe/__init__.py +++ b/src/autoplex/misc/qe/__init__.py @@ -3,9 +3,9 @@ from .schema import ( InputDoc, OutputDoc, - TaskDoc, QeKpointsSettings, QeRunSettings, + TaskDoc, ) from .utils import QeStaticInputGenerator diff --git a/src/autoplex/misc/qe/jobs.py b/src/autoplex/misc/qe/jobs.py index 2304446da..e975701de 100644 --- a/src/autoplex/misc/qe/jobs.py +++ b/src/autoplex/misc/qe/jobs.py @@ -4,9 +4,8 @@ from dataclasses import dataclass from ase import Atoms -from pymatgen.core import Structure - from jobflow import Flow, Maker +from pymatgen.core import Structure from .run import run_qe_static from .schema import QeKpointsSettings, QeRunSettings @@ -46,8 +45,8 @@ class QeStaticMaker(Maker): def make( self, - structures: Atoms | list[Atoms] | Structure | list[Structure] | str | list[str] - ) -> Flow: + structures: Atoms | list[Atoms] | Structure | list[Structure] | str | list[str], + ) -> Flow: """ Create a Flow to run static SCF calculations with QE for given structures. @@ -55,7 +54,7 @@ def make( ---------- structures : Atoms | list[Atoms] | Structure | list[Structure] | str | list[str] Single or list of ASE Atoms, pymatgen Structures, or ASE-readable files. - + Returns ------- Flow @@ -89,4 +88,4 @@ def make( jobs.append(j) tasks.append(j.output) - return Flow(jobs=jobs, output=tasks, name=self.name) \ No newline at end of file + return Flow(jobs=jobs, output=tasks, name=self.name) diff --git a/src/autoplex/misc/qe/run.py b/src/autoplex/misc/qe/run.py index 3650b1a32..e8221d852 100644 --- a/src/autoplex/misc/qe/run.py +++ b/src/autoplex/misc/qe/run.py @@ -7,9 +7,8 @@ from ase.io import read from ase.units import GPa -from pymatgen.io.ase import AseAtomsAdaptor - from jobflow import job +from pymatgen.io.ase import AseAtomsAdaptor from .schema import InputDoc, OutputDoc, TaskDoc @@ -27,7 +26,7 @@ def _parse_total_energy_ev(pwo_path: str) -> float | None: ---------- pwo_path : str Path to QE output file (.pwo) - + Returns ------- float | None @@ -61,7 +60,7 @@ def run_qe_static(input: InputDoc, command: str) -> TaskDoc: Input document containing paths and settings for the QE run. command : str Command to execute QE (e.g. "pw.x" or "mpirun -np 4 pw.x -nk 2"). - + Returns ------- TaskDoc @@ -80,12 +79,12 @@ def run_qe_static(input: InputDoc, command: str) -> TaskDoc: # # Manual parse of total energy in eV from .pwo # energy_ev = _parse_total_energy_ev(pwo_path) - + # Parse with ASE atoms = read(pwo_path) energy_ev = atoms.get_total_energy() forces_evA = atoms.get_forces() - stress_kbar = atoms.get_stress(voigt=False)*(-10/GPa) + stress_kbar = atoms.get_stress(voigt=False) * (-10 / GPa) final_structure = AseAtomsAdaptor().get_structure(atoms) output = OutputDoc( @@ -101,4 +100,4 @@ def run_qe_static(input: InputDoc, command: str) -> TaskDoc: task_label="qe_scf", input=input, output=output, - ) \ No newline at end of file + ) diff --git a/src/autoplex/misc/qe/schema.py b/src/autoplex/misc/qe/schema.py index 531d3aecc..3cb895372 100644 --- a/src/autoplex/misc/qe/schema.py +++ b/src/autoplex/misc/qe/schema.py @@ -2,11 +2,10 @@ from typing import Dict -from pymatgen.core import Structure -from emmet.core.structure import StructureMetadata from emmet.core.math import Matrix3D, Vector3D - +from emmet.core.structure import StructureMetadata from pydantic import BaseModel, Field, field_validator +from pymatgen.core import Structure class QeRunSettings(BaseModel): @@ -51,8 +50,13 @@ class InputDoc(BaseModel): workdir: str pwi_path: str seed: str - run_settings: QeRunSettings = Field(None, description="QE namelist section with: &control, &system, &electrons") - pseudo: Dict[str, str] = Field(None, description="Dictionary of atomic symbols and corresponding pseudopotential files.") + run_settings: QeRunSettings = Field( + None, description="QE namelist section with: &control, &system, &electrons" + ) + pseudo: dict[str, str] = Field( + None, + description="Dictionary of atomic symbols and corresponding pseudopotential files.", + ) kpoints: QeKpointsSettings = Field(None, description="QE K_POINTS settings") @@ -60,6 +64,7 @@ class OutputDoc(BaseModel): """ The outputs of this jobs """ + energy: float | None = Field(None, description="Total energy in units of eV.") energy_per_atom: float | None = Field( @@ -82,6 +87,7 @@ class OutputDoc(BaseModel): None, description="The stress on the cell in units of kbar." ) + class TaskDoc(StructureMetadata): """Document containing information on structure manipulation using Quantum ESPRESSO.""" @@ -102,4 +108,4 @@ class TaskDoc(StructureMetadata): dir_name: str | None = Field( None, description="Directory where the QE calculations are performed." - ) \ No newline at end of file + ) diff --git a/src/autoplex/misc/qe/utils.py b/src/autoplex/misc/qe/utils.py index fb6eabca4..c663a313d 100644 --- a/src/autoplex/misc/qe/utils.py +++ b/src/autoplex/misc/qe/utils.py @@ -1,22 +1,21 @@ from __future__ import annotations -import os import logging +import os from dataclasses import dataclass import numpy as np - -from pymatgen.core import Structure -from pymatgen.io.ase import AseAtomsAdaptor - -from ase import Atoms, Atom +from ase import Atom, Atoms from ase.data import atomic_masses, atomic_numbers from ase.io import read +from pymatgen.core import Structure +from pymatgen.io.ase import AseAtomsAdaptor from .schema import InputDoc, QeKpointsSettings, QeRunSettings logger = logging.getLogger(__name__) + @dataclass class QeStaticInputGenerator: """ @@ -60,7 +59,7 @@ def generate_for_structures( Directory used to write input/output files. seed_prefix : str Prefix for naming input/output QE files. - + Returns ------- list[InputDoc] @@ -85,7 +84,8 @@ def generate_for_structures( run_settings=self.run_settings, kpoints=self.kpoints, pseudo=self.pseudo, - )) + ) + ) return input_sets def _write_pwi( @@ -103,7 +103,7 @@ def _write_pwi( Path to output .pwi file. atoms : Atoms ASE Atoms object. - + Raises ------ ValueError @@ -199,6 +199,7 @@ def _write_pwi( fh.writelines(cell_lines) fh.writelines(pos_lines) + # --------- Utils --------- @@ -213,7 +214,7 @@ def _read_template(path: str | None) -> list[str]: def _load_structures( - structures: Atoms | list[Atoms] | Structure | list[Structure] | str | list[str] + structures: Atoms | list[Atoms] | Structure | list[Structure] | str | list[str], ) -> list[Atoms]: """ Load structures from various input types and return a list of ASE Atoms objects. @@ -222,13 +223,12 @@ def _load_structures( ---------- structures : Atoms | list[Atoms] | Structure | list[Structure] | str | list[str] Single or list of ASE Atoms, pymatgen Structures, or ASE-readable files. - + Returns ------- list[Atoms] List of ASE Atoms objects. """ - # ASE readable files # Sinlge ASE-readable file if isinstance(structures, str): @@ -239,7 +239,6 @@ def _load_structures( for fname in structures: atoms_list += read(fname, index=":") - # ASE Atoms elif isinstance(structures, Atoms): # List of ASE Atoms objects @@ -252,19 +251,19 @@ def _load_structures( raise ValueError("Unsupported ASE Atoms input.") # List of ASE Atoms objects elif isinstance(structures, list) and all(isinstance(s, Atoms) for s in structures): - atoms_list = structures - + atoms_list = structures # Pymatgen Structures # Single pymatgen structure elif isinstance(structures, Structure): atoms_list = [AseAtomsAdaptor().get_atoms(structures)] # List of pymatgen structures - elif isinstance(structures, list) and all(isinstance(s, Structure) for s in structures): + elif isinstance(structures, list) and all( + isinstance(s, Structure) for s in structures + ): adaptor = AseAtomsAdaptor() atoms_list = [adaptor.get_atoms(s) for s in structures] - else: raise ValueError("Unsupported type for structures input.") @@ -280,7 +279,7 @@ def _render_minimal_namelists(settings: QeRunSettings) -> list[str]: ---------- settings : QeRunSettings QE namelist settings. - + Returns ------- list[str] @@ -327,7 +326,7 @@ def _render_kpoints( ASE Atoms object. kpoints : QeKpointsSettings K-points settings. - + Returns ------- list[str] @@ -368,14 +367,14 @@ def _render_kpoints( def _compute_kpoints_grid(cell: np.ndarray, kspace_resolution: float) -> list[int]: """ Compute Monkhorst-Pack mesh from cell and kspace_resolution (in angstrom^-1). - + Parameters ---------- cell : np.ndarray 3x3 array with cell vectors in angstrom. kspace_resolution : float Desired k-space resolution in angstrom^-1. - + Returns ------- list[int] @@ -387,4 +386,4 @@ def _compute_kpoints_grid(cell: np.ndarray, kspace_resolution: float) -> list[in logger.debug( "QE MP mesh %s using k-resolution %s angstrom^-1", mesh, kspace_resolution ) - return mesh \ No newline at end of file + return mesh From c21b9cd2044d08c364c8cf850c772687c2dd1d54 Mon Sep 17 00:00:00 2001 From: 41bY Date: Mon, 6 Oct 2025 13:33:31 +0200 Subject: [PATCH 17/18] Removed old implementation --- .../auto/GenMLFF/QuantumEspressoSCF.py | 488 ------------------ tests/auto/GenMLFF/reference.pwi | 45 -- tests/auto/GenMLFF/test_qe.py | 41 -- 3 files changed, 574 deletions(-) delete mode 100644 src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py delete mode 100644 tests/auto/GenMLFF/reference.pwi delete mode 100644 tests/auto/GenMLFF/test_qe.py diff --git a/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py b/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py deleted file mode 100644 index d53d6da96..000000000 --- a/src/autoplex/auto/GenMLFF/QuantumEspressoSCF.py +++ /dev/null @@ -1,488 +0,0 @@ -"""Jobs to create training data for ML potentials.""" - -import logging -import os -import subprocess -from dataclasses import dataclass, field -from glob import glob - -import numpy as np -from ase import Atoms -from ase.io import read -from jobflow import Flow, Maker, Response, job - -logging.basicConfig(level=logging.INFO, format="[%(levelname)s] %(message)s") - - -def qe_params_from_config(config: dict): - """ - Return QEstaticLabelling params from a configuration dictionary. - - Args: - config (dict): Keys should match __init__ parameters. For example: - { - "qe_run_cmd": qe_run_cmd, - "num_qe_workers": num_qe_workers, - "fname_pwi_template": fname_pwi_template, - "kspace_resolution" : Kspace_resolution, - "koffset": Koffset, - "fname_structures": fname_structures, - } - - Returns - ------- - params: dict - Dictionary with parameters for QEstaticLabelling. - """ - # Get default parameters - params = { - "qe_run_cmd": "pw.x", - "num_qe_workers": 1, - "fname_pwi_template": None, - "kspace_resolution": None, - "koffset": [False, False, False], - "fname_structures": None, - } - - # Update parameters with values from the config file - if config is None: - raise ValueError("Configuration file is empty or not properly formatted.") - params.update(config) - - # Check a valid reference pwi path is provided - if not os.path.exists(params["fname_pwi_template"]): - raise ValueError( - f"Reference QE input file '{params['fname_pwi_template']}' not found." - ) - - return params - - -def run_qe(command, fname_pwi, fname_pwo): - """ - Run the QE command in a subprocess. Execute one QuantumEspresso calculation on the current input file. - """ - # Assemble QE command - run_cmd = f"{command} < {fname_pwi} >> {fname_pwo}" - - success = False - try: - # Launch QE and wait till ending - subprocess.run(run_cmd, shell=True, check=True, executable="/bin/bash") - - success = True - - except subprocess.CalledProcessError: - - success = False - - return success - - -def lock_input(pwi_fname, worker_id): - - pwi_lock_fname = "" - # Check if pwo exists - pwo_fname = pwi_fname.replace(".pwi", ".pwo") - if os.path.exists(pwo_fname): - return pwi_lock_fname, pwo_fname # If exists, skip to next pwi - - # Try to lock the pwi file by renaming it - pwi_lock_fname = f"{pwi_fname}.lock_{worker_id}" - try: - os.rename(f"{pwi_fname}", f"{pwi_lock_fname}") - except Exception: - pwi_lock_fname = "" - - return pwi_lock_fname, pwo_fname - - -@job -def run_qe_worker( - id, - command, - work_dir, -): - """ - Run the QE command in a subprocess. - """ - # Get pwi files - pwi_files = glob(os.path.join(work_dir, "*.pwi")) - - # Check pwo does not exist - worker_output = {"success": [], "output": [], "outdir": []} - for pwi in pwi_files: - # Try locking the pwi file - lock_pwi, pwo_fname = lock_input(pwi_fname=pwi, worker_id=id) - - if lock_pwi == "": - continue # Skip to next pwi if lock failed - - # Get output directory of this calculation - with open(lock_pwi) as f: - pwi_lines = f.readlines() - outdir_line = [line.split("=")[1] for line in pwi_lines if "outdir" in line][0] - outdir_line = ( - outdir_line.strip().replace("'", "").replace('"', "") - ) # Remove quotes - outdir = os.getcwd() + f"/{outdir_line}" - - # Launch QE calculation - success = run_qe(command=command, fname_pwi=lock_pwi, fname_pwo=pwo_fname) - - # Update output - worker_output["success"].append(success) - worker_output["output"].append(pwo_fname) - worker_output["outdir"].append(outdir) - - return worker_output - - -@dataclass -class QEstaticLabelling(Maker): - """ - Maker to set up and run Quantum Espresso static calculations for input structures, including bulk, isolated atoms, and dimers. - - Parameters - ---------- - name: str - Name of the flow. - qe_run_cmd: str - String with the command to run QE (including its executable path/or application name). - fname_pwi_template: str - Path to file containing the template computational parameters. - fname_structures: str - Path to ASE-readible file containing the structures to be computed. - num_qe_workers: int | None - Number of workers to use for the calculations. If None, defaults to the number of structures. - """ - - name: str = "do_qe_labelling" - qe_run_cmd: str | None = ( - None # String with the command to run QE (including its executable path/or application name) - ) - fname_pwi_template: str | None = ( - None # Path to file containing the template computational parameters - ) - fname_structures: str | list[str] | None = ( - None # Path or list[Path] to ASE-readible file containing the structures to be computed - ) - num_qe_workers: int | None = None # Number of workers to use for the calculations. - kspace_resolution: float | None = ( - None # K-space resolution in Angstrom^-1, used to set the K-points in the pwi file - ) - koffset: list[bool] = field( - default_factory=lambda: [False, False, False] - ) # K-points offset in the pwi file - - def make(self): - # Define jobs - joblist = [] - - # Load structures - structures = self.load_structures(fname_structures=self.fname_structures) - if len(structures) == 0: - logging.info("No structures found to compute with DFT. Exiting.") - return Response(replace=None, output=[]) - - # Check pwi template - pwi_reference_lines = self.check_pwi_template(self.fname_pwi_template) - - # Write pwi input files for each structure - work_dir = os.getcwd() - path_to_qe_workdir = os.path.join(work_dir, "scf_files") - os.makedirs(path_to_qe_workdir, exist_ok=True) - for i, structure in enumerate(structures): - # Get fname of the next pwi file - fname_new_pwi = os.path.join(path_to_qe_workdir, f"structure_{i}.pwi") - - # Write pwi file for the structure - self.write_pwi( - fname_pwi_output=fname_new_pwi, - structure=structure, - pwi_reference=pwi_reference_lines, - ) - - # Set number of QE workers - if ( - self.num_qe_workers is None - ): # 1 worker per structure (all DFT jobs in parallel) - num_qe_workers = len(glob(os.path.join(path_to_qe_workdir, "*.pwi"))) - else: - num_qe_workers = self.num_qe_workers - - # Launch QE workers - outputs = [] - for id_qe_worker in range(num_qe_workers): - worker_job = run_qe_worker( - id=id_qe_worker, - command=self.qe_run_cmd, - work_dir=path_to_qe_workdir, - ) - worker_job.name = f"run_qe_worker_{id_qe_worker}" - - joblist.append(worker_job) - outputs.append(worker_job.output) - - qe_wrk_flow = Flow(jobs=joblist, output=outputs, name="qe_workers") - - # Output is a list of success status, one for each worker - # The success status is a dictionary with the pwo file name as key and the calculation success status as value (True/False) - return qe_wrk_flow - - def load_structures( - self, - fname_structures: str | list[str] | None = None, - ): - """ - Load structures from a file or a list of files. - - Parameters - ---------- - fname_structures : str | list[str] | None - Path or list of paths to ASE-readable files containing the structures to be loaded. - If None, no structures will be loaded. - - Returns - ------- - list[Atoms] - List of ASE Atoms objects representing the loaded structures. - """ - # Convert fname_structures to a list if it is a string - if isinstance(fname_structures, str): - fname_structures = [fname_structures] - elif fname_structures is None: - return [] - elif not isinstance(fname_structures, list): - raise ValueError("fname_structures must be a string or a list of strings.") - - # Loop over provided files and load structures - structures = [] - for fname in fname_structures: - # Check if all files exist - if not os.path.exists(fname): - raise FileNotFoundError(f"File {fname} does not exist.") - - # Read structures from file - try: - structures += read(fname, index=":") - except Exception as e: - logging.error(f"Error reading file {fname}: {e}") - - return structures - - def check_pwi_template(self, fname_template): - """ - Check the pwi template file for the required parameters. - """ - # Read template file - tmp_pwi_lines = [] - with open(fname_template) as f: - tmp_pwi_lines = f.readlines() - - # Modify lines with structure information: - # Assume ntyp, atom_types and pseudoptentials are already defined in the template and consistent with the structures - # Assume ibrav=0 and Kspacing is already defined in the template - idx_nat_line, idx_pos_line, idx_cell_line = 0, 0, 0 - for i, line in enumerate(tmp_pwi_lines): - if "nat" in line: - idx_nat_line = i - - elif "ATOMIC_POSITIONS" in line: - idx_pos_line = i - - elif "CELL_PARAMETERS" in line: - idx_cell_line = i - - # Set nat line - if idx_nat_line == 0: # nat not defined, assume nat = 0 - raise ValueError( - "Number of atoms line not defined in the template file. Please define 'nat =' in the template file." - ) - tmp_pwi_lines[idx_nat_line] = "nat = \n" - - # Cancel lines with ATOMIC_POSITIONS and CELL_PARAMETERS - if (idx_pos_line == 0 and idx_cell_line > 0) or ( - idx_pos_line > 0 and idx_cell_line == 0 - ): - idx_to_delete = idx_pos_line - del tmp_pwi_lines[idx_to_delete:] - - elif idx_pos_line > 0 and idx_cell_line > 0: - idx_to_delete = min([idx_pos_line, idx_cell_line]) - del tmp_pwi_lines[idx_to_delete:] - - return tmp_pwi_lines - - def write_pwi( - self, - fname_pwi_output: str, - structure: Atoms, - pwi_reference: list[str], - ): - """ - Write the pwi input file for the given structure. - """ - # Duplicate the pwi template to avoid overwriting the reference lines - pwi_template = pwi_reference.copy() - - # Check pwi lines - idx_diskio, idx_outdir, idx_nat_line, idx_kpoints_line, nat = ( - 0, - 0, - 0, - 0, - len(structure), - ) - for idx, line in enumerate(pwi_template): - if "nat =" in line: - idx_nat_line = idx - elif "disk_io" in line: - idx_diskio = idx - elif "outdir" in line: - idx_outdir = idx - elif "K_POINTS" in line: - idx_kpoints_line = idx - - # Update number of atoms - pwi_template[idx_nat_line] = f"nat = {nat}\n" - - # Get identifier for this structure - structure_id = fname_pwi_output.split("/")[-1].replace(".pwi", "") - - # Update outdir based on disk_io - if ( - idx_diskio == 0 or "none" not in pwi_template[idx_diskio] - ): # disk_io is not 'none' (QE default is low for scf) - if idx_outdir == 0: # outdir not defined, define it - pwi_template.insert(idx_diskio + 1, f"outdir = {structure_id}\n") - else: # outdir is defined, update it - pwi_template[idx_outdir] = f"outdir = '{structure_id}'\n" - else: # disk_io is 'none', remove outdir line - if idx_outdir == 0: - pwi_template.insert(idx_diskio + 1, "outdir = 'OUT'\n") - - kpoints_lines = self._set_Kpoints( - tmp_pwi_lines=pwi_template, - idx_kpoints_line=idx_kpoints_line, - atoms=structure, - Kspace_resolution=self.kspace_resolution, - Koffset=self.koffset, - ) - - # Write cell lines - cell_lines = ["\nCELL_PARAMETERS (angstrom)\n"] - cell_lines += [ - f"{structure.cell[i, 0]:.10f} {structure.cell[i, 1]:.10f} {structure.cell[i, 2]:.10f}\n" - for i in range(3) - ] - - # Write positions lines - pos_lines = ["\nATOMIC_POSITIONS (angstrom)\n"] - for i, atom in enumerate(structure): - pos_lines.append( - f"{atom.symbol} {structure.positions[i, 0]:.10f} {structure.positions[i, 1]:.10f} {structure.positions[i, 2]:.10f}\n" - ) - - # Write the modified lines to the new pwi file - with open(fname_pwi_output, "w") as f: - for ( - line - ) in pwi_template: # Write reference pwi lines (computational parameters) - f.write(line) - for line in kpoints_lines: # Write K-points lines - f.write(line) - for line in cell_lines: # Write cell lines - f.write(line) - for line in pos_lines: # Write positions lines - f.write(line) - - def _set_Kpoints( - self, - tmp_pwi_lines: list[str], - idx_kpoints_line: int, - atoms: Atoms, - Kspace_resolution: float | None = None, - Koffset: list[bool] = [False, False, False], - ): - """ - Set the K-points in the pwi file based on user definition or K-space resolution. - """ - # Define K-points lines - kpoints_lines = [] - - # K_POINTS line not found - if idx_kpoints_line == 0: - if ( - Kspace_resolution is None - ): # K_POINTS line not found and Kspace_resolution is not defined - raise ValueError( - "K_POINTS line not found in the template file. Please define K_POINTS in the template file or provide Kspace_resolution." - ) - # Find k-points grid using Monkorst-Pack method based on K-space resolution - # Get real space cell - cell = atoms.cell - - # Find Kpoints grid - # TODO: use structure_type info to generalize to non-periodic systems (3d, 2d, 1d, 0d) - MP_mesh = self._compute_kpoints_grid(cell, Kspace_resolution) - - # Format k-points lines - kpoints_lines.append("\nK_POINTS automatic\n") # Header for MP-grid - Kpoint_line = ( - f"{MP_mesh[0]} {MP_mesh[1]} {MP_mesh[2]}" # K-points grid line - ) - for offset in Koffset: # Add offset - if offset: - Kpoint_line += " 1" - else: - Kpoint_line += " 0" - Kpoint_line += "\n" - kpoints_lines.append(Kpoint_line) - - # K_POINTS is defined by user in reference pwi file, keep the line/s - elif idx_kpoints_line > 0: - if ( - "gamma" in tmp_pwi_lines[idx_kpoints_line] - or "Gamma" in tmp_pwi_lines[idx_kpoints_line] - ): # KPOINT is 1 line - kpoints_lines = tmp_pwi_lines[idx_kpoints_line : idx_kpoints_line + 1] - del tmp_pwi_lines[idx_kpoints_line:] - elif "automatic" in tmp_pwi_lines[idx_kpoints_line]: # KPOINTS is 2 lines - kpoints_lines = tmp_pwi_lines[idx_kpoints_line : idx_kpoints_line + 2] - del tmp_pwi_lines[idx_kpoints_line:] - elif ( - "tpiba" in tmp_pwi_lines[idx_kpoints_line] - or "crystal" in tmp_pwi_lines[idx_kpoints_line] - ): # KPOINTS is multiple lines - num_ks = int( - tmp_pwi_lines[idx_kpoints_line + 1].split()[0] - ) # Get number of k-points - kpoints_lines = tmp_pwi_lines[ - idx_kpoints_line : idx_kpoints_line + num_ks + 2 - ] # Get k-points lines - del tmp_pwi_lines[idx_kpoints_line:] - else: - raise ValueError( - f"K_POINTS format: {tmp_pwi_lines[idx_kpoints_line]} is unknown in pwi template file" - ) - - return kpoints_lines - - def _compute_kpoints_grid(self, cell, Kspace_resolution): - """ - Compute the k-points grid using Monkhorst-Pack method based on the cell and K-space resolution. - """ - # Compute the reciprocal cell vectors: b_i = 2π * (a_j x a_k) / (a_i . (a_j x a_k)) - rec_cell = 2.0 * np.pi * np.linalg.inv(cell).T - - # Compute reciprocal lattice vecotors' lenghts - lengths = np.linalg.norm(rec_cell, axis=1) - - # Compute mesh size - mesh = [int(np.ceil(L / Kspace_resolution)) for L in lengths] - print( - f"Computed k-points mesh: {mesh} for K-space resolution: {Kspace_resolution} Angstrom^-1" - ) # DEBUG - - return mesh diff --git a/tests/auto/GenMLFF/reference.pwi b/tests/auto/GenMLFF/reference.pwi deleted file mode 100644 index 3954f485f..000000000 --- a/tests/auto/GenMLFF/reference.pwi +++ /dev/null @@ -1,45 +0,0 @@ -&CONTROL - calculation = 'scf' - restart_mode = 'from_scratch' - prefix = 'FeCOH' - tprnfor = .true. - tstress = .true. - outdir = './OUT/' - disk_io = 'none' - pseudo_dir = '/leonardo/home/userexternal/apacini0/PSEUDO' - max_seconds = 86000 -/ - -&SYSTEM - ibrav = 0 - nat = 506 - ntyp = 4 - ecutwfc = 60 - ecutrho = 480 - occupations = 'smearing' - smearing = 'gaussian' - degauss = 0.00735 - nosym=.true. - vdw_corr = 'dft-d3' - nspin = 2 - starting_magnetization(1) = 0.4 -/ - -&ELECTRONS - diagonalization = 'david' - mixing_beta = 0.15 - electron_maxstep = 150 - mixing_mode = 'local-TF' - mixing_ndim = 16 - conv_thr = 1.0d-6 -/ - -&IONS -/ - -ATOMIC_SPECIES -Fe 55.845 Fe.pbe-sp-van.UPF -C 12.011 C.pbe-n-kjpaw_psl.1.0.0.UPF -O 15.999 O.pbe-n-kjpaw_psl.1.0.0.UPF -H 2.016 H.pbe-kjpaw_psl.1.0.0.UPF - diff --git a/tests/auto/GenMLFF/test_qe.py b/tests/auto/GenMLFF/test_qe.py deleted file mode 100644 index a3bddf7c6..000000000 --- a/tests/auto/GenMLFF/test_qe.py +++ /dev/null @@ -1,41 +0,0 @@ -from jobflow_remote import submit_flow, set_run_config -from autoplex.auto.GenMLFF.QuantumEspressoSCF import qe_params_from_config, QEstaticLabelling - -if __name__ == "__main__": - - # Resources for QE - parallel_gpu_resources = { - "account": "IscrB_CNT-HARV", - "partition": "boost_usr_prod", - "qos": "boost_qos_dbg", - "time": "00:30:00", - "nodes": 1, - "ntasks_per_node": 2, - "cpus_per_task": 8, - "gres": "gpu:2", - "mem": "240000", - "job_name": "qe_auto", - "qerr_path": "JOB.err", - "qout_path": "JOB.out", - } - - - # QE job parameters (test) - qe_test_params = { - "qe_run_cmd": "mpirun -np 2 pw.x -nk 2 ", #Command to run QE scf calculation - "num_qe_workers": 2, #Number of workers to use for the calculations. If None setp up 1 worker per scf - "kspace_resolution": 0.25, #k-point spacing in 1/Angstrom - "koffset": [False, False, True], #k-point offset - "fname_pwi_template": "/leonardo_work/EUHPC_A04_113/Alberto/GenMLFF-progect/autoplex/tests/auto/GenMLFF/reference.pwi", #Path to file containing the template QE input - "fname_structures": "/leonardo_work/EUHPC_A04_113/Alberto/GenMLFF-progect/Test-QE/initial_dataset.extxyz", #Path to file containing the structures to be computed - } - - # Define QE scf workflow - qe_params = qe_params_from_config(qe_test_params) - qe_workflow = QEstaticLabelling(**qe_params).make() - - # Update flow config - set_run_config(qe_workflow, name_filter="run_qe_worker", exec_config="qe_config", worker="schedule_worker", resources=parallel_gpu_resources) - - # Submit flow - submit_flow(qe_workflow, worker="local_worker", resources={}, project="GenMLFF") \ No newline at end of file From c0611d6a406cc9675ab3cead1c8e039256059f09 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Mon, 6 Oct 2025 11:34:04 +0000 Subject: [PATCH 18/18] pre-commit auto-fixes --- src/autoplex/misc/qe/schema.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/autoplex/misc/qe/schema.py b/src/autoplex/misc/qe/schema.py index 3cb895372..e88476449 100644 --- a/src/autoplex/misc/qe/schema.py +++ b/src/autoplex/misc/qe/schema.py @@ -1,7 +1,5 @@ from __future__ import annotations -from typing import Dict - from emmet.core.math import Matrix3D, Vector3D from emmet.core.structure import StructureMetadata from pydantic import BaseModel, Field, field_validator