/
structurecontainer.py
164 lines (136 loc) · 5.86 KB
/
structurecontainer.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
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
155
156
157
158
159
160
161
162
163
164
# 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.
__author__ = "Yury Lysogorskiy, Jan Janssen, Marvin Poul"
__copyright__ = (
"Copyright 2020, Max-Planck-Institut für Eisenforschung GmbH - "
"Computational Materials Design (CM) Department"
)
__version__ = "0.1"
__maintainer__ = "Marvin Poul"
__email__ = "poul@mpie.de"
__status__ = "development"
__date__ = "Aug 12, 2020"
from pyiron_base import InputList, GenericJob
from pyiron.atomistics.job.atomistic import AtomisticGenericJob
from pyiron.atomistics.structure.atoms import Atoms
class StructureContainer(AtomisticGenericJob):
"""
Container to save a list of structures in HDF5 together with tags.
Add new structures with :meth:`.append`, they are added to
:attr:`.structure_lst`. The HDF5 is written when :meth:`.run` is called.
"""
def __init__(self, project, job_name):
super().__init__(project, job_name)
self.__version__ = "0.2.0"
self.__hdf_version__ = "0.2.0"
self._structure_lst = InputList(table_name = "structures")
self.server.run_mode.interactive = True
@property
def structure_lst(self):
"""
:class:`.InputList`: list of :class:`~.Atoms`
"""
return self._structure_lst
@staticmethod
def _to_structure(structure_or_job):
"""
Return structure from structure or atomic job.
Args:
structure_or_job (:class:`~.AtomisticGenericJob`, :class:`~.Atoms`):
if :class:`~.AtomisticGenericJob` try to get most recent structure,
copy it and set the job_id in :attr:`~.Atoms.info`
Returns:
:class:`~.Atoms`: structure from the job or given structure
Raises:
ValueError: if given :class:`~.AtomisticGenericJob` has no structure set
TypeError: if structure_or_job is of invalid type
"""
if isinstance(structure_or_job, AtomisticGenericJob):
if structure_or_job.structure:
s = structure_or_job.get_structure(-1).copy()
s.info["jobid"] = structure_or_job.job_id
return s
else:
raise ValueError(
"The job does not contain any structure to import."
)
elif isinstance(structure_or_job, Atoms):
return structure_or_job
else:
raise TypeError(
"You can only use a structure object or an "
"AtomisticGenericJob object."
)
@property
def structure(self):
"""
:class:`~.Atoms`: first (or only) structure set in the container
:setter: :class:`~.Atoms`, :class:`~.AtomisticGenericJob`
if a job is given take the last structure and add the job id to its
:attr:`pyiron.atomistics.structure.Atoms.info`
"""
return self.structure_lst.get(0, None)
@structure.setter
def structure(self, structure_or_job):
item = self._to_structure(structure_or_job)
if len(self.structure_lst) >= 1:
self.structure_lst[0] = item
else:
self.structure_lst.append(item)
def append(self, structure_or_job):
"""
Add new structure to structure list.
The added structure will available in :attr:`~.structure_lst`. If the
structure is added via a job, retrieve the latest structure and add its
id to :attr:`pyiron.atomistics.generic.Atoms.info`.
Args:
structure_or_job (:class:`~.AtomisticGenericJob`/:class:`~.Atoms`):
if :class:`~.AtomisticGenericJob` add from
:meth:`~.AtomisticGenericJob.get_structure`,
otherwise add just the given :class:`~.Atoms`
Returns:
dict: item added to :attr:`~.structure_lst`
"""
self.structure_lst.append(self._to_structure(structure_or_job))
return self.structure_lst[0]
def run_static(self):
self.status.finished = True
def run_if_interactive(self):
self.to_hdf()
self.status.finished = True
def write_input(self):
pass
def collect_output(self):
pass
def to_hdf(self, hdf = None, group_name = None):
# skip any of the AtomisticGenericJob specific serialization, since we
# handle the structures on our own and that method might just confuse
# self.structure and self.structure_lst
GenericJob.to_hdf(self, hdf = hdf, group_name = group_name)
hdf = self.project_hdf5.create_group("structures")
for i, structure in enumerate(self.structure_lst.values()):
structure.to_hdf(hdf, group_name = "structure_{}".format(i))
def from_hdf(self, hdf = None, group_name = None):
# keep hdf structure for version peeking in separate variable, so that
# the inherited from_hdf() can properly deal with it
h5 = hdf or self.project_hdf5
if group_name:
h5 = h5[group_name]
if "HDF_VERSION" in h5.list_nodes():
hdf_version = h5["HDF_VERSION"]
else:
# old versions didn't use to set a HDF version
hdf_version = "0.1.0"
if hdf_version == "0.1.0":
super().from_hdf(hdf=hdf, group_name=group_name)
with self.project_hdf5.open("input") as hdf5_input:
self.structure = Atoms().from_hdf(hdf5_input)
else:
GenericJob.from_hdf(self, hdf = hdf, group_name = group_name)
self.structure_lst.clear()
hdf = self.project_hdf5["structures"]
for group in sorted(hdf.list_groups()):
structure = Atoms()
structure.from_hdf(hdf, group_name = group)
self.structure_lst.append(structure)