#
from __future__ import annotations
from datetime import datetime
__author__= "Sabyasachi Tiwari"
__copyright__= "Copyright 2024, EPWpy project"
__version__= "1.0"
__maintainer__= "Sabyasachi Tiwari"
__maintainer_email__= "sabyasachi.tiwari@austin.utexas.edu"
__status__= "Production"
__date__= "May 03, 2024"
import numpy as np
import argparse
import pickle
from EPWpy import Logo as Logo
from EPWpy.default import code_loc as cd_loc
from EPWpy.default import default as df
from EPWpy.EPWpy_run import PyRun
from EPWpy.EPWpy_prepare import PyPrepare
from EPWpy.EPWpy_analysis import PyAnalysis
from EPWpy.BGW.EPWpy_BGW import PyBGW
from EPWpy.QE.EPWpy_QE import PyQE
from EPWpy.default.set_default import *
from EPWpy.structure.lattice import *
from EPWpy.error_handling.error_handler import *
from EPWpy.flow.transport import FlowManager
#from EPWpy.flow.transport import *
import os
import EPWpy.utilities
[docs]
class EPWpy(SetDefaultVals, FlowManager, PyBGW, PyQE, PyAnalysis, JobStatus, Lattice):
"""
The central interface class of the **EPWpy** framework.
EPWpy provides a unified Python interface for *ab initio* electronic-structure
and many-body codes, including **Quantum ESPRESSO (QE)**, **EPW**, and
**BerkeleyGW (BGW)**. It automates input preparation, execution, and output
parsing, enabling scalable workflows for electron–phonon and many-body
physics simulations.
This class can be initialized directly to set up and run complete workflows,
or used as a base for higher-level automated pipelines.
---------------------------------------------------------------------------
Collaborating Institutions
---------------------------------------------------------------------------
- **The University of Texas at Austin** (Prof. Feliciano Giustino)
- **Binghamton University** (Prof. Roxana Margine)
- **University of Michigan** (Prof. Emmanouil Kioupakis)
- **Université Catholique de Louvain** (Prof. Samuel Poncé)
---------------------------------------------------------------------------
Output Structure
---------------------------------------------------------------------------
All simulation results and parameters are stored internally in dictionaries.
These can be accessed via the respective attributes of QE, PH, EPW, or BGW
modules. Example:
>>> self.pw_atomic_species["atomic_species"]
Common dictionary attributes include:
- ``pw_system``: system inputs and outputs
- ``pw_control``: control inputs and outputs
- ``pw_electrons``: electron-related inputs and outputs
- ``pw_ions``: ion dynamics inputs/outputs (enabled if iondynamics=True)
- ``pw_cell``: cell dynamics inputs/outputs (enabled if celldynamics=True)
- ``pw_atomic_positions``: atomic positions
- ``pw_atomic_species``: atomic species, pseudopotentials, and masses
- ``pw_cell_parameters``: lattice vectors
- ``pw_bands``: band-structure inputs and outputs
- ``ph_params``: PHonon code parameters
- ``epw_params``: EPW parameters
- ``wannier_params``: Wannierization parameters
- ``pw2wann_params``: pw2wannier90 parameters
- ``BGW_*``: various BerkeleyGW stages (init, epsilon, sigma, kernel, etc.)
- ``zg_params``, ``q2r_params``, ``eps_inputpp``: miscellaneous utilities
---------------------------------------------------------------------------
Inherited Classes
---------------------------------------------------------------------------
- ``PyBGW``: Interface to the BerkeleyGW package
- ``PyQE``: Interface to Quantum ESPRESSO
- ``PyAnalysis``: Property extraction and analysis utilities
- ``JobStatus``: Status tracking for running jobs
- ``Lattice``: Lattice construction and pseudopotential management
- ``SetDefaultVals``: Handles default initialization values
---------------------------------------------------------------------------
Composed Classes
---------------------------------------------------------------------------
- ``PyRun``: Executes individual simulations (PW, PH, EPW, BGW)
- ``PyPrepare``: Prepares inputs and manages calculation setup
---------------------------------------------------------------------------
Parameters
---------------------------------------------------------------------------
type_input : dict, optional
Dictionary containing QE/EPW input variables for initialization.
system : str, default="si"
Folder name or system identifier for the current calculation.
code : str, optional
Path to the executable code (QE, EPW, or BGW). If ``None``, the code path
is determined from the global ``cd_loc.code_set`` variable.
env : str, default="mpirun"
Environment launcher for parallel execution. Typical values include:
``"mpirun"``, ``"srun"``, or ``"ibrun"``.
---------------------------------------------------------------------------
Attributes
---------------------------------------------------------------------------
Run : PyRun
Manages execution of QE, PH, EPW, or BGW binaries.
Prepare : PyPrepare
Handles creation of input files and directory setup.
verbosity : int
Level of console output verbosity.
state : dict
Internal EPWpy state dictionary containing runtime parameters.
writer : object
Writer utility for generating formatted input files.
transfer_file : list
Files to be transferred across stages.
dont_do : list
Steps to skip in automated flows.
run_serial : bool
Whether to enforce serial execution mode.
dolink : bool
Whether to create symbolic links between run directories.
---------------------------------------------------------------------------
Example
---------------------------------------------------------------------------
>>> from epwpy import EPWpy
>>> epw = EPWpy(system="graphene", env="ibrun", code="/path/to/pw.x")
>>> epw.Run # Access the run manager
<PyRun object>
>>> epw.Prepare # Prepare input files
<PyPrepare object>
"""
def __init__(self, type_input: dict = None, system: str = 'si',
code: str = None, env: str = 'mpirun',
QE_ver: float = 7.4, EPW_ver: float = 5.9):
"""
Initialize an EPWpy workflow object.
Parameters
----------
type_input : dict, optional
Input parameters for initialization.
system : str, default='si'
Name of the system or folder for this calculation.
code : str, optional
Path to the executable code; inferred from global settings if None.
env : str, default='mpirun'
Parallel environment command (e.g., mpirun, ibrun, srun).
QE_ver : str, default '7.4'
Quantum ESPRESSO version.
EPW_ver : str, default '5.9'
EPW version.
Returns
-------
None
"""
self.transfer_file = []
self.dont_do = []
self.epwpy_logo()
self.system = system
self.code = code
self.run_serial = True
self.dolink = False
self.QE_ver = QE_ver
self.EPW_ver = EPW_ver
if self.code is None:
if cd_loc.code_set is None:
print('No code chosen')
else:
self.code = cd_loc.code_set
self.QE_ver = float(cd_loc.QE_version)
self.EPW_ver = float(cd_loc.EPW_version)
self.env = env
os.system(f'mkdir -p {self.system}')
self.default_values()
self.set_values(type_input)
self.Run = PyRun(1, self.env, self.code)
self.Prepare = PyPrepare(self.prefix, self.pseudo, files=[])
self.verbosity = 1
self.state = {'epwpy_params': None}
self.writer = self.create_writer(type_input)
self.writer = self.get_default_writer()
self.Prepare.writer = self.writer
self.Run.writer = self.writer
[docs]
def new_system_reset(self, type_input = None, system='si', code=None, env='mpirun'):
"""
The EPWpy is a utility that wraps EPW
"""
#
self.reset()
self._hard_refresh()
#
del self.Run
del self.Prepare
self.transfer_file=[]
self.dont_do = []
self.system=system
self.code=code
self.code = code
self.run_serial = True
self.dolink = False
if self.code is None:
if code_set is None:
print('No code chosen')
else:
self.code = code_set
self.env=env
os.system('mkdir -p '+self.system)
self.default_values()
self.set_values(type_input)
self.Run = PyRun(1,self.env,self.code)
self.Prepare = PyPrepare(self.prefix,self.pseudo, files=[])
self.verbosity = 1
self.state = {'epwpy_params':None}
self.writer = self.create_writer(type_input)
self.writer = self.get_default_writer()
self.Prepare.writer = self.writer
self.Run.writer = self.writer
[docs]
def epwpy_logo(self):
"""
Print EPWpy logo
"""
Logo.plot_logo()
[docs]
def save_EPWpy(self, filename=None):
if filename is None:
filename = f"{self.system}.pkl"
with open(filename, "wb") as f:
pickle.dump(self, f)
print(f"EPWpy object saved (pickle) as {filename}")
[docs]
@classmethod
def load_EPWpy(cls, filename):
with open(filename, "rb") as f:
obj = pickle.load(f)
print(f"EPWpy object loaded (pickle) from {filename}")
return obj
[docs]
def prepare(self, procs=1, type_run='scf', name=None,
infile=None, transfer_file=[]):
"""
Prepare the environment, input files, and working directories
for a specific EPWpy calculation stage.
This method configures a new calculation (e.g., SCF, NSCF, PH, EPW, BGW)
by creating appropriate directories, linking pseudopotentials, assigning
input/output file names, and storing the state for subsequent execution.
Parameters
----------
procs : int, optional
Number of processors to allocate for the run.
Default is ``1``.
type_run : str, optional
Type of calculation to prepare. Common options include:
- ``"scf"`` : self-consistent field run
- ``"nscf"`` : non-self-consistent run
- ``"ph"`` : phonon calculation
- ``"epw"`` : electron–phonon coupling run
- ``"gw"`` : BerkeleyGW run
Default is ``"scf"``.
name : str, optional
Custom name for the run folder. If not provided, defaults to
``type_run``.
infile : str, optional
Name of the input file to be prepared. Defaults to ``"{type_run}.in"``.
transfer_file : list of str, optional
List of additional files to copy or link into the run directory.
These may include previously generated wavefunctions, dynamical
matrices, or intermediate outputs.
Returns
-------
None
The method updates the internal EPWpy state and sets up a
:class:`PyPrepare` object (`self.prep`) to handle input generation.
Notes
-----
- The method automatically initializes a :class:`PyPrepare` instance and
assigns it to ``self.prep``.
- Internal state is serialized to JSON using :meth:`_save_json`.
- Working directories and filenames are configured via helper methods:
:meth:`set_folds`, :meth:`set_files`, :meth:`set_work`, and :meth:`set_home`.
- Files listed in ``transfer_file`` are added to ``self.transfer_file`` for
future use in multi-stage workflows.
Example
-------
>>> epw = EPWpy(system="graphene", code="/path/to/pw.x")
>>> epw.prepare(type_run="nscf", procs=8)
>>> print(epw.prep.writer)
<InputWriter object for nscf run>
"""
if(len(transfer_file) != 0):
for file in transfer_file:
self.transfer_file.append(file)
if infile is None:
infile = f'{type_run}.in'
if name is None:
name = type_run
self.prep=PyPrepare(self.prefix,self.pseudo,self.transfer_file)
self.prep.dolink = self.dolink
self.set_folds(name,type_run,self.prep)
self.set_files(name,type_run,self.prep)
#if (self.write_script):
self.prep.writer = self.writer
self.set_work()
self._save_json('epwpy_save',self.state)#self.pw_system)
self.set_home()
if (type_run == 'scf'):
self.set_work()
self.prep.prepare_scf(name, infile)
self.set_home()
elif (type_run == 'custom'):
self.set_work()
self.prep.prepare_custom(name, infile)
self.set_home()
elif (type_run == 'nscf'):
self.set_work()
self.prep.prepare_nscf(name, infile)
self.set_home()
elif (type_run == 'bs'):
self.set_work()
self.prep.prepare_bs(name, infile)
self.set_home()
elif (type_run == 'ph'):
self.set_work()
self.prep.prepare_ph(name, infile)
self.set_home()
elif (type_run == 'pp'):
self.set_work()
self.prep.prepare_pp(name, infile)
self.set_home()
elif (type_run == 'epw1'):
self.set_work()
if (name == 'epw1'):
name = 'epw'
self.epw_fold = name
self.prep.prepare_epw1(name, infile)
self.set_home()
elif (type_run == 'epw2'):
self.set_work()
if (name == 'epw2'):
name = 'epw'
self.prep.prepare_epw2(name, infile)
self.set_home()
elif (type_run == 'epw3'):
self.set_work()
if (name == 'epw3'):
name = 'epw'
self.prep.prepare_epw3(name, infile)
self.set_home()
elif (type_run == 'q2r'):
self.set_work()
if (name == 'q2r'):
name = self.ph_fold
self.prep.prepare_q2r(name, infile)
self.set_home()
elif (type_run == 'dvscf_q2r'):
self.set_work()
if (name == 'dvscf_q2r'):
name = self.ph_fold
self.prep.prepare_dvscf_q2r(name, infile)
self.set_home()
elif (type_run == 'postahc'):
self.set_work()
if (name == 'postahc'):
name = self.ph_fold
self.prep.prepare_postahc(name, infile)
self.set_home()
elif (type_run == 'zg'):
self.set_work()
self.prep.prepare_zg(name, infile)
self.set_home()
elif (type_run == 'wannier'):
self.set_work()
self.prep.prepare_wannier(name, infile)
self.set_home()
elif (type_run == 'eps'):
self.set_work()
self.prep.prepare_eps()
self.set_home()
elif (type_run == 'bands'):
self.set_work()
self.prep.prepare_bands()
self.set_home()
elif (type_run == 'nscf2supercond'):
self.set_work()
self.prep.prepare_nscf2supercond()
self.set_home()
elif (type_run == 'matdyn'):
self.set_work()
self.prep.prepare_matdyn(name = self.ph_fold)
self.set_home()
elif (type_run == 'dynmat'):
self.set_work()
self.prep.prepare_dynmat(name = self.ph_fold)
self.set_home()
elif (type_run == 'phdos'):
self.set_work()
self.prep.prepare_phdos(name = self.ph_fold)
self.set_home()
elif (type_run == 'nscf_tetra'):
self.set_work()
self.prep.prepare_nscf_tetra()
self.set_home()
elif (type_run == 'dos'):
if (name == type_run):
name = self.nscf_fold
self.set_work()
self.prep.prepare_dos(name, infile)
self.set_home()
elif (type_run == 'pdos'):
if (name == type_run):
name = self.nscf_fold
self.set_work()
self.prep.prepare_pdos(name, infile)
self.set_home()
elif (type_run == 'fbw'):
self.set_work()
self.prep.prepare_fbw()
self.set_home()
elif (type_run == 'fbw_mu'):
self.set_work()
self.prep.prepare_fbw_mu()
self.set_home()
elif (type_run == 'nesting'):
self.set_work()
self.prep.prepare_nesting()
self.set_home()
elif (type_run == 'phselfen'):
self.set_work()
self.prep.prepare_phselfen()
self.set_home()
elif (type_run == 'epw_outerbands'):
self.set_work()
self.prep.prepare_outerbands()
self.set_home()
elif (type_run == ''):
self.set_home()
self.prep.prepare_script()
[docs]
def run(self, procs, type_run=None, infile=None, name=None,
parallelization=None, flow_parallelization=None, flow_procs=None,
flavor='cplx', util=None):
"""
Execute a specific EPWpy workflow stage such as QE, PH, EPW, or BGW.
This method serves as the unified **run interface** for all EPWpy-supported
codes, launching the appropriate binary (via :class:`PyRun`) with user-defined
parallel settings, environment, and input files. It can be invoked after
:meth:`prepare` or independently once inputs are in place.
Parameters
----------
procs : int
Number of processors to use for the execution.
type_run : str, optional
Type of calculation to execute. Common values include:
- ``"scf"`` : self-consistent DFT calculation (pw.x)
- ``"nscf"`` : non-self-consistent DFT calculation
- ``"ph"`` : phonon calculation (ph.x)
- ``"epw"`` : electron–phonon coupling calculation (epw.x)
- ``"bgw"`` : BerkeleyGW execution
Default is ``"scf"``.
infile : str, optional
Input file name to be used. Defaults to ``"{type_run}.in"`` if not provided.
name : str, optional
Label or directory name for this run. Defaults to ``type_run``.
parallelization : dict, optional
Dictionary specifying explicit parallelization flags (e.g.
``{"npool": 4, "ndiag": 2}``).
flow_parallelization : dict, optional
Parallelization scheme for distributed or multi-step workflows.
flow_procs : int, optional
Total number of processors assigned across a workflow stage.
flavor : str, optional
Computational flavor of the code to be executed. Default is ``"cplx"``
(complex version). Other possible values include ``"real"`` or code-specific
tags for EPW or BGW variants.
util : str, optional
Utility or executable to be run manually (e.g. ``"bands.x"``,
``"dos.x"``, ``"epsilon.x"``). Overrides ``type_run`` if specified.
Returns
-------
None
The method triggers execution of the selected binary and monitors
job completion status.
Notes
-----
- Automatically invokes :class:`PyRun` with the current environment
(``mpirun``, ``ibrun``, ``srun``).
- Can be used in conjunction with :meth:`prepare` to first set up and
then execute runs in an automated workflow.
- Parallelization flags are translated into command-line arguments for
the target code.
- Job progress and results are tracked through :class:`JobStatus`.
Example
-------
>>> epw = EPWpy(system="graphene", env="ibrun", code="/path/to/pw.x")
>>> epw.prepare(type_run="scf")
>>> epw.run(procs=8, type_run="scf")
To run an EPW calculation after SCF:
>>> epw.run(procs=16, type_run="epw", infile="epw.in")
"""
if type_run is None:
type_run = 'none'
print('No executable chosen')
if flow_parallelization is None:
flow_parallelization = []
if flow_procs is None:
flow_procs = {}
self.run_QE = PyRun(procs, self.env, self.code)
self.run_QE.serial = self.run_serial
self.procs = procs
self.run_QE.verbosity = self.verbosity
self.run_QE.proc_set = None
if parallelization is not None:
if (self.verbosity > 1):
print('parallelization chosen: ', parallelization)
self.run_QE.proc_set = parallelization
if util is not None:
print(f'Chosen utility {util}')
self.run_QE.writer = self.writer
if (self.verbosity > 2):
print(f"EPWpy.py def runner(): self.writer.script_params: {self.writer.script_params}")
if infile is None:
infile = type_run
if name is None:
try:
folder = df.def_folds[type_run]
except KeyError:
folder = None
if (folder == 'ph'):
folder = self.ph_fold
if (folder == 'scf'):
folder = self.scf_fold
if (folder == 'nscf'):
folder = self.nscf_fold
if (folder == 'epw'):
folder = self.epw_fold
if (type_run=='scf'):
self.set_work()
self.run_QE.run_scf(name=infile, folder=self.scf_fold)
self.set_home()
elif (type_run == 'custom'):
self.set_work()
self.run_QE.run_custom(name=infile, folder=name, util=util)
self.set_home()
elif (type_run == 'nscf'):
self.set_work()
self.run_QE.run_nscf(name=infile, folder=self.nscf_fold)
self.set_home()
elif (type_run == 'bs'):
self.set_work()
self.run_QE.run_bs(name=infile, folder=self.bs_fold)
self.set_home()
elif (type_run == 'ph'):
self.set_work()
self.run_QE.run_ph(name=infile, folder=self.ph_fold)
self.set_home()
elif (type_run == 'pp'):
self.set_work()
self.run_QE.run_pp(name=infile, folder=folder)
self.set_home()
elif (type_run == 'epw1'):
self.set_work()
self.run_QE.run_epw1(name=infile, folder=self.epw_fold)
self.set_home()
elif (type_run == 'epw2'):
self.set_work()
self.run_QE.run_epw2(name=infile, folder=self.epw_fold)
self.set_home()
elif (type_run == 'epw3'):
self.set_work()
self.run_QE.run_epw3(name=infile, folder=self.epw_fold)
self.set_home()
elif (type_run == 'q2r'):
self.set_work()
self.run_QE.run_q2r(name=infile, folder=self.ph_fold)
self.set_home()
elif (type_run == 'dvscf_q2r'):
self.set_work()
self.run_QE.run_dvscf_q2r(name=infile, folder=folder)
self.set_home()
elif (type_run == 'postahc'):
self.set_work()
self.run_QE.run_postahc(name=infile, folder=folder)
self.set_home()
elif (type_run == 'zg'):
self.set_work()
self.run_QE.run_zg(name=infile)
self.set_home()
elif (type_run == 'eps'):
self.set_work()
self.run_QE.run_eps(name=infile, folder=folder)
self.set_home()
elif (type_run == 'nscf_tetra'):
self.set_work()
self.run_QE.run_nscf_tetra(folder=folder)
self.set_home()
elif (type_run == 'bands'):
self.set_work()
self.run_QE.run_bands(name=infile, folder=folder)
self.set_home()
elif (type_run == 'nscf2supercond'):
self.set_work()
self.run_QE.run_nscf2supercond(name=infile, folder=folder)
self.set_home()
elif (type_run == 'matdyn'):
self.set_work()
self.run_QE.run_matdyn(name=infile, folder=folder)
self.set_home()
elif (type_run == 'dynmat'):
self.set_work()
self.run_QE.run_dynmat(name=infile, folder=folder)
self.set_home()
elif (type_run == 'phdos'):
self.set_work()
self.run_QE.run_phdos(name=infile, folder=folder)
self.set_home()
elif (type_run == 'dos'):
if name is None:
folder = self.nscf_fold
self.set_work()
self.run_QE.run_dos(name=infile, folder=folder)
self.set_home()
elif (type_run == 'pdos'):
if name is None:
folder = self.nscf_fold
self.set_work()
self.run_QE.run_pdos(name=infile, folder=folder)
self.set_home()
elif (type_run == 'fbw'):
self.set_work()
self.run_QE.run_fbw(name=infile, folder=folder)
self.set_home()
elif (type_run == 'fbw_mu'):
self.set_work()
self.run_QE.run_fbw_mu(name=infile, folder=folder)
self.set_home()
elif (type_run == 'nesting'):
self.set_work()
self.run_QE.run_nesting(name=infile, folder=folder)
self.set_home()
elif (type_run == 'phselfen'):
self.set_work()
self.run_QE.run_phselfen(name=infile, folder=folder)
self.set_home()
elif (type_run == 'epw_outerbands'):
self.set_work()
self.run_QE.run_epw_outerbands(name=infile, folder=folder)
self.set_home()
elif (type_run == 'wannier'):
self.set_work()
self.run_QE.run_wannier(name=self.prefix, folder=folder)
self.set_home()
elif (type_run == 'GW'):
self.set_work()
self.run_QE.run_nscf(folder='./GW/wfn', name='wfn')
self.run_QE.run_pw2bgw(folder='./', name=self.prefix)
self.set_home()
self.set_work()
self.run_QE.run_nscf(folder='./GW/wfnq', name='wfnq')
self.run_QE.run_pw2bgw(folder='./', name=self.prefix)
self.set_home()
self.set_work()
self.run_QE.run_nscf(folder='./GW/wfnfi', name='wfnfi')
self.run_QE.run_pw2bgw(folder='./', name=self.prefix)
self.set_home()
self.set_work()
self.run_QE.run_nscf(folder='./GW/wfnfiq', name='wfnfiq')
self.run_QE.run_pw2bgw(folder='./', name=self.prefix)
self.set_home()
elif (type_run == 'epsilon'):
self.set_work()
self.run_QE.run_epsilon(folder='./GW/epsilon', name='epsilon', flavor=flavor)
self.set_home()
elif (type_run == 'sigma'):
self.set_work()
self.run_QE.run_sigma(folder='./GW/sigma', name='sigma', flavor=flavor)
self.set_home()
elif (type_run == 'sigma2wan'):
self.set_work()
self.run_QE.run_sig2wan(folder='./GW/sigma', name='sig2wan')
self.set_home()
elif (type_run == 'kernel'):
self.set_work()
self.run_QE.run_kernel(folder='./GW/kernel', name='kernel', flavor=flavor)
self.set_home()
elif (type_run == 'absorption'):
self.set_work()
self.run_QE.run_absorption(folder='./GW/absorption', name='absorption', flavor=flavor)
self.set_home()
elif (type_run == 'transport'):
#self.fm = flow_manager(self)
self.flow_parallelization = flow_parallelization
if (len(self.flow_parallelization) == 0):
self.flow_parallelization=[None, None, None]
self.flow_procs = {}
for key in ['scf','nscf','ph','epw1','epw2']:
if (key in flow_procs.keys()):
self.flow_procs[key] = int(flow_procs[key])
else:
self.flow_procs[key] = self.procs
self.set_work()
self.transport_flow()
self.set_home()
elif (type_run == 'optics'):
#self.fm = flow_manager(self)
self.flow_parallelization = flow_parallelization
if (len(self.flow_parallelization) == 0):
self.flow_parallelization=[None, None, None]
self.flow_procs = {}
for key in ['scf','nscf','ph','epw1','epw2']:
if (key in flow_procs.keys()):
self.flow_procs[key] = int(flow_procs[key])
else:
self.flow_procs[key] = self.procs
self.set_work()
self.optics_flow()
self.set_home()
