import os.path from typing import Callable from qgis._core import QgsVectorLayer, QgsProject, QgsLayerTreeLayer from .pegelonline_dockwidget import PegelonlineDockWidget from .pomodules.poqgscurrentw import PoQgsCurrentW from .pomodules.poqgsstations import PoQgsStations # noinspection PyMethodMayBeStatic class PoRunner(object): def __init__(self, ui: PegelonlineDockWidget, iface): self.ui: PegelonlineDockWidget = ui self.iface = iface self.local_dir = os.path.dirname(os.path.realpath(__file__)) # Layer Variablen self.stations = None self.waterlevels = None self.lines = None self.areas = None self.connect_basemap_signals() self.connect_stations_signals() self.connect_waterlevels_signals() # stations ---------------------------------------------------------------- def loadStations(self): print("loadStations") reader = PoQgsStations() features = reader.getStationsFeatures() self.stations = self._layerFromReader(reader.fields, reader.crs, features, "Stationen") self._layerShow(self.stations, "styles/label_stations.qml", self.disconnectStations) def loadWaterlevels(self): print("loadWaterlevels") reader = PoQgsCurrentW() features = reader.getCurrentWFeatures() self.waterlevels = self._layerFromReader(reader.fields, reader.crs, features, "Wasserstandinformationen") self._layerShow(self.waterlevels, "styles/label_currentw.qml", self.disconnectWaterlevels) def _layerFromReader(self, fields, crs, features, title) -> None | QgsVectorLayer: print("_layerFromReader") if features is None: return None layer_path = "Point?crs=%s" % (crs.authid(),) print("layer_path: " + layer_path) layer = QgsVectorLayer(layer_path, title, "memory") provider = layer.dataProvider() provider.addAttributes(fields) layer.updateFields() provider.addFeatures(features) layer.updateExtents() layer.commitChanges() if layer.isValid(): return layer return None def _layerShow(self, layer: QgsVectorLayer, styles_path: str, disconnect: Callable[[], None]): print("_layerShow") if layer is None: print("_layerShow: Übergebener Layer ist None") return # Styles laden layer.loadNamedStyle(os.path.join(self.local_dir, styles_path)) # disconnect setzen 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)) self._layerSetVisible(layer, True) self._layerRefresh(layer) # stations signals -------------------------------------------------------- def connect_stations_signals(self): print("connect_stations_signals") self.ui.cbStationsVisible.toggled.connect(self.cbStationsVisibleToggled) self.ui.cbStationsName.toggled.connect(self.cbStationsNameToggled) self.ui.cbStationsNumber.toggled.connect(self.cbStationsNumberToggled) self.ui.cbStationsAgency.toggled.connect(self.cbStationsAgencyToggled) self.ui.cbStationsKm.toggled.connect(self.cbStationsKmToggled) self.ui.cbStationsWater.toggled.connect(self.cbStationsWaterToggled) def cbStationsVisibleToggled(self): visible = self.ui.cbStationsVisible.isChecked() print("cbStationsVisibleToggled: %s" % (visible,)) if self.stations is None and visible: self.loadStations() if self.stations is not None: self.stations.setLabelsEnabled(visible) self._layerSetVisible(self.stations, visible) self._layerRefresh(self.stations) def cbStationsNameToggled(self): checked = self.ui.cbStationsName.isChecked() print("cbStationsNameToggled: %s" % (checked,)) # TODO def cbStationsNumberToggled(self): checked = self.ui.cbStationsNumber.isChecked() print("cbStationsNumberToggled: %s" % (checked,)) # TODO def cbStationsAgencyToggled(self): checked = self.ui.cbStationsAgency.isChecked() print("cbStationsAgencyToggled: %s" % (checked,)) # TODO def cbStationsKmToggled(self): checked = self.ui.cbStationsKm.isChecked() print("cbStationsKmToggled: %s" % (checked,)) # TODO def cbStationsWaterToggled(self): checked = self.ui.cbStationsWater.isChecked() print("cbStationsWaterToggled: %s" % (checked,)) # TODO def disconnectStations(self): print("disconnectStations") self.stations = None self.ui.cbStationsVisible.setChecked(False) # basemap ----------------------------------------------------------------- def _basemapCreate(self, path, name, disconnect: Callable[[], None]) -> None | QgsVectorLayer: print("_basemapCreate: %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 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)) return basemap # basemap signals --------------------------------------------------------- def connect_basemap_signals(self): print("connect_basemap_signals") self.ui.cbBasemapLines.toggled.connect(self.cbBasemapLinesToggled) self.ui.cbBasemapAreas.toggled.connect(self.cbBasemapAreasToggled) def cbBasemapLinesToggled(self): checked = self.ui.cbBasemapLines.isChecked() print("cbBasemapLinesToggled: %s" % (checked,)) if self.lines is None and checked: self.lines = self._basemapCreate("waters.gpkg|layername=water_l", "Flüsse", self.disconnectBasemapLines) if self.lines is not None: self._layerSetVisible(self.lines, checked) def cbBasemapAreasToggled(self): checked = self.ui.cbBasemapAreas.isChecked() print("cbBasemapAreasToggled: %s" % (checked,)) if self.areas is None and checked: self.areas = self._basemapCreate("waters.gpkg|layername=water_f", "Flächen", self.disconnectBasemapAreas) if self.areas is not None: self._layerSetVisible(self.areas, checked) def disconnectBasemapLines(self): print("disconnectBasemapLines") self.lines = None self.ui.cbBasemapLines.setChecked(False) def disconnectBasemapAreas(self): print("disconnectBasemapAreas") self.areas = None self.ui.cbBasemapAreas.setChecked(False) # layers ------------------------------------------------------------------ def _layerRefresh(self, layer): print("_layerRefresh") if self.iface.mapCanvas().isCachingEnabled(): layer.triggerRepaint() else: self.iface.mapCanvas().refresh() def _layerSetVisible(self, basemap: QgsVectorLayer, visible): print("_layerSetVisible: %s => %s" % (basemap.name, visible)) layer_tree = QgsProject.instance().layerTreeRoot().findLayer(basemap.id()) layer_tree.setItemVisibilityChecked(visible) self._layerRefresh(basemap) # waterlevels signals ----------------------------------------------------- def connect_waterlevels_signals(self): print("connect_waterlevels_signals") self.ui.cbWaterlevelsVisible.toggled.connect(self.cbWaterlevelsVisibleToggled) self.ui.cbWaterlevelsName.toggled.connect(self.cbWaterlevelsNameToggled) self.ui.cbWaterlevelsTimestamp.toggled.connect(self.cbWaterlevelsTimestampToggled) self.ui.cbWaterlevelsValue.toggled.connect(self.cbWaterlevelsValueToggled) self.ui.cbWaterlevelsMean.toggled.connect(self.cbWaterlevelsMeanToggled) self.ui.cbWaterlevelsAbsolute.toggled.connect(self.cbWaterlevelsAbsoluteToggled) def cbWaterlevelsVisibleToggled(self): visible = self.ui.cbWaterlevelsVisible.isChecked() print("cbWaterlevelsVisibleToggled: %s" % (visible,)) if self.waterlevels is None: self.loadWaterlevels() if self.waterlevels is not None: self.waterlevels.setLabelsEnabled(visible) self._layerSetVisible(self.waterlevels, visible) self._layerRefresh(self.waterlevels) def cbWaterlevelsNameToggled(self): checked = self.ui.cbWaterlevelsName.isChecked() print("cbWaterlevelsNameToggled: %s" % (checked,)) # TODO def cbWaterlevelsTimestampToggled(self): checked = self.ui.cbWaterlevelsTimestamp.isChecked() print("cbWaterlevelsTimestampToggled: %s" % (checked,)) # TODO def cbWaterlevelsValueToggled(self): checked = self.ui.cbWaterlevelsValue.isChecked() print("cbWaterlevelsValueToggled: %s" % (checked,)) # TODO def cbWaterlevelsMeanToggled(self): checked = self.ui.cbWaterlevelsMean.isChecked() print("cbWaterlevelsMeanToggled: %s" % (checked,)) # TODO def cbWaterlevelsAbsoluteToggled(self): checked = self.ui.cbWaterlevelsAbsolute.isChecked() print("cbWaterlevelsAbsoluteToggled: %s" % (checked,)) # TODO def disconnectWaterlevels(self): print("disconnectWaterlevels") self.waterlevels = None self.ui.cbWaterlevelsVisible.setChecked(False) # def rbValueChanged(self, button): # print("rbValueChanged") # if self.layer_waterlevels is None: # self.loadWaterlevels() # # if button.objectName() == "rbShowTrend": # self.layer_waterlevels.loadNamedStyle(os.path.join(self.local_dir, "styles/label_currentw_trend.qml")) # elif button.objectName() == "rbShowAbsValues": # self.layer_waterlevels.loadNamedStyle(os.path.join(self.local_dir, "styles/label_currentw_absvalue.qml")) # # self.cbWaterlevelsVisibleToggled()