-
-
- - Get Started - - -
- - Blog - - -
- - Tutorials - - -
- - Docs - - -
-
- - - Resources - - --
-
-
- - Github - -
-from qcengine.util import execute
-from mmic.components.blueprints import GenericComponent
-from typing import Any, Dict, Tuple
-from ..models import CmdOutput, CmdInput
-import ntpath
-import os
-
-__all__ = ["CmdComponent"]
-
-
-[docs]class CmdComponent(GenericComponent):
- @classmethod
- def input(cls):
- return CmdInput
-
- @classmethod
- def output(cls):
- return CmdOutput
-
- def execute(
- self,
- inputs: Dict[str, Any],
- ) -> Tuple[bool, Dict[str, Any]]:
-
- if isinstance(inputs, dict):
- inputs = self.input()(**inputs)
-
- env = os.environ.copy()
- scratch_directory = inputs.scratch_directory
-
- if inputs.config:
- env["MKL_NUM_THREADS"] = str(config.ncores)
- env["OMP_NUM_THREADS"] = str(config.ncores)
-
- scratch_directory = config.scratch_directory
-
- infiles = {}
- if inputs.infiles:
- flag = inputs.as_binary or "r"
- for fpath in inputs.infiles:
- fname = ntpath.basename(fpath)
- with open(fpath, flag) as fp:
- infiles[fname] = fp.read()
-
- exe_success, proc = execute(
- command=inputs.command,
- infiles=infiles,
- outfiles=inputs.outfiles,
- outfiles_track=inputs.outfiles_track,
- scratch_directory=scratch_directory,
- scratch_name=inputs.scratch_name,
- scratch_messy=inputs.scratch_messy,
- timeout=inputs.timeout,
- environment=env,
- )
-
- if exe_success:
- if proc.get("stderr"):
- if inputs.raise_err:
- raise RuntimeError(proc.get("stderr"))
- return exe_success, self.output()(
- outfiles=proc.get("outfiles"),
- stdout=proc.get("stdout"),
- stderr=proc.get("stderr"),
- scratch_directory=proc.get("scratch_directory"),
- )
- else:
- raise RuntimeError(proc.get("stderr"))
-
-from mmelemental.models import ProtoModel
-from pydantic import Field
-from typing import Optional, List, Dict
-
-__all__ = ["CmdInput"]
-
-
-[docs]class CmdInput(ProtoModel):
- """Input model for command-line execution."""
-
- command: List[str] = Field(..., description=".")
- infiles: List[str] = Field(
- None,
- description="Input files (abs paths) to be written in scratch dir. May be {}.",
- )
- outfiles: Optional[List[str]] = Field(
- None,
- description="Output file name(s).",
- )
- outfiles_track: Optional[List[str]] = Field(
- [],
- description="Specifies which output files (by name) are tracked and not loaded in memory. The posix paths "
- "(instead of file contents) are returned instead in `outfiles`.",
- )
- raise_err: Optional[bool] = Field(
- False,
- description="If set to True, a runtime exception is raised when stderr is not empty.",
- )
- as_binary: Optional[List[str]] = Field(
- None, description="Keys of `infiles` or `outfiles` to be treated as bytes."
- )
- scratch_name: Optional[str] = Field(
- None, description="Passed to temporary_directory."
- )
- scratch_directory: Optional[str] = Field(
- None, description="Passed to temporary_directory."
- )
- scratch_suffix: Optional[str] = Field(
- None, description="Passed to temporary_directory."
- )
- scratch_messy: bool = Field(False, description="Passed to temporary_directory")
- scratch_exist_ok: bool = Field(False, description="Passed to temporary_directory.")
- blocking_files: Optional[List[str]] = Field(
- None, description="Files which should stop execution if present beforehand."
- )
- timeout: Optional[int] = Field(
- None, description="Stop the process after n seconds."
- )
- interupt_after: Optional[int] = Field(
- None, description="Interupt the process (not hard kill) after n seconds."
- )
- environment: Optional[Dict[str, str]] = Field(
- None, description="The environment to run in."
- )
- shell: Optional[bool] = Field(False, description="Run command through the shell.")
- exit_code: Optional[int] = Field(
- 0, description="The exit code above which the process is considered failure."
- )
- config: Optional["TaskConfig"] = Field(
- None,
- description="Config class for executing tasks. See :class:`mmic.components.base.config.TaskConfig`.",
- )
-
-from mmelemental.models import ProtoModel
-from pydantic import Field
-from typing import Optional, Dict, Union, Any
-from pathlib import PosixPath
-from subprocess import Popen
-
-__all__ = ["CmdOutput"]
-
-
-[docs]class CmdOutput(ProtoModel):
- """Output model for commend-line execution."""
-
- stdout: str = Field(..., description="Standard output.")
- stderr: Optional[str] = Field(None, description="Standard error.")
- log: Optional[str] = Field(None, description="Logging output.")
- outfiles: Optional[Union[Dict[str, Any], None]] = Field(
- None,
- description="List of FileOutput objects. See the :class: ``mmelemental.models.FileOutput``.",
- )
- proc: Optional[Popen]
- scratch_directory: Optional[PosixPath]
-
- class Config(ProtoModel.Config):
- arbitrary_types_allowed = True
-