# -*- coding: utf-8 -*-
# /***************************************************************************
# Irmt
# A QGIS plugin
# OpenQuake Integrated Risk Modelling Toolkit
# -------------------
# begin : 2015-06-15
# copyright : (C) 2015 by GEM Foundation
# email : devops@openquake.org
# ***************************************************************************/
#
# OpenQuake 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.
#
# OpenQuake 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 General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with OpenQuake. If not, see <http://www.gnu.org/licenses/>.
import unittest
import os
import sys
from copy import deepcopy
from qgis.core import QgsVectorLayer, QgsVectorFileWriter
from utilities import get_qgis_app
from svir.calculations.calculate_utils import (calculate_node,
get_node_attr_id_and_name,
calculate_composite_variable,
)
from svir.calculations.process_layer import ProcessLayer
from svir.utilities.utils import (set_operator,
get_node,
save_layer_as_shapefile,
)
from svir.utilities.shared import OPERATORS_DICT, DiscardedFeature
QGIS_APP, CANVAS, IFACE, PARENT = get_qgis_app()
PROJ_DEF_STD_OPERATORS = {
"name": "IRI",
"weight": 1.0,
"level": "1.0",
"operator": "Weighted sum",
"type": "Integrated Risk Index",
"children": [
{"name": "RI",
"type": "Risk Index",
"children": [],
"weight": 0.5,
"level": "2.0"
},
{"name": "SVI",
"type": "Social Vulnerability Index",
"children": [
{"name": "Education",
"weight": 0.5,
"level": "3.0",
"operator": "Weighted sum",
"type": "Social Vulnerability Theme",
"children": [
{"name": ("Female population without secondary"
" education or higher"),
"isInverted": True,
"weight": 0.2,
"level": 4.0,
"field": "EDUEOCSAF",
"type": "Social Vulnerability Indicator",
"children": []
},
{"name": ("Male population without secondary"
" education or higher"),
"isInverted": True,
"weight": 0.3,
"level": 4.0,
"field": "EDUEOCSAM",
"type": "Social Vulnerability Indicator",
"children": []
},
{"name": "Scientific and technical journal articles",
"weight": 0.5,
"level": 4.0,
"field": "EDUEOCSTJ",
"type": "Social Vulnerability Indicator",
"children": []
}
]
},
{"name": "Environment",
"weight": 0.5,
"level": "3.0",
"operator": "Weighted sum",
"type": "Social Vulnerability Theme",
"children": [
{"name": "Droughts, floods, extreme temperatures",
"weight": 0.5,
"level": 4.1,
"field": "ENVDIPDFT",
"type": "Social Vulnerability Indicator",
"children": []
},
{"name": "Natural disasters - Number of deaths",
"weight": 0.5,
"level": 4.1,
"field": "ENVDIPIND",
"type": "Social Vulnerability Indicator",
"children": []
}
]
}
],
"weight": 0.5,
"level": "2.0"
}
],
"description": ""
}
[docs]class CalculateCompositeVariableTestCase(unittest.TestCase):
[docs] def setUp(self):
self.project_definition = PROJ_DEF_STD_OPERATORS
# Load layer
curr_dir_name = os.path.dirname(__file__)
self.data_dir_name = os.path.join(curr_dir_name,
'data/calculate_indices')
layer_path = os.path.join(
self.data_dir_name, 'socioeconomic_data.shp')
orig_layer = QgsVectorLayer(layer_path, 'Zonal Layer', 'ogr')
# Avoid modifying the original files
self.layer = ProcessLayer(orig_layer).duplicate_in_memory()
[docs] def test_custom_operator(self):
proj_def = deepcopy(self.project_definition)
operator = OPERATORS_DICT['CUSTOM']
# set economy's operator to custom and use a custom formula
proj_def['children'][1]['children'][0]['operator'] = operator
proj_def['children'][1]['children'][0]['fieldDescription'] = \
'EDUEOCSAF plus one'
proj_def['children'][1]['children'][0]['customFormula'] = \
'"EDUEOCSAF" + 1'
node_attr_id, node_attr_name, discarded_feats = \
calculate_education_node(proj_def, operator, self.layer)
# # to rebuild the outputs
# res_layer_name = 'custom_operator'
# write_output(self.layer, self.data_dir_name, res_layer_name)
expected_layer_path = os.path.join(
self.data_dir_name, 'custom_operator.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'custom_operator', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
try:
self.assertEqual(res, True)
except AssertionError:
sys.stderr.write("The resulting layer is different than expected")
sys.stderr.write("\n\n\nCalculated EDUCATION as EDUEOCSAF+1:\n")
ProcessLayer(self.layer).pprint(usage='testing')
sys.stderr.write("\n\n\nExpected layer (custom_operator.shp):\n")
ProcessLayer(expected_layer).pprint(usage='testing')
raise
[docs] def test_simple_sum(self):
proj_def = deepcopy(self.project_definition)
operator = OPERATORS_DICT['SUM_S']
node_attr_id, node_attr_name, discarded_feats = \
calculate_education_node(proj_def, operator, self.layer)
expected_layer_path = os.path.join(
self.data_dir_name, 'simple_sum.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'simple_sum', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
self.assertEqual(res, True)
# # to rebuild the outputs
# res_layer_name = 'simple_sum'
# write_output(self.layer, self.data_dir_name, res_layer_name)
[docs] def test_weighted_sum(self):
proj_def = deepcopy(self.project_definition)
operator = OPERATORS_DICT['SUM_W']
node_attr_id, node_attr_name, discarded_feats = \
calculate_education_node(proj_def, operator, self.layer)
expected_layer_path = os.path.join(
self.data_dir_name, 'weighted_sum.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'weighted_sum', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
self.assertEqual(res, True)
# to rebuild the outputs
# res_layer_name = 'weighted_sum'
# write_output(self.layer, self.data_dir_name, res_layer_name)
[docs] def test_simple_multiplication(self):
proj_def = deepcopy(self.project_definition)
operator = OPERATORS_DICT['MUL_S']
node_attr_id, node_attr_name, discarded_feats = \
calculate_education_node(proj_def, operator, self.layer)
expected_layer_path = os.path.join(
self.data_dir_name, 'simple_multiplication.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'simple_multiplication', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
self.assertEqual(res, True)
# # to rebuild the outputs
# res_layer_name = 'simple_multiplication'
# write_output(self.layer, self.data_dir_name, res_layer_name)
[docs] def test_weighted_multiplication(self):
proj_def = deepcopy(self.project_definition)
operator = OPERATORS_DICT['MUL_W']
node_attr_id, node_attr_name, discarded_feats = \
calculate_education_node(proj_def, operator, self.layer)
expected_layer_path = os.path.join(
self.data_dir_name, 'weighted_multiplication.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'weighted_multiplication', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
self.assertEqual(res, True)
# # to rebuild the outputs
# res_layer_name = 'weighted_multiplication'
# write_output(self.layer, self.data_dir_name, res_layer_name)
[docs] def test_average(self):
proj_def = deepcopy(self.project_definition)
operator = OPERATORS_DICT['AVG']
node_attr_id, node_attr_name, discarded_feats = \
calculate_education_node(proj_def, operator, self.layer)
expected_layer_path = os.path.join(
self.data_dir_name, 'average.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'average', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
self.assertEqual(res, True)
# # to rebuild the outputs
# res_layer_name = 'average'
# write_output(self.layer, self.data_dir_name, res_layer_name)
[docs] def test_geometric_mean_positive_argument(self):
proj_def = deepcopy(self.project_definition)
operator = OPERATORS_DICT['GEOM_MEAN']
node_attr_id, node_attr_name, discarded_feats = \
calculate_education_node(proj_def, operator, self.layer)
expected_layer_path = os.path.join(
self.data_dir_name, 'geometric_mean_positive_argument.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'geometric_mean_positive_argument', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
self.assertEqual(res, True)
# # to rebuild the outputs
# res_layer_name = 'geometric_mean_positive_argument'
# write_output(self.layer, self.data_dir_name, res_layer_name)
[docs] def test_geometric_mean_negative_argument(self):
proj_def = deepcopy(self.project_definition)
# do not invert EDUEOCSAM ==> it should cause the geometric mean to
# attempt calculating the root of a negative number, so we should have
# the corresponding field discarded with 'Invalid value' reason
assert proj_def['children'][1]['children'][0]['children'][1]['field'] \
== 'EDUEOCSAM'
# SVI Education EDUEOCSAM
proj_def['children'][1]['children'][0]['children'][1]['isInverted'] \
= False
operator = OPERATORS_DICT['GEOM_MEAN']
node_attr_id, node_attr_name, discarded_feats = \
calculate_education_node(proj_def, operator, self.layer)
expected_layer_path = os.path.join(
self.data_dir_name, 'geometric_mean_negative_argument.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'geometric_mean_negative_argument', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
self.assertEqual(res, True)
# # to rebuild the outputs
# res_layer_name = 'geometric_mean_negative_argument'
# write_output(self.layer, self.data_dir_name, res_layer_name)
[docs] def test_calculate_svi(self):
proj_def = deepcopy(self.project_definition)
svi_node = proj_def['children'][1]
added_attrs_ids, discarded_feats, edited_node, any_change = \
calculate_composite_variable(IFACE, self.layer, svi_node)
proj_def['children'][1] = edited_node
self.assertEqual(added_attrs_ids, set([9, 10, 11]))
expected_discarded_feats = set([DiscardedFeature(1, 'Missing value'),
DiscardedFeature(2, 'Missing value')])
self.assertEqual(discarded_feats,
expected_discarded_feats)
self.assertEqual(any_change, True)
self.assertEqual(proj_def, proj_def_svi_calc_first_round)
expected_layer_path = os.path.join(
self.data_dir_name, 'svi_calculation_first_round.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'svi_calculation_first_round', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
self.assertEqual(res, True)
# # to rebuild the outputs
# res_layer_name = 'svi_calculation_first_round'
# write_output(self.layer, self.data_dir_name, res_layer_name)
# If the attributes have already been added to the layer, they should
# be re-used instead of adding new ones
# therefore the project definition should be the same as it was after
# the first round
svi_node = proj_def['children'][1]
added_attrs_ids, discarded_feats, edited_node, any_change = \
calculate_composite_variable(IFACE, self.layer, svi_node)
proj_def['children'][1] = edited_node
self.assertEqual(added_attrs_ids, set([]))
expected_discarded_feats = set([DiscardedFeature(1, 'Missing value'),
DiscardedFeature(2, 'Missing value')])
self.assertEqual(discarded_feats,
expected_discarded_feats)
self.assertEqual(any_change, True)
self.assertEqual(proj_def, proj_def_svi_calc_first_round)
expected_layer_path = os.path.join(
self.data_dir_name, 'svi_calculation_first_round.shp')
expected_layer = QgsVectorLayer(
expected_layer_path, 'svi_calculation_first_round', 'ogr')
res = ProcessLayer(self.layer).has_same_content_as(expected_layer)
self.assertEqual(res, True)
[docs]def calculate_education_node(proj_def, operator, layer):
"""
Use the calculate_node function to compute the 'Education' node using the
given project definition and operator.
The layer is updated as a side effect.
"""
# set all operators of proj_def to the one selected
proj_def = set_operator(proj_def, operator)
# we are testing the calculation for a single node
education_node = get_node(proj_def, 'Education')
node_attr_id, node_attr_name, field_was_added = \
get_node_attr_id_and_name(education_node, layer)
discarded_feats = set()
discarded_feats = calculate_node(
education_node, node_attr_name, node_attr_id,
layer, discarded_feats)
return node_attr_id, node_attr_name, discarded_feats
[docs]def write_output(res_layer, data_dir_name, res_layer_name):
res_layer_path = os.path.join(data_dir_name, res_layer_name + '.shp')
write_success = save_layer_as_shapefile(res_layer, res_layer_path)
if write_success != QgsVectorFileWriter.NoError:
raise RuntimeError('Could not save shapefile')
proj_def_svi_calc_first_round = {
'children': [
{'children': [],
'level': '2.0',
'name': 'RI',
'type': 'Risk Index',
'weight': 0.5},
{'children': [
{'children': [
{'children': [],
'field': 'EDUEOCSAF',
'isInverted': True,
'level': 4.0,
'name': ('Female population without secondary'
' education or higher'),
'type': 'Social Vulnerability Indicator',
'weight': 0.2},
{'children': [],
'field': 'EDUEOCSAM',
'isInverted': True,
'level': 4.0,
'name': ('Male population without secondary'
' education or higher'),
'type': 'Social Vulnerability Indicator',
'weight': 0.3},
{'children': [],
'field': 'EDUEOCSTJ',
'level': 4.0,
'name': 'Scientific and technical journal articles',
'type': 'Social Vulnerability Indicator',
'weight': 0.5}],
'field': u'EDUCATION',
'level': '3.0',
'name': 'Education',
'operator': 'Weighted sum',
'type': 'Social Vulnerability Theme',
'weight': 0.5},
{'children': [
{'children': [],
'field': 'ENVDIPDFT',
'level': 4.1,
'name': 'Droughts, floods, extreme temperatures',
'type': 'Social Vulnerability Indicator',
'weight': 0.5},
{'children': [],
'field': 'ENVDIPIND',
'level': 4.1,
'name': 'Natural disasters - Number of deaths',
'type': 'Social Vulnerability Indicator',
'weight': 0.5}],
'field': u'ENVIRONMEN',
'level': '3.0',
'name': 'Environment',
'operator': 'Weighted sum',
'type': 'Social Vulnerability Theme',
'weight': 0.5}],
'field': u'SVI',
'level': '2.0',
'name': 'SVI',
'type': 'Social Vulnerability Index',
'weight': 0.5}],
'description': '',
'level': '1.0',
'name': 'IRI',
'operator': 'Weighted sum',
'type': 'Integrated Risk Index',
'weight': 1.0
}