encyclopedia.py 12.8 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# Copyright 2018 Markus Scheidgen
#
# Licensed under the Apache License, Version 2.0 (the 'License');
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an'AS IS' BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from typing import Any

17
from nomad.normalizing.normalizer import Normalizer
18
19
from nomad.datamodel.encyclopedia import (
    EncyclopediaMetadata,
20
21
22
23
24
25
26
27
28
    Material,
    Method,
    Properties,
    Calculation,
)
from nomad.normalizing.encyclopedia.context import Context
from nomad.normalizing.encyclopedia.material import MaterialBulkNormalizer, Material2DNormalizer, Material1DNormalizer
from nomad.normalizing.encyclopedia.method import MethodDFTNormalizer, MethodGWNormalizer
from nomad.normalizing.encyclopedia.properties import PropertiesNormalizer
29
from nomad import config
30
31
32
33
34
35


class EncyclopediaNormalizer(Normalizer):
    """
    This normalizer emulates the functionality of the old Encyclopedia backend.
    The data used by the encyclopedia have been assigned under new metainfo
36
    within a new section called "encyclopedia". In the future these separate
37
38
39
40
41
42
43
44
45
46
    metainfos could be absorbed into the existing metainfo hiearchy.
    """
    def calc_type(self, calc: Calculation) -> str:
        """Decides what type of calculation this is: single_point, md,
        geometry_optimization, etc.
        """
        calc_enums = Calculation.calculation_type.type
        calc_type = calc_enums.unavailable

        try:
47
            sccs = self.section_run.section_single_configuration_calculation
48
49
50
        except Exception:
            sccs = []
        try:
51
            frame_sequences = self.section_run.section_frame_sequence
52
53
54
55
56
57
58
59
        except Exception:
            frame_sequences = []

        n_scc = len(sccs)
        n_frame_seq = len(frame_sequences)

        # No sequences, only a few calculations
        if n_scc <= 3 and n_frame_seq == 0:
60
            program_name = self.section_run.program_name
61
62
63
64
65
66
67
68
69
70
71
72
            if program_name == "elastic":
                # TODO move to taylor expansion as soon as data is correct in archive
                calc_type = calc_enums.elastic_constants
            else:
                calc_type = calc_enums.single_point

        # One sequence. Currently calculations with multiple sequences are
        # unsupported.
        elif n_frame_seq == 1:
            frame_seq = frame_sequences[0]

            # See if sampling_method is present
73
74
            section_sampling_method = frame_seq.frame_sequence_to_sampling_ref
            if section_sampling_method is None:
75
76
77
78
79
80
81
                self.logger.info(
                    "Cannot determine encyclopedia run type because missing "
                    "value for frame_sequence_to_sampling_ref."
                )
                return calc_type

            # See if local frames are present
82
83
            frames = frame_seq.frame_sequence_local_frames_ref
            if not frames:
84
85
86
                self.logger.info("No frames referenced in section_frame_sequence_local_frames.")
                return calc_type

87
            sampling_method = section_sampling_method.sampling_method
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

            if sampling_method == "molecular_dynamics":
                calc_type = calc_enums.molecular_dynamics
            if sampling_method == "geometry_optimization":
                calc_type = calc_enums.geometry_optimization
            if sampling_method == "taylor_expansion":
                calc_type = calc_enums.phonon_calculation

        calc.calculation_type = calc_type
        return calc_type

    def material_type(self, material: Material) -> tuple:
        # Try to fetch representative system
        system = None
        material_type = config.services.unavailable_value
        material_enums = Material.material_type.type
104
        try:
105
            system_idx = self.section_run.m_cache["representative_system_idx"]
106
107
108
        except (AttributeError, KeyError):
            pass
        else:
109
110
            # Try to find system type information from backend for the selected system.
            try:
111
112
                system = self.section_run.section_system[system_idx]
                stype = system.system_type
113
114
115
116
117
118
119
120
            except KeyError:
                pass
            else:
                if stype == material_enums.one_d or stype == material_enums.two_d:
                    material_type = stype
                # For bulk systems we also ensure that the symmetry information is available
                if stype == material_enums.bulk:
                    try:
121
                        system.section_symmetry[0]
122
123
124
125
126
127
128
129
                    except (KeyError, IndexError):
                        self.logger.info("Symmetry information is not available for a bulk system. No Encylopedia entry created.")
                    else:
                        material_type = stype

        material.material_type = material_type
        return system, material_type

130
    def method_type(self, method: Method) -> tuple:
131
132
        repr_method = None
        method_id = config.services.unavailable_value
133
        methods = self.section_run.section_method
134
135
136
137
        n_methods = len(methods)

        if n_methods == 1:
            repr_method = methods[0]
138
139
140
            method_id = repr_method.electronic_structure_method
            if method_id is None:
                method_id = config.services.unavailable_value
141
        elif n_methods > 1:
142
            for sec_method in methods:
143
                # GW
144
                electronic_structure_method = sec_method.electronic_structure_method
145
146
147
148
149
150
151
152
153
                if electronic_structure_method in {"G0W0", "scGW"}:
                    repr_method = sec_method
                    method_id = "GW"
                    break

                # Methods linked to each other through references. Get all
                # linked methods, try to get electronic_structure_method from
                # each.
                try:
154
                    refs = sec_method.section_method_to_method_refs
155
156
157
158
159
                except KeyError:
                    pass
                else:
                    linked_methods = [sec_method]
                    for ref in refs:
160
161
                        method_to_method_kind = ref.method_to_method_kind
                        method_to_method_ref = ref.method_to_method_ref
162
                        if method_to_method_kind == "core_settings":
163
                            linked_methods.append(method_to_method_ref)
164
165

                    for i_method in linked_methods:
166
167
                        electronic_structure_method = i_method.electronic_structure_method
                        if electronic_structure_method is not None:
168
169
170
171
172
173
174
175
176
177
                            repr_method = sec_method
                            method_id = electronic_structure_method

        method.method_type = method_id
        return repr_method, method_id

    def fill(self, context: Context):
        # Fill structure related metainfo
        struct: Any = None
        if context.material_type == Material.material_type.type.bulk:
178
            struct = MaterialBulkNormalizer(self.entry_archive, self.logger)
179
        elif context.material_type == Material.material_type.type.two_d:
180
            struct = Material2DNormalizer(self.entry_archive, self.logger)
181
        elif context.material_type == Material.material_type.type.one_d:
182
            struct = Material1DNormalizer(self.entry_archive, self.logger)
183
184
185
186
187
188
        if struct is not None:
            struct.normalize(context)

        # Fill method related metainfo
        method = None
        if context.method_type == Method.method_type.type.DFT or context.method_type == Method.method_type.type.DFTU:
189
            method = MethodDFTNormalizer(self.entry_archive, self.logger)
190
        elif context.method_type == Method.method_type.type.GW:
191
            method = MethodGWNormalizer(self.entry_archive, self.logger)
192
193
194
195
        if method is not None:
            method.normalize(context)

        # Fill properties related metainfo
196
        properties = PropertiesNormalizer(self.entry_archive, self.logger)
197
198
199
200
201
202
        properties.normalize(context)

    def normalize(self, logger=None) -> None:
        """The caller will automatically log if the normalizer succeeds or ends
        up with an exception.
        """
203
        sec_enc = self.entry_archive.section_metadata.m_create(EncyclopediaMetadata)
204
        status_enums = EncyclopediaMetadata.status.type
205
        calc_enums = Calculation.calculation_type.type
206

207
208
        # Do nothing if section_run is not present
        if self.section_run is None:
209
210
            status = status_enums.invalid_metainfo
            sec_enc.status = status
211
            self.logger.info(
212
213
                "required metainfo is missing or is invalid.",
                enc_status=status,
214
215
216
217
                invalid_metainfo="section_run",
            )
            return

218
219
220
221
222
223
224
225
226
227
228
        try:
            super().normalize(logger)
            # Initialise metainfo structure
            material = sec_enc.m_create(Material)
            method = sec_enc.m_create(Method)
            sec_enc.m_create(Properties)
            calc = sec_enc.m_create(Calculation)

            # Determine run type, stop if unknown
            calc_type = self.calc_type(calc)
            if calc_type == config.services.unavailable_value:
229
230
                status = status_enums.unsupported_calculation_type
                sec_enc.status = status
231
                self.logger.info(
232
233
                    "unsupported calculation type for encyclopedia",
                    enc_status=status,
234
235
236
                )
                return

237
            # Get the system type.
238
239
240
            material_enums = Material.material_type.type
            representative_system, material_type = self.material_type(material)
            if material_type != material_enums.bulk and material_type != material_enums.two_d and material_type != material_enums.one_d:
241
242
                status = status_enums.unsupported_material_type
                sec_enc.status = status
243
                self.logger.info(
244
245
                    "unsupported material type for encyclopedia",
                    enc_status=status,
246
                )
247
                return
248

249
250
251
            # Get the method type. For now, we allow unknown method type for
            # phonon calculations, as the method information is resolved at a
            # later stage.
252
            representative_method, method_type = self.method_type(method)
253
            if method_type == config.services.unavailable_value:
254
                sec_enc.status = status_enums.unsupported_method_type
255
256
                self.logger.info(
                    "unsupported method type for encyclopedia",
257
                    enc_status=status_enums.unsupported_method_type,
258
                )
259
260
                if calc_type != calc_enums.phonon_calculation:
                    return
261
262
263

            # Get representative scc
            try:
264
265
                representative_scc_idx = self.section_run.m_cache["representative_scc_idx"]
                representative_scc = self.section_run.section_single_configuration_calculation[representative_scc_idx]
266
            except Exception:
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
                representative_scc = None
                representative_scc_idx = None

            # Create one context that holds all details
            context = Context(
                material_type=material_type,
                method_type=method_type,
                calc_type=calc_type,
                representative_system=representative_system,
                representative_method=representative_method,
                representative_scc=representative_scc,
                representative_scc_idx=representative_scc_idx,
            )

            # Put the encyclopedia section into backend
            self.fill(context)
283

284
285
286
287
288
            # Check that the necessary information is in place
            functional_type = method.functional_type
            if functional_type is None:
                sec_enc.status = status_enums.unsupported_method_type
                self.logger.info(
289
                    "unsupported functional type for encyclopedia",
290
291
292
293
                    enc_status=status_enums.unsupported_method_type,
                )
                return

294
        except Exception:
295
296
            status = status_enums.failure
            sec_enc.status = status
297
            self.logger.error(
298
299
                "failed to process encyclopedia data due to an unhandlable exception",
                enc_status=status,
300
301
302
            )
            raise  # Reraise for the caller to log the exception as well
        else:
303
304
            status = status_enums.success
            sec_enc.status = status
305
            self.logger.info(
306
307
                "successfully created metainfo for encyclopedia.",
                enc_status=status,
308
            )