# -*- coding: utf-8 -*-
# /***************************************************************************
# Irmt
# A QGIS plugin
# OpenQuake Integrated Risk Modelling Toolkit
# -------------------
# begin : 2013-10-24
# copyright : (C) 2014 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 numpy
from qgis.core import (QgsVectorLayer,
QgsMapLayerRegistry,
QgsSymbolV2,
QgsSymbolLayerV2Registry,
QgsOuterGlowEffect,
QgsSingleSymbolRendererV2,
QgsVectorGradientColorRampV2,
QgsGraduatedSymbolRendererV2,
QgsRendererRangeV2,
QgsProject,
QgsMapUnitScale,
QGis,
)
from PyQt4.QtCore import pyqtSlot, QDir, QSettings, QFileInfo, Qt
from PyQt4.QtGui import (QDialogButtonBox,
QDialog,
QFileDialog,
QColor,
QComboBox,
QSpinBox,
QLabel,
QCheckBox,
QHBoxLayout,
)
from svir.utilities.shared import (OQ_CSV_LOADABLE_TYPES,
OQ_NPZ_LOADABLE_TYPES,
OQ_ALL_LOADABLE_TYPES,
)
from svir.utilities.utils import (get_ui_class,
get_style,
clear_widgets_from_layout,
log_msg,
)
FORM_CLASS = get_ui_class('ui_load_output_as_layer.ui')
[docs]class LoadOutputAsLayerDialog(QDialog, FORM_CLASS):
"""
Modal dialog to load an oq-engine output as layer
"""
def __init__(self, iface, viewer_dock, output_type=None,
path=None, mode=None):
# sanity check
if output_type not in OQ_ALL_LOADABLE_TYPES:
raise NotImplementedError(output_type)
self.iface = iface
self.viewer_dock = viewer_dock
self.path = path
self.output_type = output_type
self.mode = mode # if 'testing' it will avoid some user interaction
QDialog.__init__(self)
# Set up the user interface from Designer.
self.setupUi(self)
# Disable ok_button until all user options are set
self.ok_button = self.buttonBox.button(QDialogButtonBox.Ok)
self.ok_button.setDisabled(True)
self.file_browser_tbn.setEnabled(True)
if self.path:
self.path_le.setText(self.path)
clear_widgets_from_layout(self.output_dep_vlayout)
[docs] def create_rlz_or_stat_selector(self):
self.rlz_or_stat_cbx = QComboBox()
self.rlz_or_stat_cbx.setEnabled(False)
self.rlz_or_stat_cbx.currentIndexChanged['QString'].connect(
self.on_rlz_or_stat_changed)
self.num_sites_msg = 'Number of sites: %s'
self.rlz_or_stat_num_sites_lbl = QLabel(self.num_sites_msg % '')
self.rlz_or_stat_h_layout = QHBoxLayout()
self.rlz_or_stat_h_layout.addWidget(self.rlz_or_stat_cbx)
self.rlz_or_stat_h_layout.addWidget(self.rlz_or_stat_num_sites_lbl)
self.output_dep_vlayout.addLayout(self.rlz_or_stat_h_layout)
[docs] def create_imt_selector(self):
self.imt_lbl = QLabel('Intensity Measure Type')
self.imt_cbx = QComboBox()
self.imt_cbx.setEnabled(False)
self.imt_cbx.currentIndexChanged['QString'].connect(
self.on_imt_changed)
self.output_dep_vlayout.addWidget(self.imt_lbl)
self.output_dep_vlayout.addWidget(self.imt_cbx)
[docs] def create_poe_selector(self):
self.poe_lbl = QLabel('Probability of Exceedance')
self.poe_cbx = QComboBox()
self.poe_cbx.setEnabled(False)
self.poe_cbx.currentIndexChanged['QString'].connect(
self.on_poe_changed)
self.output_dep_vlayout.addWidget(self.poe_lbl)
self.output_dep_vlayout.addWidget(self.poe_cbx)
[docs] def create_loss_type_selector(self):
self.loss_type_lbl = QLabel('Loss Type')
self.loss_type_cbx = QComboBox()
self.loss_type_cbx.setEnabled(False)
self.loss_type_cbx.currentIndexChanged['QString'].connect(
self.on_loss_type_changed)
self.output_dep_vlayout.addWidget(self.loss_type_lbl)
self.output_dep_vlayout.addWidget(self.loss_type_cbx)
[docs] def create_eid_selector(self):
self.eid_lbl = QLabel('Event ID')
self.eid_sbx = QSpinBox()
self.eid_sbx.setEnabled(False)
self.output_dep_vlayout.addWidget(self.eid_lbl)
self.output_dep_vlayout.addWidget(self.eid_sbx)
[docs] def create_dmg_state_selector(self):
self.dmg_state_lbl = QLabel('Damage state')
self.dmg_state_cbx = QComboBox()
self.dmg_state_cbx.setEnabled(False)
self.dmg_state_cbx.currentIndexChanged['QString'].connect(
self.on_dmg_state_changed)
self.output_dep_vlayout.addWidget(self.dmg_state_lbl)
self.output_dep_vlayout.addWidget(self.dmg_state_cbx)
[docs] def create_taxonomy_selector(self):
self.taxonomy_lbl = QLabel('Taxonomy')
self.taxonomy_cbx = QComboBox()
self.taxonomy_cbx.setEnabled(False)
# TODO: it might be needed when npz risk outputs are available
# self.taxonomy_cbx.currentIndexChanged['QString'].connect(
# self.on_taxonomy_changed)
self.output_dep_vlayout.addWidget(self.taxonomy_lbl)
self.output_dep_vlayout.addWidget(self.taxonomy_cbx)
[docs] def create_load_selected_only_ckb(self):
self.load_selected_only_ckb = QCheckBox("Load only the selected items")
self.load_selected_only_ckb.setChecked(True)
self.output_dep_vlayout.addWidget(self.load_selected_only_ckb)
[docs] def create_save_as_shp_ckb(self):
self.save_as_shp_ckb = QCheckBox("Save the loaded layer as shapefile")
self.save_as_shp_ckb.setChecked(False)
self.output_dep_vlayout.addWidget(self.save_as_shp_ckb)
[docs] def on_output_type_changed(self):
if self.output_type in OQ_NPZ_LOADABLE_TYPES:
self.create_load_selected_only_ckb()
elif self.output_type in OQ_CSV_LOADABLE_TYPES:
self.create_save_as_shp_ckb()
# elif self.output_type == 'loss_maps':
# self.setWindowTitle('Load loss maps from NPZ, as layer')
# self.create_rlz_or_stat_selector()
# self.create_loss_type_selector()
# self.create_poe_selector()
# self.adjustSize()
# elif self.output_type == 'loss_curves':
# self.setWindowTitle('Load loss curves from NPZ, as layer')
# self.create_rlz_or_stat_selector()
# self.adjustSize()
self.set_ok_button()
@pyqtSlot()
[docs] def on_file_browser_tbn_clicked(self):
path = self.open_file_dialog()
if path:
if self.output_type in OQ_NPZ_LOADABLE_TYPES:
self.npz_file = numpy.load(self.path, 'r')
self.populate_out_dep_widgets()
self.set_ok_button()
[docs] def on_rlz_or_stat_changed(self):
self.dataset = self.npz_file[self.rlz_or_stat_cbx.currentText()]
# TODO: change as soon as npz files for these become available
# if self.output_type in ('loss_maps'):
# # FIXME: likely, self.npz_file.keys()
# self.loss_types = self.npz_file.dtype.fields
# self.loss_type_cbx.clear()
# self.loss_type_cbx.setEnabled(True)
# self.loss_type_cbx.addItems(self.loss_types.keys())
# elif self.output_type == 'loss_curves':
# # FIXME: likely, self.npz_file.keys()
# self.loss_types = self.npz_file.dtype.names
self.set_ok_button()
[docs] def on_loss_type_changed(self):
# TODO: change as soon as npz becomes available
# self.loss_type = self.loss_type_cbx.currentText()
# if self.output_type == 'loss_maps':
# poe_names = self.loss_types[self.loss_type][0].names
# poe_thresholds = [name.split('poe-')[1] for name in poe_names]
# self.poe_cbx.clear()
# self.poe_cbx.setEnabled(True)
# self.poe_cbx.addItems(poe_thresholds)
self.set_ok_button()
[docs] def on_imt_changed(self):
self.set_ok_button()
[docs] def on_poe_changed(self):
self.set_ok_button()
[docs] def on_eid_changed(self):
self.set_ok_button()
[docs] def on_dmg_state_changed(self):
self.set_ok_button()
[docs] def open_file_dialog(self):
"""
Open a file dialog to select the data file to be loaded
"""
text = self.tr('Select the OQ-Engine output file to import')
if self.output_type in OQ_NPZ_LOADABLE_TYPES:
filters = self.tr('NPZ files (*.npz)')
elif self.output_type in OQ_CSV_LOADABLE_TYPES:
filters = self.tr('CSV files (*.csv)')
else:
raise NotImplementedError(self.output_type)
default_dir = QSettings().value('irmt/load_as_layer_dir',
QDir.homePath())
path = QFileDialog.getOpenFileName(
self, text, default_dir, filters)
if not path:
return
selected_dir = QFileInfo(path).dir().path()
QSettings().setValue('irmt/load_as_layer_dir', selected_dir)
self.path = path
self.path_le.setText(self.path)
return path
# self.populate_dmg_states()
[docs] def get_taxonomies(self):
raise NotImplementedError()
# TODO: change as soon as npz risk outputs are available
# if self.output_type in (
# 'loss_curves', 'loss_maps'):
# self.taxonomies = self.npz_file[
# 'assetcol/taxonomies'][:].tolist()
[docs] def populate_rlz_or_stat_cbx(self):
# TODO: change as soon as npz files for these become available
# if self.output_type in ('loss_curves', 'loss_maps'):
# if self.output_type == 'loss_curves':
# self.hdata = self.npz_file['loss_curves-rlzs']
# elif self.output_type == 'loss_maps':
# self.hdata = self.npz_file['loss_maps-rlzs']
# _, n_rlzs = self.hdata.shape
# self.rlzs = [str(i+1) for i in range(n_rlzs)]
self.rlzs_or_stats = [key for key in sorted(self.npz_file)
if key != 'imtls']
self.rlz_or_stat_cbx.clear()
self.rlz_or_stat_cbx.setEnabled(True)
self.rlz_or_stat_cbx.addItems(self.rlzs_or_stats)
[docs] def populate_loss_type_cbx(self, loss_types):
self.loss_type_cbx.clear()
self.loss_type_cbx.setEnabled(True)
self.loss_type_cbx.addItems(loss_types)
[docs] def show_num_sites(self):
# NOTE: we are assuming all realizations have the same number of sites,
# which currently is always true.
# If different realizations have a different number of sites, we
# need to move this block of code inside on_rlz_or_stat_changed()
rlz_or_stat_data = self.npz_file[self.rlz_or_stat_cbx.currentText()]
self.rlz_or_stat_num_sites_lbl.setText(
self.num_sites_msg % rlz_or_stat_data.shape)
# TODO: change as soon as npz files for these become available
# if self.output_type == 'loss_maps':
# self.ok_button.setEnabled(self.poe_cbx.currentIndex() != -1)
# elif self.output_type == 'loss_curves':
# self.ok_button.setEnabled(
# self.rlz_or_stat_cbx.currentIndex() != -1)
[docs] def get_layer_group(self, npz_key):
# get the root of layerTree, in order to add groups of layers
# (one group for each realization or statistic)
root = QgsProject.instance().layerTreeRoot()
npz_key_group = root.findGroup(npz_key)
if not npz_key_group:
npz_key_group = root.insertGroup(0, npz_key)
return npz_key_group
[docs] def build_layer_name(self, rlz_or_stat, **kwargs):
raise NotImplementedError()
# TODO: change as soon as npz files for these become available
# elif self.output_type == 'loss_curves':
# layer_name = "loss_curves_%s_%s" % (rlz_or_stat, taxonomy)
# elif self.output_type == 'loss_maps':
# layer_name = "loss_maps_%s_%s" % (rlz_or_stat, taxonomy)
# elif self.output_type == 'dmg_by_asset':
# # TODO: probably to be removed
# layer_name = "dmg_by_asset_%s_%s" % (rlz_or_stat, taxonomy)
# return layer_name
[docs] def get_field_names(self, **kwargs):
raise NotImplementedError()
# TODO: change as soon as npz files for these become available
# if self.output_type == 'loss_maps':
# field_names = self.loss_types.keys()
# elif self.output_type == 'loss_curves':
# field_names = list(self.loss_types)
# if self.output_type in ('loss_curves', 'loss_maps'):
# self.taxonomy_idx = self.taxonomies.index(self.taxonomy)
# return field_names
[docs] def add_field_to_layer(self, field_name):
raise NotImplementedError()
# TODO: change as soon as npz files for these become available
# if self.output_type == 'loss_maps':
# # NOTE: add_numeric_attribute uses LayerEditingManager
# added_field_name = add_numeric_attribute(
# field_name, self.layer)
# elif self.output_type == 'loss_curves':
# # NOTE: probably we need a different type with more capacity
# added_field_name = add_textual_attribute(
# field_name, self.layer)
# else:
# raise NotImplementedError(self.output_type)
# return added_field_name
[docs] def read_npz_into_layer(self, field_names, **kwargs):
raise NotImplementedError()
# TODO: change as soon as npz files for these become available
# with LayerEditingManager(self.layer, 'Reading npz', DEBUG):
# feats = []
# TODO: change as soon as npz files for these become available
# elif self.output_type == 'loss_curves':
# # We need to select rows from loss_curves-rlzs where the
# # row index (the asset idx) has the given taxonomy. The
# # taxonomy is found in the assetcol/array, together with
# # the coordinates lon and lat of the asset.
# # From the selected rows, we extract loss_type -> losses
# # and loss_type -> poes
# asset_array = self.npz_file['assetcol/array']
# loss_curves = self.npz_file[
# 'loss_curves-rlzs'][:, rlz_or_stat_idx]
# for asset_idx, row in enumerate(loss_curves):
# asset = asset_array[asset_idx]
# if asset['taxonomy_id'] != taxonomy_idx:
# continue
# else:
# lon = asset['lon']
# lat = asset['lat']
# # add a feature
# feat = QgsFeature(self.layer.pendingFields())
# # NOTE: field names are loss types
# # (normalized to 10 chars)
# for field_name_idx, field_name in enumerate(field_names):
# losses = row[field_name_idx]['losses'].tolist()
# poes = row[field_name_idx]['poes'].tolist()
# dic = dict(losses=losses, poes=poes)
# value = json.dumps(dic)
# feat.setAttribute(field_name, value)
# feat.setGeometry(QgsGeometry.fromPoint(
# QgsPoint(lon, lat)))
# feats.append(feat)
# elif self.output_type == 'loss_maps':
# # We need to select rows from loss_maps-rlzs where the
# # row index (the asset idx) has the given taxonomy. The
# # taxonomy is found in the assetcol/array, together with
# # the coordinates lon and lat of the asset.
# # From the selected rows, we extract loss_type -> poes
# # TODO: with npz, the following needs to be changed
# asset_array = self.npz_file['assetcol/array']
# loss_maps = self.npz_file[
# 'loss_maps-rlzs'][:, rlz_or_stat_idx]
# for asset_idx, row in enumerate(loss_maps):
# asset = asset_array[asset_idx]
# if asset['taxonomy_id'] != taxonomy_idx:
# continue
# else:
# lon = asset['lon']
# lat = asset['lat']
# # add a feature
# feat = QgsFeature(self.layer.pendingFields())
# # NOTE: field names are loss types
# # (normalized to 10 chars)
# for field_name_idx, field_name in enumerate(field_names):
# loss = row[field_name_idx][poe]
# feat.setAttribute(field_name, float(loss))
# feat.setGeometry(QgsGeometry.fromPoint(
# QgsPoint(lon, lat)))
# feats.append(feat)
# added_ok = self.layer.addFeatures(feats, makeSelected=False)
# if not added_ok:
# msg = 'There was a problem adding features to the layer.'
# log_msg(msg, level='C', message_bar=self.iface.messageBar())
[docs] def load_from_npz(self):
raise NotImplementedError()
# for rlz_or_stat in self.rlzs_or_stats:
# if (self.load_selected_only_ckb.isChecked()
# and rlz_or_stat != self.rlz_or_stat_cbx.currentText()):
# continue
# # TODO: change as soon as the npz outputs are available
# if self.output_type in ('loss_curves', 'loss_maps'):
# for taxonomy in self.taxonomies:
# if (self.load_selected_only_ckb.isChecked()
# and taxonomy != self.taxonomy_cbx.currentText()):
# continue
# with WaitCursorManager(
# 'Creating layer for "%s" '
# ' and taxonomy "%s"...' % (rlz_or_stat,
# taxonomy),
# self.iface):
# self.build_layer(rlz_or_stat, taxonomy=taxonomy)
# if self.output_type == 'loss_curves':
# self.style_curves()
# elif self.output_type == 'loss_maps':
# self.style_maps()
# else:
# with WaitCursorManager('Creating layer for '
# ' "%s"...' % rlz_or_stat,
# self.iface):
# self.build_layer(rlz_or_stat)
# if self.npz_file is not None:
# self.npz_file.close()
[docs] def get_investigation_time(self):
if self.output_type in ('hcurves', 'uhs', 'hmaps'):
try:
investigation_time = self.npz_file['investigation_time']
except KeyError:
msg = ('investigation_time not found. It is mandatory for %s.'
' Please check if the ouptut was produced by an'
' obsolete version of the OpenQuake Engine'
' Server.') % self.output_type
log_msg(msg, level='C', message_bar=self.iface.messageBar())
else:
return investigation_time
else:
# some output do not need the investigation time
return None
[docs] def build_layer(self, rlz_or_stat, taxonomy=None, poe=None, loss_type=None,
dmg_state=None):
rlz_or_stat_group = self.get_layer_group(rlz_or_stat)
layer_name = self.build_layer_name(
rlz_or_stat, taxonomy=taxonomy, poe=poe, loss_type=loss_type,
dmg_state=dmg_state)
# TODO: change as soon as npz files for these become available
# if self.output_type in ('loss_maps',
# 'loss_curves'):
# # 'dmg_by_asset'):
# # NOTE: realizations in the npz file start counting from 1, but
# # we need to refer to column indices that start from 0
# rlz_or_stat_idx = int(rlz_or_stat) - 1
# if self.output_type == 'loss_maps':
# loss_type = self.loss_type_cbx.currentText()
# poe = "poe-%s" % self.poe_cbx.currentText()
# self.default_field_name = loss_type
field_names = self.get_field_names(
rlz_or_stat=rlz_or_stat, taxonomy=taxonomy, poe=poe,
loss_type=loss_type, dmg_state=dmg_state)
# create layer
self.layer = QgsVectorLayer(
"Point?crs=epsg:4326", layer_name, "memory")
for field_name in field_names:
if field_name in ['lon', 'lat']:
continue
added_field_name = self.add_field_to_layer(field_name)
if field_name != added_field_name:
if field_name == self.default_field_name:
self.default_field_name = added_field_name
# replace field_name with the actual added_field_name
field_name_idx = field_names.index(field_name)
field_names.remove(field_name)
field_names.insert(field_name_idx, added_field_name)
self.read_npz_into_layer(
field_names, rlz_or_stat=rlz_or_stat, taxonomy=taxonomy, poe=poe,
loss_type=loss_type, dmg_state=dmg_state)
self.layer.setCustomProperty('output_type', self.output_type)
investigation_time = self.get_investigation_time()
if investigation_time is not None:
self.layer.setCustomProperty('investigation_time',
investigation_time)
# add self.layer to the legend
# False is to avoid adding the layer to the tree root, but only to the
# group
QgsMapLayerRegistry.instance().addMapLayer(self.layer, False)
rlz_or_stat_group.insertLayer(0, self.layer)
self.iface.setActiveLayer(self.layer)
self.iface.zoomToActiveLayer()
def _set_symbol_size(self, symbol):
if self.iface.mapCanvas().mapUnits() == QGis.Degrees:
point_size = 0.05
elif self.iface.mapCanvas().mapUnits() == QGis.Meters:
point_size = 4000
else:
# it is not obvious how to choose the point size in the other
# cases, so we conservatively keep the default sizing
return
symbol.setOutputUnit(symbol.MapUnit)
symbol.setSize(point_size)
map_unit_scale = QgsMapUnitScale()
map_unit_scale.maxSizeMMEnabled = True
map_unit_scale.minSizeMMEnabled = True
map_unit_scale.minSizeMM = 0.5
map_unit_scale.maxSizeMM = 10
symbol.setMapUnitScale(map_unit_scale)
[docs] def style_maps(self):
symbol = QgsSymbolV2.defaultSymbol(self.layer.geometryType())
# see properties at:
# https://qgis.org/api/qgsmarkersymbollayerv2_8cpp_source.html#l01073
symbol.setAlpha(1) # opacity
self._set_symbol_size(symbol)
symbol.symbolLayer(0).setOutlineStyle(Qt.PenStyle(Qt.NoPen))
style = get_style(self.layer, self.iface.messageBar())
ramp = QgsVectorGradientColorRampV2(
style['color_from'], style['color_to'])
graduated_renderer = QgsGraduatedSymbolRendererV2.createRenderer(
self.layer,
self.default_field_name,
style['classes'],
style['mode'],
symbol,
ramp)
graduated_renderer.updateRangeLowerValue(0, 0.0)
symbol_zeros = QgsSymbolV2.defaultSymbol(self.layer.geometryType())
symbol_zeros.setColor(QColor(222, 255, 222))
self._set_symbol_size(symbol_zeros)
symbol_zeros.symbolLayer(0).setOutlineStyle(Qt.PenStyle(Qt.NoPen))
zeros_min = 0.0
zeros_max = 0.0
range_zeros = QgsRendererRangeV2(
zeros_min, zeros_max, symbol_zeros,
" %.4f - %.4f" % (zeros_min, zeros_max), True)
graduated_renderer.addClassRange(range_zeros)
graduated_renderer.moveClass(style['classes'], 0)
self.layer.setRendererV2(graduated_renderer)
self.layer.setLayerTransparency(30) # percent
self.layer.triggerRepaint()
self.iface.legendInterface().refreshLayerSymbology(self.layer)
self.iface.mapCanvas().refresh()
[docs] def style_curves(self):
registry = QgsSymbolLayerV2Registry.instance()
cross = registry.symbolLayerMetadata("SimpleMarker").createSymbolLayer(
{'name': 'cross2', 'color': '0,0,0', 'color_border': '0,0,0',
'offset': '0,0', 'size': '1.5', 'angle': '0'})
symbol = QgsSymbolV2.defaultSymbol(self.layer.geometryType())
symbol.deleteSymbolLayer(0)
symbol.appendSymbolLayer(cross)
self._set_symbol_size(symbol)
renderer = QgsSingleSymbolRendererV2(symbol)
effect = QgsOuterGlowEffect()
effect.setSpread(0.5)
effect.setTransparency(0)
effect.setColor(QColor(255, 255, 255))
effect.setBlurLevel(1)
renderer.paintEffect().appendEffect(effect)
renderer.paintEffect().setEnabled(True)
self.layer.setRendererV2(renderer)
self.layer.setLayerTransparency(30) # percent
self.layer.triggerRepaint()
self.iface.legendInterface().refreshLayerSymbology(
self.layer)
self.iface.mapCanvas().refresh()
[docs] def accept(self):
if self.output_type in OQ_NPZ_LOADABLE_TYPES:
self.load_from_npz()
elif self.output_type in OQ_CSV_LOADABLE_TYPES:
self.load_from_csv()
super(LoadOutputAsLayerDialog, self).accept()
[docs] def reject(self):
if (hasattr(self, 'npz_file') and self.npz_file is not None
and self.output_type in OQ_NPZ_LOADABLE_TYPES):
self.npz_file.close()
super(LoadOutputAsLayerDialog, self).reject()