Source code for svir.test.unit.test_load_oq_engine_output_as_layer

# -*- coding: utf-8 -*-
# /***************************************************************************
# Irmt
#                                 A QGIS plugin
# OpenQuake Integrated Risk Modelling Toolkit
#                              -------------------
#        begin                : 201-10-24
#        copyright            : (C) 2014-2017 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 qgis libs so that we set the correct sip api version
import qgis  # NOQA

import os
import unittest
import tempfile
import csv
from mock import Mock
from numpy.testing import assert_almost_equal

from qgis.PyQt.QtGui import QAction
from qgis.core import QgsVectorLayer
from svir.dialogs.load_dmg_by_asset_as_layer_dialog import (
    LoadDmgByAssetAsLayerDialog)
from svir.dialogs.load_ruptures_as_layer_dialog import (
    LoadRupturesAsLayerDialog)
from svir.dialogs.load_gmf_data_as_layer_dialog import (
    LoadGmfDataAsLayerDialog)
from svir.dialogs.load_losses_by_asset_as_layer_dialog import (
    LoadLossesByAssetAsLayerDialog)
from svir.dialogs.viewer_dock import ViewerDock
from svir.calculations.process_layer import ProcessLayer
from svir.test.utilities import get_qgis_app

QGIS_APP, CANVAS, IFACE, PARENT = get_qgis_app()


[docs]class LoadOQEngineOutputAsLayerTestCase(unittest.TestCase):
[docs] def setUp(self): IFACE.newProject() curr_dir_name = os.path.dirname(__file__) self.data_dir_name = os.path.join( curr_dir_name, os.pardir, 'data') mock_action = QAction(IFACE.mainWindow()) self.viewer_dock = ViewerDock(IFACE, mock_action)
[docs] def test_load_gmf(self): filepath = os.path.join(self.data_dir_name, 'hazard', 'output-195-gmf_data_70.npz') # TODO: in the future, we will move this to integration tests, using # session, hostname and calc_id and the extract api, instead of # mocking dlg = LoadGmfDataAsLayerDialog( IFACE, self.viewer_dock, Mock(), Mock(), Mock(), 'gmf_data', filepath) dlg.accept()
# ground motion fields have nothing to do with the Data Viewer
[docs] def test_load_ruptures(self): filepath = os.path.join( self.data_dir_name, 'hazard', 'ruptures', 'output-607-ruptures_162.csv') # TODO: in the future, we will move this to integration tests, using # session, hostname and calc_id and the extract api, instead of # mocking dlg = LoadRupturesAsLayerDialog( IFACE, self.viewer_dock, Mock(), Mock(), Mock(), 'ruptures', filepath, mode='testing') dlg.save_as_shp_ckb.setChecked(True) dlg.accept() current_layer = IFACE.activeLayer() reference_path = os.path.join( self.data_dir_name, 'hazard', 'ruptures', 'output-607-ruptures_162.shp') reference_layer = QgsVectorLayer( reference_path, 'reference_ruptures', 'ogr') ProcessLayer(current_layer).has_same_content_as(reference_layer)
[docs] def test_load_losses_by_asset_only_selected_taxonomy_and_loss_type(self): filepath = os.path.join(self.data_dir_name, 'risk', 'output-399-losses_by_asset_123.npz') # TODO: in the future, we will move this to integration tests, using # session, hostname and calc_id and the extract api, instead of # mocking dlg = LoadLossesByAssetAsLayerDialog( IFACE, self.viewer_dock, Mock(), Mock(), Mock(), 'losses_by_asset', filepath) dlg.load_selected_only_ckb.setChecked(True) taxonomy_idx = dlg.taxonomy_cbx.findText('"Concrete"') self.assertNotEqual(taxonomy_idx, -1, 'Taxonomy "Concrete" was not found') dlg.taxonomy_cbx.setCurrentIndex(taxonomy_idx) loss_type_idx = dlg.loss_type_cbx.findText('structural') self.assertNotEqual(loss_type_idx, -1, 'Loss type structural was not found') dlg.loss_type_cbx.setCurrentIndex(loss_type_idx) dlg.accept()
[docs] def test_load_losses_by_asset_all_taxonomies_only_selected_loss_type(self): filepath = os.path.join(self.data_dir_name, 'risk', 'output-399-losses_by_asset_123.npz') # TODO: in the future, we will move this to integration tests, using # session, hostname and calc_id and the extract api, instead of # mocking dlg = LoadLossesByAssetAsLayerDialog( IFACE, self.viewer_dock, Mock(), Mock(), Mock(), 'losses_by_asset', filepath) dlg.load_selected_only_ckb.setChecked(True) taxonomy_idx = dlg.taxonomy_cbx.findText('All') self.assertNotEqual(taxonomy_idx, -1, 'Taxonomy All was not found') dlg.taxonomy_cbx.setCurrentIndex(taxonomy_idx) loss_type_idx = dlg.loss_type_cbx.findText('structural') self.assertNotEqual(loss_type_idx, -1, 'Loss type structural was not found') dlg.loss_type_cbx.setCurrentIndex(loss_type_idx) dlg.accept()
@unittest.skip("Causing segfault")
[docs] def test_load_losses_by_asset_aggregate_by_zone(self): loss_layer_path = os.path.join(self.data_dir_name, 'risk', 'output-399-losses_by_asset_123.npz') zonal_layer_path = os.path.join(self.data_dir_name, 'risk', 'zonal_layer.shp') # TODO: in the future, we will move this to integration tests, using # session, hostname and calc_id and the extract api, instead of # mocking dlg = LoadLossesByAssetAsLayerDialog( IFACE, self.viewer_dock, Mock(), Mock(), Mock(), 'losses_by_asset', loss_layer_path, zonal_layer_path=zonal_layer_path) dlg.load_selected_only_ckb.setChecked(True) dlg.zonal_layer_gbx.setChecked(True) taxonomy_idx = dlg.taxonomy_cbx.findText('All') self.assertNotEqual(taxonomy_idx, -1, 'Taxonomy All was not found') dlg.taxonomy_cbx.setCurrentIndex(taxonomy_idx) loss_type_idx = dlg.loss_type_cbx.findText('structural') self.assertNotEqual(loss_type_idx, -1, 'Loss type structural was not found') dlg.loss_type_cbx.setCurrentIndex(loss_type_idx) self.assertTrue(dlg.zonal_layer_cbx.currentText(), 'The zonal layer was not loaded') dlg.accept() zonal_layer_plus_stats = [layer for layer in IFACE.layers() if layer.name() == 'Zonal data (copy)'][0] zonal_layer_plus_stats_first_feat = \ zonal_layer_plus_stats.getFeatures().next() expected_zonal_layer_path = os.path.join( self.data_dir_name, 'risk', 'zonal_layer_plus_losses_by_asset_stats.shp') expected_zonal_layer = QgsVectorLayer( expected_zonal_layer_path, 'Zonal data', 'ogr') expected_zonal_layer_first_feat = \ expected_zonal_layer.getFeatures().next() assert_almost_equal( zonal_layer_plus_stats_first_feat.attributes(), expected_zonal_layer_first_feat.attributes())
[docs] def test_load_dmg_by_asset_only_selected_taxonomy(self): filepath = os.path.join(self.data_dir_name, 'risk', 'output-1614-dmg_by_asset_356.npz') # TODO: in the future, we will move this to integration tests, using # session, hostname and calc_id and the extract api, instead of # mocking dlg = LoadDmgByAssetAsLayerDialog( IFACE, self.viewer_dock, Mock(), Mock(), Mock(), 'dmg_by_asset', filepath) dlg.load_selected_only_ckb.setChecked(True) taxonomy_idx = dlg.taxonomy_cbx.findText('"Concrete"') self.assertNotEqual(taxonomy_idx, -1, 'Taxonomy "Concrete" was not found') dlg.taxonomy_cbx.setCurrentIndex(taxonomy_idx) loss_type_idx = dlg.loss_type_cbx.findText('structural') self.assertNotEqual(loss_type_idx, -1, 'Loss type structural was not found') dlg.loss_type_cbx.setCurrentIndex(loss_type_idx) dmg_state_idx = dlg.dmg_state_cbx.findText('moderate') self.assertNotEqual(dmg_state_idx, -1, 'Damage state moderate was not found') dlg.dmg_state_cbx.setCurrentIndex(dmg_state_idx) dlg.accept()
[docs] def test_load_dmg_by_asset_all_taxonomies(self): filepath = os.path.join(self.data_dir_name, 'risk', 'output-1614-dmg_by_asset_356.npz') # TODO: in the future, we will move this to integration tests, using # session, hostname and calc_id and the extract api, instead of # mocking dlg = LoadDmgByAssetAsLayerDialog( IFACE, self.viewer_dock, Mock(), Mock(), Mock(), 'dmg_by_asset', filepath) dlg.load_selected_only_ckb.setChecked(True) taxonomy_idx = dlg.taxonomy_cbx.findText('All') self.assertNotEqual(taxonomy_idx, -1, 'Taxonomy All was not found') dlg.taxonomy_cbx.setCurrentIndex(taxonomy_idx) loss_type_idx = dlg.loss_type_cbx.findText('structural') self.assertNotEqual(loss_type_idx, -1, 'Loss type structural was not found') dlg.loss_type_cbx.setCurrentIndex(loss_type_idx) dmg_state_idx = dlg.dmg_state_cbx.findText('moderate') self.assertNotEqual(dmg_state_idx, -1, 'Damage state moderate was not found') dlg.dmg_state_cbx.setCurrentIndex(dmg_state_idx) dlg.accept()
@unittest.skip("Causing segfault")
[docs] def test_load_dmg_by_asset_aggregate_by_zone(self): dmg_layer_path = os.path.join(self.data_dir_name, 'risk', 'output-1614-dmg_by_asset_356.npz') zonal_layer_path = os.path.join(self.data_dir_name, 'risk', 'zonal_layer.shp') # TODO: in the future, we will move this to integration tests, using # session, hostname and calc_id and the extract api, instead of # mocking dlg = LoadDmgByAssetAsLayerDialog( IFACE, self.viewer_dock, Mock(), Mock(), Mock(), 'dmg_by_asset', dmg_layer_path, zonal_layer_path=zonal_layer_path) dlg.load_selected_only_ckb.setChecked(True) dlg.zonal_layer_gbx.setChecked(True) taxonomy_idx = dlg.taxonomy_cbx.findText('All') self.assertNotEqual(taxonomy_idx, -1, 'Taxonomy All was not found') dlg.taxonomy_cbx.setCurrentIndex(taxonomy_idx) loss_type_idx = dlg.loss_type_cbx.findText('structural') self.assertNotEqual(loss_type_idx, -1, 'Loss type structural was not found') dlg.loss_type_cbx.setCurrentIndex(loss_type_idx) dmg_state_idx = dlg.dmg_state_cbx.findText('moderate') self.assertNotEqual(dmg_state_idx, -1, 'Damage state moderate was not found') dlg.dmg_state_cbx.setCurrentIndex(dmg_state_idx) dlg.accept() zonal_layer_plus_stats = [layer for layer in IFACE.layers() if layer.name() == 'Zonal data (copy)'][0] zonal_layer_plus_stats_first_feat = \ zonal_layer_plus_stats.getFeatures().next() expected_zonal_layer_path = os.path.join( self.data_dir_name, 'risk', 'zonal_layer_plus_dmg_by_asset_stats.shp') expected_zonal_layer = QgsVectorLayer( expected_zonal_layer_path, 'Zonal data', 'ogr') expected_zonal_layer_first_feat = \ expected_zonal_layer.getFeatures().next() assert_almost_equal( zonal_layer_plus_stats_first_feat.attributes(), expected_zonal_layer_first_feat.attributes())
def _test_export(self): _, exported_file_path = tempfile.mkstemp(suffix=".csv") layer = IFACE.activeLayer() # select the first 2 features (the same used to produce the reference # csv) layer.select([1, 2]) # probably we have the wrong layer selected (uhs produce many layers) self.viewer_dock.write_export_file(exported_file_path) # NOTE: we are only checking that the exported CSV has at least 3 rows # and 3 columns per row. We are avoiding more precise checks, because # CSV tests are very fragile. On different platforms the numbers could # be slightly different. With different versions of # shapely/libgeos/numpy/etc the numbers could be slightly different. # The parameters of the demos could change in the future and the # numbers (even the number of rows and columns) could change. with open(exported_file_path, 'r') as got: got_reader = csv.reader(got) n_rows = 0 for got_line in got_reader: n_rows += 1 n_cols = 0 for got_element in got_line: n_cols += 1 self.assertGreaterEqual( n_cols, 3, "The following line of the exported file %s has" " only %s columns:\n%s" % ( exported_file_path, n_cols, got_line)) self.assertGreaterEqual( n_rows, 3, "The exported file %s has only %s rows" % ( exported_file_path, n_rows)) def _set_output_type(self, output_type): idx = self.viewer_dock.output_type_cbx.findText(output_type) self.assertNotEqual(idx, -1, 'Output type %s not found' % output_type) self.viewer_dock.output_type_cbx.setCurrentIndex(idx) def _change_selection(self): layer = IFACE.activeLayer() # the behavior should be slightly different (pluralizing labels, etc) # depending on the amount of features selected layer.select(1) layer.removeSelection() layer.select(2) layer.selectAll() layer.removeSelection()