configuration.py 7 KB
Newer Older
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
165
166
167
168
169
170
171
172
#!/usr/bin/env python3

# Copyright (C) 2021:
#   Helmholtz-Zentrum Potsdam Deutsches GeoForschungsZentrum GFZ
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or (at
# your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
# General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see http://www.gnu.org/licenses/.

import logging
import yaml

logger = logging.getLogger()


class Configuration:
    """This class handles the configuration parameters of the gde-importer.

    Attributes:
        self.model_name (str):
            Name of the input aggregated model.
        self.exposure_format (str):
            Format of the input aggregated model. Currently supported values: "esrm20".
        self.metadata_filepath (str):
            Full file path to the .xlsx file that contains metadata on the input aggregated
            exposure model.
        self.occupancy_cases (dictionary):
            Dictionary in which each first level key corresponds to an occupancy case (e.g.
            "residential", "commercial", "industrial"). Each first level key contains two
            sub-keys: "sheet_name" and "data_units_types_field":
                sheet_name (str):
                    Name of the sheet in the meatadata file of the input aggregated exposure
                    model from which info on this occupancy case can be retrieved. E.g.: "RES",
                    "COM", "IND".
                data_units_types_field (str):
                    Name of the field in sheet_name from which to retrieve information on the
                    types of data_units. E.g.: "Resolution".
    """

    REQUIRES = ["metadata_filepath", "occupancy_cases"]

    def __init__(self, filepath):
        """
        Args:
            filepath (str):
                Full file path to the .yml configuration file.
        """

        config = self.read_config_file(filepath)

        self.model_name = self._assign_parameter(config, "model_name")

        self.exposure_format = self._assign_parameter(config, "exposure_format")
        self.metadata_filepath = self._assign_parameter(config, "metadata_filepath")
        self.occupancy_cases = self._assign_hierarchical_parameters(
            config, "occupancy_cases", ["sheet_name", "data_units_types_field"]
        )

        # Terminate if critical parameters are missing (not all parameters are critical)
        for key_parameter in self.REQUIRES:
            if not getattr(self, key_parameter):
                error_message = (
                    "ERROR: PARAMETER '%s' COULD NOT BE RETRIEVED FROM "
                    "CONFIGURATION FILE. THE PROGRAM CANNOT RUN." % (key_parameter)
                )
                logger.critical(error_message)
                raise OSError(error_message)

    def read_config_file(self, filepath):
        """This function attempts to open the configuration file. If not found, it returns an
        empty dictionary and logs a critical error.

        Args:
            filepath (str):
                Full file path to the .yml configuration file.

        Returns:
            config (dictionary):
                The configuration file read as a dictionary, or an empty dictionary if the
                configuration file was not found.
        """

        try:
            with open(filepath, "r") as ymlfile:
                config = yaml.load(ymlfile, Loader=yaml.FullLoader)
        except FileNotFoundError:
            logger.critical("ERROR instantiating Configuration: configuration file not found")
            config = {}

        return config

    def _assign_parameter(self, config, input_parameter):
        """This function searches for the key input_parameter in the dictionary config. If
        found, it returns its value (a string or a dictionary). If not found, it returns None.

        Args:
            config (dictionary):
                The configuration file read as a dictionary. It may be an empty dictionary.
            input_parameter (str):
                Name of the desired parameter, to be searched for as a primary key of config.
        Returns:
            assigned_parameter (str, dictionary or None):
                The content of config[input_parameter], which can be a string or a dictionary.
                It is None if input_parameter is not a key of config.
        """

        try:
            assigned_parameter = config[input_parameter]
        except KeyError:
            logger.warning(
                "WARNING: parameter '%s' is missing from configuration file" % (input_parameter)
            )
            assigned_parameter = None

        return assigned_parameter

    def _assign_hierarchical_parameters(self, config, input_parameter, requested_nested):
        """This function searches for the key input_parameter in the dictionary config, and for
        each of the elements of requested_nested as keys of config[input_parameter].

        If input_parameter is not a key of config, the output is None.

        If input_parameter is a key of config, but one of the elements of requested_nested is
        not a key of config[input_parameter]

        Args:
            config (dictionary):
                The configuration file read as a dictionary. It may be an empty dictionary.
            input_parameter (str):
                Name of the desired parameter, to be searched for as a primary key of config.
            requested_nested (list of str):
                List of the names of the desired nested parameters, to be searched for as keys
                of config[input_parameter].

        Returns:
            assigned_parameter (dictionary or None):
                The content of config[input_parameter], if input_parameter is a key of config
                and all elements of requested_nested are keys of config[input_parameter], or
                None otherwise.
        """

        assigned_parameter = self._assign_parameter(config, input_parameter)

        if assigned_parameter is None:
            return None

        if not isinstance(assigned_parameter, dict):
            return None

        sub_parameters_missing = False
        for case in assigned_parameter.keys():
            for requested_parameter in requested_nested:
                if requested_parameter not in assigned_parameter[case].keys():
                    logger.critical(
                        "ERROR instantiating Configuration: occupancy case '%s' does not "
                        "contain a '%s' parameter" % (case, requested_parameter)
                    )
                    sub_parameters_missing = True

        if sub_parameters_missing is True:
            assigned_parameter = None

        return assigned_parameter