flowtk Package

flowtk Package

abipy.flowtk.flow_main(main)

This decorator is used to decorate main functions producing Flows. It adds the initialization of the logger and an argument parser that allows one to select the loglevel, the workdir of the flow as well as the YAML file with the parameters of the TaskManager. The main function shall have the signature:

main(options)

where options in the container with the command line options generated by ArgumentParser.

Parameters:

main – main function.

abipy.flowtk.build_flow_main_parser()

Build and return the parser used in the abipy/data/runs scripts.

abiinspect Module

This module provides objects to inspect the status of the Abinit tasks at run-time. by extracting information from the main output file (text format).

abipy.flowtk.abiinspect.straceback() → str

Returns a string with the traceback.

abipy.flowtk.abiinspect.plottable_from_outfile(filepath: str)

Factory function that returns a plottable object by inspecting the main output file of abinit Returns None if it is not able to detect the class to instantiate.

class abipy.flowtk.abiinspect.ScfCycle(fields: dict)

Bases: Mapping

It essentially consists of a dictionary mapping string to list of floats containing the data at the different iterations.

MAGIC = 'Must be defined by the subclass.'

to_string(verbose=0) → str

String representation.

property last_iteration: dict

Returns a dictionary with the values of the last iteration.

classmethod from_file(filepath: str) → ScfCycle

Read the first occurrence of ScfCycle from file.

classmethod from_stream(stream) → ScfCycle | None

Read the first occurrence of ScfCycle from stream.

Returns:

None if no ScfCycle entry is found.

plot(ax_list=None, fontsize=8, **kwargs) → Any

Uses matplotlib to plot the evolution of the SCF cycle.

Parameters:

• ax_list – List of axes. If None a new figure is produced.

• fontsize – legend fontsize.

• kwargs – keyword arguments are passed to ax.plot

Returns: matplotlib figure

Keyword arguments controlling the display of the figure:

kwargs	Meaning
title	Title of the plot (Default: None).
show	True to show the figure (default: True).
savefig	“abc.png” or “abc.eps” to save the figure to a file.
size_kwargs	Dictionary with options passed to fig.set_size_inches e.g. size_kwargs=dict(w=3, h=4)
tight_layout	True to call fig.tight_layout (default: False)
ax_grid	True (False) to add (remove) grid from all axes in fig. Default: None i.e. fig is left unchanged.
ax_annotate	Add labels to subplots e.g. (a), (b). Default: False
fig_close	Close figure. Default: False.
plotly	Try to convert mpl figure to plotly.

plotly(fontsize=12, **kwargs)

Uses plotly to plot the evolution of the SCF cycle.

Parameters:

• fontsize – legend fontsize.

• kwargs – keyword arguments are passed to go.Scatter.

Returns: plotly figure

Keyword arguments controlling the display of the figure: ================ ==================================================================== kwargs Meaning ================ ==================================================================== title Title of the plot (Default: None). show True to show the figure (default: True). hovermode True to show the hover info (default: False) savefig “abc.png” , “abc.jpeg” or “abc.webp” to save the figure to a file. write_json Write plotly figure to write_json JSON file.

Inside jupyter-lab, one can right-click the write_json file from the file menu and open with “Plotly Editor”. Make some changes to the figure, then use the file menu to save the customized plotly plot. Requires jupyter labextension install jupyterlab-chart-editor. See https://github.com/plotly/jupyterlab-chart-editor

renderer (str or None (default None)) –

A string containing the names of one or more registered renderers (separated by ‘+’ characters) or None. If None, then the default renderers specified in plotly.io.renderers.default are used. See https://plotly.com/python-api-reference/generated/plotly.graph_objects.Figure.html

config (dict) A dict of parameters to configure the figure. The defaults are set in plotly.js. chart_studio True to push figure to chart_studio server. Requires authenticatios.

Default: False.

template Plotly template. See https://plotly.com/python/templates/

[“plotly”, “plotly_white”, “plotly_dark”, “ggplot2”,

“seaborn”, “simple_white”, “none”]

Default is None that is the default template is used.

class abipy.flowtk.abiinspect.GroundStateScfCycle(fields: dict)

Bases: ScfCycle

Result of the Ground State self-consistent cycle.

MAGIC = 'iter   Etot(hartree)'

property last_etotal: float

The total energy at the last iteration.

class abipy.flowtk.abiinspect.D2DEScfCycle(fields: dict)

Bases: ScfCycle

Result of the Phonon self-consistent cycle.

MAGIC = 'iter   2DEtotal(Ha)'

property last_etotal: float

The 2-nd order derivative of the energy at the last iteration.

class abipy.flowtk.abiinspect.PhononScfCycle(fields: dict)

Bases: D2DEScfCycle

Iterations of the DFPT SCF cycle for phonons.

class abipy.flowtk.abiinspect.CyclesPlotter

Bases: object

Relies on the plot method of cycle objects to build multiple subfigures.

items()

To iterate over (label, cycle).

add_label_cycle(label, cycle)

Add new cycle to the plotter with label label.

combiplot(fontsize=8, **kwargs) → Any

Compare multiple cycels on a grid: one subplot per quantity, all cycles on the same subplot.

Parameters:

fontsize – Legend fontsize.

Keyword arguments controlling the display of the figure:

kwargs	Meaning
title	Title of the plot (Default: None).
show	True to show the figure (default: True).
savefig	“abc.png” or “abc.eps” to save the figure to a file.
size_kwargs	Dictionary with options passed to fig.set_size_inches e.g. size_kwargs=dict(w=3, h=4)
tight_layout	True to call fig.tight_layout (default: False)
ax_grid	True (False) to add (remove) grid from all axes in fig. Default: None i.e. fig is left unchanged.
ax_annotate	Add labels to subplots e.g. (a), (b). Default: False
fig_close	Close figure. Default: False.
plotly	Try to convert mpl figure to plotly.

slideshow(**kwargs)

Produce slides show of the different cycles. One plot per cycle.

class abipy.flowtk.abiinspect.Relaxation(cycles: list[GroundStateScfCycle])

Bases: Iterable

A list of GroundStateScfCycle objects.

Note

Forces, stresses and crystal structures are missing. This object is mainly used to analyze the behavior of the Scf cycles during the structural relaxation. A more powerful and detailed analysis can be obtained by using the HIST.nc file.

to_string(verbose: int = 0) → str

String representation.

classmethod from_file(filepath: str) → Relaxation | None

Initialize the object from the Abinit main output file.

classmethod from_stream(stream) → Relaxation | None

Extract data from stream. Returns None if some error occurred.

history()

dictionary of lists with the evolution of the data as function of the relaxation step.

slideshow(**kwargs)

Uses matplotlib to plot the evolution of the structural relaxation.

Parameters:

ax_list – List of axes. If None a new figure is produced.

Returns: matplotlib figure

plot(ax_list=None, fontsize=8, **kwargs) → Any

Plot relaxation history i.e. the results of the last iteration of each SCF cycle.

Parameters:

• ax_list – List of axes. If None a new figure is produced.

• fontsize – legend fontsize.

• kwargs – keyword arguments are passed to ax.plot

Returns: matplotlib figure

Keyword arguments controlling the display of the figure:

kwargs	Meaning
title	Title of the plot (Default: None).
show	True to show the figure (default: True).
savefig	“abc.png” or “abc.eps” to save the figure to a file.
size_kwargs	Dictionary with options passed to fig.set_size_inches e.g. size_kwargs=dict(w=3, h=4)
tight_layout	True to call fig.tight_layout (default: False)
ax_grid	True (False) to add (remove) grid from all axes in fig. Default: None i.e. fig is left unchanged.
ax_annotate	Add labels to subplots e.g. (a), (b). Default: False
fig_close	Close figure. Default: False.
plotly	Try to convert mpl figure to plotly.

exception abipy.flowtk.abiinspect.YamlTokenizerError

Bases: Exception

Exceptions raised by YamlTokenizer.

class abipy.flowtk.abiinspect.YamlTokenizer(filename: str)

Bases: Iterator

Provides context-manager support so you can use it in a with statement.

Error

alias of YamlTokenizerError

close() → None

Close the stream.

seek(offset[, whence]) → None.  Move to new file position.

Argument offset is a byte count. Optional argument whence defaults to 0 (offset from start of file, offset should be >= 0); other values are 1 (move relative to current position, positive or negative), and 2 (move relative to end of file, usually negative, although many platforms allow seeking beyond the end of a file). If the file is opened in text mode, only offsets returned by tell() are legal. Use of other offsets causes undefined behavior. Note that not all file objects are seekable.

next()

Returns the first YAML document in stream.

Warning

Assume that the YAML document are closed explicitely with the sentinel ‘…’

all_yaml_docs() → list

Returns a list with all the YAML docs found in stream. Seek the stream before returning.

Warning

Assume that all the YAML docs (with the exception of the last one) are closed explicitely with the sentinel ‘…’

next_doc_with_tag(doc_tag: str)

Returns the next document with the specified tag. Empty string is no doc is found.

all_docs_with_tag(doc_tag: str) → list

Returns all the documents with the specified tag.

abipy.flowtk.abiinspect.yaml_read_kpoints(filename: str, doc_tag: str = '!Kpoints') → ndarray

Read the K-points from file. Return numpy array

abipy.flowtk.abiinspect.yaml_read_irred_perts(filename: str, doc_tag='!IrredPerts') → list[AttrDict]

Read the list of irreducible perturbations from file.

class abipy.flowtk.abiinspect.YamlDoc(text: str, lineno: int, tag=None)

Bases: object

Handy object that stores that YAML document, its main tag and the position inside the file.

text

lineno

tag

property text_notag: str

Returns the YAML text without the tag. Useful if we don’t have any constructor registered for the tag (we used the tag just to locate the document).

as_dict() → dict

Use Yaml to parse the text (without the tag) and returns a dictionary.

abiobjects Module

class abipy.flowtk.abiobjects.LujForSpecie(l, u, j, unit)

Bases: LdauForSpecie

This object stores the value of l, u, j used for a single atomic specie.

class abipy.flowtk.abiobjects.LdauParams(usepawu, structure)

Bases: object

This object stores the parameters for LDA+U calculations with the PAW method It facilitates the specification of the U-J parameters in the Abinit input file. (see to_abivars). The U-J operator will be applied only on the atomic species that have been selected by calling lui_for_symbol.

To setup the Abinit variables for a LDA+U calculation in NiO with a U value of 5 eV applied on the nickel atoms:

luj_params = LdauParams(usepawu=1, structure=nio_structure)
# Apply U-J on Ni only.
u = 5.0
luj_params.luj_for_symbol("Ni", l=2, u=u, j=0.1*u, unit="eV")

print(luj_params.to_abivars())

property symbols_by_typat: list

luj_for_symbol(symbol, l, u, j, unit='eV') → None

Parameters:

• symbol – Chemical symbol of the atoms on which LDA+U should be applied.

• l – Angular momentum.

• u – Value of U.

• j – Value of J.

• unit – Energy unit of U and J.

to_abivars() → dict

Returns a dict with the Abinit variables.

class abipy.flowtk.abiobjects.LexxParams(structure)

Bases: object

This object stores the parameters for local exact exchange calculations with the PAW method It facilitates the specification of the LEXX parameters in the Abinit input file. (see to_abivars). The LEXX operator will be applied only on the atomic species that have been selected by calling lexx_for_symbol.

To perform a LEXX calculation for NiO in which the LEXX is computed only for the l=2 channel of the nickel atoms:

lexx_params = LexxParams(nio_structure)
lexx_params.lexx_for_symbol("Ni", l=2)

print(lexc_params.to_abivars())

property symbols_by_typat: list

lexx_for_symbol(symbol, l) → None

Enable LEXX for the given chemical symbol and the angular momentum l

Parameters:

• symbol – Chemical symbol of the atoms on which LEXX should be applied.

• l – Angular momentum.

to_abivars() → dict

Returns a dict with the Abinit variables.

abiphonopy Module

Interface between phonopy and the AbiPy workflow model.

abipy.flowtk.abiphonopy.atoms_from_structure(structure: Structure)

Convert a pymatgen Structure into a phonopy Atoms object.

abipy.flowtk.abiphonopy.structure_from_atoms(atoms) → Structure

Convert a phonopy Atoms object into a abipy Structure.

class abipy.flowtk.abiphonopy.PhonopyWork(workdir=None, manager=None)

Bases: Work

This work computes the inter-atomic force constants with phonopy.

scdims(3)

numpy arrays with the number of cells in the supercell along the three reduced directions

phonon

Phonopy object used to construct the supercells with displaced atoms.

phonopy_tasks

List of ScfTask. Each task compute the forces in one perturbed supercell.

bec_tasks

cpdata2dst

If not None, the work will copy the output results to the outdir of the flow once all_ok is invoked. Note that cpdata2dst must be an absolute path.

classmethod from_gs_input(gs_inp, scdims, phonopy_kwargs=None, displ_kwargs=None) → PhonopyWork

Build the work from an AbinitInput object representing a GS calculations.

Parameters:

• gs_inp – AbinitInput object representing a GS calculation in the initial unit cell.

• scdims – Number of unit cell replicas along the three reduced directions.

• phonopy_kwargs – (Optional) dictionary with arguments passed to Phonopy constructor.

• displ_kwargs – (Optional) dictionary with arguments passed to generate_displacements.

Return: PhonopyWork instance.

on_all_ok()

This method is called once the Work is completed i.e. when all the tasks have reached status S_OK. Here we get the forces from the output files, and we call phonopy to compute the inter-atomic force constants.

class abipy.flowtk.abiphonopy.PhonopyGruneisenWork(workdir=None, manager=None)

Bases: Work

This work computes the Grüneisen parameters with phonopy. The workflow is as follows:

It is necessary to run three phonon calculations. One is calculated at the equilibrium volume and the remaining two are calculated at the slightly larger volume and smaller volume than the equilibrium volume. The unitcells at these volumes have to be fully relaxed under the constraint of each volume.

scdims(3)

numpy arrays with the number of cells in the supercell along the three reduced directions.

classmethod from_gs_input(gs_inp, voldelta, scdims, phonopy_kwargs=None, displ_kwargs=None) → PhonopyGruneisenWork

Build the work from an AbinitInput object representing a GS calculations.

Parameters:

• gs_inp – AbinitInput object representing a GS calculation in the initial unit cell.

• voldelta – Absolute increment for unit cell volume. The three volumes are: [v0 - voldelta, v0, v0 + voldelta] where v0 is taken from gs_inp.structure.

• scdims – Number of unit cell replicas along the three reduced directions.

• phonopy_kwargs – (Optional) dictionary with arguments passed to Phonopy constructor.

• displ_kwargs – (Optional) dictionary with arguments passed to generate_displacements.

Return: PhonopyGruneisenWork instance.

on_all_ok()

This method is called once the Work is completed i.e. when all the tasks have reached status S_OK.

add_phonopy_works_and_build()

Get relaxed structures from the tasks, build Phonopy works with supercells constructed from the new structures, add them to the flow and build new directories.

abitimer Module

dfpt_works Module

Work subclasses related to DFTP.

class abipy.flowtk.dfpt_works.ElasticWork(workdir=None, manager=None)

Bases: Work, MergeDdb

This Work computes the elastic constants and (optionally) the piezoelectric tensor. It consists of response function calculations for:

• rigid-atom elastic tensor

• rigid-atom piezoelectric tensor

• interatomic force constants at gamma

• Born effective charges

The structure is assumed to be already relaxed

Create a Flow for phonon calculations. The flow has one works with:

• 1 GS Task

• 3 DDK Task

• 4 Phonon Tasks (Gamma point)

• 6 Elastic tasks (3 uniaxial + 3 shear strain)

The Phonon tasks and the elastic task will read the DDK produced at the beginning

classmethod from_scf_input(scf_input: AbinitInput, with_relaxed_ion=True, with_piezo=False, with_dde=False, tolerances=None, den_deps=None, manager=None) → ElasticWork

Parameters:

• scf_input

• with_relaxed_ion

• with_piezo

• with_dde – Compute electric field perturbations.

• tolerances – Dict of tolerances

• den_deps

• manager

Similar to from_scf_task, the difference is that this method requires an input for SCF calculation instead of a ScfTask. All the tasks (Scf + Phonon) are packed in a single Work whereas in the previous case we usually have multiple works.

on_all_ok()

This method is called when all the tasks of the Work reach S_OK. Ir runs mrgddb in sequential on the local machine to produce the final DDB file in the outdir of the Work.

class abipy.flowtk.dfpt_works.NscfDdksWork(workdir=None, manager=None)

Bases: Work

This work requires a DEN file and computes the KS energies with a non self-consistent task with a dense k-mesh and empty states. This task is then followed by the computation of the DDK matrix elements with nstep = 1 (the first order change of the wavefunctions is not converged but we only need the matrix elements) Mainly used to prepare optic calculations or other post-processing steps requiring the DDKs.

classmethod from_scf_task(scf_task: ScfTask, ddk_ngkpt, ddk_shiftk, ddk_nband, manager=None) → NscfDdksWork

Build NscfDdksWork from a scf_task.

Parameters:

• scf_task – GS task. Must produce the DEN file required for the NSCF run.

• ddk_ngkpt – k-mesh used for the NSCF run and the non self-consistent DDK tasks.

• ddk_shiftk – k-mesh shifts

• ddk_nband – Number of bands (occupied + empty) used in the NSCF task and the DDKs tasks.

• manager – TaskManager instance. Use default if None.

effmass_works Module

Work subclasses related to effective mass calculations.

class abipy.flowtk.effmass_works.EffMassLineWork(workdir=None, manager=None)

Bases: Work

Work for the computation of effective masses via finite differences along a k-line. Useful for cases such as NC+SOC where DFPT is not implemented or if one is interested in the non-parabolic behaviour of the energy dispersion.

Inheritance Diagram

classmethod from_scf_input(scf_input: AbinitInput, k0_list, step=0.01, npts=15, red_dirs=[[1, 0, 0], [0, 1, 0], [0, 0, 1]], ndivsm=-20, cart_dirs=None, den_node=None, manager=None) → EffMassLineWork

Build the Work from an abipy.abio.inputs.AbinitInput representing a GS-SCF calculation.

Parameters:

• scf_input – abipy.abio.inputs.AbinitInput for GS-SCF used as template to generate the other inputs.

• k0_list – List with the reduced coordinates of the k-points where effective masses are wanted.

• step – Step for finite difference in Angstrom^-1

• npts – Number of points sampled around each k-point for each direction.

• red_dirs – List of reduced directions used to generate the segments passing through the k-point

• cart_dirs – List of Cartesian directions used to generate the segments passing through the k-point

• den_node – Path to the DEN file or Task object producing a DEN file. Can be used to avoid the initial SCF calculation if a DEN file is already available. If None, a GS calculation is performed.

• ndivsm – if > 0, it’s the number of divisions for the smallest segment of the path (Abinit variable). if < 0, it’s interpreted as the pymatgen line_density parameter in which the number of points in the segment is proportional to its length. Typical value: -20.

• manager – abipy.flowtk.tasks.TaskManager instance. Use default if None.

class abipy.flowtk.effmass_works.EffMassDFPTWork(workdir=None, manager=None)

Bases: Work

Work for the computation of effective masses with DFPT. Requires explicit list of k-points and range of bands.

Inheritance Diagram

classmethod from_scf_input(scf_input: AbinitInput, k0_list, effmass_bands_f90, ngfft=None, den_node=None, manager=None) → EffMassDFPTWork

Build the Work from an abipy.abio.inputs.AbinitInput representing a GS-SCF calculation.

Parameters:

• scf_input – abipy.abio.inputs.AbinitInput for GS-SCF used as template to generate the other inputs.

• k0_list – List with the reduced coordinates of the k-points where effective masses are wanted.

• effmass_bands_f90 – (nkpt, 2) array with band range for effmas computation. WARNING: Assumes Fortran convention with indices starting from 1.

• ngfft – FFT divisions (3 integers). Used to enforce the same FFT mesh in the NSCF run as the one used for GS.

• den_node – Path to the DEN file or Task object producing a DEN file. Can be used to avoid the initial SCF calculation if a DEN file is already available. If None, a GS calculation is performed.

• manager – abipy.flowtk.tasks.TaskManager instance. Use default if None.

class abipy.flowtk.effmass_works.EffMassAutoDFPTWork(workdir=None, manager=None)

Bases: Work

Work for the automatic computation of effective masses with DFPT. Band extrema are automatically detected by performing a NSCF calculation along a high-symmetry k-path with ndivsm. Requires more computation that EffMassWork since input variables (kpoints and band range) are computed at runtime.

Inheritance Diagram

classmethod from_scf_input(scf_input: AbinitInput, ndivsm=15, tolwfr=1e-20, den_node=None, manager=None) → EffMassAutoDFPTWork

Build the Work from an abipy.abio.inputs.AbinitInput representing a GS-SCF calculation.

Parameters:

• scf_input – abipy.abio.inputs.AbinitInput for GS-SCF used as template to generate the other inputs.

• ndivsm – if > 0, it’s the number of divisions for the smallest segment of the path (Abinit variable). if < 0, it’s interpreted as the pymatgen line_density parameter in which the number of points in the segment is proportional to its length. Typical value: -20. This option is the recommended one if the k-path contains two high symmetry k-points that are very close as ndivsm > 0 may produce a very large number of wavevectors.

• tolwfr – Tolerance on residuals for NSCF calculation

• den_node – Path to the DEN file or Task object producing a DEN file. Can be used to avoid the initial SCF calculation if a DEN file is already available. If None, a GS calculation is performed.

• manager – abipy.flowtk.tasks.TaskManager instance. Use default if None.

on_all_ok()

This method is called once the Work is completed i.e. when all tasks have reached status S_OK. Here, we read the band structure from GSR to find the position of the band edges and use these values to generate a new work for effective masses with DFPT.

class abipy.flowtk.effmass_works.FrohlichZPRFlow(workdir, manager=None, pickle_protocol=-1, remove=False)

Bases: Flow

Inheritance Diagram

classmethod from_scf_input(workdir: str, scf_input: AbinitInput, ddb_node=None, ndivsm=15, tolwfr=1e-20, metadata=None, manager=None) → FrohlichZPRFlow

Build the Flow from an abipy.abio.inputs.AbinitInput representing a GS-SCF calculation. Final results are stored in the “zprfrohl_results.json” in the outdata directory of the flow.

Parameters:

• workdir – Working directory.

• scf_input – abipy.abio.inputs.AbinitInput for GS-SCF used as template to generate the other inputs.

• ddb_node – Path to an external DDB file that is used to avoid the calculation of BECS/eps_inf and phonons. If None, a DFPT calculation is automatically performed by the flow.

• ndivsm – Number of divisions used to sample the smallest segment of the k-path.

• tolwfr – Tolerance on residuals for NSCF calculation

• manager – abipy.flowtk.tasks.TaskManager instance. Use default if None.

• metadata – Dictionary with metadata addeded to the final JSON file.

on_all_ok()

This method is called when all the works in the flow have reached S_OK. This method shall return True if the calculation is completed or False if the execution should continue due to side-effects such as adding a new work to the flow.

finalize()

This method is called when the flow is completed. Here we write the final results in the “zprfrohl_results.json” file in the outdata directory of the flow

eph_flows Module

Flows for electron-phonon calculations (high-level interface)

class abipy.flowtk.eph_flows.EphPotFlow(workdir, manager=None, pickle_protocol=-1, remove=False)

Bases: Flow

This flow computes the e-ph scattering potentials on a q-mesh defined by ngqpt and a list of q-points (usually a q-path) specified by the user. The DFPT potentials on the q-mesh are merged in the DVDB located in the outdata of the second work while the DFPT potentials on the q-path are merged in the DVDB located in the outdata of the third work. These DVDB files are then passed to the EPH code to compute the average over the unit cell of the periodic part of the scattering potentials as a function of q. Results are stored in the V1QAVG.nc files of the outdata of the tasks in the fourth work.

classmethod from_scf_input(workdir: str, scf_input: AbinitInput, ngqpt, qbounds, ndivsm=5, with_becs=True, with_quad=True, dvdb_add_lr_list=(0, 1, 2), ddb_filepath=None, dvdb_filepath=None, ddk_tolerance=None, prepgkk=0, manager=None) → EphPotFlow

Build the flow from an input file representing a GS calculation.

Parameters:

• workdir – Working directory.

• scf_input – Input for the GS SCF run.

• ngqpt – 3 integers defining the q-mesh.

• qbounds – List of boundaries defining the q-path used for the computation of the GKQ files. The q-path is automatically generated using ndivsm and the reciprocal-space metric. If ndivsm is 0, the code assumes that qbounds contains the full list of q-points and no pre-processing is performed.

• ndivsm – Number of points in the smallest segment of the path defined by qbounds. Use 0 to pass full list of q-points.

• with_becs – Activate calculation of Electric field and Born effective charges.

• with_quad – Activate calculation of dynamical quadrupoles. Require with_becs Note that only selected features are compatible with dynamical quadrupoles. Please consult <https://docs.abinit.org/topics/longwave/>

• dvdb_add_lr_list – List of dvdb_add_lr values to consider in the interpolation.

• ddb_filepath – Paths to the DDB/DVDB files that will be used to bypass the DFPT computation on the ngqpt mesh.

• dvdb_filepath – Paths to the DDB/DVDB files that will be used to bypass the DFPT computation on the ngqpt mesh.

• ddk_tolerance – dict {“varname”: value} with the tolerance used in the DDK run if with_becs.

• prepgkk – 1 to activate computation of all 3 * natom perts (debugging option).

• manager – abipy.flowtk.tasks.TaskManager object.

class abipy.flowtk.eph_flows.GkqPathFlow(workdir, manager=None, pickle_protocol=-1, remove=False)

Bases: Flow

This flow computes the gkq e-ph matrix elements <k+q|Delta V_q|k> for a list of q-points (usually a q-path). The results are stored in the GKQ.nc files for the different q-points. These files can be used to analyze the behaviour of the e-ph matrix elements as a function of qpts with the objects provided by the abipy.eph.gkq module. It is also possible to compute the e-ph matrix elements using the interpolated DFPT potentials if test_ft_interpolation is set to True.

classmethod from_scf_input(workdir: str, scf_input: AbinitInput, ngqpt, qbounds, ndivsm=5, with_becs=True, with_quad=True, dvdb_add_lr_list=(0, 1, 2), ddb_filepath=None, dvdb_filepath=None, ddk_tolerance=None, test_ft_interpolation=False, prepgkk=0, manager=None) → GkqPathFlow

Build the flow from an input file representing a GS calculation.

Parameters:

• workdir – Working directory.

• scf_input – Input for the GS SCF run.

• ngqpt – 3 integers defining the q-mesh.

• qbounds – List of boundaries defining the q-path used for the computation of the GKQ files. The q-path is automatically generated using ndivsm and the reciprocal-space metric. If ndivsm is 0, the code assumes that qbounds contains the full list of q-points and no pre-processing is performed.

• ndivsm – Number of points in the smallest segment of the path defined by qbounds. Use 0 to pass list of q-points.

• with_becs – Activate calculation of Electric field and Born effective charges.

• with_quad – Activate calculation of dynamical quadrupoles. Require with_becs Note that only selected features are compatible with dynamical quadrupoles. Please consult <https://docs.abinit.org/topics/longwave/>

• dvdb_add_lr_list – List of dvdb_add_lr values to consider in the interpolation.

• ddb_filepath – Paths to the DDB/DVDB files that will be used to bypass the DFPT computation on the ngqpt mesh.

• dvdb_filepath – Paths to the DDB/DVDB files that will be used to bypass the DFPT computation on the ngqpt mesh.

• ddk_tolerance – dict {“varname”: value} with the tolerance used in the DDK run if with_becs.

• test_ft_interpolation – True to add an extra Work in which the GKQ files are computed using the interpolated DFPT potentials and the q-mesh defined by ngqpt. The quality of the interpolation depends on the convergence of the BECS, epsinf and ngqpt. and the treatment of the LR part of the e-ph scattering potentials.

• prepgkk – 1 to activate computation of all 3 * natom perts (debugging option).

• manager – abipy.flowtk.tasks.TaskManager object.

events Module

This module defines the events signaled by abinit during the execution. It also provides a parser to extract these events form the main output file and the log file.

class abipy.flowtk.events.EventsParser

Bases: object

Parses the output or the log file produced by ABINIT and extract the list of events.

Error

alias of EventsParserError

parse(filename: str, verbose: int = 0) → EventReport

Parse the given file. Return EventReport.

report_exception(filename, exc) → EventReport

This method is used when self.parser raises an Exception so that we can report a customized EventReport object with info the exception.

abipy.flowtk.events.get_event_handler_classes(categories=None)

Return the list of handler classes.

class abipy.flowtk.events.ScfConvergenceWarning(src_file: str, src_line: int, message: str)

Bases: AbinitCriticalWarning

Warning raised when the GS SCF cycle did not converge.

yaml_tag: Any = '!ScfConvergenceWarning'

class abipy.flowtk.events.NscfConvergenceWarning(src_file: str, src_line: int, message: str)

Bases: AbinitCriticalWarning

Warning raised when the GS NSCF cycle did not converge.

yaml_tag: Any = '!NscfConvergenceWarning'

class abipy.flowtk.events.RelaxConvergenceWarning(src_file: str, src_line: int, message: str)

Bases: AbinitCriticalWarning

Warning raised when the structural relaxation did not converge.

yaml_tag: Any = '!RelaxConvergenceWarning'

class abipy.flowtk.events.Correction(handler, actions, event, reset=False)

Bases: MSONable

as_dict() → dict

A JSON serializable dict representation of an object.

classmethod from_dict(d: dict) → Correction

Parameters:

d – Dict representation.

Returns:

MSONable class.

class abipy.flowtk.events.DilatmxError(src_file: str, src_line: int, message: str)

Bases: AbinitError

This Error occurs in variable cell calculations when the increase in the unit cell volume is too large.

yaml_tag: Any = '!DilatmxError'

class abipy.flowtk.events.DilatmxErrorHandler(max_dilatmx=1.3)

Bases: ErrorHandler

Handle DilatmxError. Abinit produces a netcdf file with the last structure before aborting The handler changes the structure in the input with the last configuration and modify the value of dilatmx.

event_class

alias of DilatmxError

can_change_physics = False

as_dict() → dict

Basic implementation of as_dict if __init__ has no arguments. Subclasses may need to overwrite.

classmethod from_dict(d: dict) → DilatmxErrorHandler

Basic implementation of from_dict if __init__ has no arguments. Subclasses may need to overwrite.

handle_task_event(task, event)

Method to handle Abinit events.

Parameters:

• task – Task object.

• event – AbinitEvent found in the log file.

Returns:

0 if no action has been applied, 1 if the problem has been fixed.

handle_input_event(abi_input, outdir, event)

flows Module

A Flow is a container for Works, and works consist of tasks. Flows are the final objects that can be dumped directly to a pickle file on disk Flows are executed using abirun (abipy).

class abipy.flowtk.flows.Flow(workdir, manager=None, pickle_protocol=-1, remove=False)

Bases: Node, NodeContainer, MSONable

This object is a container of work. Its main task is managing the possible inter-dependencies among the work and the creation of dynamic workflows that are generated by callbacks registered by the user.

creation_date

String with the creation_date

pickle_protocol

Protocol for Pickle database (default: -1 i.e. latest protocol)

Important methods for constructing flows:

register_work: register (add) a work to the flow resister_task: register a work that contains only this task returns the work allocate: propagate the workdir and manager of the flow to all the registered tasks build: build_and_pickle_dump:

VERSION = '0.1'

PICKLE_FNAME = '__AbinitFlow__.pickle'

Error

alias of FlowError

Results

alias of FlowResults

classmethod from_inputs(workdir, inputs, manager=None, pickle_protocol=-1, task_class=<class 'abipy.flowtk.tasks.ScfTask'>, work_class=<class 'abipy.flowtk.works.Work'>, remove=False) → Flow

Construct a simple flow from a list of inputs. The flow contains a single Work with tasks whose class is given by task_class.

Warning

Don’t use this interface if you have dependencies among the tasks.

Parameters:

• workdir – String specifying the directory where the works will be produced.

• inputs – List of inputs.

• manager – abipy.flowtk.tasks.TaskManager object responsible for the submission of the jobs. If manager is None, the object is initialized from the yaml file located either in the working directory or in the user configuration dir.

• pickle_protocol – Pickle protocol version used for saving the status of the object. -1 denotes the latest version supported by the python interpreter.

• task_class – The class of the abipy.flowtk.tasks.Task.

• work_class – The class of the abipy.flowtk.works.Work.

• remove – attempt to remove working directory workdir if directory already exists.

classmethod as_flow(obj: Any) → Flow

Convert obj into a Flow. Accepts filepath, dict, or Flow object.

as_dict(**kwargs) → dict

JSON serialization, note that we only need to save a string with the working directory since the object will be reconstructed from the pickle file located in workdir

to_dict(**kwargs) → dict

JSON serialization, note that we only need to save a string with the working directory since the object will be reconstructed from the pickle file located in workdir

classmethod from_dict(d: dict, **kwargs) → Flow

Reconstruct the flow from the pickle file.

classmethod temporary_flow(workdir=None, manager=None) → Flow

Return a Flow in a temporary directory. Useful for unit tests.

set_status(status, msg: str) → None

Set and return the status of the flow

Parameters:

• status – Status object or string representation of the status

• msg – string with human-readable message used in the case of errors.

set_workdir(workdir: str, chroot=False) → None

Set the working directory. Cannot be set more than once unless chroot is True

reload() → None

Reload the flow from the pickle file. Used when we are monitoring the flow executed by the scheduler. In this case, indeed, the flow might have been changed by the scheduler and we have to reload the new flow in memory.

classmethod pickle_load(filepath: str, spectator_mode=True, remove_lock=False) → Flow

Loads the object from a pickle file and performs initial setup.

Parameters:

• filepath – Filename or directory name. It filepath is a directory, we scan the directory tree starting from filepath and we read the first pickle database. Raise RuntimeError if multiple databases are found.

• spectator_mode – If True, the nodes of the flow are not connected by signals. This option is usually used when we want to read a flow in read-only mode and we want to avoid callbacks that can change the flow.

• remove_lock – True to remove the file lock if any (use it carefully).

classmethod from_file(filepath: str, spectator_mode=True, remove_lock=False) → Flow

Loads the object from a pickle file and performs initial setup.

Parameters:

• filepath – Filename or directory name. It filepath is a directory, we scan the directory tree starting from filepath and we read the first pickle database. Raise RuntimeError if multiple databases are found.

• spectator_mode – If True, the nodes of the flow are not connected by signals. This option is usually used when we want to read a flow in read-only mode and we want to avoid callbacks that can change the flow.

• remove_lock – True to remove the file lock if any (use it carefully).

classmethod pickle_loads(s: str) → Flow

Reconstruct the flow from a string.

change_manager(new_manager: Any) → TaskManager

Change the manager at runtime.

get_panel(**kwargs)

Build panel with widgets to interact with the abipy.flowtk.flows.Flow either in a notebook or in panel app.

set_pyfile(pyfile: str) → None

Set the path of the python script used to generate the flow.

property pyfile: str | None

Absolute path of the python script used to generate the flow. None if unset. Set by set_pyfile

property pid_file: str

The path of the pid file created by PyFlowScheduler.

property has_scheduler: bool

True if there’s a scheduler running the flow.

check_pid_file() → int

This function checks if we are already running the abipy.flowtk.flows.Flow with a PyFlowScheduler. Raises: Flow.Error if the pid file of the scheduler exists.

property pickle_file: str

The path of the pickle file.

property mongo_id

validate_json_schema()

Validate the JSON schema. Return list of errors.

get_mongo_info()

Return a JSON dictionary with information on the flow. Mainly used for constructing the info section in FlowEntry. The default implementation is empty. Subclasses must implement it

mongo_assimilate()

This function is called by client code when the flow is completed Return a JSON dictionary with the most important results produced by the flow. The default implementation is empty. Subclasses must implement it

property works: list[Work]

List of abipy.flowtk.works.Work objects contained in self..

property all_ok: bool

True if all the works in the flow have reached S_OK.

property num_tasks: int

Total number of tasks

property errored_tasks: set[Task]

Set errored tasks.

property num_errored_tasks: int

The number of tasks whose status is S_ERROR.

property unconverged_tasks: int

List of unconverged tasks.

property num_unconverged_tasks: int

The number of tasks whose status is S_UNCONVERGED.

property status_counter

Returns a Counter object that counts the number of tasks with given status (use the string representation of the status as key).

property ncores_reserved: int

Returns the number of cores reserved in this moment. A core is reserved if the task is not running but we have submitted the task to the queue manager.

property ncores_allocated: int

Returns the number of cores allocated at this moment. A core is allocated if it’s running a task or if we have submitted a task to the queue manager but the job is still in pending state.

property ncores_used: int

Returns the number of cores used at this moment. A core is used if there’s a job that is running on it.

property has_chrooted: str

Returns a string that evaluates to True if we have changed the workdir for visualization purposes e.g. we are using sshfs. to mount the remote directory where the Flow is located. The string gives the previous workdir of the flow.

chroot(new_workdir: str) → None

Change the workir of the abipy.flowtk.flows.Flow. Mainly used for allowing the user to open the GUI on the local host and access the flow from remote via sshfs.

Note

Calling this method will make the flow go in read-only mode.

groupby_status() → dict

Returns dictionary mapping the task status to the list of named tuples (task, work_index, task_index).

groupby_work_class() → dict

Returns a dictionary mapping the work class to the list of works in the flow

groupby_task_class() → dict

Returns a dictionary mapping the task class to the list of tasks in the flow

iflat_nodes(status=None, op='==', nids=None) → Generator[None]

Generators that produces a flat sequence of nodes. if status is not None, only the tasks with the specified status are selected. nids is an optional list of node identifiers used to filter the nodes.

node_from_nid(nid: int) → Node

Return the node in the Flow with the given nid identifier

iflat_tasks_wti(status=None, op='==', nids=None) → Generator[tuple[Task, int, int]]

Generator to iterate over all the tasks of the Flow. :Yields: (task, work_index, task_index)

If status is not None, only the tasks whose status satisfies the condition (task.status op status) are selected status can be either one of the flags defined in the abipy.flowtk.tasks.Task class (e.g Task.S_OK) or a string e.g “S_OK” nids is an optional list of node identifiers used to filter the tasks.

iflat_tasks(status=None, op='==', nids=None) → Generator[Task]

Generator to iterate over all the tasks of the abipy.flowtk.flows.Flow.

If status is not None, only the tasks whose status satisfies the condition (task.status op status) are selected status can be either one of the flags defined in the abipy.flowtk.tasks.Task class (e.g Task.S_OK) or a string e.g “S_OK” nids is an optional list of node identifiers used to filter the tasks.

abivalidate_inputs() → tuple

Run ABINIT in dry mode to validate all the inputs of the flow.

Returns:

(isok, tuples)

isok is True if all inputs are ok. tuples is List of namedtuple objects, one for each task in the flow. Each namedtuple has the following attributes:

retcode: Return code. 0 if OK. log_file: log file of the Abinit run, use log_file.read() to access its content. stderr_file: stderr file of the Abinit run. use stderr_file.read() to access its content.

Raises:

RuntimeError –

check_dependencies() → None

Test the dependencies of the nodes for possible deadlocks and raise RuntimeError

find_deadlocks()

This function detects deadlocks

Returns:

deadlocks, runnables, running

Return type:

named tuple with the tasks grouped in

check_status(**kwargs) → None

Check the status of the works in self.

Parameters:

• show – True to show the status of the flow.

• kwargs – keyword arguments passed to show_status

property status: Status

Gives the status of the Flow.

fix_abicritical() → int

This function tries to fix critical events originating from ABINIT. Returns the number of tasks that have been fixed.

fix_queue_critical() → int

This function tries to fix critical events originating from the queue submission system.

Returns the number of tasks that have been fixed.

show_info(**kwargs) → None

Print info on the flow i.e. total number of tasks, works, tasks grouped by class.

Example

Task Class Number ———— ——– ScfTask 1 NscfTask 1 ScrTask 2 SigmaTask 6

compare_abivars(varnames, nids=None, wslice=None, printout=False, with_colors=False) → DataFrame

Print the input of the tasks to the given stream.

Parameters:

• varnames – List of Abinit variables. If not None, only the variable in varnames are selected and printed.

• nids – List of node identifiers. By defaults all nodes are shown

• wslice – Slice object used to select works.

• printout – True to print dataframe.

• with_colors – True if task status should be colored.

get_dims_dataframe(nids=None, printout=False, with_colors=False) → DataFrame

Analyze output files produced by the tasks. Print pandas DataFrame with dimensions.

Parameters:

• nids – List of node identifiers. By defaults all nodes are shown

• printout – True to print dataframe.

• with_colors – True if task status should be colored.

write_fix_flow_script() → None

Write python script in the flow workdir that can be used by expert users to change the input variables of the task according to their status.

compare_structures(nids=None, with_spglib=False, what='io', verbose=0, precision=3, printout=False, with_colors=False)

Analyze structures of the tasks (input and output structures if it’s a relaxation task. Print pandas DataFrame

Parameters:

• nids – List of node identifiers. By defaults all nodes are shown

• with_spglib – If True, spglib is invoked to get the spacegroup symbol and number

• what (str) – “i” for input structures, “o” for output structures.

• precision – Floating point output precision (number of significant digits). This is only a suggestion

• printout – True to print dataframe.

• with_colors – True if task status should be colored.

compare_ebands(nids=None, with_path=True, with_ibz=True, with_spglib=False, verbose=0, precision=3, printout=False, with_colors=False) → tuple

Analyze electron bands produced by the tasks. Return pandas DataFrame and abipy.electrons.ebands.ElectronBandsPlotter.

Parameters:

• nids – List of node identifiers. By default, all nodes are shown

• with_path – Select files with ebands along k-path.

• with_ibz – Select files with ebands in the IBZ.

• with_spglib – If True, spglib is invoked to get the spacegroup symbol and number

• precision – Floating point output precision (number of significant digits). This is only a suggestion

• printout – True to print dataframe.

• with_colors – True if task status should be colored.

Return: (df, ebands_plotter)

compare_hist(nids=None, with_spglib=False, verbose=0, precision=3, printout=False, with_colors=False) → tuple

Analyze HIST nc files produced by the tasks. Print pandas DataFrame with final results. Return: (df, hist_plotter)

Parameters:

• nids – List of node identifiers. By defaults all nodes are shown

• with_spglib – If True, spglib is invoked to get the spacegroup symbol and number

• precision – Floating point output precision (number of significant digits). This is only a suggestion

• printout – True to print dataframe.

• with_colors – True if task status should be colored.

show_summary(**kwargs) → None

Print a short summary with the status of the flow and a counter task_status –> number_of_tasks

Parameters:

stream – File-like object, Default: sys.stdout

Example

Status Count ——— ——- Completed 10

<Flow, node_id=27163, workdir=flow_gwconv_ecuteps>, num_tasks=10, all_ok=True

get_dataframe(as_dict=False) → DataFrame

Return pandas dataframe task info or dictionary if as_dict is True. This function should be called after flow.get_status to update the status.

show_status(return_df=False, **kwargs)

Report the status of the works and the status of the different tasks on the specified stream.

Parameters:

• stream – File-like object, Default: sys.stdout

• nids – List of node identifiers. By defaults all nodes are shown

• wslice – Slice object used to select works.

• verbose – Verbosity level (default 0). > 0 to show only the works that are not finalized.

Returns:

task –> dict(report=report, timedelta=timedelta)

Return type:

data_task dictionary with mapping

show_events(status=None, nids=None, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Print the Abinit events (ERRORS, WARNIING, COMMENTS) to stdout

Parameters:

• status – if not None, only the tasks with this status are select

• nids – optional list of node identifiers used to filter the tasks.

• stream – File-like object, Default: sys.stdout

Returns:

task –> report

Return type:

data_task dictionary with mapping

show_corrections(status=None, nids=None, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>) → int

Show the corrections applied to the flow at run-time.

Parameters:

• status – if not None, only the tasks with this status are select.

• nids – optional list of node identifiers used to filter the tasks.

• stream – File-like object, Default: sys.stdout

Return: The number of corrections found.

show_history(status=None, nids=None, full_history=False, metadata=False, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>) → None

Print the history of the flow to stream

Parameters:

• status – if not None, only the tasks with this status are select

• full_history – Print full info set, including nodes with an empty history.

• nids – optional list of node identifiers used to filter the tasks.

• metadata – print history metadata (experimental)

• stream – File-like object, Default: sys.stdout

show_inputs(varnames=None, nids=None, wslice=None, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>) → None

Print the input of the tasks to the given stream.

Parameters:

• varnames – List of Abinit variables. If not None, only the variable in varnames are selected and printed.

• nids – List of node identifiers. By defaults all nodes are shown

• wslice – Slice object used to select works.

• stream – File-like object, Default: sys.stdout

listext(ext, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>) → None

Print to the given stream a table with the list of the output files with the given ext produced by the flow.

select_tasks(nids=None, wslice=None, task_class=None) → list[Task]

Return a list with a subset of tasks.

Parameters:

• nids – List of node identifiers.

• wslice – Slice object used to select works.

• task_class – String or class used to select tasks. Ignored if None.

Note

nids and wslice are mutually exclusive. If no argument is provided, the full list of tasks is returned.

get_task_scfcycles(nids=None, wslice=None, task_class=None, exclude_ok_tasks=False) → list[Task]

Return list of (taks, scfcycle) tuples for all the tasks in the flow with a SCF algorithm e.g. electronic GS-SCF iteration, DFPT-SCF iterations etc.

Parameters:

• nids – List of node identifiers.

• wslice – Slice object used to select works.

• task_class – String or class used to select tasks. Ignored if None.

• exclude_ok_tasks – True if only running tasks should be considered.

Returns:

List of ScfCycle subclass instances.

show_tricky_tasks(verbose=0, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>) → None

Print list of tricky tasks i.e. tasks that have been restarted or launched more than once or tasks with corrections.

Parameters:

• verbose – Verbosity level. If > 0, task history and corrections (if any) are printed.

• stream – File-like object. Default: sys.stdout

inspect(nids=None, wslice=None, **kwargs)

Inspect the tasks (SCF iterations, Structural relaxation …) and produces matplotlib plots.

Parameters:

• nids – List of node identifiers.

• wslice – Slice object used to select works.

• kwargs – keyword arguments passed to task.inspect method.

Note

nids and wslice are mutually exclusive. If nids and wslice are both None, all tasks in self are inspected.

Returns:

List of matplotlib figures.

get_results(**kwargs)

look_before_you_leap() → str

This method should be called before running the calculation to make sure that the most important requirements are satisfied.

Return: string with inconsistencies/errors.

tasks_from_nids(nids) → list[Task]

Return the list of tasks associated to the given list of node identifiers (nids).

Note

Invalid ids are ignored

wti_from_nids(nids) → list[Task]

Return the list of (w, t) indices from the list of node identifiers nids.

open_files(what='o', status=None, op='==', nids=None, editor=None)

Open the files of the flow inside an editor (command line interface).

Parameters:

• what –

  string with the list of characters selecting the file type Possible choices:

  i ==> input_file, o ==> output_file, f ==> files_file, j ==> job_file, l ==> log_file, e ==> stderr_file, q ==> qout_file, all ==> all files.

• status – if not None, only the tasks with this status are select

• op – status operator. Requires status. A task is selected if task.status op status evaluates to true.

• nids – optional list of node identifiers used to filter the tasks.

• editor – Select the editor. None to use the default editor ($EDITOR shell env var)

parse_timing(nids=None)

Parse the timer data in the main output file(s) of Abinit. Requires timopt /= 0 in the input file, usually timopt = -1.

Parameters:

nids – optional list of node identifiers used to filter the tasks.

Return: AbinitTimerParser instance, None if error.

show_abierrors(nids=None, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Write to the given stream the list of ABINIT errors for all tasks whose status is S_ABICRITICAL.

Parameters:

• nids – optional list of node identifiers used to filter the tasks.

• stream – File-like object. Default: sys.stdout

show_qouts(nids=None, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

Write to the given stream the content of the queue output file for all tasks whose status is S_QCRITICAL.

Parameters:

• nids – optional list of node identifiers used to filter the tasks.

• stream – File-like object. Default: sys.stdout

debug(status=None, nids=None, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)

This method is usually used when the flow didn’t completed succesfully It analyzes the files produced the tasks to facilitate debugging. Info are printed to stdout.

Parameters:

• status – If not None, only the tasks with this status are selected

• nids – optional list of node identifiers used to filter the tasks.

• stream – File-like object. Default: sys.stdout

cancel(nids=None) → int

Cancel all the tasks that are in the queue. nids is an optional list of node identifiers used to filter the tasks.

Returns:

Number of jobs cancelled, negative value if error

get_njobs_in_queue(username=None) → int

Returns the number of jobs in the queue, None when the number of jobs cannot be determined.

Parameters:

username – (str) the username of the jobs to count (default is to autodetect)

rmtree(ignore_errors=False, onerror=None) → None

Remove workdir (same API as shutil.rmtree).

rm_and_build() → None

Remove the workdir and rebuild the flow.

build(*args, **kwargs) → None

Make directories and files of the Flow.

build_and_pickle_dump(abivalidate=False)

Build dirs and file of the Flow and save the object in pickle format. Returns 0 if success

Parameters:

abivalidate – If True, all the input files are validate by calling the abinit parser. If the validation fails, ValueError is raise.

pickle_dump()

Save the status of the object in pickle format. Returns 0 if success

pickle_dumps(protocol=None)

Return a string with the pickle representation. protocol selects the pickle protocol. self.pickle_protocol is used if protocol is None

register_task(input: AbinitInput, deps=None, manager=None, task_class=None, append=False) → Work

Utility function that generates a Work made of a single task

Parameters:

• input – abipy.abio.inputs.AbinitInput

• deps – List of Dependency objects specifying the dependency of this node. An empy list of deps implies that this node has no dependencies.

• manager – The abipy.flowtk.tasks.TaskManager responsible for the submission of the task. If manager is None, we use the abipy.flowtk.tasks.TaskManager specified during the creation of the work.

• task_class – Task subclass to instantiate. Default: abipy.flowtk.tasks.AbinitTask

• append – If true, the task is added to the last work (a new Work is created if flow is empty)

Returns:

The generated abipy.flowtk.works.Work for the task, work[0] is the actual task.

new_work(deps=None, manager=None, workdir=None) → Work

Helper function to add a new empty abipy.flowtk.works.Work and add it to the internal list. Client code is responsible for filling the new work.

Parameters:

• deps – List of Dependency objects specifying the dependency of this node. An empy list of deps implies that this node has no dependencies.

• manager – The abipy.flowtk.tasks.TaskManager responsible for the submission of the task. If manager is None, we use the TaskManager specified during the creation of the work.

• workdir – The name of the directory used for the abipy.flowtk.works.Work.

Returns:

The registered abipy.flowtk.works.Work.

register_work(work: Work, deps=None, manager=None, workdir=None) → Work

Register a new abipy.flowtk.works.Work and add it to the internal list, taking into account possible dependencies.

Parameters:

• work – abipy.flowtk.works.Work object.

• deps – List of Dependency objects specifying the dependency of this node. An empy list of deps implies that this node has no dependencies.

• manager – The abipy.flowtk.tasks.TaskManager responsible for the submission of the task. If manager is None, we use the TaskManager specified during the creation of the work.

• workdir – The name of the directory used for the abipy.flowtk.works.Work.

Returns: The registered abipy.flowtk.works.Work.

register_work_from_cbk(cbk_name, cbk_data, deps, work_class, manager=None)

Registers a callback function that will generate the Task of the Work.

Parameters:

• cbk_name – Name of the callback function (must be a bound method of self)

• cbk_data – Additional data passed to the callback function.

• deps – List of Dependency objects specifying the dependency of the work.

• work_class – abipy.flowtk.works.Work class to instantiate.

• manager – The abipy.flowtk.tasks.TaskManager responsible for the submission of the task. If manager is None, we use the TaskManager specified during the creation of the abipy.flowtk.flows.Flow.

Returns:

The abipy.flowtk.works.Work that will be finalized by the callback.

property allocated: int

Numer of allocations. Set by allocate.

allocate(workdir=None, use_smartio: bool = False, build: bool = False) → Flow

Allocate the Flow i.e. assign the workdir and (optionally) the abipy.flowtk.tasks.TaskManager to the different tasks in the Flow.

Parameters:

• workdir – Working directory of the flow. Must be specified here if we haven’t initialized the workdir in the __init__.

• build – True to build the flow and save status to pickle file.

use_smartio()

This function should be called when the entire Flow has been built. It tries to reduce the pressure on the hard disk by using Abinit smart-io capabilities for those files that are not needed by other nodes. Smart-io means that big files (e.g. WFK) are written only if the calculation is unconverged so that we can restart from it. No output is produced if convergence is achieved.

Return: self

show_dependencies(stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>) → None

Writes to the given stream the ASCII representation of the dependency tree.

on_all_ok()

This method is called when all the works in the flow have reached S_OK. This method shall return True if the calculation is completed or False if the execution should continue due to side-effects such as adding a new work to the flow.

This methods allows subclasses to implement customized logic such as extending the flow by adding new works. The flow has an internal counter: on_all_ok_num_calls that shall be incremented by client code when subclassing this method. This counter can be used to decide if futher actions are needed or not.

An example of flow that adds a new work (only once) when all_ok is reached for the first time:

def on_all_ok(self):

if self.on_all_ok_num_calls > 0: return True self.on_all_ok_num_calls += 1

self.register_work(work) self.allocate() self.build_and_pickle_dump()

# The scheduler will keep on running the flow. return False

on_dep_ok(signal, sender)

finalize()

This method is called when the flow is completed. Return 0 if success

set_garbage_collector(exts=None, policy='task') → None

Enable the garbage collector that will remove the big output files that are not needed.

Parameters:

• exts – string or list with the Abinit file extensions to be removed. A default is provided if exts is None

• policy – Either flow or task. If policy is set to ‘task’, we remove the output files as soon as the task reaches S_OK. If ‘flow’, the files are removed only when the flow is finalized. This option should be used when we are dealing with a dynamic flow with callbacks generating other tasks since a abipy.flowtk.tasks.Task might not be aware of its children when it reached S_OK.

connect_signals() → None

Connect the signals within the Flow. The Flow is responsible for catching the important signals raised from its works.

disconnect_signals() → None

Disable the signals within the Flow.

show_receivers(sender=None, signal=None) → None

set_spectator_mode(mode=True) → None

When the flow is in spectator_mode, we have to disable signals, pickle dump and possible callbacks A spectator can still operate on the flow but the new status of the flow won’t be saved in the pickle file. Usually the flow is in spectator mode when we are already running it via the scheduler or other means and we should not interfere with its evolution. This is the reason why signals and callbacks must be disabled. Unfortunately preventing client-code from calling methods with side-effects when the flow is in spectator mode is not easy (e.g. flow.cancel will cancel the tasks submitted to the queue and the flow used by the scheduler won’t see this change!

rapidfire(check_status=True, max_nlaunch=-1, max_loops=1, sleep_time=5, **kwargs)

Use PyLauncher to submits tasks in rapidfire mode. kwargs contains the options passed to the launcher.

Parameters:

• check_status

• max_nlaunch – Maximum number of launches. default: no limit.

• max_loops – Maximum number of loops

• sleep_time – seconds to sleep between rapidfire loop iterations

Return: Number of tasks submitted.

single_shot(check_status=True, **kwargs)

Use PyLauncher to submit one task. kwargs contains the options passed to the launcher.

Return: Number of tasks submitted.

make_scheduler(**kwargs)

Build and return a PyFlowScheduler to run the flow.

Parameters:

kwargs – if empty we use the user configuration file. if filepath in kwargs we init the scheduler from filepath. else pass kwargs to PyFlowScheduler __init__ method.

make_light_tarfile(name=None)

Lightweight tarball file. Mainly used for debugging. Return the name of the tarball file.

make_tarfile(name=None, max_filesize=None, exclude_exts=None, exclude_dirs=None, verbose=0, **kwargs)

Create a tarball file.

Parameters:

• name – Name of the tarball file. Set to os.path.basename(flow.workdir) + “tar.gz”` if name is None.

• max_filesize (int or string with unit) – a file is included in the tar file if its size <= max_filesize Can be specified in bytes e.g. max_files=1024 or with a string with unit e.g. max_filesize=”1 MB”. No check is done if max_filesize is None.

• exclude_exts – List of file extensions to be excluded from the tar file.

• exclude_dirs – List of directory basenames to be excluded.

• verbose (int) – Verbosity level.

• kwargs – keyword arguments passed to the TarFile constructor.

Returns: The name of the tarfile.

explain(what='all', nids=None, verbose=0) → str

Return string with the docstrings of the works/tasks in the Flow grouped by class.

Parameters:

• what – “all” to print all nodes, “works” for Works only, “tasks” for tasks only.

• nids – list of node identifiers used to filter works or tasks.

• verbose – Verbosity level

show_autoparal(nids=None, verbose=0) → None

Print to terminal the autoparal configurations for each task in the Flow.

Parameters:

• nids – list of node identifiers used to filter works or tasks.

• verbose – Verbosity level

get_graphviz(engine='automatic', graph_attr=None, node_attr=None, edge_attr=None)

Generate flow graph in the DOT language.

Parameters:

• engine – Layout command used. [‘dot’, ‘neato’, ‘twopi’, ‘circo’, ‘fdp’, ‘sfdp’, ‘patchwork’, ‘osage’]

• graph_attr – Mapping of (attribute, value) pairs for the graph.

• node_attr – Mapping of (attribute, value) pairs set for all nodes.

• edge_attr – Mapping of (attribute, value) pairs set for all edges.

Returns: graphviz.Digraph <https://graphviz.readthedocs.io/en/stable/api.html#digraph>

graphviz_imshow(ax=None, figsize=None, dpi=300, fmt='png', **kwargs) → Any

Generate flow graph in the DOT language and plot it with matplotlib.

Parameters:

• ax – matplotlib.axes.Axes or None if a new figure should be created.

• figsize – matplotlib figure size (None to use default)

• dpi – DPI value.

• fmt – Select format for output image

Return: matplotlib.figure.Figure

Keyword arguments controlling the display of the figure:

kwargs	Meaning
title	Title of the plot (Default: None).
show	True to show the figure (default: True).
savefig	“abc.png” or “abc.eps” to save the figure to a file.
size_kwargs	Dictionary with options passed to fig.set_size_inches e.g. size_kwargs=dict(w=3, h=4)
tight_layout	True to call fig.tight_layout (default: False)
ax_grid	True (False) to add (remove) grid from all axes in fig. Default: None i.e. fig is left unchanged.
ax_annotate	Add labels to subplots e.g. (a), (b). Default: False
fig_close	Close figure. Default: False.
plotly	Try to convert mpl figure to plotly.

plot_networkx(mode='network', with_edge_labels=False, ax=None, arrows=False, node_size='num_cores', node_label='name_class', layout_type='spring', **kwargs) → Any

Use networkx to draw the flow with the connections among the nodes and the status of the tasks.

Parameters:

• mode – networkx to show connections, status to group tasks by status.

• with_edge_labels – True to draw edge labels.

• ax – matplotlib.axes.Axes or None if a new figure should be created.

• arrows – if True draw arrowheads.

• node_size – By default, the size of the node is proportional to the number of cores used.

• node_label – By default, the task class is used to label node.

• layout_type – Get positions for all nodes using layout_type. e.g. pos = nx.spring_layout(g)

Warning

Requires networkx package.

Keyword arguments controlling the display of the figure:

kwargs	Meaning
title	Title of the plot (Default: None).
show	True to show the figure (default: True).
savefig	“abc.png” or “abc.eps” to save the figure to a file.
size_kwargs	Dictionary with options passed to fig.set_size_inches e.g. size_kwargs=dict(w=3, h=4)
tight_layout	True to call fig.tight_layout (default: False)
ax_grid	True (False) to add (remove) grid from all axes in fig. Default: None i.e. fig is left unchanged.
ax_annotate	Add labels to subplots e.g. (a), (b). Default: False
fig_close	Close figure. Default: False.
plotly	Try to convert mpl figure to plotly.

write_open_notebook(foreground) → int

Generate an ipython notebook and open it in the browser. Return system exit code.

class abipy.flowtk.flows.G0W0WithQptdmFlow(workdir, scf_input, nscf_input, scr_input, sigma_inputs, manager=None)

Bases: Flow

cbk_qptdm_workflow(cbk)

This callback is executed by the flow when bands_work.nscf_task reaches S_OK.

It computes the list of q-points for the W(q,G,G’), creates nqpt tasks in the second work (QptdmWork), and connect the signals.

abipy.flowtk.flows.bandstructure_flow(workdir: str, scf_input: AbinitInput, nscf_input: AbinitInput, dos_inputs=None, manager=None, flow_class=<class 'abipy.flowtk.flows.Flow'>, allocate=True)

Build a abipy.flowtk.flows.Flow for band structure calculations.

Parameters:

• workdir – Working directory.

• scf_input – Input for the GS SCF run.

• nscf_input – Input for the NSCF run (band structure run).

• dos_inputs – Input(s) for the NSCF run (dos run).

• manager – abipy.flowtk.tasks.TaskManager object used to submit the jobs. Initialized from manager.yml if manager is None.

• flow_class – Flow subclass

• allocate – True if the flow should be allocated before returning.

Returns: abipy.flowtk.flows.Flow object

abipy.flowtk.flows.g0w0_flow(workdir: str, scf_input: AbinitInput, nscf_input: AbinitInput, scr_input: AbinitInput, sigma_inputs, manager=None, flow_class=<class 'abipy.flowtk.flows.Flow'>, allocate=True)

Build a abipy.flowtk.flows.Flow for one-shot $G_0W_0$ calculations.

Parameters:

• workdir – Working directory.

• scf_input – Input for the GS SCF run.

• nscf_input – Input for the NSCF run (band structure run).

• scr_input – Input for the SCR run.

• sigma_inputs – List of inputs for the SIGMA run.

• flow_class – Flow class

• manager – abipy.flowtk.tasks.TaskManager object used to submit the jobs. Initialized from manager.yml if manager is None.

• allocate – True if the flow should be allocated before returning.

Returns: abipy.flowtk.flows.Flow object

gw_works Module

Works and Flows for GW calculations with the quartic-scaling implementation.

class abipy.flowtk.gw_works.ScreeningWork(workdir=None, manager=None)

Bases: Work

This work parallelizes the calculation of the q-points of the screening. It also provides the callback on_all_ok that calls the mrgscr tool to merge all the partial SCR files produced.

The final SCR file is available in the outdata directory of the work. To use this SCR file as dependency of other tasks, use the syntax:

deps={scr_work: “SCR”}

Inheritance Diagram

classmethod from_nscf_task(nscf_task: NscfTask, scr_input, manager=None) → ScreeningWork

Construct a ScreeningWork from a abipy.flowtk.tasks.NscfTask object.

Parameters:

• nscf_task – abipy.flowtk.tasks.NscfTask object which produced the WFK file with empty states

• scr_input – abipy.abio.inputs.AbinitInput object representing a SCREENING calculation.

• manager – abipy.flowtk.tasks.TaskManager object.

classmethod from_wfk_filepath(wfk_filepath, scr_input, manager=None) → ScreeningWork

Construct a ScreeningWork from a WFK filepath and a screening input.

WARNING: the parameters reported in scr_input e.g. k-mesh, nband, ecut etc must be consisten with those used to generate the WFK file. No consistencty check is done at this level. You have been warned!

Parameters:

• wfk_filepath – Path to the WFK file.

• scr_input – abipy.abio.inputs.AbinitInput object representing a SCREENING calculation.

• manager – abipy.flowtk.tasks.TaskManager object.

merge_scr_files(remove_scrfiles=True, verbose=0) → str

This method is called when all the q-points have been computed. It runs mrgscr in sequential on the local machine to produce the final SCR file in the outdir of the Work. If remove_scrfiles is True, the partial SCR files are removed after the merge.

on_all_ok()

This method is called when all the q-points have been computed. It runs mrgscr in sequential on the local machine to produce the final SCR file in the outdir of the Work.

gwr_works Module

Works and Flows for GWR calculations (GW with supercells).

NB: An Abinit build with Scalapack is required to run GWR.

class abipy.flowtk.gwr_works.DirectDiagoWork(workdir=None, manager=None)

Bases: Work

This work performs the direct diagonalization of the KS Hamiltonian using the density produced by a GS-SCF run and produces a WFK file with empty states in the outdir of the second task.

Inheritance Diagram

classmethod from_scf_input(scf_input: AbinitInput, green_nband: int, manager: TaskManager = None) → DirectDiagoWork

Build object from an input representing a GS-SCF calculation.

Parameters:

• scf_input – Input for the GS-SCF calculation.

• green_nband – Number of bands to compute in the direct diagonalization. A negative value activate full diagonalization with nband equal to the number of PWs.

class abipy.flowtk.gwr_works.GWRSigmaConvWork(workdir=None, manager=None)

Bases: _BaseGWRWork

This work performs multiple QP calculations with the GWR code and produces xlsx files in its outdata directory with the QP results obtained with the different parameters.

Inheritance Diagram

on_all_ok()

Generate qp_dirgaps.csv file in the outdata directory.

class abipy.flowtk.gwr_works.GWRChiCompareWork(workdir=None, manager=None)

Bases: _BaseGWRWork

This work computes the irreducibile polarizability along the imaginary axis using the GWR code and the quartic-scaling algorithm using the same minimax mesh so that one can compare the two quantities.

Inheritance Diagram

classmethod from_scf_input(scf_input: AbinitInput, gwr_ntau, nband, ecuteps, den_node: Node, wfk_node: Node, gwr_kwargs=None, scr_kwargs=None, manager: TaskManager = None)

Build Work from an input for GS-SCF calculation

Parameters:

• scf_input – input for GS run.

• gwr_ntau – Number of points in minimax mesh.

• nband – Number of bands to build G and chi.

• ecuteps – Cutoff energy for chi.

• den_node – The Node who produced the DEN file.

• wfk_node – The Node who produced the WFK file.

• gwr_kwargs – Extra kwargs used to build the GWR input.

• scr_kwargs – Extra kwargs used to build the SCR input.

• manager – Abipy Task Manager.

on_all_ok()

Write python script to compare chi0 matrix elements.

class abipy.flowtk.gwr_works.GWRRPAConvWork(workdir=None, manager=None)

Bases: _BaseGWRWork

This work computes the RPA correlated energy for different number of points in the minimax mesh.

classmethod from_scf_input_ntaus(scf_input: AbinitInput, gwr_ntau_list, nband, ecuteps, den_node: Node, wfk_node: Node, gwr_kwargs=None, manager: TaskManager = None)

Build Work from an input for GS-SCF calculation

Parameters:

• scf_input – input for GS run.

• gwr_ntau_list – List with number of points in minimax mesh.

• nband – Number of bands to build G and chi.

• ecuteps – Cutoff energy for chi.

• den_node – The Node who produced the DEN file.

• wfk_node – The Node who produced the WFK file.

• gwr_kwargs – Extra kwargs used to build the GWR input.

• manager – Abipy Task Manager.

on_all_ok()

Generate gwr_robot.py script in the outdata directory.

gruneisen Module

Work for computing the Grüneisen parameters with finite differences of DFPT phonons.

WARNING: This code must be tested more carefully.

class abipy.flowtk.gruneisen.GruneisenWork(workdir=None, manager=None)

Bases: Work

This work computes the Grüneisen parameters (derivative of frequencies wrt volume) using finite differences and the phonons obtained with the DFPT part of Abinit. The Anaddb input file needed to compute Grüneisen parameters will be generated in the outdata directory of the flow.

It is necessary to run three DFPT phonon calculations. One is calculated at the equilibrium volume and the remaining two are calculated at the slightly larger volume and smaller volume than the equilibrium volume. The unitcells at these volumes have to be fully relaxed under the constraint of each volume. This Work automates the entire procedure starting from an input for GS calculations.

classmethod from_gs_input(gs_inp, voldelta, ngqpt, tolerance=None, with_becs=False, ddk_tolerance=None, workdir=None, manager=None) → GruneisenWork

Build the work from an abipy.abio.inputs.AbinitInput representing a GS calculations.

Parameters:

• gs_inp – abipy.abio.inputs.AbinitInput representing a GS calculation in the initial unit cell.

• voldelta – Absolute increment for unit cell volume. The three volumes are: [v0 - voldelta, v0, v0 + voldelta] where v0 is taken from gs_inp.structure.

• ngqpt – three integers defining the q-mesh for phonon calculations.

• tolerance – dict {“varname”: value} with the tolerance to be used in the phonon run. None to use AbiPy default.

• with_becs – Activate calculation of Electric field and Born effective charges.

• ddk_tolerance – dict {“varname”: value} with the tolerance used in the DDK run if with_becs. None to use AbiPy default.

on_all_ok()

This method is called once the Work is completed i.e. when all the tasks have reached status S_OK.

add_phonon_works_and_build() → None

Get relaxed structures from the tasks, build Phonons works with new structures. Add works to the flow and build new directories.

gs_works Module

Work subclasses related to GS calculations.

class abipy.flowtk.gs_works.EosWork(workdir=None, manager=None)

Bases: Work

Work to compute the Equation of State. The EOS is obtained by computing E(V) for several volumes around the input V0, The initial volumes are obtained by rescaling the input lattice vectors so that length proportions and angles are preserved. This guess is exact for cubic materials while other Bravais lattices require a constant-volume optimization of the cell geometry.

If lattice_type==”cubic” and atomic positions are fixed by symmetry. use can use move_atoms=False to perform standard GS-SCF calculations. In all the other cases, E(V) is obtained by relaxing the atomic positions at fixed volume.

The E(V) points are fitted at the end of the work and the results are saved in the eos_data.json file produced in the outdata directory. The file contains the energies, the volumes and the values of V0, B0, B1 obtained with different EOS models.

Inheritance Diagram

classmethod from_scf_input(scf_input: AbinitInput, npoints=4, deltap_vol=0.25, ecutsm=0.5, move_atoms=True, manager=None) → EosWork

Build an EosWork from an AbinitInput for GS-SCF.

Parameters:

• scf_input – AbinitInput for GS-SCF used as template to generate the other inputs.

• npoints – Number of volumes generated on the right (left) of the equilibrium volume The total number of points is therefore 2 * n + 1.

• deltap_vol – Step of the linear mesh given in relative percentage of the equilibrium volume The step is thus: v0 * deltap_vol / 100.

• ecutsm – Value of ecutsm input variable. If scf_input does not provide ecutsm, this value will be used else the vale in scf_input.

• move_atoms – If True, a structural relaxation of ions is performed for each volume This is needed if the atomic positions are non fixed by symmetry.

• manager – TaskManager instance. Use default if None.

classmethod from_inputs(inputs: list[AbinitInput], manager=None) → EosWork

Advanced interface to build an EosWork from an list of AbinitInputs.

get_and_write_eosdata(write_json=True) → dict

Compute the EOS and produce a JSON file eos_data.json in outdata.

on_all_ok()

This method is called when all tasks have reached S_OK. It reads the energies and the volumes from the GSR file, computes the EOS and produce a JSON file eos_data.json in outdata.

launcher Module

Tools for the submission of Tasks.

class abipy.flowtk.launcher.ScriptEditor

Bases: object

Simple editor to simplify the writing of shell scripts

property shell: str

reset() → None

Reset the editor.

shebang() → None

Adds the shebang line.

declare_var(key: str, val: str) → None

Declare a env variable. If val is None the variable is unset.

declare_vars(d: dict) → None

Declare the variables defined in the dictionary d.

export_envar(key: str, val: str) → None

Export an environment variable.

export_envars(env: dict) → None

Export the environment variables contained in the dict env.

add_emptyline() → None

Add an empty line.

add_comment(comment: str) → None

Add a comment

load_modules(modules: list[str]) → None

Load the list of specified modules.

load_module(module: str) → None

add_line(line: str) → None

add_lines(lines: list[str]) → None

get_script_str(reset=True) → str

Returns a string with the script and reset the editor if reset is True

class abipy.flowtk.launcher.PyLauncher(flow: Flow, **kwargs)

Bases: object

This object handle the submission of the tasks contained in a abipy.flowtk.flows.Flow.

Error

alias of PyLauncherError

single_shot()

Run the first Task than is ready for execution.

Returns:

Number of jobs launched.

rapidfire(max_nlaunch=-1, max_loops=1, sleep_time=5)

Keeps submitting Tasks until we are out of jobs or no job is ready to run.

Parameters:

• max_nlaunch – Maximum number of launches. default: no limit.

• max_loops – Maximum number of loops

• sleep_time – seconds to sleep between rapidfire loop iterations

Returns:

The number of tasks launched.

fetch_tasks_to_run() → list[Task]

Return the list of tasks that can be submitted. Empty list if no task has been found.

class abipy.flowtk.launcher.PyFlowScheduler(**kwargs)

Bases: BaseScheduler

property pid_file: str

Absolute path of the file with the pid. The file is located in the workdir of the flow

property flow: Flow

Flow.

add_flow(flow: Flow) → None

Add a flow to the scheduler.

start() → int

Starts the scheduler. Returns 0 if success. This method blocks until the flow is completed or some exception is raised.

callback()

The function that will be executed by the scheduler.

cleanup() → None

Cleanup routine: remove the pid file and save the pickle database

shutdown(msg: str) → None

Shutdown the scheduler.

class abipy.flowtk.launcher.MultiFlowScheduler(sqldb_path, **kwargs)

Bases: BaseScheduler

add_flow(flow: Flow, user_message, priority=None) → None

Add a flow to the scheduler.

register_flow_exception(flow_idx, exc) → None

handle_flow_exception() → None

start() → int

Starts the scheduler. Returns 0 if success. This method blocks until the flow is completed or some exception is raised.

sql_connect()

create_sqldb() → None

get_incoming_flows() → list[Flow]

get_dataframe() → DataFrame

get_json_status() → dict

get_flow_and_status_by_nodeid(node_id)

get_sql_rows_with_node_ids(node_id_list)

groupby_status()

remove_flows_with_status(status)

callback()

The function that will be executed by the scheduler.

update_flows_and_slqdb()

lumi_works Module

Work subclasses for the computation of luminiscent properties.

class abipy.flowtk.lumi_works.LumiWork(workdir=None, manager=None)

Bases: Work

This Work implements Fig 1 of https://arxiv.org/abs/2010.00423.

Client code is responsible for the preparation of the supercell and of the GS SCF input files with the fixed electronic occupations associated to the two configurations. By default, the work computes the two relaxed structures and the four total energies corresponding to the Ag, Ag*, Ae*, Ae configurations. Optionally, one can activate the computation of four electronic band structures. See docstring of from_scf_inputs for further info.

classmethod from_scf_inputs(gs_scf_inp, ex_scf_inp, relax_kwargs_gs, relax_kwargs_ex, ndivsm=0, nb_extra=10, tolwfr=1e-12, four_points=True, meta=None, manager=None) → LumiWork

Parameters:

• gs_scf_inp – abipy.abio.inputs.AbinitInput representing a GS SCF run for the ground-state.

• ex_scf_inp – abipy.abio.inputs.AbinitInput representing a GS SCF run for the excited-state.

• relax_kwargs_gs – Dictonary with input variables to be added to gs_scf_inp when generating input files for ground state structural relaxations.

• relax_kwargs_ex – Dictonary with input variables to be added to ex_scf_inp when generating input files for excited state structural relaxations.

• ndivsm – Activates the computation of band structure if different from zero. if > 0, it’s the number of divisions for the smallest segment of the path (Abinit variable). if < 0, it’s interpreted as the pymatgen line_density parameter in which the number of points in the segment is proportional to its length. Typical value: -20. This option is the recommended one if the k-path contains two high symmetry k-points that are very close as ndivsm > 0 may produce a very large number of wavevectors.

• nb_extra – Number of extra bands added to the input nband when computing band structures (ndivsm != 0).

• tolwfr – Tolerance of the residuals used for the NSCF band structure calculations.

• four_points – if True, compute the two relaxations and the four points energies. If false, only the two relaxations.

• meta – dict corresponding to the metadata of a lumiwork (supercell size, dopant type,…)

• manager – abipy.flowtk.tasks.TaskManager of the task. If None, the manager is initialized from the config file.

on_all_ok()

This method is called when all the works in the flow have reached S_OK.

This is the section in which we implement most of the workflow logic at runtime. since we need to generate input files with relaxed structures.

class abipy.flowtk.lumi_works.LumiWork_relaxations(workdir=None, manager=None)

Bases: Work

This Work implements the ground and excited state relaxations only.

The relaxations run simultaneously. No task creation at run-time.

classmethod from_scf_inputs(gs_scf_inp, ex_scf_inp, relax_kwargs_gs, relax_kwargs_ex, meta=None, manager=None)

Parameters:

• gs_scf_inp – abipy.abio.inputs.AbinitInput representing a GS SCF run for the ground-state.

• ex_scf_inp – abipy.abio.inputs.AbinitInput representing a GS SCF run for the excited-state.

• relax_kwargs_gs – Dictonary with input variables to be added to gs_scf_inp when generating input files for ground state structural relaxations.

• relax_kwargs_ex – Dictonary with input variables to be added to ex_scf_inp when generating input files for excited state structural relaxations.

• meta – dict corresponding to the metadata of a lumiwork (supercell size, dopant type,…)

• manager – abipy.flowtk.tasks.TaskManager of the task. If None, the manager is initialized from the config file.

on_all_ok()

This method is called when all the works in the flow have reached S_OK.

class abipy.flowtk.lumi_works.LumiWorkFromRelax(workdir=None, manager=None)

Bases: Work

Same as LumiWork, without the relaxations. Typically used after a LumiWork_relaxations work. The two relaxed structures (in ground and excited state) are given as input. No creation at run-time

classmethod from_scf_inputs(gs_scf_inp, ex_scf_inp, gs_structure, ex_structure, ndivsm=0, nb_extra=10, tolwfr=1e-12, meta=None, manager=None)

Parameters:

• gs_scf_inp – abipy.abio.inputs.AbinitInput representing a GS SCF run for the ground-state.

• exc_scf_inp – abipy.abio.inputs.AbinitInput representing a GS SCF run for the excited-state.

• gs_structure – object representing the relaxed ground state structure

• ex_structure – object representing the excited ground state structure

• ndivsm – Activates the computation of band structure if different from zero. if > 0, it’s the number of divisions for the smallest segment of the path (Abinit variable). if < 0, it’s interpreted as the pymatgen line_density parameter in which the number of points in the segment is proportional to its length. Typical value: -20. This option is the recommended one if the k-path contains two high symmetry k-points that are very close as ndivsm > 0 may produce a very large number of wavevectors.

• nb_extra – Number of extra bands added to the input nband when computing band structures (ndivsm != 0).

• tolwfr – Tolerance of the residuals used for the NSCF band structure calculations.

• manager – abipy.flowtk.tasks.TaskManager of the task. If None, the manager is initialized from the config file.

on_all_ok()

This method is called once the Work is completed i.e. when all tasks have reached status S_OK. Subclasses should provide their own implementation

Returns:

returncode: 0 on success. message: a string that should provide a human-readable description of what has been performed.

Return type:

Dictionary that must contain at least the following entries

mocks Module

Mock objects for unit tests.

abipy.flowtk.mocks.change_task_start(task: Task, mocked_status='Error')

Return a AbinitTaskMockedStart object.

class abipy.flowtk.mocks.AbinitTaskMockedStart(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: AbinitTask

A Task whose status is always self.mocked_status.

start(**kwargs)

Starts the calculation by performing the following steps:

• build dirs and files

• call the _setup method

• execute the job file by executing/submitting the job script.

Main entry point for the Launcher.

kwargs	Meaning
autoparal	False to skip the autoparal step (default True)
exec_args	List of arguments passed to executable.

Returns:

1 if task was started, 0 otherwise.

abipy.flowtk.mocks.infinite_flow(flow: Flow)

Return an InfiniteFlow.

class abipy.flowtk.mocks.InfiniteFlow(workdir, manager=None, pickle_protocol=-1, remove=False)

Bases: Flow

A Flow that will never reach all_ok

check_status(**kwargs)

Check the status of the works in self.

Parameters:

• show – True to show the status of the flow.

• kwargs – keyword arguments passed to show_status

netcdf Module

nodes Module

This module defines the Node base class inherited by Task, Work and Flow objects.

class abipy.flowtk.nodes.Status

Bases: int

This object is an integer representing the status of the Node.

classmethod as_status(obj: Any) → Status

Convert obj into Status.

classmethod from_string(s: str) → str

Return a Status instance from its string representation.

classmethod all_status_strings() → list[str]

List of strings with all possible values status.

property is_critical: bool

True if status is critical.

property color_opts: dict

property colored: str

Return colorized text used to print the status if the stream supports it.

class abipy.flowtk.nodes.Dependency(node, exts=None)

Bases: object

This object describes the dependencies among the nodes of a calculation.

A Dependency consists of a Node that produces a list of products (files) that are used by the other nodes (Task or Work) to start the calculation. One usually creates the object by calling work.register

Example

# Register the SCF task in work. scf_task = work.register(scf_strategy)

# Register the NSCF calculation and its dependency on the SCF run via deps. nscf_task = work.register(nscf_strategy, deps={scf_task: “DEN”})

property info: str

property node: Node

The abipy.flowtk.nodes.Node associated to the dependency.

property status: Status

The status of the dependency, i.e. the status of the abipy.flowtk.nodes.Node.

products()

List of output files produces by self.

apply_getters(task)

This function is called when we specify the task dependencies with the syntax:

deps={node: “@property”}

In this case the task has to the get property from node before starting the calculation.

At present, the following properties are supported:

• @structure

connecting_vars() → dict

Returns a dictionary with the variables that must be added to the input file in order to connect this abipy.flowtk.nodes.Node to its dependencies.

get_filepaths_and_exts() → tuple[list, list]

Returns the paths of the output files produced by self and its extensions

class abipy.flowtk.nodes.Product(ext: str, path: str)

Bases: object

A product represents an output file produced by ABINIT instance. This file is needed to start another Task or another Work.

classmethod from_file(filepath: str) → Product

Build a Product instance from a filepath.

property filepath: str

Absolute path of the file.

connecting_vars() → dict

Returns a dictionary with the ABINIT variables that must be used to make the code use this file.

class abipy.flowtk.nodes.GridFsFile(path, fs_id=None, mode='b')

Bases: AttrDict

Information on a file that will stored in the MongoDb gridfs collection.

class abipy.flowtk.nodes.NodeResults(node, **kwargs)

Bases: dict, MSONable

Dictionary used to store the most important results produced by a abipy.flowtk.nodes.Node.

JSON_SCHEMA = {'properties': {'exceptions': {'required': True, 'type': 'array'}, 'files': {'required': True, 'type': 'object'}, 'in': {'description': 'dictionary with input parameters', 'required': True, 'type': 'object'}, 'node_class': {'required': True, 'type': 'string'}, 'node_finalized': {'required': True, 'type': 'boolean'}, 'node_history': {'required': True, 'type': 'array'}, 'node_id': {'required': True, 'type': 'integer'}, 'node_name': {'required': True, 'type': 'string'}, 'node_status': {'required': True, 'type': 'string'}, 'out': {'description': 'dictionary with the output results', 'required': True, 'type': 'object'}}, 'type': 'object'}

classmethod from_node(node)

Initialize an instance of NodeResults from a Node subclass.

property exceptions

property gridfs_files

List with the absolute paths of the files to be put in GridFs.

register_gridfs_files(**kwargs)

This function registers the files that will be saved in GridFS. kwargs is a dictionary mapping the key associated to the file (usually the extension) to the absolute path. By default, files are assumed to be in binary form, for formatted files one should pass a tuple (“filepath”, “t”).

Example:

results.register_gridfs(GSR="path/to/GSR.nc", text_file=("/path/to/txt_file", "t"))

The GSR file is a binary file, whereas text_file is a text file.

push_exceptions(*exceptions)

as_dict()

A JSON serializable dict representation of an object.

classmethod from_dict(d)

Parameters:

d – Dict representation.

Returns:

MSONable class.

json_dump(filename)

classmethod json_load(filename)

update_collection(collection)

Update a mongodb collection.

abipy.flowtk.nodes.check_spectator(node_method)

Decorator for abipy.flowtk.nodes.Node methods. Raise SpectatorNodeError.

exception abipy.flowtk.nodes.NodeError

Bases: Exception

Base Exception raised by abipy.flowtk.nodes.Node subclasses

exception abipy.flowtk.nodes.SpectatorNodeError

Bases: NodeError

Exception raised by abipy.flowtk.nodes.Node methods when the node is in spectator mode and we are calling a method with side effects.

class abipy.flowtk.nodes.Node

Bases: object

Abstract base class defining the interface that must be implemented by the nodes of the calculation.

Nodes are hashable and can be tested for equality

Results

alias of NodeResults

Error

alias of NodeError

SpectatorError

alias of SpectatorNodeError

S_INIT = <Status: Initialized, at 139955597801216>

S_LOCKED = <Status: Locked, at 139955597801856>

S_READY = <Status: Ready, at 139955597793472>

S_SUB = <Status: Submitted, at 139955597792256>

S_RUN = <Status: Running, at 139955597793024>

S_DONE = <Status: Done, at 139955597792384>

S_ABICRITICAL = <Status: AbiCritical, at 139955597792576>

S_QCRITICAL = <Status: QCritical, at 139955597791808>

S_UNCONVERGED = <Status: Unconverged, at 139955597792064>

S_ERROR = <Status: Error, at 139955597792000>

S_OK = <Status: Completed, at 139955597790976>

ALL_STATUS = [<Status: Initialized, at 139955597801216>, <Status: Locked, at 139955597801856>, <Status: Ready, at 139955597793472>, <Status: Submitted, at 139955597792256>, <Status: Running, at 139955597793024>, <Status: Done, at 139955597792384>, <Status: AbiCritical, at 139955597792576>, <Status: QCritical, at 139955597791808>, <Status: Unconverged, at 139955597792064>, <Status: Error, at 139955597792000>, <Status: Completed, at 139955597790976>]

color_rgb = array([0.41176471, 0.41176471, 0.41176471])

color_hex()

Node color as Hex Triplet https://en.wikipedia.org/wiki/Web_colors#Hex_triplet

isinstance(class_or_string)

Check whether the node is a instance of class_or_string. Unlinke the standard isinstance builtin, the method accepts either a class or a string. In the later case, the string is compared with self.__class__.__name__ (case insensitive).

classmethod as_node(obj: Any) → Node | None

Convert obj into a Node instance.

Returns:

obj if obj is a Node instance, cast obj to FileNode instance of obj is a string. None if obj is None

property name: str

The name of the node (only used for facilitating its identification in the user interface).

property relworkdir: str

Return a relative version of the workdir

set_name(name: str) → None

Set the name of the Node.

property node_id: int

Node identifier.

set_readme(md_string: str) → None

Set the value of readme_md.

set_abipy_meta_json(data: dict) → None

Set the value of abipy_meta_json

set_user_message(user_message: str) → None

Set the value of user_message

property user_message: str

set_node_id(node_id: int) → None

Set the node identifier. Use it carefully!

property finalized: bool

True if the Node has been finalized.

property in_spectator_mode: bool

True if we are in spectator mode.

property corrections: list[dict]

List of dictionaries with infornation on the actions performed to solve AbiCritical Events. Each dictionary contains the AbinitEvent who triggered the correction and a human-readable message with the description of the operation performed.

property num_corrections: int

Number of corrections performed.

log_correction(event, action: str) → None

This method should be called once we have fixed the problem associated to this event. It adds a new entry in the correction history of the node.

Parameters:

• event – AbinitEvent that triggered the correction.

• action (str) – Human-readable string with info on the action perfomed to solve the problem.

property is_file: bool

True if this node is a file

property is_task: bool

True if this node is a Task

property is_work: bool

True if this node is a Work

property is_flow: bool

True if this node is a Flow

property deps: list[Dependency]

List of Dependency objects defining the dependencies of this Node. Empty list if this abipy.flowtk.nodes.Node does not have dependencies.

add_deps(deps) → None

Add a list of dependencies to the abipy.flowtk.nodes.Node.

Parameters:

deps – List of Dependency objects specifying the dependencies of the node. or dictionary mapping nodes to file extensions e.g. {task: “DEN”}

merge_deps() → None

Group all extensions associated to the same node in a single list. Useful for cases in which we may end up with the same node appearing more than once in self.deps. See e.g. add_deps.

remove_deps(deps) → None

Remove a list of dependencies from the abipy.flowtk.nodes.Node.

Parameters:

deps – List of Dependency objects specifying the dependencies of the node.

property deps_status: list

Returns a list with the status of the dependencies.

depends_on(other: Node) → bool

True if this node depends on the other node.

find_parent_with_ext(ext: str) → Node

Return the parent (usually a abipy.flowtk.tasks.Task) that produces the file with extension ext. Raises ValueError if multiple parents are found. Return None if no parent is found.

get_parents() → list[Node]

Return the list of nodes in the abipy.flowtk.flows.Flow required by this abipy.flowtk.nodes.Node

get_children() → list[Node]

Return the list of nodes in the abipy.flowtk.flows.Flow that depends on this abipy.flowtk.nodes.Node

Note

This routine assumes the entire flow has been allocated.

str_deps() → str

Return the string representation of the dependencies of the node.

get_vars_dataframe(*varnames) → DataFrame

Return pandas DataFrame with the value of the variables specified in varnames. Can be used for task/works/flow. It’s recursive!

get_graphviz_dirtree(engine='automatic', **kwargs)

Generate directory graph in the DOT language. The graph show the files and directories in the node workdir.

Returns: graphviz.Digraph <https://graphviz.readthedocs.io/en/stable/api.html#digraph>

set_gc(gc)

Set the garbage collector.

property gc

Garbage collector. None if garbage collection is deactivated. Use flow.set_garbage_collector to initialize the object.

property event_handlers: list

The list of handlers registered for this node. If the node is not a Flow and does not have its own list of handlers the handlers registered at the level of the flow are returned.

This trick allows one to registered different handlers at the level of the Task for testing purposes. By default, we have a common list of handlers for all the nodes in the flow. This choice facilitates the automatic installation of the handlers when we use callbacks to generate new Works and Tasks!

install_event_handlers(categories=None, handlers=None) → None

Install the EventHandlers for this `Node. If no argument is provided the default list of handlers is installed.

Parameters:

• categories – List of categories to install e.g. base + can_change_physics

• handlers – explicit list of EventHandler instances. This is the most flexible way to install handlers.

Note

categories and handlers are mutually exclusive.

show_event_handlers(stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>, verbose=0) → None

Print to stream the event handlers installed for this flow.

send_signal(signal) → None

Send signal from this node to all connected receivers unless the node is in spectator mode.

signal – (hashable) signal value, see dispatcher connect for details

Return a list of tuple pairs [(receiver, response), … ] or None if the node is in spectator mode.

if any receiver raises an error, the error propagates back through send, terminating the dispatch loop, so it is quite possible to not have all receivers called if a raises an error.

write_json_in_outdir(filename: str, data: dict) → str

Write data to json file of basename filename inside the outdir directory of the node. Support MSONable objects. Return path of json file.

write_json_in_workdir(filename: str, data: dict) → str

Write data to json file of basename filename inside the outdir directory of the node. Support MSONable objects. Return path of json file.

abstract property status

The status of the Node.

abstract check_status()

Check the status of the Node.

class abipy.flowtk.nodes.FileNode(filename: str)

Bases: Node

A Node that consists of a file. May be not yet existing

Mainly used to connect abipy.flowtk.tasks.Task objects to external files produced in previous runs.

color_rgb = array([0.4, 0.2, 1. ])

basename()

Basename of the file.

property products: list[Product]

opath_from_ext(ext: str) → str

property status: Status

The status of the Node.

check_status() → Status

Check the status of the Node.

get_results(**kwargs)

add_filechild(node: Node) → None

Add a node (usually Task) to the children of this FileNode.

property filechildren: list

List with the children (nodes) of this FileNode.

abiopen()

open_gsr()

class abipy.flowtk.nodes.HistoryRecord(level, pathname, lineno, msg, args, exc_info, func=None)

Bases: object

A HistoryRecord instance represents an entry in the NodeHistory.

HistoryRecord instances are created every time something is logged. They contain all the information pertinent to the event being logged. The main information passed in is in msg and args, which are combined using str(msg) % args to create the message field of the record. The record also includes information such as when the record was created, the source line where the logging call was made

levelno

Numeric logging level for the message (DEBUG, INFO, WARNING, ERROR, CRITICAL)

levelname

Text logging level for the message (“DEBUG”, “INFO”, “WARNING”, “ERROR”, “CRITICAL”)

pathname

Full pathname of the source file where the logging call was issued (if available)

filename

Filename portion of pathname

module

Module (name portion of filename)

lineno

Source line number where the logging call was issued (if available)

func_name

Function name

created

Time when the HistoryRecord was created (time.time() return value)

asctime

Textual time when the HistoryRecord was created

message

The result of record.getMessage(), computed just as the record is emitted

get_message(metadata=False, asctime=True) → str

Return the message after merging any user-supplied arguments with the message.

Parameters:

• metadata – True if function and module name should be added.

• asctime – True if time string should be added.

as_dict() → dict

classmethod from_dict(d: dict) → HistoryRecord

class abipy.flowtk.nodes.NodeHistory

Bases: deque

Logger-like object

to_string(metadata=False) → str

Returns a string with the history. Set metadata to True to have info on function and module.

info(msg: str, *args, **kwargs) → None

Log ‘msg % args’ with the info severity level

warning(msg: str, *args, **kwargs) → None

Log ‘msg % args’ with the warning severity level

critical(msg: str, *args, **kwargs) → None

Log ‘msg % args’ with the critical severity level

debug(msg: str, *args, **kwargs) → None

Log ‘msg % args’ with the critical severity level

class abipy.flowtk.nodes.NodeCorrections(iterable=(), /)

Bases: list

Iterable storing the correctios performed by the EventHandler

class abipy.flowtk.nodes.GarbageCollector(exts, policy)

Bases: object

This object stores information on the

abipy.flowtk.nodes.init_counter() → None

abipy.flowtk.nodes.get_newnode_id() → int

Returns a new node identifier used for abipy.flowtk.tasks.Task, abipy.flowtk.works.Work and abipy.flowtk.flows.Flow objects.

abipy.flowtk.nodes.save_lastnode_id() → None

Save the id of the last node created.

pseudos Module

qadapters Module

The initial version of this module was based on a similar implementation present in FireWorks (https://pypi.python.org/pypi/FireWorks). Work done by D. Waroquiers, A. Jain, and M. Kocher.

The main difference wrt the Fireworks implementation is that the QueueAdapter objects provide a programmatic interface for setting important attributes such as the number of MPI nodes, the number of OMP threads and the memory requirements. This programmatic interface is used by the TaskManager for optimizing the parameters of the run before submitting the job (Abinit provides the autoparal option that allows one to get a list of parallel configuration and their expected efficiency).

abipy.flowtk.qadapters.make_qadapter(**kwargs)

Return the concrete QueueAdapter class from a string. Note that one can register a customized version with:

from qadapters import SlurmAdapter

class MyAdapter(SlurmAdapter):
    QTYPE = "myslurm"
    # Add your customized code here

# Register your class.
SlurmAdapter.register(MyAdapter)

make_qadapter(qtype="myslurm", **kwargs)

Warning

MyAdapter should be pickleable, hence one should declare it at the module level so that pickle can import it at run-time.

qha Module

Workflows for calculations within the quasi-harmonic approximation.

class abipy.flowtk.qha.RelaxAndAddPhononWorks(workdir=None, manager=None)

Bases: Work

This work performs a structural relaxation of the initial structure, then a set of distorted structures is genenerated and the relaxed structures are used to compute phonons, BECS and the dielectric tensor with DFPT.

Inheritance Diagram

classmethod from_scf_input(scf_input, eps, ngqpt, with_becs: bool, optcell: int, ionmov: int, edos_ngkpt=None) → RelaxAndAddPhononWorks

Build the work from an abipy.abio.inputs.AbinitInput representing a GS-SCF calculation.

Parameters:

• scf_input – abipy.abio.inputs.AbinitInput for GS-SCF used as template to generate the other inputs.

• with_becs – Activate calculation of Electric field and Born effective charges.

• optcell – Abinit input variables.

• ionmov – Abinit input variables.

• edos_ngkpt – Three integers defining the the k-sampling for the computation of the electron DOS with the relaxed structures. Useful for metals in which the electronic contribution should be included. None disables the computation of the e-DOS.

on_ok(sender)

This method is called when one task reaches status S_OK. It executes on_all_ok when all tasks in self have reached S_OK.

on_all_ok()

This callback is called when all tasks in the Work reach status S_OK. Here we add a new PhononWork for each volume using the relaxed structure.

class abipy.flowtk.qha.ThermalRelaxTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: RelaxTask

class abipy.flowtk.qha.ThermalRelaxWork(workdir=None, manager=None)

Bases: Work

Inheritance Diagram

classmethod from_relax_input(relax_input, qha, temperatures, pressures)

on_all_ok()

Implement the post-processing step at the end of the Work.

class abipy.flowtk.qha.QhaFlow(workdir, manager=None, pickle_protocol=-1, remove=False)

Bases: Flow

Flow for QHA calculations.

Inheritance Diagram

classmethod from_scf_input(workdir, scf_input, ngqpt, with_becs, eps=0.005, edos_ngkpt=None, manager=None) → QhaFlow

Build a flow for QHA calculations from an abipy.abio.inputs.AbinitInput for GS-SCF calculations.

Parameters:

• workdir – Working directory of the flow.

• scf_input – abipy.abio.inputs.AbinitInput for GS-SCF run used as template to generate the other inputs.

• ngqpt – Three integers defining the q-mesh for phonon calculation.

• with_becs – Activate calculation of Electric field and Born effective charges.

• eps

• edos_ngkpt – Three integers defining the the k-sampling for the computation of the electron DOS with the relaxed structures. Useful for metals or small gap semiconductors in which the electronic contribution should be included. None disables the computation of the e-DOS.

• manager – abipy.flowtk.tasks.TaskManager instance. Use default if None.

finalize()

This method is called when the flow is completed. It performs some basic post-processing of the results to facilitate further analysis.

qjobs Module

Objects and methods to contact the resource manager to get info on the status of the job and useful statistics. Note that this is not a wrapper for the C API but a collection of simple wrappers around the shell commands provided by the resource manager (qsub, qdel and qstat for PBS, sinfo, squeue… for Slurm). The main goal indeed is providing a simplified common interface for different resource managers without having to rely on external libraries.

class abipy.flowtk.qjobs.JobStatus

Bases: int

This object is an integer representing the status of a QueueJob.

Slurm API, see man squeue.

JOB STATE CODES

Jobs typically pass through several states in the course of their execution. The typical states are PENDING, RUNNING, SUSPENDED, COMPLETING, and COMPLETED. An explanation of each state follows:

BF  BOOT_FAIL       Job  terminated  due  to launch failure, typically due to a hardware failure (e.g.
                    unable to boot the node or block and the job can not be requeued).
CA  CANCELLED       Job was explicitly cancelled by the user or system administrator.
                    The job may or may not have been initiated.
CD  COMPLETED       Job has terminated all processes on all nodes.
CF  CONFIGURING     Job has been allocated resources, but are waiting for them to become ready for use (e.g. booting).
CG  COMPLETING      Job is in the process of completing. Some processes on some  nodes may still be active.
F   FAILED          Job terminated with non-zero exit code or other failure condition.
NF  NODE_FAIL       Job terminated due to failure of one or more allocated nodes.
PD  PENDING         Job is awaiting resource allocation.
PR  PREEMPTED       Job terminated due to preemption.
R   RUNNING         Job currently has an allocation.
S   SUSPENDED       Job has an allocation, but execution has been suspended.
TO  TIMEOUT         Job terminated upon reaching its time limit.
SE SPECIAL_EXIT     The job was requeued in a special state. This state can be set by users, typically
                    in EpilogSlurmctld, if the job has terminated with a particular exit value.

classmethod from_string(s: str) → JobStatus

Return an instance from its string representation.

class abipy.flowtk.qjobs.QueueJob(queue_id, qname='UnknownQueue')

Bases: object

This object provides methods to contact the resource manager to get info on the status of the job and useful statistics. This is an abstract class.

QTYPE = None

S_UNKNOWN = <JobStatus: UNKNOWN, at 139955598822208>

S_PENDING = <JobStatus: PENDING, at 139955598822976>

S_RUNNING = <JobStatus: RUNNING, at 139955598823104>

S_RESIZING = <JobStatus: RESIZING, at 139955598823232>

S_SUSPENDED = <JobStatus: SUSPENDED, at 139955598823936>

S_COMPLETED = <JobStatus: COMPLETED, at 139955598823488>

S_CANCELLED = <JobStatus: CANCELLED, at 139955598823744>

S_FAILED = <JobStatus: FAILED, at 139955598824320>

S_TIMEOUT = <JobStatus: TIMEOUT, at 139955598824128>

S_PREEMPTED = <JobStatus: PREEMPTED, at 139955598824192>

S_NODEFAIL = <JobStatus: NODEFAIL, at 139955598824896>

static from_qtype_and_id(qtype: str, queue_id, qname=None)

Return a new istance of the appropriate subclass.

Parameters:

• qtype – String specifying the Resource manager type.

• queue_id – Job identifier.

• qname – Name of the queue (optional).

property is_completed: bool

property is_running: bool

property is_failed: bool

property timeout: bool

property has_node_failures: bool

property unknown_status: bool

set_status_exitcode_signal(status, exitcode, signal)

likely_code_error()

See <http://man7.org/linux/man-pages/man7/signal.7.html>

SIGHUP	1	Term	Hangup detected on controlling terminal or death of controlling process
SIGINT	2	Term	Interrupt from keyboard
SIGQUIT	3	Core	Quit from keyboard
SIGILL	4	Core	Illegal Instruction
SIGABRT	6	Core	Abort signal from abort(3)
SIGFPE	8	Core	Floating point exception
SIGKILL	9	Term	Kill signal
SIGSEGV	11	Core	Invalid memory reference
SIGPIPE	13	Term	Broken pipe: write to pipe with no readers
SIGALRM	14	Term	Timer signal from alarm(2)
SIGTERM	15	Term	Termination signal
SIGUSR1	30,10,16	Term	User-defined signal 1
SIGUSR2	31,12,17	Term	User-defined signal 2
SIGCHLD	20,17,18	Ign	Child stopped or terminated
SIGCONT	19,18,25	Cont	Continue if stopped
SIGSTOP	17,19,23	Stop	Stop process
SIGTSTP	18,20,24	Stop	Stop typed at terminal
SIGTTIN	21,21,26	Stop	Terminal input for background process
SIGTTOU	22,22,27	Stop	Terminal output for background process

The signals SIGKILL and SIGSTOP cannot be caught, blocked, or ignored. Next the signals not in the POSIX.1-1990 standard but described in SUSv2 and POSIX.1-2001.

Signal	Value	Action	Comment
SIGBUS	10,7,10	Core	Bus error (bad memory access)
SIGPOLL		Term	Pollable event (Sys V). Synonym for SIGIO
SIGPROF	27,27,29	Term	Profiling timer expired
SIGSYS	12,31,12	Core	Bad argument to routine (SVr4)
SIGTRAP	5	Core	Trace/breakpoint trap
SIGURG	16,23,21	Ign	Urgent condition on socket (4.2BSD)
SIGVTALRM	26,26,28	Term	Virtual alarm clock (4.2BSD)
SIGXCPU	24,24,30	Core	CPU time limit exceeded (4.2BSD)
SIGXFSZ	25,25,31	Core	File size limit exceeded (4.2BSD)

received_signal(sig_name: str) → bool

estimated_start_time()

Return date with estimated start time. None if it cannot be detected

get_info(**kwargs)

get_nodes(**kwargs)

get_stats(**kwargs)

class abipy.flowtk.qjobs.ShellJob(queue_id, qname='UnknownQueue')

Bases: QueueJob

Handler for Shell jobs.

QTYPE = 'shell'

class abipy.flowtk.qjobs.SlurmJob(queue_id, qname='UnknownQueue')

Bases: QueueJob

Handler for Slurm jobs.

QTYPE = 'slurm'

estimated_start_time()

Return date with estimated start time. None if it cannot be detected

get_info(**kwargs)

get_stats(**kwargs) → dict

class abipy.flowtk.qjobs.PbsProJob(queue_id, qname='UnknownQueue')

Bases: QueueJob

Handler for PbsPro Jobs.

See also https://github.com/plediii/pbs_util for a similar project.

QTYPE = 'pbspro'

PBSSTAT_TO_SLURM = {'E': <JobStatus: FAILED, at 139955598824320>, 'F': <JobStatus: COMPLETED, at 139955598823488>, 'Q': <JobStatus: PENDING, at 139955598822976>, 'R': <JobStatus: RUNNING, at 139955598823104>, 'S': <JobStatus: SUSPENDED, at 139955598823936>}

estimated_start_time()

Return date with estimated start time. None if it cannot be detected

get_info(**kwargs)

class abipy.flowtk.qjobs.TorqueJob(queue_id, qname='UnknownQueue')

Bases: QueueJob

Not supported

QTYPE = 'torque'

class abipy.flowtk.qjobs.SgeJob(queue_id, qname='UnknownQueue')

Bases: QueueJob

Not supported

QTYPE = 'sge'

class abipy.flowtk.qjobs.MoabJob(queue_id, qname='UnknownQueue')

Bases: QueueJob

Not supported

QTYPE = 'moab'

class abipy.flowtk.qjobs.BlueGeneJob(queue_id, qname='UnknownQueue')

Bases: QueueJob

Not supported

QTYPE = 'bluegene'

qutils Module

Collection of low-level tools to facilitate the interface with resource managers.

The preferred way of importing this module is:

import qutils as qu

abipy.flowtk.qutils.slurm_parse_timestr(s: str) -> functools.partial(<class 'pymatgen.core.units.FloatWithUnit'>, unit_type='time')

A slurm time parser. Accepts a string in one the following forms:

# “days-hours”, # “days-hours:minutes”, # “days-hours:minutes:seconds”. # “minutes”, # “minutes:seconds”, # “hours:minutes:seconds”,

Returns: Time in seconds.

Raises:

ValueError –

abipy.flowtk.qutils.time2slurm(timeval: float, unit='s') → str

Convert a number representing a time value in the given unit (Default: seconds) to a string following the slurm convention: “days-hours:minutes:seconds”.

>>> assert time2slurm(61) == '0-0:1:1' and time2slurm(60*60+1) == '0-1:0:1'
>>> assert time2slurm(0.5, unit="h") == '0-0:30:0'

abipy.flowtk.qutils.time2pbspro(timeval: float, unit='s') → str

Convert a number representing a time value in the given unit (Default: seconds) to a string following the PbsPro convention: “hours:minutes:seconds”.

>>> assert time2pbspro(2, unit="d") == '48:0:0'

abipy.flowtk.qutils.time2loadlever(timeval: float, unit='s') → str

Convert a number representing a time value in the given unit (Default: seconds) to a string following the LoadLever convention. format hh:mm:ss (hours:minutes:seconds)

>>> assert time2loadlever(2, unit="d") == '48:00:00'

abipy.flowtk.qutils.timelimit_parser(s)

Convert a float or a string into time in seconds.

abipy.flowtk.qutils.any2mb(s)

Convert string or number to memory in megabytes.

abipy.flowtk.qutils.slurm_get_jobs() → dict[int, dict]

Invoke squeue, parse output and return list of dictionaries with job info indexed by job id.

class abipy.flowtk.qutils.SlurmJobArray(header: str, command: str, arr_options: list[str])

Bases: object

Example:

header = ‘’’#!/bin/bash

#SBATCH –account=battab #SBATCH –job-name=abiml_md #SBATCH –time=0-16:0:0 #SBATCH –partition=batch #SBATCH –nodes=1 # 1 node has 128 cores #SBATCH –ntasks-per-node=1 #SBATCH –cpus-per-task=1

conda activate env3.10 export OMP_NUM_THREADS=$SLURM_CPUS_PER_TASK ulimit -s unlimited ‘’’

command = “abiml.py md” arr_options = [”–help”, “–version”] job_array = SlurmJobArray(header, command, arr_options) print(job_array) queue_id = job_array.sbatch(“job_array.sh”)

sbatch(slurm_filepath: str | PathLike) → int

Write slurm submission script to slurm_filepath and submits it. Return Slurm JOB id.

abipy.flowtk.qutils.slurm_write_and_sbatch(script_filepath: str, slurm_script_str: str) → int

Write job script and submit it to the queue with Slurm sbatch. Return Slurm JOB ID.

abipy.flowtk.qutils.slurm_sbatch(slurm_filepath: str | PathLike) → int

Submit a job script to the queue with Slurm sbatch. Return Slurm JOB ID.

abipy.flowtk.qutils.get_sacct_info()

Run the sacct command to get the job information

abipy.flowtk.qutils.get_completed_job_info(job_id: int | str)

abipy.flowtk.qutils.get_slurm_template(body: str) → str

Return template for slurm submission that is supposed to be customized by the user.

abipy.flowtk.qutils.get_custodian_template() → str

tasks Module

This module provides functions and classes related to Task objects.

class abipy.flowtk.tasks.TaskManager(**kwargs)

Bases: MSONable

A TaskManager is responsible for the generation of the job script, the specification of the parameters passed to the resource manager (e.g. Slurm, PBS, etc). and the submission of the task. A TaskManager uses a QueueAdapter to perform most of this work A TaskManager has a TaskPolicy object that governs the specification of the parameters for the parallel executions. Ideally, the TaskManager should be the main entry point used by the task to deal with job submission/optimization

YAML_FILE = 'manager.yml'

USER_CONFIG_DIR = '/home/runner/.abinit/abipy'

ENTRIES = {'policy', 'qadapters'}

classmethod autodoc() → str

classmethod get_simple_manager() → str

classmethod from_user_config() → TaskManager

Initialize the abipy.flowtk.tasks.TaskManager from the YAML file ‘manager.yaml’. Search first in the working directory and then in the AbiPy configuration directory.

Raises:

RuntimeError if file is not found. –

classmethod from_file(filepath: str) → TaskManager

Read the configuration parameters from the Yaml file filepath.

classmethod from_string(s: str) → TaskManager

Create an instance from string s containing a YAML dictionary.

classmethod as_manager(obj: Any) → TaskManager

Convert obj into TaskManager instance. Accepts string, filepath, dictionary, TaskManager object. If obj is None, the manager is initialized from the user config file.

classmethod from_dict(d: dict) → TaskManager

Create an instance from a dictionary.

as_dict() → dict

A JSON serializable dict representation of an object.

abinit_build()

AbinitBuild object with Abinit version and options used to build the code

to_shell_manager(mpi_procs: int = 1) → TaskManager

Returns a new abipy.flowtk.tasks.TaskManager with the same parameters as self but replace the QueueAdapter with a ShellAdapter with mpi_procs so that we can submit the job without passing through the queue.

new_with_fixed_mpi_omp(mpi_procs: int, omp_threads: int) → TaskManager

Return a new TaskManager in which autoparal has been disabled. The Task will be executed with mpi_procs MPI processes and omp_threads OpenMP threads. Useful for generating input files for benchmarks or enforcing a certain number of procs if the Task does not support autoparal.

property has_queue: bool

True if we are submitting jobs via a queue manager.

property qads: list[QueueAdapter]

List of QueueAdapter objects sorted according to priorities (highest one comes first)

property qadapter: QueueAdapter

The qadapter used to submit jobs.

select_qadapter(pconfs)

Given a list of parallel configurations, pconfs, this method select an optimal configuration according to some criterion as well as the QueueAdapter to use.

Parameters:

pconfs – ParalHints object with the list of parallel configurations

Returns:

ParallelConf object with the optimal configuration.

property has_omp: bool

True if we are using OpenMP parallelization.

property num_cores: int

Total number of CPUs used to run the task.

property mpi_procs: int

Number of MPI processes.

property mem_per_proc: float

Memory per MPI process.

property omp_threads: int

Number of OpenMP threads

deepcopy() → TaskManager

Deep copy of self.

set_mpi_procs(mpi_procs: int) → None

Set the number of MPI processes to use.

set_omp_threads(omp_threads: int) → None

Set the number of OpenMp threads to use.

set_mem_per_proc(mem_mb: float) → None

Set the memory (in Megabytes) per CPU.

property max_cores: int

Maximum number of cores that can be used. This value is mainly used in the autoparal part to get the list of possible configurations.

get_njobs_in_queue(username=None)

returns the number of jobs in the queue, returns None when the number of jobs cannot be determined.

Parameters:

username – (str) the username of the jobs to count (default is to autodetect)

cancel(job_id)

Cancel the job. Returns exit status.

write_jobfile(task: Task, **kwargs) → str

Write the submission script. Return the path of the script

kwargs	Meaning
exec_args	List of arguments passed to task.executable. Default: no arguments.

launch(task: Task, **kwargs)

Build the input files and submit the task via the Qadapter

Parameters:

task – abipy.flowtk.tasks.Task object.

Returns: Process object.

increase_mem()

increase_ncpus()

increase the number of cpus, first ask the current qadapter, if that one raises a QadapterIncreaseError switch to the next qadapter. If all fail raise an ManagerIncreaseError

increase_resources()

exclude_nodes(nodes)

increase_time()

class abipy.flowtk.tasks.AbinitBuild(workdir=None, manager=None)

Bases: object

This object stores information on the options used to build Abinit

info

String with build information as produced by `abinit -b`

version

Abinit version number e.g 8.0.1 (string)

has_netcdf

True if netcdf is enabled.

has_libxc

True if libxc is enabled.

has_omp

True if OpenMP is enabled.

has_mpi

True if MPI is enabled.

has_mpiio

True if MPI-IO is supported.

version_ge(version_string)

True is Abinit version is >= version_string

compare_version(version_string, op)

Compare Abinit version to version_string with operator op

class abipy.flowtk.tasks.ParalHintsParser

Bases: object

Error

alias of ParalHintsError

add_error(errmsg: str) → None

parse(filename: str) → ParalHints

Read the AutoParal section (YAML format) from filename. Assumes the file contains only one section.

class abipy.flowtk.tasks.ParalHints(info: dict, confs: list[dict])

Bases: Iterable

Iterable with the hints for the parallel execution reported by ABINIT.

Error

alias of ParalHintsError

classmethod from_mpi_omp_lists(mpi_procs: int, omp_threads: int) → ParalHints

Build a list of Parallel configurations from two lists with the number of MPI processes and the number of OpenMP threads i.e. product(mpi_procs, omp_threads).

The configuration have parallel efficiency set to 1.0 and no input variables. Mainly used for preparing benchmarks.

max_cores()

Maximum number of cores.

max_mem_per_proc()

Maximum memory per MPI process.

max_speedup()

Maximum speedup.

max_efficiency()

Maximum parallel efficiency.

as_dict(**kwargs) → dict

classmethod from_dict(d: dict) → ParalHints

copy() → ParalHints

Shallow copy

get_dataframe() → DataFrame

select_with_condition(condition, key=None) → None

Remove all the configurations that do not satisfy the given condition.

Parameters:

• condition – dict or Condition object with operators expressed with a Mongodb-like syntax

• key – Selects the sub-dictionary on which condition is applied, e.g. key=”vars” if we have to filter the configurations depending on the values in vars

sort_by_efficiency(reverse=True) → ParalHints

Sort the configurations in place. items with highest efficiency come first

sort_by_speedup(reverse=True) → ParalHints

Sort the configurations in place. items with highest speedup come first

sort_by_mem_per_proc(reverse=False) → ParalHints

Sort the configurations in place. items with lowest memory per proc come first.

multidimensional_optimization(priorities=('speedup', 'efficiency'))

get_ordered_with_policy(policy, max_ncpus) → ParalHints

Sort and return a new list of configurations ordered according to the TaskPolicy policy.

class abipy.flowtk.tasks.AbinitTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: Task

Base class for ABINIT Tasks.

Results

alias of TaskResults

classmethod from_input(input: AbinitInput, workdir=None, manager=None) → AbinitTask

Create an instance of AbinitTask from an ABINIT input.

Parameters:

• ainput – abipy.abio.inputs.AbinitInput object.

• workdir – Path to the working directory.

• manager – abipy.flowtk.tasks.TaskManager object.

classmethod temp_shell_task(inp: AbinitInput, mpi_procs=1, workdir=None, manager=None) → AbinitTask

Build a Task with a temporary workdir. The task is executed via the shell with 1 MPI proc. Mainly used for invoking Abinit to get important parameters needed to prepare the real task.

Parameters:

mpi_procs – Number of MPI processes to use.

setup() → None

Abinit has the very bad habit of changing the file extension by appending the characters in [A,B …, Z] to the output file, and this breaks a lot of code that relies of the use of a unique file extension. Here we fix this issue by renaming run.abo to run.abo_[number] if the output file “run.abo” already exists. A few lines of code in python, a lot of problems if you try to implement this trick in Fortran90.

property executable: str

Path to the executable required for running the Task.

property pseudos

List of pseudos used in the calculation.

property isnc: bool

True if norm-conserving calculation.

property ispaw: bool

True if PAW calculation

property is_gs_task: bool

True if task is GsTask subclass.

property is_dfpt_task: bool

True if task is a DftpTask subclass.

cycle_class()

Return the subclass of ScfCycle associated to the task or None if no SCF algorithm if associated to the task.

property filesfile_string: str

String with the list of files and prefixes needed to execute ABINIT.

set_pconfs(pconfs) → None

Set the list of autoparal configurations.

property pconfs

List of autoparal configurations.

uses_paral_kgb(value=1)

True if the task is a GS Task and uses paral_kgb with the given value.

get_panel(**kwargs)

Build panel with widgets to interact with the Task either in a notebook or in panel app. This is the implementation provided by the base class. Subclasses may provide specialized implementations.

autoparal_run() → int

Find an optimal set of parameters for the execution of the task This method can change the ABINIT input variables and/or the submission parameters e.g. the number of CPUs for MPI and OpenMp.

Set:

self.pconfs where pconfs is a ParalHints object with the configuration reported by autoparal and optimal is the optimal configuration selected. Returns 0 if success

find_optconf(pconfs)

Find the optimal Parallel configuration.

select_files(what='o')

Helper function used to select the files of a task.

Parameters:

what – string with the list of characters selecting the file type Possible choices: i ==> input_file, o ==> output_file, f ==> files_file, j ==> job_file, l ==> log_file, e ==> stderr_file, q ==> qout_file, all ==> all files.

restart() → None

general restart used when scheduler problems have been taken care of

reset_from_scratch()

Restart from scratch, this is to be used if a job is restarted with more resources after a crash

Move output files produced in workdir to _reset otherwise check_status continues to see the task as crashed even if the job did not run

fix_abicritical() → int

method to fix crashes/error caused by abinit

Returns:

1 if task has been fixed else 0.

fix_queue_critical() → int

This function tries to fix critical events originating from the queue submission system.

General strategy, first try to increase resources in order to fix the problem, if this is not possible, call a task specific method to attempt to decrease the demands.

Returns:

1 if task has been fixed else 0.

parse_timing() → AbinitTimerParser | None

Parse the timer data in the main output file of Abinit. Requires timopt /= 0 in the input file (usually timopt = -1)

Return: AbinitTimerParser instance, None if error.

get_output_file()

Parse the main output file in text format, return AbinitOutputFile object.

Example

with task.get_output_file() as out:

dims_dataset, spginfo_dataset = out.get_dims_spginfo_dataset(verbose=0)

class abipy.flowtk.tasks.ScfTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: GsTask

Task for self-consistent GS calculations. Provide support for in-place restart via (WFK|DEN) files.

CRITICAL_EVENTS = [<class 'abipy.flowtk.events.ScfConvergenceWarning'>]

color_rgb = array([1., 0., 0.])

restart()

SCF calculations can be restarted if we have either the WFK file or the DEN file.

inspect(**kwargs)

Plot the SCF cycle results with matplotlib.

Returns: matplotlib.figure.Figure or None if some error occurred.

add_ebands_task_to_work(work, ndivsm=15, tolwfr=1e-20, nscf_nband=None, nb_extra=10)

Generate an NSCF task for band structure calculation from a GS SCF task and add it to the work.

Parameters:

• work

• ndivsm – if > 0, it’s the number of divisions for the smallest segment of the path (Abinit variable). if < 0, it’s interpreted as the pymatgen line_density parameter in which the number of points in the segment is proportional to its length. Typical value: -20. This option is the recommended one if the k-path contains two high symmetry k-points that are very close as ndivsm > 0 may produce a very large number of wavevectors.

• tolwfr – Tolerance on residuals for NSCF calculation.

• nscf_nband – Number of bands for NSCF calculation. If None, use nband + nb_extra

return: NscfTask object.

class abipy.flowtk.tasks.NscfTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: GsTask

Task for non-self-consistent GS calculations. Provides in-place restart via the WFK file and a specialized setup method that enforces the same ngfft FFT-mesh as the one used in the previous GS task.

CRITICAL_EVENTS = [<class 'abipy.flowtk.events.NscfConvergenceWarning'>]

color_rgb = array([0.78431373, 0.31372549, 0.39215686])

setup()

NSCF calculations should use the same FFT mesh as the one employed in the GS task (in principle, it’s possible to interpolate inside Abinit but tests revealed some numerical noise Here we change the input file of the NSCF task to have the same FFT mesh.

restart()

NSCF calculations can be restarted only if we have the WFK file.

class abipy.flowtk.tasks.RelaxTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: GsTask, ProduceHist

Task for structural optimizations.

CRITICAL_EVENTS = [<class 'abipy.flowtk.events.RelaxConvergenceWarning'>]

color_rgb = array([1.        , 0.23921569, 1.        ])

get_final_structure() → Structure

Read the final structure from the GSR file.

restart()

Restart the structural relaxation.

Structure relaxations can be restarted only if we have the WFK file or the DEN or the GSR file from which we can read the last structure (mandatory) and the wavefunctions (not mandatory but useful). Prefer WFK over other files since we can reuse the wavefunctions.

Note

The problem in the present approach is that some parameters in the input are computed from the initial structure and may not be consistent with the modification of the structure done during the structure relaxation.

inspect(**kwargs)

Plot the evolution of the structural relaxation with matplotlib.

Parameters:

what – Either “hist” or “scf”. The first option (default) extracts data from the HIST file and plot the evolution of the structural parameters, forces, pressures and energies. The second option, extracts data from the main output file and plot the evolution of the SCF cycles (etotal, residuals, etc).

Returns: matplotlib.figure.Figure or None if some error occurred.

reduce_dilatmx(target: float = 1.01) → None

fix_ofiles() → None

Note that ABINIT produces lots of out_TIM1_DEN files for each step. Here we list all TIM*_DEN files, we select the last one and we rename it as out_DEN

This change is needed so that we can specify dependencies with the syntax {node: “DEN”} without having to know the number of iterations needed to converge.

class abipy.flowtk.tasks.DdkTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: DfptTask

Task for DDK calculations.

color_rgb = array([0. , 0.8, 0.8])

class abipy.flowtk.tasks.EffMassTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: DfptTask

Task for effective mass calculations with DFPT.

color_rgb = array([0.        , 0.47843137, 0.8       ])

class abipy.flowtk.tasks.PhononTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: DfptTask

DFPT calculations for a single atomic perturbation. Implements in-place restart via (1WF|1DEN) files and inspect method.

color_rgb = array([0.        , 0.58823529, 0.98039216])

inspect(**kwargs)

Plot the Phonon SCF cycle results with matplotlib.

Returns: matplotlib.figure.Figure or None if some error occurred.

class abipy.flowtk.tasks.ElasticTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: DfptTask

DFPT calculations for a single strain perturbation (uniaxial or shear strain). Provide support for in-place restart via (1WF|1DEN) files

color_rgb = array([1. , 0.8, 1. ])

class abipy.flowtk.tasks.SigmaTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: ManyBodyTask

Self-energy calculations with the quartic GW code. Provides support for in-place restart via QPS files.

CRITICAL_EVENTS = [<class 'abipy.flowtk.events.QPSConvergenceWarning'>]

color_rgb = array([0., 1., 0.])

restart()

general restart used when scheduler problems have been taken care of

property sigres_path: str

Absolute path of the SIGRES.nc file. Empty string if file is not present.

open_sigres()

Open the SIGRES file located in the in self.outdir. Returns abipy.electrons.gw.SigresFile object, None if file could not be found or file is not readable.

get_scissors_builder()

Returns an instance of ScissorsBuilder from the SIGRES file.

Raises:

RuntimeError –

class abipy.flowtk.tasks.EphTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: AbinitTask

Task for electron-phonon calculations with the EPH code.

color_rgb = array([1.        , 0.50196078, 0.        ])

class abipy.flowtk.tasks.KerangeTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: AbinitTask

Task for kerange calculations.

color_rgb = array([1.        , 0.50196078, 0.50196078])

class abipy.flowtk.tasks.OpticTask(optic_input: OpticInput, nscf_node: Node, ddk_nodes: list[Node], use_ddknc=False, workdir=None, manager=None)

Bases: Task

Task for the computation of optical spectra with optics tool i.e. RPA without local-field effects and velocity operator computed from DDK files.

color_rgb = array([1. , 0.8, 0.4])

set_workdir(workdir: str, chroot=False) → None

Set the working directory of the task.

set_vars(*args, **kwargs) → None

Optic does not use get or ird variables hence we should never try to change the input when we connect this task

property executable: str

Path to the executable required for running the OpticTask.

property filesfile_string: str

String with the list of files and prefixes needed to execute ABINIT.

property wfk_filepath: None

Returns (at runtime) the absolute path of the WFK file produced by the NSCF run.

property ddk_filepaths: list[str]

Returns (at runtime) the absolute path of the DDK files produced by the DDK runs.

make_input() → str

Construct and write the input file of the calculation.

setup()

Public method called before submitting the task.

make_links()

Optic allows the user to specify the paths of the input file. hence we don’t need to create symbolic links.

fix_abicritical()

Cannot fix abicritical errors for optic

reset_from_scratch()

restart from scratch, this is to be used if a job is restarted with more resources after a crash

fix_queue_critical()

This function tries to fix critical events originating from the queue submission system.

General strategy, first try to increase resources in order to fix the problem, if this is not possible, call a task specific method to attempt to decrease the demands.

Returns:

1 if task has been fixed else 0.

autoparal_run()

Find an optimal set of parameters for the execution of the Optic task This method can change the submission parameters e.g. the number of CPUs for MPI and OpenMp.

Returns 0 if success

class abipy.flowtk.tasks.AnaddbTask(anaddb_input, ddb_node, gkk_node=None, md_node=None, ddk_node=None, workdir=None, manager=None)

Bases: Task

Task for Anaddb runs (post-processing of DFPT calculations).

color_rgb = array([0.8, 0.4, 1. ])

classmethod temp_shell_task(inp, ddb_node, mpi_procs=1, gkk_node=None, md_node=None, ddk_node=None, workdir=None, manager=None) → AnaddbTask

Build a abipy.flowtk.tasks.AnaddbTask with a temporary workdir. The task is executed via the shell with 1 MPI proc. Mainly used for post-processing the DDB files.

Parameters:

• mpi_procs – Number of MPI processes to use.

• anaddb_input – string with the anaddb variables.

• ddb_node – The node that will produce the DDB file. Accept abipy.flowtk.tasks.Task, abipy.flowtk.works.Work or filepath.

See AnaddbInit for the meaning of the other arguments.

property executable: str

Path to the executable required for running the abipy.flowtk.tasks.AnaddbTask.

property filesfile_string: str

String with the list of files and prefixes needed to execute ABINIT.

property ddb_filepath: str

Returns (at runtime) the absolute path of the input DDB file.

property md_filepath: str

Returns (at runtime) the absolute path of the input MD file.

property gkk_filepath: str

Returns (at runtime) the absolute path of the input GKK file.

property ddk_filepath: str

Returns (at runtime) the absolute path of the input DKK file.

setup()

Public method called before submitting the task.

make_links()

Anaddb allows the user to specify the paths of the input file. hence we don’t need to create symbolic links.

outpath_from_ext(ext)

open_phbst()

Open PHBST file produced by Anaddb and returns abipy.dfpt.phonons.PhbstFile object.

open_phdos()

Open PHDOS file produced by Anaddb and returns abipy.dfpt.phonons.PhdosFile object.

make_input(with_header=False) → str

return string the input file of the calculation.

class abipy.flowtk.tasks.GwrTask(input: AbinitInput, workdir=None, manager=None, deps=None)

Bases: AbinitTask

Class for calculations with the GWR code. Provide open_gwr method to open GWR.nc

color_rgb = array([1.        , 0.50196078, 0.        ])

setup()

Abinit has the very bad habit of changing the file extension by appending the characters in [A,B …, Z] to the output file, and this breaks a lot of code that relies of the use of a unique file extension. Here we fix this issue by renaming run.abo to run.abo_[number] if the output file “run.abo” already exists. A few lines of code in python, a lot of problems if you try to implement this trick in Fortran90.

property gwr_path: str

Absolute path of the GWR.nc file. Empty string if file is not present.

open_gwr()

Open the GWR file located in the in self.outdir. Returns GwrFile object, None if file could not be found or file is not readable.

abipy.flowtk.tasks.set_user_config_taskmanager(task_manager: TaskManager) → None

Change the default manager returned by TaskManager.from_user_config.

utils Module

Tools and helper functions for abinit calculations

abipy.flowtk.utils.as_bool(s: str | bool) → bool

Convert a string into a boolean value.

>>> assert as_bool(True) is True and as_bool("Yes") is True and as_bool("false") is False

class abipy.flowtk.utils.File(path: str)

Bases: object

Very simple class used to store file basenames, absolute paths and directory names. Provides wrappers for the most commonly used functions defined in os.path.

property path: str

Absolute path of the file.

property basename: str

File basename.

property relpath: str

Relative path.

property dirname: str

Absolute path of the directory where the file is located.

property exists: bool

True if file exists.

property isncfile: bool

True if self is a NetCDF file

chmod(mode: str) → None

Change the access permissions of a file.

read() → str

Read data from file.

readlines() → list[str]

Read lines from files.

write(string: str)

Write string to file.

writelines(lines: list[str])

Write a list of strings to file.

make_dir() → None

Make the directory where the file is located.

remove() → None

Remove the file.

move(dst: str) → None

Recursively move a file or directory to another location. This is similar to the Unix “mv” command.

get_stat()

Results from os.stat

getsize() → int

Return the size, in bytes, of path. Return 0 if the file is empty or it does not exist.

class abipy.flowtk.utils.Directory(path: str)

Bases: object

Very simple class that provides helper functions wrapping the most commonly used functions defined in os.path.

property path: str

Absolute path of the directory.

property relpath: str

Relative path.

property basename: str

Directory basename.

path_join(*p) → str

Join two or more pathname components, inserting ‘/’ as needed. If any component is an absolute path, all previous path components will be discarded.

property exists: bool

True if file exists.

makedirs() → None

Super-mkdir; create a leaf directory and all intermediate ones. Works like mkdir, except that any intermediate path segment (not just the rightmost) will be created if it does not exist.

rmtree() → None

Recursively delete the directory tree

copy_r(dst: str) → None

Implements a recursive copy function similar to Unix’s “cp -r” command.

clean() → None

Remove all files in the directory tree while preserving the directory

path_in(file_basename: str) → str

Return the absolute path of filename in the directory.

list_filepaths(wildcard: str | None = None) → list[str]

Return the list of absolute filepaths in the directory.

Parameters:

wildcard –

String of tokens separated by “|”. Each token represents a pattern. If wildcard is not None, we return only those files whose basename matches the given shell pattern (uses fnmatch). .. rubric:: Example

wildcard=”.nc|.pdf” selects only those files that end with .nc or .pdf

need_abiext(ext: str) → str

Returns the absolute path of the ABINIT file with extension ext. Support both Fortran files and netcdf files. In the later case, we check whether a file with extension ext + “.nc” is present in the directory.

Raises: FileNotFoundError if file cannot be found.

has_abiext(ext: str, single_file: bool = True) → str

Returns the absolute path of the ABINIT file with extension ext. Support both Fortran files and netcdf files. In the later case, we check whether a file with extension ext + “.nc” is present in the directory. Returns empty string is file is not present.

Parameters:

• ext – File extension. .nc is not needed unless you enforce netcdf format.

• single_file – If None, allow for multiple matches and return the first one.

Raises:

• ValueError –

• This implies that this method is not compatible with multiple datasets. –

symlink_abiext(inext: str, outext: str) → int

Create a simbolic link (outext –> inext). The file names are implicitly given by the ABINIT file extension.

Example

outdir.symlink_abiext(‘1WF’, ‘DDK’)

creates the link out_DDK that points to out_1WF

Return: 0 if success.

Raise: RuntimeError

rename_abiext(inext: str, outext: str) → int

Rename the Abinit file with extension inext with the new extension outext

copy_abiext(inext: str, outext: str) → int

Copy the Abinit file with extension inext to a new file with the extension outext

remove_exts(exts: str | list[str]) → list[str]

Remove the files with the given extensions. Unlike rmtree, this function preserves the directory path. Return list with the absolute paths of the files that have been removed.

find_last_timden_file()

ABINIT produces lots of out_TIM1_DEN files for each step and we need to find the lat one in order to prepare the restart or to connect other tasks to the structural relaxation.

This function finds all the TIM?_DEN files in self and return a namedtuple (path, step) where path is the path of the last TIM?_DEN file and step is the iteration number. Returns None if the directory does not contain TIM?_DEN files.

find_1wf_files()

Abinit adds the idir-ipert index at the end of the 1WF file and this breaks the extension e.g. out_1WF4. This method scans the files in the directories and returns a list of namedtuple Each named tuple gives the path of the 1FK file and the pertcase index.

find_1den_files()

Abinit adds the idir-ipert index at the end of the 1DEN file and this breaks the extension e.g. out_DEN1. This method scans the files in the directories and returns a list of namedtuple Each named tuple gives the path of the 1DEN file and the pertcase index.

abipy.flowtk.utils.irdvars_for_ext(ext) → dict

Returns a dictionary with the ABINIT variables that must be used to read the file with extension ext.

abipy.flowtk.utils.abi_extensions() → list

List with all the ABINIT extensions that are registered.

abipy.flowtk.utils.abi_splitext(filename: str) → tuple[str, str]

Split the ABINIT extension from a filename. “Extension” are found by searching in an internal database.

Returns “(root, ext)” where ext is the registered ABINIT extension The final “.nc” is included (if any)

>>> assert abi_splitext("foo_WFK") == ('foo_', 'WFK')
>>> assert abi_splitext("/home/guido/foo_bar_WFK.nc") == ('foo_bar_', 'WFK.nc')

class abipy.flowtk.utils.FilepathFixer

Bases: object

This object modifies the names of particular output files produced by ABINIT so that the file extension is preserved. Having a one-to-one mapping between file extension and data format is indeed fundamental for the correct behaviour of abinit since:

• We locate the output file by just inspecting the file extension

• We select the variables that must be added to the input file on the basis of the extension specified by the user during the initialization of the AbinitFlow.

Unfortunately, ABINIT developers like to append extra stuff to the initial extension therefore we have to call FilepathFixer to fix the output files produced by the run.

Example

fixer = FilepathFixer() fixer.fix_paths(‘/foo/out_1WF17’) == {‘/foo/out_1WF17’: ‘/foo/out_1WF’} fixer.fix_paths(‘/foo/out_1WF5.nc’) == {‘/foo/out_1WF5.nc’: ‘/foo/out_1WF.nc’}

fix_paths(paths) → dict

Fix the filenames in the iterable paths

Returns:

Mapping old_path –> new_path

Return type:

old2new

abipy.flowtk.utils.map2rpn(map, obj)

Convert a Mongodb-like dictionary to an RPN list of operands and operators.

Reverse Polish notation (RPN) is a mathematical notation in which every operator follows all of its operands, e.g.

3 - 4 + 5 –> 3 4 - 5 +

>>> d = {2.0: {'$eq': 1.0}}
>>> assert map2rpn(d, None) == [2.0, 1.0, '$eq']

abipy.flowtk.utils.evaluate_rpn(rpn)

Evaluates the RPN form produced my map2rpn.

Returns: bool

class abipy.flowtk.utils.Condition(cmap=None)

Bases: object

This object receives a dictionary that defines a boolean condition whose syntax is similar to the one used in mongodb (albeit not all the operators available in mongodb are supported here).

Example:

$gt: {field: {$gt: value} }

$gt selects those documents where the value of the field is greater than (i.e. >) the specified value.

$and performs a logical AND operation on an array of two or more expressions (e.g. <expression1>, <expression2>, etc.) and selects the documents that satisfy all the expressions in the array.

{ $and: [ { <expression1> }, { <expression2> } , … , { <expressionN> } ] }

Consider the following example:

db.inventory.find( { qty: { $gt: 20 } } ) This query will select all documents in the inventory collection where the qty field value is greater than 20. Consider the following example:

db.inventory.find( { qty: { $gt: 20 } } ) db.inventory.find({ $and: [ { price: 1.99 }, { qty: { $lt: 20 } }, { sale: true } ] } )

classmethod as_condition(obj)

Convert obj into Condition

class abipy.flowtk.utils.Editor(editor=None)

Bases: object

Wrapper class that calls the editor specified by the user or the one specified in the $EDITOR env variable.

edit_files(fnames, ask_for_exit=True)

edit_file(fname)

static user_wants_to_exit()

Show an interactive prompt asking if exit is wanted.

class abipy.flowtk.utils.SparseHistogram(items, key=None, num=None, step=None)

Bases: object

plot(ax=None, **kwargs)

Plot the histogram with matplotlib, returns matplotlib figure.

Keyword arguments controlling the display of the figure:

kwargs	Meaning
title	Title of the plot (Default: None).
show	True to show the figure (default: True).
savefig	“abc.png” or “abc.eps” to save the figure to a file.
size_kwargs	Dictionary with options passed to fig.set_size_inches e.g. size_kwargs=dict(w=3, h=4)
tight_layout	True to call fig.tight_layout (default: False)
ax_grid	True (False) to add (remove) grid from all axes in fig. Default: None i.e. fig is left unchanged.
ax_annotate	Add labels to subplots e.g. (a), (b). Default: False
fig_close	Close figure. Default: False.
plotly	Try to convert mpl figure to plotly.

class abipy.flowtk.utils.Dirviz(top)

Bases: object

get_cluster_graph(engine='fdp', graph_attr=None, node_attr=None, edge_attr=None)

Generate directory graph in the DOT language. Directories are shown as clusters

Warning

This function scans the entire directory tree starting from top so the resulting graph can be really big.

Parameters:

• engine – Layout command used. [‘dot’, ‘neato’, ‘twopi’, ‘circo’, ‘fdp’, ‘sfdp’, ‘patchwork’, ‘osage’]

• graph_attr – Mapping of (attribute, value) pairs for the graph.

• node_attr – Mapping of (attribute, value) pairs set for all nodes.

• edge_attr – Mapping of (attribute, value) pairs set for all edges.

Returns: graphviz.Digraph <https://graphviz.readthedocs.io/en/stable/api.html#digraph>

works Module

Works for Abinit

class abipy.flowtk.works.Work(workdir=None, manager=None)

Bases: BaseWork, NodeContainer

A Work is a list of (possibly connected) tasks.

Inheritance Diagram

set_manager(manager: TaskManager) → None

Set the abipy.flowtk.tasks.TaskManager to be used to launch the abipy.flowtk.tasks.Task.

property flow: Flow

The flow containing this abipy.flowtk.works.Work.

set_flow(flow: Flow) → None

Set the flow associated to this abipy.flowtk.works.Work.

pos()

The position of work in the abipy.flowtk.flows.Flow

property pos_str: str

String representation of self.pos

set_workdir(workdir: str, chroot=False) → None

Set the working directory. Cannot be set more than once unless chroot is True

chroot(new_workdir: str) → None

postpone_on_all_ok()

This method should be called when additional tasks are added to the Work at runtime. e.g inside an on_all_ok method. This call triggers the allocations of new tasks so that the scheduler will continue execution since new tasks have been added and the finalized flag is set to False.

chunks(chunk_size) → list[Task]

Yield successive chunks of tasks of lenght chunk_size.

ipath_from_ext(ext: str) → str

Returns the path of the input file with extension ext. Use it when the file does not exist yet.

opath_from_ext(ext: str) → str

Returns the path of the output file with extension ext. Use it when the file does not exist yet.

get_all_outdata_files_with_ext(ext: str) → list

Returns list with all the output files produced in outdata with extension ext.

property processes: list

Return a list of objects that support the subprocess.Popen protocol.

property all_done: bool

True if all the abipy.flowtk.tasks.Task objects in the abipy.flowtk.works.Work are done.

property isnc: bool

True if norm-conserving calculation.

property ispaw: bool

True if PAW calculation.

property status_counter: Counter

Returns a Counter object that counts the number of task with given status (use the string representation of the status as key).

allocate(manager=None) → None

This function is called once we have completed the initialization of the abipy.flowtk.works.Work. It sets the manager of each task (if not already done) and defines the working directories of the tasks.

Parameters:

manager – abipy.flowtk.tasks.TaskManager object or None

register(obj: AbinitInput | Task, deps=None, required_files=None, manager=None, task_class=None) → Task

Registers a new abipy.flowtk.tasks.Task and add it to the internal list, taking into account possible dependencies.

Parameters:

• obj – abipy.abio.inputs.AbinitInput instance or abipy.flowtk.tasks.Task object.

• deps – Dictionary specifying the dependency of this node or list of dependencies None means that this obj has no dependency.

• required_files – List of strings with the path of the files used by the task. Note that the files must exist when the task is registered. Use the standard approach based on Works, Tasks and deps if the files will be produced in the future.

• manager – The abipy.flowtk.tasks.TaskManager responsible for the submission of the task. If manager is None, we use the TaskManager specified during the creation of the abipy.flowtk.works.Work.

• task_class – Task subclass to instantiate. Default: AbinitTask

Returns: abipy.flowtk.tasks.Task object

register_task(obj: AbinitInput | Task, deps=None, required_files=None, manager=None, task_class=None) → Task

Registers a new abipy.flowtk.tasks.Task and add it to the internal list, taking into account possible dependencies.

Parameters:

• obj – abipy.abio.inputs.AbinitInput instance or abipy.flowtk.tasks.Task object.

• deps – Dictionary specifying the dependency of this node or list of dependencies None means that this obj has no dependency.

• required_files – List of strings with the path of the files used by the task. Note that the files must exist when the task is registered. Use the standard approach based on Works, Tasks and deps if the files will be produced in the future.

• manager – The abipy.flowtk.tasks.TaskManager responsible for the submission of the task. If manager is None, we use the TaskManager specified during the creation of the abipy.flowtk.works.Work.

• task_class – Task subclass to instantiate. Default: AbinitTask

Returns: abipy.flowtk.tasks.Task object

path_in_workdir(filename: str) → str

Create the absolute path of filename in the working directory.

setup(*args, **kwargs) → None

Method called before running the calculations. The default implementation is empty.

build(*args, **kwargs) → None

Creates the top level directory.

property status: Status

Returns the status of the work i.e. the minimum of the status of the tasks.

get_all_status(only_min=False)

Returns a list with the status of the tasks in self.

Parameters:

only_min – If True, the minimum of the status is returned.

check_status() → None

Check the status of the tasks.

has_different_structures(rtol=1e-05, atol=1e-08) → str

Check if structures are equivalent, return string with info about differences (if any).

get_panel(**kwargs)

Build panel with widgets to interact with the Work either in a notebook or in panel app. This is the implementation provided by the base class. Subclasses may provide specialized implementations.

get_dataframe(as_dict=False) → DataFrame

Return pandas dataframe task info or dictionary if as_dict is True. This function should be called after work.get_status to update the status.

rmtree(exclude_wildcard: str = '') → None

Remove all files and directories in the working directory

Parameters:

exclude_wildcard – Optional string with regular expressions separated by |. Files matching one of the regular expressions will be preserved. example: exclude_wildard=”.nc|.txt” preserves all the files whose extension is in [“nc”, “txt”].

rm_indatadir() → None

Remove all the indata directories.

rm_outdatadir() → None

Remove all the indata directories.

rm_tmpdatadir() → None

Remove all the tmpdata directories.

move(dest, isabspath=False) → None

Recursively move self.workdir to another location. This is similar to the Unix “mv” command. The destination path must not already exist. If the destination already exists but is not a directory, it may be overwritten depending on os.rename() semantics.

Be default, dest is located in the parent directory of self.workdir, use isabspath=True to specify an absolute path.

submit_tasks(wait=False) → None

Submits the task in self and wait. TODO: change name.

start(*args, **kwargs) → None

Start the work. Calls build and _setup first, then submit the tasks. Non-blocking call unless wait is set to True

read_etotals(unit='Ha') -> functools.partial(<class 'pymatgen.core.units.ArrayWithUnit'>, unit_type='energy')

Reads the total energy from the GSR file produced by the task.

Return a numpy array with the total energies in Hartree The array element is set to np.inf if an exception is raised while reading the GSR file.

parse_timers() → AbinitTimerParser

Parse the TIMER section reported in the ABINIT output files.

Returns:

AbinitTimerParser object

class abipy.flowtk.works.BandStructureWork(scf_input: AbinitInput, nscf_input: AbinitInput, dos_inputs=None, workdir=None, manager=None)

Bases: Work

Work for band structure calculations.

The first task performs the GS-SCF calculations, the second one computes the band dispersion along an high-symmetry k-path. Finally, we have a list of NSCF tasks computing the e-DOS.

Inheritance Diagram

classmethod from_scf_input(scf_input: AbinitInput, dos_ngkpt, nb_extra=10, ndivsm=-20, dos_shiftk=(0, 0, 0), prtdos=3) → BandStructureWork

Build a BandStructureWork from an abipy.abio.inputs.AbinitInput representing a GS-SCF calculation.

Parameters:

• scf_input – Input for the GS-SCF run.

• dos_ngkpt – K-mesh for e-DOS. Set it to None to skip this task.

• nb_extra – Extra bands to be added to the input nband.

• ndivsm – if > 0, it’s the number of divisions for the smallest segment of the path (Abinit variable). if < 0, it’s interpreted as the pymatgen line_density parameter in which the number of points in the segment is proportional to its length. Typical value: -20. This option is the recommended one if the k-path contains two consecutive high symmetry k-points that are very close as ndivsm > 0 may produce a very large number of wavevectors.

• prtdos – By default, we compute L-projections with tetrahedron method. Set prtdos to zero to deactivate L-projections.

plot_ebands(**kwargs) → Any

Plot the band structure. kwargs are passed to the plot method of abipy.electrons.ebands.ElectronBands.

Return: matplotlib.figure.Figure

plot_ebands_with_edos(dos_pos=0, method='gaussian', step=0.01, width=0.1, **kwargs) → Any

Plot the band structure and the DOS.

Parameters:

• dos_pos – Index of the task from which the DOS should be obtained (note: 0 refers to the first DOS task).

• method – String defining the method for the computation of the DOS.

• step – Energy step (eV) of the linear mesh.

• width – Standard deviation (eV) of the gaussian.

• kwargs – Keyword arguments passed to plot_with_edos method to customize the plot.

Return: matplotlib.figure.Figure

plot_edoses(dos_pos=None, method='gaussian', step=0.01, width=0.1, **kwargs) → Any

Plot the band structure and the DOS.

Parameters:

• dos_pos – Index of the task from which the DOS should be obtained. None is all DOSes should be displayed. Accepts integer or list of integers.

• method – String defining the method for the computation of the DOS.

• step – Energy step (eV) of the linear mesh.

• width – Standard deviation (eV) of the gaussian.

• kwargs – Keyword arguments passed to plot method to customize the plot.

Return: matplotlib.figure.Figure

class abipy.flowtk.works.RelaxWork(ion_input: AbinitInput, ioncell_input: AbinitInput, workdir=None, manager=None, target_dilatmx=None)

Bases: Work

Work for structural relaxations. The first task relaxes the atomic position while keeping the unit cell parameters fixed. The second task uses the final structure to perform a structural relaxation in which both the atomic positions and the lattice parameters are optimized.

Inheritance Diagram

on_ok(sender)

This callback is called when one task reaches status S_OK. If sender == self.ion_task, we update the initial structure used by self.ioncell_task and we unlock it so that the job can be submitted.

plot_ion_relaxation(**kwargs) → Any

Plot the history of the ion-cell relaxation. kwargs are passed to the plot method of abipy.dynamics.hist.HistFile

Return: matplotlib.figure.Figure or None if hist file is not found.

plot_ioncell_relaxation(**kwargs) → Any

Plot the history of the ion-cell relaxation. kwargs are passed to the plot method of abipy.dynamics.hist.HistFile

Return: matplotlib.figure.Figure or None if hist file is not found.

class abipy.flowtk.works.G0W0Work(scf_inputs, nscf_inputs, scr_inputs, sigma_inputs, workdir=None, manager=None)

Bases: Work

Work for generic G0W0 calculations. All input can be either single inputs or lists of inputs

Inheritance Diagram

class abipy.flowtk.works.QptdmWork(workdir=None, manager=None)

Bases: Work

This work parallelizes the calculation of the q-points of the screening. It also provides the callback on_all_ok that calls mrgscr to merge all the partial screening files produced.

Inheritance Diagram

create_tasks(wfk_file: str, scr_input: AbinitInput)

Create the SCR tasks and register them in self.

Parameters:

• wfk_file – Path to the ABINIT WFK file to use for the computation of the screening.

• scr_input – Input for the screening calculation.

merge_scrfiles(remove_scrfiles=True) → str

This method is called when all the q-points have been computed. It runs mrgscr in sequential on the local machine to produce the final SCR file in the outdir of the Work. If remove_scrfiles is True, the partial SCR files are removed after the merge.

on_all_ok()

This method is called when all the q-points have been computed. It runs mrgscr in sequential on the local machine to produce the final SCR file in the outdir of the Work.

class abipy.flowtk.works.SigmaConvWork(wfk_node, scr_node, sigma_inputs, workdir=None, manager=None)

Bases: Work

Work for self-energy convergence studies.

Inheritance Diagram

class abipy.flowtk.works.BseMdfWork(scf_input: AbinitInput, nscf_input: AbinitInput, bse_inputs, workdir=None, manager=None)

Bases: Work

Work for BSE calculations in which the self-energy corrections are approximated by the scissors operator and the screening is modeled with the model dielectric function.

Inheritance Diagram

get_mdf_robot()

Builds and returns a MdfRobot for analyzing the results in the MDF files.

class abipy.flowtk.works.PhononWork(workdir=None, manager=None)

Bases: Work, MergeDdb

This work consists of nirred Phonon tasks where nirred is the number of irreducible atomic perturbations for a given set of q-points. It provides the callback method (on_all_ok) that calls mrgddb and mrgdv to merge all the partial DDB (POT) files produced. The two files are available in the output directory of the Work.

Inheritance Diagram

classmethod from_scf_task(scf_task: ScfTask, qpoints, is_ngqpt=False, with_becs=False, with_quad=False, with_flexoe=False, with_dvdb=True, tolerance=None, ddk_tolerance=None, ndivsm=0, qptopt=1, prtwf=-1, prepgkk=0, manager=None) → PhononWork

Construct a PhononWork from a abipy.flowtk.tasks.ScfTask object. The input file for phonons is automatically generated from the input of the ScfTask. Each phonon task depends on the WFK file produced by the scf_task.

Parameters:

• scf_task – abipy.flowtk.tasks.ScfTask object.

• qpoints – q-points in reduced coordinates. Accepts single q-point, list of q-points or three integers defining the q-mesh if is_ngqpt.

• is_ngqpt – True if qpoints should be interpreted as divisions instead of q-points.

• with_becs – Activate calculation of Electric field and Born effective charges.

• with_quad – Activate calculation of dynamical quadrupoles. Require with_becs Note that only selected features are compatible with dynamical quadrupoles. Please consult <https://docs.abinit.org/topics/longwave/>

• with_flexoe – True to activate computation of flexoelectric tensor. Require with_becs

• with_dvdb – True to merge POT1 files associated to atomic perturbations in the DVDB file at the end of the calculation

• tolerance – dict {“varname”: value} with the tolerance to be used in the phonon run. None to use AbiPy default.

• ddk_tolerance – dict {“varname”: value} with the tolerance used in the DDK run if with_becs. None to use AbiPy default.

• ndivsm – If different from zero, activate computation of electron band structure. if > 0, it’s the number of divisions for the smallest segment of the path (Abinit variable). if < 0, it’s interpreted as the pymatgen line_density parameter in which the number of points in the segment is proportional to its length. Typical value: -20. This option is the recommended one if the k-path contains two high symmetry k-points that are very close as ndivsm > 0 may produce a very large number of wavevectors.

• qptopt – Option for the generation of q-points. Default: 1

• prtwf – Controls the output of the first-order WFK. By default we set it to -1 when q != 0 so that AbiPy is still able to restart the DFPT task if the calculation is not converged (worst case scenario) but we avoid the output of the 1-st order WFK if the calculation converged successfully. Non-linear DFPT tasks should not be affected since they assume q == 0.

• prepgkk – option to compute only the irreducible preturbations or all perturbations for the chosen set of q-point. Default: 0 (irred. pret. only)

• manager – abipy.flowtk.tasks.TaskManager object.

classmethod from_scf_input(scf_input: AbinitInput, qpoints, is_ngqpt=False, with_becs=False, with_quad=False, with_flexoe=False, with_dvdb=True, tolerance=None, ddk_tolerance=None, ndivsm=0, qptopt=1, prtwf=-1, prepgkk=0, manager=None) → PhononWork

Similar to from_scf_task, the difference is that this method requires an input for SCF calculation. A new abipy.flowtk.tasks.ScfTask is created and added to the Work. This API should be used if the DDB of the GS task with the forces should be merged. This is needed for the computation of relaxed-atom elastic constants.

on_all_ok()

This method is called when all the q-points have been computed. It runs mrgddb in sequential on the local machine to produce the final DDB file in the outdir of the abipy.flowtk.works.Work.

class abipy.flowtk.works.PhononWfkqWork(workdir=None, manager=None)

Bases: Work, MergeDdb

This work computes phonons with DFPT on an arbitrary q-mesh (usually denser than the k-mesh for electrons) by computing WKQ files for each q-point. The number of irreducible atomic perturbations for each q-point is taken into account. It provides the callback method (on_all_ok) that calls mrgddb (mrgdv) to merge all the partial DDB (POT) files produced. The two files are available in the output directory of the Work. The WKQ files are removed at runtime.

Inheritance Diagram

classmethod from_scf_task(scf_task: ScfTask, ngqpt, ph_tolerance=None, tolwfr=1e-22, nband=None, with_becs=False, with_quad=False, ddk_tolerance=None, shiftq=(0, 0, 0), is_ngqpt=True, qptopt=1, remove_wfkq=True, prepgkk=0, manager=None) → PhononWfkqWork

Construct a PhononWfkqWork from a abipy.flowtk.tasks.ScfTask object. The input files for WFQ and phonons are automatically generated from the input of the ScfTask. Each phonon task depends on the WFK file produced by scf_task and the associated WFQ file.

Parameters:

• scf_task – abipy.flowtk.tasks.ScfTask object.

• ngqpt – three integers defining the q-mesh

• with_becs – Activate calculation of Electric field and Born effective charges.

• with_quad – Activate calculation of dynamical quadrupoles. Note that only selected features are compatible with dynamical quadrupoles. Please consult <https://docs.abinit.org/topics/longwave/>

• ph_tolerance – dict {“varname”: value} with the tolerance for the phonon run. None to use AbiPy default.

• tolwfr – tolerance used to compute WFQ.

• ddk_tolerance – dict {“varname”: value} with the tolerance used in the DDK run if with_becs. None to use AbiPy default.

• shiftq – Q-mesh shift. Multiple shifts are not supported.

• is_ngqpt – the ngqpt is interpreted as a set of integers defining the q-mesh, otherwise is an explicit list of q-points

• qptopt – Option for the generation of q-points. Default: 1

• remove_wfkq – Remove WKQ files when the children are completed.

• prepgkk – 1 to activate computation of all 3*natom perts (debugging option).

• manager – abipy.flowtk.tasks.TaskManager object.

on_ok(sender)

This callback is called when one task reaches status S_OK. It removes the WFKQ file if all its children have reached S_OK.

on_all_ok()

This method is called when all the q-points have been computed. Ir runs mrgddb in sequential on the local machine to produce the final DDB file in the outdir of the abipy.flowtk.works.Work.

class abipy.flowtk.works.GKKPWork(workdir=None, manager=None)

Bases: Work

This work computes electron-phonon matrix elements for all the q-points present in a DVDB and a DDB file

Inheritance Diagram

classmethod from_den_ddb_dvdb(inp, den_path, ddb_path, dvdb_path, mpiprocs=1, remove_wfkq=True, qpath=None, with_ddk=True, expand=True, manager=None)

Construct a GKKPWork from a DDB and DVDB file. For each q found, a WFQ task and an EPH task computing the matrix elements are created.

classmethod from_phononwfkq_work(phononwfkq_work, nscf_vars={}, remove_wfkq=True, with_ddk=True, manager=None)

Construct a GKKPWork from a PhononWfkqWork object. The WFQ are the ones used for PhononWfkqWork so in principle have only valence bands

on_ok(sender)

This callback is called when one task reaches status S_OK. It removes the WFKQ file if all its children have reached S_OK.

class abipy.flowtk.works.BecWork(workdir=None, manager=None)

Bases: Work, MergeDdb

Work for the computation of the Born effective charges.

This work consists of DDK tasks and phonon + electric field perturbation It provides the callback method (on_all_ok) that calls mrgddb to merge the partial DDB files produced by the work.

Inheritance Diagram

classmethod from_scf_task(scf_task: ScfTask, ddk_tolerance=None, ph_tolerance=None, manager=None)

Build tasks for the computation of Born effective charges from a ground-state task.

Parameters:

• scf_task – abipy.flowtk.tasks.ScfTask object.

• ddk_tolerance – tolerance used in the DDK run if with_becs. None to use AbiPy default.

• ph_tolerance – dict {“varname”: value} with the tolerance used in the phonon run. None to use AbiPy default.

• manager – abipy.flowtk.tasks.TaskManager object.

on_all_ok()

This method is called when all tasks reach S_OK Ir runs mrgddb in sequential on the local machine to produce the final DDB file in the outdir of the abipy.flowtk.works.Work.

class abipy.flowtk.works.DteWork(workdir=None, manager=None)

Bases: Work, MergeDdb

Work for the computation of the third derivative of the energy.

This work consists of DDK tasks and electric field perturbation. It provides the callback method (on_all_ok) that calls mrgddb to merge the partial DDB files produced

Inheritance Diagram

classmethod from_scf_task(scf_task: ScfTask, ddk_tolerance=None, manager=None) → DteWork

Build a DteWork from a ground-state task.

Parameters:

• scf_task – abipy.flowtk.tasks.ScfTask object.

• ddk_tolerance – tolerance used in the DDK run if with_becs. None to use AbiPy default.

• manager – abipy.flowtk.tasks.TaskManager object.

on_all_ok()

This method is called when all tasks reach S_OK It runs mrgddb in sequential on the local machine to produce the final DDB file in the outdir of the Work.

class abipy.flowtk.works.ConducWork(workdir=None, manager=None)

Bases: Work

Workflow for the computation of electrical conductivity.

Can be called from:

1. MultiDataset and PhononWork

2. MultiDataset, DDB filepath and DVDB filepath.

Can use Kerange Capability using withKerange=True

This work consists of 3 tasks or 5 tasks with kerange:

1. SCF GS

2. NSCF

3. Kerange (Kerange only)

4. WFK Interpolation (Kerange only)

5. Electrical Conductivity Calculation.

Inheritance Diagram

classmethod from_phwork(phwork, multi, nbr_proc=None, flow=None, with_kerange=False, omp_nbr_thread=1, manager=None) → ConducWork

Construct a ConducWork from a abipy.flowtk.works.PhononWork and abipy.abio.inputs.MultiDataset.

Parameters:

• phwork – abipy.flowtk.works.PhononWork object calculating the DDB and DVDB files.

• multi – abipy.abio.inputs.MultiDataset object containing a list of 3 datasets or 5 with Kerange. See abipy/abio/factories.py -> conduc_from_scf_nscf_inputs for details about multi.

• nbr_proc – Required if with_kerange since autoparal doesn’t work with optdriver=8.

• flow – The flow calling the work. Used for with_fixed_mpi_omp.

• with_kerange – True if using Kerange.

• omp_nbr_thread – Number of omp_thread to use.

• manager – abipy.flowtk.tasks.TaskManager of the task. If None, the manager is initialized from the config file.

classmethod from_filepath(ddb_path, dvdb_path, multi, nbr_proc=None, flow=None, with_kerange=False, omp_nbr_thread=1, manager=None) → ConducWork

Construct a ConducWork from previously calculated DDB/DVDB file and abipy.abio.inputs.MultiDataset.

Parameters:

• multi – a abipy.abio.inputs.MultiDataset object containing a list of 3 datasets or 5 with Kerange. See abipy/abio/factories.py -> conduc_from_scf_nscf_inputs for details about multi.

• ddb_path – a string containing the path to the DDB file.

• dvdb_path – a string containing the path to the DVDB file.

• nbr_proc – Required if with_kerange since autoparal doesn’t work with optdriver=8.

• flow – The flow calling the work, only needed if with_kerange.

• with_kerange – True if using Kerange.

• omp_nbr_thread – Number of omp_thread to use.

• manager – abipy.flowtk.tasks.TaskManager of the task. If None, the manager is initialized from the config file.

wrappers Module

Wrappers for ABINIT main executables

class abipy.flowtk.wrappers.Mrgscr(manager=None, executable=None, verbose=0)

Bases: ExecWrapper

Wraps the mrgddb Fortran executable.

merge_qpoints(workdir, files_to_merge, out_prefix)

Execute mrgscr inside directory workdir to merge files_to_merge. Produce new file with prefix out_prefix

class abipy.flowtk.wrappers.Mrgddb(manager=None, executable=None, verbose=0)

Bases: ExecWrapper

Wraps the mrgddb Fortran executable.

merge(workdir, ddb_files, out_ddb, description, delete_source_ddbs=True) → str

Merge DDB file, return the absolute path of the new database in workdir.

class abipy.flowtk.wrappers.Mrgdvdb(manager=None, executable=None, verbose=0)

Bases: ExecWrapper

Wraps the mrgdvdb Fortran executable.

merge(workdir, pot_files, out_dvdb, delete_source=True) → str

Merge POT files containing 1st order DFPT potential return the absolute path of the new database in workdir.

Parameters:

delete_source – True if POT1 files should be removed after (successful) merge.

class abipy.flowtk.wrappers.Cut3D(manager=None, executable=None, verbose=0)

Bases: ExecWrapper

Wraps the cut3d Fortran executable.

cut3d(cut3d_input, workdir) → tuple[str, str]

Runs cut3d with a Cut3DInput

Parameters:

• cut3d_input – a Cut3DInput object.

• workdir – directory where cut3d is executed.

Returns:

(string) absolute path to the standard output of the cut3d execution. (string) absolute path to the output filepath. None if output is required.

class abipy.flowtk.wrappers.Fold2Bloch(manager=None, executable=None, verbose=0)

Bases: ExecWrapper

Wraps the fold2Bloch Fortran executable.

unfold(wfkpath, folds, workdir=None) → str