430 lines
17 KiB
Python
430 lines
17 KiB
Python
import os.path
|
|
from typing import Callable
|
|
|
|
from PyQt5 import QtGui
|
|
from qgis._core import QgsVectorLayer, QgsProject, QgsLayerTreeLayer, QgsPalLayerSettings, QgsVectorLayerSimpleLabeling
|
|
|
|
from .pomodules.po_history import PoHistory
|
|
from .pomodules.po_stations import PoStations
|
|
from .pomodules.po_stations_qgs import PoStationsQgs
|
|
from .pomodules.po_waterlevels_qgs import PoWaterlevelsQgs
|
|
|
|
INFIX = ", ' - ', "
|
|
|
|
|
|
# noinspection PyMethodMayBeStatic
|
|
class PoRunner(object):
|
|
|
|
def __init__(self, ui, iface):
|
|
self.ui = ui
|
|
self.iface = iface
|
|
self.local_dir = os.path.dirname(os.path.realpath(__file__))
|
|
|
|
# Layer
|
|
self.stations = None
|
|
self.waterlevels = None
|
|
self.lines = None
|
|
self.areas = None
|
|
|
|
# ui Signale verbinden
|
|
self._connect_basemap_signals()
|
|
self._connect_stations_signals()
|
|
self._waterlevels_connect_signals()
|
|
self._history_connect_signals()
|
|
|
|
self._history_load_stations()
|
|
|
|
# basemap -----------------------------------------------------------------
|
|
|
|
def _basemap_create(self, path, name, disconnect: Callable[[], None]) -> None | QgsVectorLayer:
|
|
print("_basemap_create: %s" % (name,))
|
|
path = os.path.join(self.local_dir, "basemap", path)
|
|
basemap = QgsVectorLayer(path, name, "ogr")
|
|
|
|
if not basemap.isValid():
|
|
print("_basemapCreate: QgsVectorLayer nicht gültig: path=%s, name=%s" % (path, name))
|
|
return None
|
|
|
|
# disconnect setzen
|
|
# noinspection PyUnresolvedReferences
|
|
basemap.willBeDeleted.connect(disconnect)
|
|
|
|
# zur Instanz hinzufügen
|
|
QgsProject.instance().addMapLayer(basemap, False)
|
|
|
|
# zum LayerTree hinzufügen
|
|
layer_tree = self.iface.layerTreeCanvasBridge().rootGroup()
|
|
layer_tree.insertChildNode(-1, QgsLayerTreeLayer(basemap)) # am unteren Ende anhängen → liegt somit unter stations/waterlevels
|
|
|
|
return basemap
|
|
|
|
# basemap signals ---------------------------------------------------------
|
|
|
|
def _connect_basemap_signals(self):
|
|
print("_connect_basemap_signals")
|
|
self.ui.cbBasemapLines.toggled.connect(self._cbBasemapLines_toggled)
|
|
self.ui.cbBasemapAreas.toggled.connect(self._cbBasemapAreas_toggled)
|
|
|
|
def _cbBasemapLines_toggled(self):
|
|
checked = self.ui.cbBasemapLines.isChecked()
|
|
print("_cbBasemapLines_toggled: %s" % (checked,))
|
|
|
|
if self.lines is None and checked:
|
|
self.lines = self._basemap_create("waters.gpkg|layername=water_l", "Flüsse", self._basemap_disconnect_lines)
|
|
|
|
if self.lines is not None:
|
|
self._layer_set_visible(self.lines, checked)
|
|
self._layer_refresh(self.lines)
|
|
|
|
def _cbBasemapAreas_toggled(self):
|
|
checked = self.ui.cbBasemapAreas.isChecked()
|
|
print("_cbBasemapAreas_toggled: %s" % (checked,))
|
|
|
|
if self.areas is None and checked:
|
|
self.areas = self._basemap_create("waters.gpkg|layername=water_f", "Flächen", self._basemap_disconnect_areas)
|
|
|
|
if self.areas is not None:
|
|
self._layer_set_visible(self.areas, checked)
|
|
self._layer_refresh(self.areas)
|
|
|
|
def _basemap_disconnect_lines(self):
|
|
print("_basemap_disconnect_lines")
|
|
self.lines = None
|
|
self.ui.cbBasemapLines.setChecked(False)
|
|
|
|
def _basemap_disconnect_areas(self):
|
|
print("_basemap_disconnect_areas")
|
|
self.areas = None
|
|
self.ui.cbBasemapAreas.setChecked(False)
|
|
|
|
# stations ----------------------------------------------------------------
|
|
|
|
def _connect_stations_signals(self):
|
|
print("_connect_stations_signals")
|
|
# noinspection DuplicatedCode
|
|
self.ui.cbStationsVisible.toggled.connect(self._cbStationsVisible_toggled)
|
|
self.ui.cbStationsName.toggled.connect(self._cbStationsName_toggled)
|
|
self.ui.cbStationsNumber.toggled.connect(self._cbStationsNumber_toggled)
|
|
self.ui.cbStationsAgency.toggled.connect(self._cbStationsAgency_toggled)
|
|
self.ui.cbStationsKm.toggled.connect(self._cbStationsKm_toggled)
|
|
self.ui.cbStationsWater.toggled.connect(self._cbStationsWater_toggled)
|
|
|
|
def _cbStationsVisible_toggled(self):
|
|
visible = self.ui.cbStationsVisible.isChecked()
|
|
print("_cbStationsVisible_toggled: %s" % (visible,))
|
|
|
|
if self.stations is None and visible:
|
|
reader = PoStationsQgs()
|
|
features = reader.getStationsFeatures()
|
|
self.stations = self._layer_create_from_reader(reader.fields, reader.crs, features, "Stationen")
|
|
self._layer_add_to_instance(self.stations, "styles/label_stations.qml", self._stations_disconnect)
|
|
|
|
if self.stations is not None:
|
|
self._layer_set_visible(self.stations, visible)
|
|
self._stations_update_labels()
|
|
|
|
def _cbStationsName_toggled(self):
|
|
checked = self.ui.cbStationsName.isChecked()
|
|
print("_cbStationsName_toggled: %s" % (checked,))
|
|
self._stations_update_labels()
|
|
|
|
def _cbStationsNumber_toggled(self):
|
|
checked = self.ui.cbStationsNumber.isChecked()
|
|
print("_cbStationsNumber_toggled: %s" % (checked,))
|
|
self._stations_update_labels()
|
|
|
|
def _cbStationsAgency_toggled(self):
|
|
checked = self.ui.cbStationsAgency.isChecked()
|
|
print("_cbStationsAgency_toggled: %s" % (checked,))
|
|
self._stations_update_labels()
|
|
|
|
def _cbStationsKm_toggled(self):
|
|
checked = self.ui.cbStationsKm.isChecked()
|
|
print("_cbStationsKm_toggled: %s" % (checked,))
|
|
self._stations_update_labels()
|
|
|
|
def _cbStationsWater_toggled(self):
|
|
checked = self.ui.cbStationsWater.isChecked()
|
|
print("_cbStationsWater_toggled: %s" % (checked,))
|
|
self._stations_update_labels()
|
|
|
|
def _stations_disconnect(self):
|
|
print("_stations_disconnect")
|
|
self.stations = None
|
|
self.ui.cbStationsVisible.setChecked(False)
|
|
|
|
# noinspection DuplicatedCode
|
|
def _stations_update_labels(self):
|
|
print("_stations_update_labels")
|
|
if self.stations is None:
|
|
return
|
|
|
|
fields = []
|
|
if self.ui.cbStationsName.isChecked():
|
|
fields.append("shortname")
|
|
if self.ui.cbStationsNumber.isChecked():
|
|
fields.append("number")
|
|
if self.ui.cbStationsAgency.isChecked():
|
|
fields.append("agency")
|
|
if self.ui.cbStationsKm.isChecked():
|
|
fields.append("km")
|
|
if self.ui.cbStationsWater.isChecked():
|
|
fields.append("water")
|
|
|
|
self._layer_update_labels(self.stations, fields)
|
|
|
|
# waterlevels -------------------------------------------------------------
|
|
|
|
def _waterlevels_connect_signals(self):
|
|
print("_waterlevels_connect_signals")
|
|
# noinspection DuplicatedCode
|
|
self.ui.cbWaterlevelsVisible.toggled.connect(self._cbWaterlevelsVisible_toggled)
|
|
self.ui.cbWaterlevelsName.toggled.connect(self._cbWaterlevelsName_toggled)
|
|
self.ui.cbWaterlevelsNumber.toggled.connect(self._cbWaterlevelsNumber_toggled)
|
|
self.ui.cbWaterlevelsAgency.toggled.connect(self._cbWaterlevelsAgency_toggled)
|
|
self.ui.cbWaterlevelsTimestamp.toggled.connect(self._cbWaterlevelsTimestamp_toggled)
|
|
self.ui.cbWaterlevelsValue.toggled.connect(self._cbWaterlevelsValue_toggled)
|
|
self.ui.cbWaterlevelsMean.toggled.connect(self._cbWaterlevelsMean_toggled)
|
|
self.ui.cbWaterlevelsAbsolute.toggled.connect(self._cbWaterlevelsAbsolute_toggled)
|
|
|
|
def _cbWaterlevelsVisible_toggled(self):
|
|
visible = self.ui.cbWaterlevelsVisible.isChecked()
|
|
print("_cbWaterlevelsVisible_toggled: %s" % (visible,))
|
|
|
|
if self.waterlevels is None and visible:
|
|
reader = PoWaterlevelsQgs()
|
|
features = reader.getWaterlevelsFeatures()
|
|
self.waterlevels = self._layer_create_from_reader(reader.fields, reader.crs, features, "Pegelstände")
|
|
self._layer_add_to_instance(self.waterlevels, "styles/label_currentw.qml", self.waterlevels_disconnect)
|
|
|
|
if self.waterlevels is not None:
|
|
self._layer_set_visible(self.waterlevels, visible)
|
|
self._waterlevels_update_labels()
|
|
|
|
def _cbWaterlevelsName_toggled(self):
|
|
checked = self.ui.cbWaterlevelsName.isChecked()
|
|
print("_cbWaterlevelsName_toggled: %s" % (checked,))
|
|
self._waterlevels_update_labels()
|
|
|
|
def _cbWaterlevelsNumber_toggled(self):
|
|
checked = self.ui.cbWaterlevelsNumber.isChecked()
|
|
print("_cbWaterlevelsNumber_toggled: %s" % (checked,))
|
|
self._waterlevels_update_labels()
|
|
|
|
def _cbWaterlevelsAgency_toggled(self):
|
|
checked = self.ui.cbWaterlevelsAgency.isChecked()
|
|
print("_cbWaterlevelsAgency_toggled: %s" % (checked,))
|
|
self._waterlevels_update_labels()
|
|
|
|
def _cbWaterlevelsTimestamp_toggled(self):
|
|
checked = self.ui.cbWaterlevelsTimestamp.isChecked()
|
|
print("_cbWaterlevelsTimestamp_toggled: %s" % (checked,))
|
|
self._waterlevels_update_labels()
|
|
|
|
def _cbWaterlevelsValue_toggled(self):
|
|
checked = self.ui.cbWaterlevelsValue.isChecked()
|
|
print("_cbWaterlevelsValue_toggled: %s" % (checked,))
|
|
self._waterlevels_update_labels()
|
|
|
|
def _cbWaterlevelsMean_toggled(self):
|
|
checked = self.ui.cbWaterlevelsMean.isChecked()
|
|
print("_cbWaterlevelsMean_toggled: %s" % (checked,))
|
|
self._waterlevels_update_labels()
|
|
|
|
def _cbWaterlevelsAbsolute_toggled(self):
|
|
checked = self.ui.cbWaterlevelsAbsolute.isChecked()
|
|
print("_cbWaterlevelsAbsolute_toggled: %s" % (checked,))
|
|
self._waterlevels_update_labels()
|
|
|
|
def waterlevels_disconnect(self):
|
|
print("waterlevels_disconnect")
|
|
self.waterlevels = None
|
|
self.ui.cbWaterlevelsVisible.setChecked(False)
|
|
|
|
# noinspection DuplicatedCode
|
|
def _waterlevels_update_labels(self):
|
|
print("_waterlevels_update_labels")
|
|
if self.waterlevels is None:
|
|
return
|
|
|
|
fields = []
|
|
if self.ui.cbWaterlevelsName.isChecked():
|
|
fields.append("shortname")
|
|
if self.ui.cbWaterlevelsNumber.isChecked():
|
|
fields.append("number")
|
|
if self.ui.cbWaterlevelsAgency.isChecked():
|
|
fields.append("agency")
|
|
if self.ui.cbWaterlevelsTimestamp.isChecked():
|
|
fields.append("timestamp")
|
|
if self.ui.cbWaterlevelsValue.isChecked():
|
|
fields.append("value")
|
|
if self.ui.cbWaterlevelsMean.isChecked():
|
|
fields.append("stateMnwMhw")
|
|
if self.ui.cbWaterlevelsAbsolute.isChecked():
|
|
fields.append("stateNswHsw")
|
|
self._layer_update_labels(self.waterlevels, fields)
|
|
|
|
# layers ------------------------------------------------------------------
|
|
|
|
def _layer_set_visible(self, basemap: QgsVectorLayer, visible):
|
|
print("_layer_set_visible: %s => %s" % (basemap.name, visible))
|
|
layer_tree = QgsProject.instance().layerTreeRoot().findLayer(basemap.id())
|
|
layer_tree.setItemVisibilityChecked(visible)
|
|
|
|
def _layer_create_from_reader(self, fields, crs, features, title) -> None | QgsVectorLayer:
|
|
print("_layer_create_from_reader")
|
|
if features is None:
|
|
return None
|
|
|
|
layer_path = "Point?crs=%s" % (crs.authid(),)
|
|
print("_layer_create_from_reader: layer_path: " + layer_path)
|
|
|
|
layer = QgsVectorLayer(layer_path, title, "memory")
|
|
provider = layer.dataProvider()
|
|
|
|
provider.addAttributes(fields)
|
|
layer.updateFields()
|
|
|
|
provider.addFeatures(features)
|
|
for error in provider.errors():
|
|
print("_layer_create_from_reader: Fehler beim Hinzufügen von Features: " + error)
|
|
|
|
layer.updateExtents()
|
|
|
|
layer.commitChanges()
|
|
|
|
if layer.isValid():
|
|
return layer
|
|
|
|
return None
|
|
|
|
def _layer_add_to_instance(self, layer: QgsVectorLayer, styles_path: str, disconnect: Callable[[], None]):
|
|
print("_layer_add_to_instance")
|
|
if layer is None:
|
|
print("_layer_add_to_instance: Fehler: Übergebener Layer ist None")
|
|
return
|
|
|
|
# Styles laden
|
|
layer.loadNamedStyle(os.path.join(self.local_dir, styles_path))
|
|
|
|
# disconnect Signal verbinden
|
|
layer.willBeDeleted.connect(disconnect)
|
|
|
|
# zur Instanz hinzufügen
|
|
QgsProject.instance().addMapLayer(layer, False)
|
|
|
|
# zum LayerTree hinzufügen
|
|
layer_tree = self.iface.layerTreeCanvasBridge().rootGroup()
|
|
layer_tree.insertChildNode(0, QgsLayerTreeLayer(layer)) # am oberen Ende anhängen → liegt somit über basemap
|
|
|
|
def _layer_update_labels(self, layer, fields):
|
|
print("_layer_update_labels")
|
|
labeling = QgsVectorLayerSimpleLabeling(QgsPalLayerSettings())
|
|
|
|
# Anführungszeichen um Feldnamen anbringen
|
|
fields_quoted = ['"%s"' % (field,) for field in fields]
|
|
|
|
# Feldnamen zu einem Minus-getrennten String zusammenbauen
|
|
expression = ", ' - ', ".join(fields_quoted)
|
|
|
|
settings = labeling.settings()
|
|
settings.fieldName = "concat(" + expression + ")"
|
|
settings.isExpression = True
|
|
|
|
layer.setLabeling(QgsVectorLayerSimpleLabeling(settings))
|
|
layer.setLabelsEnabled(True)
|
|
|
|
self._layer_refresh(layer)
|
|
|
|
def _layer_refresh(self, layer):
|
|
print("_layerRefresh")
|
|
if self.iface.mapCanvas().isCachingEnabled():
|
|
layer.triggerRepaint()
|
|
else:
|
|
self.iface.mapCanvas().refresh()
|
|
|
|
# history signals ---------------------------------------------------------
|
|
|
|
def _history_connect_signals(self):
|
|
print("_history_connect_signals")
|
|
self.ui.slHistoryStation.currentTextChanged.connect(self._slHistoryStation_changed)
|
|
self.ui.btnHistoryStationsRefresh.clicked.connect(self._btnHistoryStationsRefresh_clicked)
|
|
self.ui.numHistoryDays.valueChanged.connect(self._numHistoryDays_changed)
|
|
self.ui.btnHistoryGo.clicked.connect(self._history_load_graph)
|
|
|
|
def _slHistoryStation_changed(self):
|
|
print("_slHistoryStation_changed: %s" % (self.ui.slHistoryStation.currentText(),))
|
|
self._history_load_graph()
|
|
|
|
def _btnHistoryStationsRefresh_clicked(self):
|
|
print("_btnHistoryStationsRefresh_clicked")
|
|
self._history_load_stations()
|
|
|
|
def _numHistoryDays_changed(self):
|
|
print("_numHistoryDays_changed: %s" % (self.ui.numHistoryDays.value(),))
|
|
|
|
def _history_load_graph(self):
|
|
print("_history_load_graph")
|
|
station = self.ui.slHistoryStation.currentText()
|
|
days = self.ui.numHistoryDays.value()
|
|
|
|
print("_history_load_graph: station=%s days=%s" % (station, days))
|
|
|
|
self.ui.lbHistory.clear()
|
|
|
|
if station == '' or station is None:
|
|
print("_history_load_graph: Fehler: Ungültige Station: %s" % (station,))
|
|
self.ui.lbHistory.setText("Bitte Station wählen...")
|
|
return
|
|
|
|
if days is None or days < 1 or days > 30:
|
|
print("_history_load_graph: Fehler: Ungültige Anzahl von Tagen: %s" % (days,))
|
|
self.ui.lbHistory.setText("Bitte Tage [1, 30] wählen...")
|
|
return
|
|
|
|
history = PoHistory(station, days)
|
|
image_data = history.download()
|
|
|
|
if image_data is None or len(image_data) == 0:
|
|
print("_history_load_graph: Fehler: Keine Daten erhalten")
|
|
self.ui.lbHistory.setText("Fehler beim Download!")
|
|
return
|
|
|
|
pixmap = QtGui.QPixmap()
|
|
pixmap.loadFromData(image_data)
|
|
self.ui.lbHistory.setPixmap(pixmap)
|
|
self.ui.lbHistory.resize(pixmap.width(), pixmap.height())
|
|
|
|
print("_history_load_graph: Bild erfolgreich geladen")
|
|
|
|
def _history_load_stations(self):
|
|
print("_history_load_stations")
|
|
|
|
# behalte die aktuelle Station, um sie (mit eventuell neuem Index) wiederherzustellen
|
|
current_station = self.ui.slHistoryStation.currentText()
|
|
print("_history_load_stations: bisherige_station=%s" % current_station)
|
|
|
|
self.ui.slHistoryStation.clear()
|
|
|
|
stations = PoStations().getStations()
|
|
if stations is None or len(stations) == 0:
|
|
print("_history_load_stations: Fehler: Keine Stationen erhalten")
|
|
return
|
|
|
|
index = 0
|
|
neuer_index = None
|
|
for station in stations:
|
|
shortname = station['attributes']['shortname']
|
|
if shortname == current_station:
|
|
neuer_index = index
|
|
self.ui.slHistoryStation.addItem(shortname)
|
|
index += 1
|
|
|
|
if neuer_index is not None:
|
|
self.ui.slHistoryStation.setCurrentIndex(neuer_index)
|
|
print("_history_load_stations: Bisherige Station \"%s\" mit neuem index=%d wiederhergestellt" % (current_station, neuer_index))
|
|
else:
|
|
self.ui.slHistoryStation.setCurrentIndex(0)
|
|
station = self.ui.slHistoryStation.currentText()
|
|
print("_history_load_stations: Bisherige Station \"%s\" nicht wiedergefunden. Nehme erste Station: %s" % (current_station, station))
|