Labeling API

Labeling API#

Labeling package public exports.

This package provides reusable labeling components consumed by LabelingWorkflow and related orchestration modules.

class dpeva.labeling.AbacusGenerator(config: Dict[str, Any])[源代码]#

基类：object

Generates ABACUS input files for a set of atomic structures.

generate(atoms: Atoms, output_dir: str | Path, task_name: str, stru_type: str | None = None, vacuum_status: List[bool] | None = None, dataset_name: str = 'unknown', system_name: str = 'unknown')[源代码]#

Generate input files for a single structure.

参数:

  • atoms -- ASE Atoms object (preprocessed).

  • output_dir -- Directory to save files.

  • task_name -- Name of the task.

  • stru_type -- Pre-determined structure type (optional).

  • vacuum_status -- Pre-determined vacuum status (optional).

  • dataset_name -- Name of the dataset this task belongs to.

  • system_name -- Name of the system this task belongs to.

class dpeva.labeling.AbacusPostProcessor(config: Dict[str, Any])[源代码]#

基类：object

Process ABACUS results: Check convergence, Parse data, Clean data.

static build_no_contribution_hint(conv: int, clean: int) → str | None[源代码]#
check_convergence(task_dir: Path) → bool[源代码]#
classify_task_status(task_dir: Path) → Tuple[str, str][源代码]#
compute_metrics(systems: MultiSystems) → DataFrame[源代码]#

Compute metrics (Cohesive Energy, Pressure, Max Force) for a MultiSystems object. Returns a DataFrame.

export_data(systems: Sequence[LabeledSystem], df_clean: DataFrame, output_dir: Path, format: str = 'deepmd/npy')[源代码]#

Export cleaned data to disk.

filter_data(df: DataFrame) → DataFrame[源代码]#

Filter DataFrame based on thresholds.

load_data(task_dir: Path) → LabeledSystem | None[源代码]#

Load data from a converged task directory using dpdata.

class dpeva.labeling.ResubmissionStrategy(attempt_params: List[Dict[str, Any]] | None = None)[源代码]#

基类：object

Manages parameters for each calculation attempt.

This class handles the retrieval and application of parameters for different attempts in the calculation workflow. It allows for modifying input files based on the attempt index.

apply(task_dir: str, attempt: int) → bool[源代码]#

Apply the parameters for the given attempt to the INPUT file.

参数:

  • task_dir -- Directory containing the INPUT file.

  • attempt -- The attempt index (0-based).

返回:

True if applied successfully, False if no params for this attempt.

get_params(attempt: int) → Dict[str, Any][源代码]#

Get parameters for a specific attempt index.

目录