Skip to content

Engine

The engine module provides interfaces to external simulation engines.

Base

Engine base classes for molecular simulation engines.

This module provides abstract base classes for running external computational chemistry programs like LAMMPS, CP2K, etc. It integrates with the core Script class for input file management.

Engine

Engine(executable, *, check_executable=True)

Bases: ABC

Abstract base class for computational chemistry engines.

Provides a common interface for running external programs like LAMMPS, CP2K, etc. Each engine handles setup, execution, and output processing for its specific program.

The Engine class integrates with the core Script class for input file management. Scripts can be created from text, loaded from files, or loaded from URLs.

Attributes:

Name Type Description
executable

Path or command to the executable

work_dir

Working directory for calculations

scripts

List of Script objects for input files

input_script

Primary input script (first script or script with 'input' tag)

output_file

Primary output file from the calculation

Example

from molpy.core.script import Script from molpy.engine import LAMMPSEngine

Create input script

script = Script.from_text( ... name="input", ... text="units real\natom_style full\n", ... language="other" ... )

Create engine and prepare

engine = LAMMPSEngine(executable="lmp") engine.prepare(work_dir="./calc", scripts=[script])

Run calculation

result = engine.run() print(result.returncode) 0

Initialize the engine.

Parameters:

Name Type Description Default
executable str

Path or command to the executable

required
check_executable bool

Whether to check if executable exists in PATH (default: True)

True

Raises:

Type Description
FileNotFoundError

If check_executable is True and executable not found

Source code in src/molpy/engine/base.py
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(self, executable: str, *, check_executable: bool = True):
    """
    Initialize the engine.

    Args:
        executable: Path or command to the executable
        check_executable: Whether to check if executable exists in PATH
                         (default: True)

    Raises:
        FileNotFoundError: If check_executable is True and executable not found
    """
    self.executable = executable
    if check_executable:
        self.check_executable()

name abstractmethod property

name

Return the name of the engine.

check_executable

check_executable()

Check if the executable is available in the system PATH.

Raises:

Type Description
FileNotFoundError

If the executable is not found

Source code in src/molpy/engine/base.py
82
83
84
85
86
87
88
89
90
91
92
93
def check_executable(self) -> None:
    """
    Check if the executable is available in the system PATH.

    Raises:
        FileNotFoundError: If the executable is not found
    """
    if not shutil.which(self.executable):
        raise FileNotFoundError(
            f"Executable '{self.executable}' not found in PATH. "
            f"Please install the engine or set the correct executable path."
        )

clean

clean(keep_scripts=True)

Clean up calculation files.

Parameters:

Name Type Description Default
keep_scripts bool

Whether to keep input scripts (default: True)

True
Source code in src/molpy/engine/base.py
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def clean(self, keep_scripts: bool = True) -> None:
    """
    Clean up calculation files.

    Args:
        keep_scripts: Whether to keep input scripts (default: True)
    """
    if not hasattr(self, "work_dir") or not self.work_dir.exists():
        return

    if not keep_scripts and hasattr(self, "scripts"):
        for script in self.scripts:
            if script.path is not None:
                script_path = self.work_dir / script.path.name
                if script_path.exists():
                    script_path.unlink()

get_output_file

get_output_file(name=None)

Get the output file path.

Parameters:

Name Type Description Default
name str | None

Name of the output file. If None, returns default output file.

None

Returns:

Type Description
Path | None

Path to the output file or None if not found

Source code in src/molpy/engine/base.py
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def get_output_file(self, name: str | None = None) -> Path | None:
    """
    Get the output file path.

    Args:
        name: Name of the output file. If None, returns default output file.

    Returns:
        Path to the output file or None if not found
    """
    if not hasattr(self, "work_dir") or not self.work_dir.exists():
        return None

    if name is not None:
        output_path = self.work_dir / name
        if output_path.exists():
            return output_path
        return None

    # Try to find default output file
    output_files = self.list_output_files()
    if output_files:
        # Return the first output file (could be improved with heuristics)
        return output_files[0]

    return None

get_script

get_script(name=None, tag=None)

Get a script by name or tag.

Parameters:

Name Type Description Default
name str | None

Name of the script (logical name or filename)

None
tag str | None

Tag to search for

None

Returns:

Type Description
Script | None

Script object or None if not found

Example

script = engine.get_script(name="input") script = engine.get_script(tag="input")

Source code in src/molpy/engine/base.py
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def get_script(
    self, name: str | None = None, tag: str | None = None
) -> Script | None:
    """
    Get a script by name or tag.

    Args:
        name: Name of the script (logical name or filename)
        tag: Tag to search for

    Returns:
        Script object or None if not found

    Example:
        >>> script = engine.get_script(name="input")
        >>> script = engine.get_script(tag="input")
    """
    if not hasattr(self, "scripts"):
        return None

    for script in self.scripts:
        if name is not None:
            # Match by logical name or filename
            if script.name == name:
                return script
            if script.path is not None and script.path.name == name:
                return script

        if tag is not None:
            # Match by tag
            if tag in script.tags:
                return script

    return None

list_output_files

list_output_files()

List all output files in the working directory.

Returns:

Type Description
list[Path]

List of output file paths

Source code in src/molpy/engine/base.py
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
def list_output_files(self) -> list[Path]:
    """
    List all output files in the working directory.

    Returns:
        List of output file paths
    """
    if not hasattr(self, "work_dir") or not self.work_dir.exists():
        return []

    # Get all script paths to exclude them
    script_paths = set()
    if hasattr(self, "scripts"):
        for script in self.scripts:
            if script.path is not None:
                script_paths.add(self.work_dir / script.path.name)

    # Return all files except scripts
    return [
        f for f in self.work_dir.iterdir() if f.is_file() and f not in script_paths
    ]

prepare

prepare(work_dir, scripts, *, auto_save=True)

Prepare the engine for execution by setting up the working directory and scripts.

Parameters:

Name Type Description Default
work_dir str | Path

Path to the working directory

required
scripts Script | Sequence[Script]

Single Script object or sequence of Script objects

required
auto_save bool

Whether to automatically save scripts to the working directory (default: True)

True

Returns:

Name Type Description
Self Self

The engine instance for method chaining

Example

script = Script.from_text("input", "units real\natom_style full\n") engine.prepare("./calc", script)

Or with multiple scripts

engine.prepare("./calc", [script1, script2])

Source code in src/molpy/engine/base.py
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def prepare(
    self,
    work_dir: str | Path,
    scripts: Script | Sequence[Script],
    *,
    auto_save: bool = True,
) -> "Self":
    """
    Prepare the engine for execution by setting up the working directory and scripts.

    Args:
        work_dir: Path to the working directory
        scripts: Single Script object or sequence of Script objects
        auto_save: Whether to automatically save scripts to the working directory
                  (default: True)

    Returns:
        Self: The engine instance for method chaining

    Example:
        >>> script = Script.from_text("input", "units real\\natom_style full\\n")
        >>> engine.prepare("./calc", script)
        >>> # Or with multiple scripts
        >>> engine.prepare("./calc", [script1, script2])
    """
    self.work_dir = Path(work_dir)
    self.work_dir.mkdir(parents=True, exist_ok=True)

    # Normalize scripts to a list
    if isinstance(scripts, Script):
        self.scripts = [scripts]
    else:
        self.scripts = list(scripts)

    if not self.scripts:
        raise ValueError("At least one script is required")

    # Save scripts to the working directory if auto_save is True
    if auto_save:
        for script in self.scripts:
            # Determine filename from script path or name
            if script.path is not None:
                # Use the filename from the script's path
                script_path = self.work_dir / script.path.name
            else:
                # Generate filename from script name
                # Try to guess extension from language or use default
                ext = self._get_default_extension()
                script_path = self.work_dir / f"{script.name}{ext}"

            # Save script to working directory
            script.save(script_path)

    # Set input_script (first script or script with 'input' tag)
    self.input_script = self._find_input_script()

    return self

run abstractmethod

run(**kwargs)

Execute the engine calculation.

Parameters:

Name Type Description Default
**kwargs Any

Additional arguments for the specific engine (e.g., input_file, output_file, etc.)

{}

Returns:

Type Description
CompletedProcess

CompletedProcess object with execution results

Raises:

Type Description
RuntimeError

If engine is not prepared (prepare() not called)

Source code in src/molpy/engine/base.py
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
@abstractmethod
def run(self, **kwargs: Any) -> subprocess.CompletedProcess:
    """
    Execute the engine calculation.

    Args:
        **kwargs: Additional arguments for the specific engine
                 (e.g., input_file, output_file, etc.)

    Returns:
        CompletedProcess object with execution results

    Raises:
        RuntimeError: If engine is not prepared (prepare() not called)
    """
    if not hasattr(self, "work_dir"):
        raise RuntimeError("Engine not prepared. Call prepare() first.")
    pass

CP2K

CP2K quantum chemistry engine.

This module provides the CP2KEngine class for running CP2K calculations.

CP2KEngine

CP2KEngine(executable, *, check_executable=True)

Bases: Engine

CP2K quantum chemistry engine.

This engine runs CP2K calculations with input scripts.

Example

from molpy.core.script import Script from molpy.engine import CP2KEngine

Create input script

script = Script.from_text( ... name="input", ... text="&GLOBAL\n PROJECT water\n&END GLOBAL\n", ... language="other" ... )

Create engine and prepare

engine = CP2KEngine(executable="cp2k.psmp") engine.prepare(work_dir="./calc", scripts=script)

Run calculation

result = engine.run() print(result.returncode) 0

Source code in src/molpy/engine/base.py
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(self, executable: str, *, check_executable: bool = True):
    """
    Initialize the engine.

    Args:
        executable: Path or command to the executable
        check_executable: Whether to check if executable exists in PATH
                         (default: True)

    Raises:
        FileNotFoundError: If check_executable is True and executable not found
    """
    self.executable = executable
    if check_executable:
        self.check_executable()

name property

name

Return engine name.

get_output_file

get_output_file(name=None)

Get the CP2K output file path.

Parameters:

Name Type Description Default
name str | None

Name of the output file. If None, uses default "cp2k.out".

None

Returns:

Type Description
Path | None

Path to the output file or None if not found

Source code in src/molpy/engine/cp2k.py
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def get_output_file(self, name: str | None = None) -> Path | None:
    """
    Get the CP2K output file path.

    Args:
        name: Name of the output file. If None, uses default "cp2k.out".

    Returns:
        Path to the output file or None if not found
    """
    if name is None:
        name = "cp2k.out"

    output_path = self.work_dir / name
    if output_path.exists():
        return output_path
    return None

run

run(input_file=None, output_file=None, **kwargs)

Execute CP2K calculation.

Parameters:

Name Type Description Default
input_file str | Path | None

Name of the input file. If None, uses the input script from prepare() (default: None)

None
output_file str | Path | None

Name of the output file. If None, uses default "cp2k.out" (default: None)

None
**kwargs Any

Additional arguments passed to subprocess.run (e.g., capture_output, text, check, etc.)

{}

Returns:

Type Description
CompletedProcess

CompletedProcess object with execution results

Raises:

Type Description
RuntimeError

If engine is not prepared (prepare() not called)

Source code in src/molpy/engine/cp2k.py
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def run(
    self,
    input_file: str | Path | None = None,
    output_file: str | Path | None = None,
    **kwargs: Any,
) -> subprocess.CompletedProcess:
    """
    Execute CP2K calculation.

    Args:
        input_file: Name of the input file. If None, uses the input script
                   from prepare() (default: None)
        output_file: Name of the output file. If None, uses default "cp2k.out"
                    (default: None)
        **kwargs: Additional arguments passed to subprocess.run
                 (e.g., capture_output, text, check, etc.)

    Returns:
        CompletedProcess object with execution results

    Raises:
        RuntimeError: If engine is not prepared (prepare() not called)
    """
    if not hasattr(self, "work_dir"):
        raise RuntimeError("Engine not prepared. Call prepare() first.")

    # Use input script from prepare() if not specified
    if input_file is None:
        if (
            not hasattr(self, "input_script")
            or self.input_script is None
            or self.input_script.path is None
        ):
            raise RuntimeError(
                "No input file specified and no input script found. "
                "Either specify input_file or ensure prepare() was called with a script."
            )
        input_file = self.input_script.path.name
    else:
        input_file = Path(input_file).name

    # Default output file name
    output_file = "cp2k.out" if output_file is None else Path(output_file).name

    # Build command
    command = [self.executable, "-i", str(input_file), "-o", str(output_file)]

    # Default subprocess arguments
    run_kwargs = {
        "cwd": self.work_dir,
        "capture_output": True,
        "text": True,
        "check": False,
    }
    run_kwargs.update(kwargs)

    return subprocess.run(command, **run_kwargs)

LAMMPS

LAMMPS molecular dynamics engine.

This module provides the LAMMPSEngine class for running LAMMPS calculations.

LAMMPSEngine

LAMMPSEngine(executable, *, check_executable=True)

Bases: Engine

LAMMPS molecular dynamics engine.

This engine runs LAMMPS calculations with input scripts.

Example

from molpy.core.script import Script from molpy.engine import LAMMPSEngine

Create input script

script = Script.from_text( ... name="input", ... text="units real\natom_style full\n", ... language="other" ... )

Create engine and prepare

engine = LAMMPSEngine(executable="lmp") engine.prepare(work_dir="./calc", scripts=script)

Run calculation

result = engine.run() print(result.returncode) 0

Source code in src/molpy/engine/base.py
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(self, executable: str, *, check_executable: bool = True):
    """
    Initialize the engine.

    Args:
        executable: Path or command to the executable
        check_executable: Whether to check if executable exists in PATH
                         (default: True)

    Raises:
        FileNotFoundError: If check_executable is True and executable not found
    """
    self.executable = executable
    if check_executable:
        self.check_executable()

name property

name

Return engine name.

get_log_file

get_log_file()

Get the LAMMPS log file path.

Returns:

Type Description
Path | None

Path to the log file or None if not found

Source code in src/molpy/engine/lammps.py
108
109
110
111
112
113
114
115
116
117
118
def get_log_file(self) -> Path | None:
    """
    Get the LAMMPS log file path.

    Returns:
        Path to the log file or None if not found
    """
    log_path = self.work_dir / "log.lammps"
    if log_path.exists():
        return log_path
    return None

run

run(input_file=None, log_file=None, **kwargs)

Execute LAMMPS calculation.

Parameters:

Name Type Description Default
input_file str | Path | None

Name of the input script file. If None, uses the input script from prepare() (default: None)

None
log_file str | Path | None

Name of the log file. If None, uses default "log.lammps" (default: None)

None
**kwargs Any

Additional arguments passed to subprocess.run (e.g., capture_output, text, check, etc.)

{}

Returns:

Type Description
CompletedProcess

CompletedProcess object with execution results

Raises:

Type Description
RuntimeError

If engine is not prepared (prepare() not called)

Source code in src/molpy/engine/lammps.py
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def run(
    self,
    input_file: str | Path | None = None,
    log_file: str | Path | None = None,
    **kwargs: Any,
) -> subprocess.CompletedProcess:
    """
    Execute LAMMPS calculation.

    Args:
        input_file: Name of the input script file. If None, uses the input script
                   from prepare() (default: None)
        log_file: Name of the log file. If None, uses default "log.lammps"
                 (default: None)
        **kwargs: Additional arguments passed to subprocess.run
                 (e.g., capture_output, text, check, etc.)

    Returns:
        CompletedProcess object with execution results

    Raises:
        RuntimeError: If engine is not prepared (prepare() not called)
    """
    if not hasattr(self, "work_dir"):
        raise RuntimeError("Engine not prepared. Call prepare() first.")

    # Use input script from prepare() if not specified
    if input_file is None:
        if (
            not hasattr(self, "input_script")
            or self.input_script is None
            or self.input_script.path is None
        ):
            raise RuntimeError(
                "No input file specified and no input script found. "
                "Either specify input_file or ensure prepare() was called with a script."
            )
        input_file = self.input_script.path.name
    else:
        input_file = Path(input_file).name

    # Default log file name
    log_file = "log.lammps" if log_file is None else Path(log_file).name

    # Build command
    command = [self.executable, "-in", str(input_file), "-log", str(log_file)]

    # Default subprocess arguments
    run_kwargs = {
        "cwd": self.work_dir,
        "capture_output": True,
        "text": True,
        "check": False,
    }
    run_kwargs.update(kwargs)

    return subprocess.run(command, **run_kwargs)