-
Notifications
You must be signed in to change notification settings - Fork 13
/
factory.py
124 lines (107 loc) · 4.37 KB
/
factory.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
# coding: utf-8
# Copyright (c) Max-Planck-Institut für Eisenforschung GmbH - Computational Materials Design (CM) Department
# Distributed under the terms of "New BSD License", see the LICENSE file.
"""
Factories for creating jobs, which may already exist in the database/storage.
"""
from pyiron_base.project.generic import Project
from pyiron_base.interfaces.factory import PyironFactory
from abc import ABC, abstractmethod
from pyiron_base.jobs.job.jobtype import JobType
from pyiron_base.jobs.job.generic import GenericJob
from typing import Type, Dict, List, Union
from pyiron_base.jobs.master.flexible import FlexibleMaster
from pyiron_base.jobs.script import ScriptJob
from pyiron_base.jobs.master.serial import SerialMasterBase
from pyiron_base.jobs.datamining import TableJob
from pyiron_base.storage.hdfio import ProjectHDFio
from pyiron_base.state import state
__author__ = "Liam Huber, Jan Janssen"
__copyright__ = (
"Copyright 2021, Max-Planck-Institut für Eisenforschung GmbH - "
"Computational Materials Design (CM) Department"
)
__version__ = "1.0"
__maintainer__ = "Liam Huber"
__email__ = "huber@mpie.de"
__status__ = "production"
__date__ = "Sep 7, 2021"
class JobFactoryCore(PyironFactory, ABC):
def __init__(self, project: Project):
self._project = project
@property
@abstractmethod
def _job_class_dict(self) -> Dict:
pass
def __dir__(self) -> List:
"""
Enable autocompletion by overwriting the __dir__() function.
"""
return list(self._job_class_dict.keys())
def __getattr__(self, name) -> Type[GenericJob]:
if name in self._job_class_dict.keys():
def wrapper(
job_name: str,
delete_existing_job: bool = False,
delete_aborted_job: bool = False,
) -> GenericJob:
"""
Create a job.
Args:
job_name (str): name of the job
delete_existing_job (bool): delete an existing job - default false
delete_aborted_job (bool): delete an existing and aborted job - default false
Returns:
GenericJob: job object depending on the job_type selected
"""
return JobType(
class_name=name, # Pass the class directly, JobType can handle that
project=ProjectHDFio(
project=self._project.copy(), file_name=job_name
),
job_name=job_name,
job_class_dict=self._job_class_dict,
delete_existing_job=delete_existing_job,
delete_aborted_job=delete_aborted_job,
)
return wrapper
else:
raise AttributeError("no job class named '{}' defined".format(name))
def __call__(
self,
job_type: Union[str, Type[GenericJob]],
job_name: str,
delete_existing_job: bool = False,
delete_aborted_job: bool = False,
) -> GenericJob:
"""
Create a job.
Args:
job_type (str|Type[GenericJob]): The job class to be instantiated, either the string from a known class, or
an actual class, e.g. in the case of custom user-made jobs.
job_name (str): name of the job.
delete_existing_job (bool): delete an existing job. (Default is False.)
delete_aborted_job (bool): delete an existing and aborted job. (Default is False.)
Returns:
GenericJob: job object depending on the job_type selected
"""
job = JobType(
class_name=job_type, # Pass the class directly, JobType can handle that
project=ProjectHDFio(project=self._project.copy(), file_name=job_name),
job_name=job_name,
job_class_dict=self._job_class_dict,
delete_existing_job=delete_existing_job,
delete_aborted_job=delete_aborted_job,
)
if state.settings.login_user is not None:
job.user = state.settings.login_user
return job
class JobFactory(JobFactoryCore):
@property
def _job_class_dict(self) -> Dict:
return {
"FlexibleMaster": FlexibleMaster,
"ScriptJob": ScriptJob,
"SerialMasterBase": SerialMasterBase,
"TableJob": TableJob,
}