ai2_kit.core.executor module

class ai2_kit.core.executor.BaseExecutorConfig(*, ssh: Optional[SshConfig] = None, queue_system: QueueSystemConfig, work_dir: str, python_cmd: str = 'python')

Bases: BaseModel

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'python_cmd': FieldInfo(annotation=str, required=False, default='python'), 'queue_system': FieldInfo(annotation=QueueSystemConfig, required=True), 'ssh': FieldInfo(annotation=Union[SshConfig, NoneType], required=False, default=None), 'work_dir': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

python_cmd: str

queue_system: QueueSystemConfig

ssh: Optional[SshConfig]

work_dir: str

class ai2_kit.core.executor.Executor

Bases: ABC

abstract download(from_path: str, to_dir: str) → str

abstract dump_text(text: str, path: str)

abstract glob(pattern: str) → List[str]

init()

abstract load_text(path: str) → str

abstract mkdir(path: str)

name: str

python_cmd: str

abstract resolve_artifact(artifact: Artifact) → List[str]

abstract run(script: str, **kwargs) → Result

abstract run_python_fn(fn: FnType, python_cmd=None, cwd=None) → FnType

abstract run_python_script(script: str, python_cmd=None)

setup_workspace(workspace_dir: str, dirs: List[str])

abstract submit(script: str, **kwargs) → JobFuture

tmp_dir: str

abstract upload(from_path: str, to_dir: str) → str

work_dir: str

class ai2_kit.core.executor.ExecutorManager(executor_configs: Mapping[str, BaseExecutorConfig])

Bases: object

get_executor(name: str)

class ai2_kit.core.executor.HpcExecutor(connector: BaseConnector, queue_system: BaseQueueSystem, work_dir: str, python_cmd: str, name: str)

Bases: Executor

download(from_path: str, to_dir: str)

dump_text(text: str, path: str)

classmethod from_config(config: Union[dict, BaseExecutorConfig], name: str = '')

glob(pattern: str)

init()

property is_local

load_text(path: str) → str

mkdir(path: str)

name: str

python_cmd: str

resolve_artifact(artifact: Artifact) → List[str]

run(script: str, **kwargs)

run_python_fn(fn: FnType, python_cmd=None, cwd=None) → FnType

run_python_script(script: str, python_cmd=None, cwd=None)

submit(script: str, cwd: str, **kwargs)

tmp_dir: str

upload(from_path: str, to_dir: str)

upload_python_pkg(pkg: str)

upload python package to remote server

work_dir: str

ai2_kit.core.executor.create_executor(config: BaseExecutorConfig, name: str) → Executor

ai2_kit.core.executor.fn_to_script(fn: Callable, args, kwargs, delimiter='@')

ai2_kit.core.executor.pickle_converts(obj, pickle_module='cp', bz2_module='bz2', base64_module='base64')

convert an object to its pickle string form