PoRunner DocStrings + Kommentare

This commit is contained in:
Patrick Haßel 2025-09-28 16:11:07 +02:00
parent 1549ea3d90
commit b4e1b88521
2 changed files with 382 additions and 56 deletions

View File

@ -193,7 +193,7 @@
<item> <item>
<layout class="QVBoxLayout" name="verticalLayout_11"> <layout class="QVBoxLayout" name="verticalLayout_11">
<item> <item>
<widget class="QRadioButton" name="rbStationsStyleSimple"> <widget class="QRadioButton" name="rbStationsStylePreset">
<property name="text"> <property name="text">
<string>Vorgabe</string> <string>Vorgabe</string>
</property> </property>
@ -673,7 +673,7 @@
<number>0</number> <number>0</number>
</property> </property>
<item> <item>
<widget class="QRadioButton" name="rbWaterlevelsStyleSimple"> <widget class="QRadioButton" name="rbWaterlevelsStylePreset">
<property name="text"> <property name="text">
<string>Vorgabe</string> <string>Vorgabe</string>
</property> </property>

View File

@ -19,12 +19,27 @@ WATERLEVELS_QML = "styles/waterlevels.qml"
class PoRunner(object): class PoRunner(object):
"""
Initialisiert ein neues PoRunner-Objekt:
- Initialisiert Objekt-Variablen
- Verbindet UI-Actions
- Verbindet UI-Signale
:param ui: Haupt-Widget
:type ui: PegelonlineDockWidget
:param graph: Pegelverlauf-Widget
:type graph: PegelonlineDockWidgetGraph
:param iface: QGIS-Interface
:type iface: QgisInterface
"""
def __init__(self, ui: PegelonlineDockWidget, graph: PegelonlineDockWidgetGraph, iface): def __init__(self, ui: PegelonlineDockWidget, graph: PegelonlineDockWidgetGraph, iface):
# Widget-Referenzen
self.ui = ui self.ui = ui
self.graph = graph self.graph = graph
# QGIS-Referenz
self.iface = iface self.iface = iface
self.local_dir = os.path.dirname(os.path.realpath(__file__)) self.local_dir = os.path.dirname(os.path.realpath(__file__))
# während dem Aktualisieren der Stationsliste treten change-signale auf, die werden so abgefangen # während dem Aktualisieren der Stationsliste treten change-signale auf, die werden so abgefangen
@ -53,12 +68,24 @@ class PoRunner(object):
# basemap ----------------------------------------------------------------- # basemap -----------------------------------------------------------------
"""
Erzeugt einen neuen Basemap-Layer (ogr),
verbindet UI Signale damit und fügt den erzeugten Layer
ans untere Ende der Layer-Liste an.
:param path: Pfad zu dem Layer-Daten
:type path: str
:param disconnect: Methode die Aufgerufen wird, wenn der Layer gelöscht wird
:type disconnect: Callable[[], None]
:param map_tips: HTML Code für die QGIS-Map-Tips
"""
def _basemap_create(self, path, name, disconnect: Callable[[], None], map_tips) -> None | QgsVectorLayer: def _basemap_create(self, path, name, disconnect: Callable[[], None], map_tips) -> None | QgsVectorLayer:
print("_basemap_create: %s" % (name,)) print("_basemap_create: %s" % (name,))
path = os.path.join(self.local_dir, "basemap", path) path = os.path.join(self.local_dir, "basemap", path)
basemap = QgsVectorLayer(path, name, "ogr") basemap = QgsVectorLayer(path, name, "ogr")
if not basemap.isValid(): if not basemap.isValid():
# Fehler beim Erzeugen des Layers → Abbruch
print("_basemapCreate: QgsVectorLayer nicht gültig: path=%s, name=%s" % (path, name)) print("_basemapCreate: QgsVectorLayer nicht gültig: path=%s, name=%s" % (path, name))
return None return None
@ -77,40 +104,62 @@ class PoRunner(object):
return basemap return basemap
# basemap signals --------------------------------------------------------- """
Verbindet alle Basemap-Signale mit der UI
"""
def _basemap_connect_signals(self): def _basemap_connect_signals(self):
print("_connect_basemap_signals") print("_connect_basemap_signals")
self.ui.cbBasemapLines.toggled.connect(self._cbBasemapLines_toggled) self.ui.cbBasemapLines.toggled.connect(self._cbBasemapLines_toggled)
self.ui.cbBasemapAreas.toggled.connect(self._cbBasemapAreas_toggled) self.ui.cbBasemapAreas.toggled.connect(self._cbBasemapAreas_toggled)
"""
Behandelt die Sichbarkeitsänderung der Flüsse durch die UI-Checkbox
"""
def _cbBasemapLines_toggled(self): def _cbBasemapLines_toggled(self):
checked = self.ui.cbBasemapLines.isChecked() checked = self.ui.cbBasemapLines.isChecked()
print("_cbBasemapLines_toggled: %s" % (checked,)) print("_cbBasemapLines_toggled: %s" % (checked,))
if self.lines is None and checked: if self.lines is None and checked:
# Flüsse sind noch nicht als Layer angelegt → Anlegen
self.lines = self._basemap_create("waters.gpkg|layername=water_l", "Flüsse", self._basemap_disconnect_lines, BASEMAP_MAP_TIPS) self.lines = self._basemap_create("waters.gpkg|layername=water_l", "Flüsse", self._basemap_disconnect_lines, BASEMAP_MAP_TIPS)
if self.lines is not None: if self.lines is not None:
# Flüsse sind (bereits) angelegt → setze Sichtbarkeit wie gefordert
self._layer_set_visible(self.lines, checked) self._layer_set_visible(self.lines, checked)
self._layer_refresh(self.lines) self._layer_refresh(self.lines)
"""
Behandelt die Sichtbarkeitsänderung der Flächen durch die UI-Checkbox
"""
def _cbBasemapAreas_toggled(self): def _cbBasemapAreas_toggled(self):
checked = self.ui.cbBasemapAreas.isChecked() checked = self.ui.cbBasemapAreas.isChecked()
print("_cbBasemapAreas_toggled: %s" % (checked,)) print("_cbBasemapAreas_toggled: %s" % (checked,))
if self.areas is None and checked: if self.areas is None and checked:
# Flächen sind noch nicht als Layer angelegt → Anlegen
self.areas = self._basemap_create("waters.gpkg|layername=water_f", "Flächen", self._basemap_disconnect_areas, BASEMAP_MAP_TIPS) self.areas = self._basemap_create("waters.gpkg|layername=water_f", "Flächen", self._basemap_disconnect_areas, BASEMAP_MAP_TIPS)
if self.areas is not None: if self.areas is not None:
# Flächen sind (bereits) angelegt → setze Sichtbarkeit wie gefordert
self._layer_set_visible(self.areas, checked) self._layer_set_visible(self.areas, checked)
self._layer_refresh(self.areas) self._layer_refresh(self.areas)
"""
Löscht die Flüsse-Layer-Referenz nachdem der Layer aus QGIS gelöscht wurde und passt die Checkbox an
"""
def _basemap_disconnect_lines(self): def _basemap_disconnect_lines(self):
print("_basemap_disconnect_lines") print("_basemap_disconnect_lines")
self.lines = None self.lines = None
self.ui.cbBasemapLines.setChecked(False) self.ui.cbBasemapLines.setChecked(False)
"""
Löscht die Flächen-Layer-Referenz nachdem der Layer aus QGIS gelöscht wurde und passt die Checkbox an
"""
def _basemap_disconnect_areas(self): def _basemap_disconnect_areas(self):
print("_basemap_disconnect_areas") print("_basemap_disconnect_areas")
self.areas = None self.areas = None
@ -118,6 +167,10 @@ class PoRunner(object):
# stations ---------------------------------------------------------------- # stations ----------------------------------------------------------------
"""
Verbindet alle Stations-Signale mit der UI
"""
def _stations_connect_signals(self): def _stations_connect_signals(self):
print("_connect_stations_signals") print("_connect_stations_signals")
self.ui.cbStationsVisible.toggled.connect(self._cbStationsVisible_toggled) self.ui.cbStationsVisible.toggled.connect(self._cbStationsVisible_toggled)
@ -128,24 +181,41 @@ class PoRunner(object):
self.ui.cbStationsWater.toggled.connect(self._cbStationsWater_toggled) self.ui.cbStationsWater.toggled.connect(self._cbStationsWater_toggled)
self.ui.bgStationsStyle.buttonClicked.connect(self._bgStationsStyle_clicked) self.ui.bgStationsStyle.buttonClicked.connect(self._bgStationsStyle_clicked)
"""
Wendet eingestellte Styles auf den Stations-Layer an
"""
def _stations_apply_style(self): def _stations_apply_style(self):
button = self.ui.bgStationsStyle.checkedButton() button = self.ui.bgStationsStyle.checkedButton()
self._bgStationsStyle_clicked(button) self._bgStationsStyle_clicked(button)
"""
Behandlung eines Klicks auf einen der Stations-Style-Radio-Buttons
Wendet Styles auf Stations-Layer an
:param button: Geklickter Radio-Button
:type button: QRadioButton
"""
def _bgStationsStyle_clicked(self, button): def _bgStationsStyle_clicked(self, button):
print("_bgStationsStyle_clicked: %s" % (button.objectName(),)) print("_bgStationsStyle_clicked: %s" % (button.objectName(),))
if self.stations is None: if self.stations is None:
# Stations-Layer nicht geladen → Keine Änderung
return return
if self.ui.rbStationsStyleSimple == button: if self.ui.rbStationsStylePreset == button:
# Vorgabe ausgewählt → lade aus Datei
self._layer_apply_style_from_file(self.stations, STATIONS_QML) self._layer_apply_style_from_file(self.stations, STATIONS_QML)
return return
elif self.ui.rbStationsStyleAgency == button:
if self.ui.rbStationsStyleAgency == button:
# Behörde ausgewählt
field = "agency" field = "agency"
elif self.ui.rbStationsStyleWater == button: elif self.ui.rbStationsStyleWater == button:
# Gewässer ausgewählt
field = "water" field = "water"
else: else:
# Programmierfehler: Unbekannter Button
print("_bgStationsStyle_clicked: Style nicht implementiert: %s" % (button.objectName(),)) print("_bgStationsStyle_clicked: Style nicht implementiert: %s" % (button.objectName(),))
self._layer_apply_style_from_file(self.stations, STATIONS_QML) self._layer_apply_style_from_file(self.stations, STATIONS_QML)
return return
@ -153,51 +223,85 @@ class PoRunner(object):
self._stations_update_labels() self._stations_update_labels()
self._layer_apply_style_per_category(self.stations, field, STATIONS_QML) self._layer_apply_style_per_category(self.stations, field, STATIONS_QML)
"""
Schaltet Sichtbarkeit des Stations-Layers um (und erstellt ihn, falls nötig)
"""
def _cbStationsVisible_toggled(self): def _cbStationsVisible_toggled(self):
visible = self.ui.cbStationsVisible.isChecked() visible = self.ui.cbStationsVisible.isChecked()
print("_cbStationsVisible_toggled: %s" % (visible,)) print("_cbStationsVisible_toggled: %s" % (visible,))
if self.stations is None and visible: if self.stations is None and visible:
# Stationen sind noch nicht als Layer angelegt → Anlegen
reader = PoStationReaderQgs() reader = PoStationReaderQgs()
features = reader.get_features() features = reader.get_features()
self.stations = self._layer_create_from_reader(reader.fields, reader.crs, features, "Stationen", STATIONS_MAP_TIPS) self.stations = self._layer_create_from_features(reader.fields, reader.crs, features, "Stationen", STATIONS_MAP_TIPS)
self._layer_add_to_instance(self.stations, self._stations_disconnect) self._layer_add_to_instance(self.stations, self._stations_disconnect)
if self.stations is not None: if self.stations is not None:
# Stationen sind (bereits) angelegt → setze Sichtbarkeit wie gefordert
self._layer_set_visible(self.stations, visible) self._layer_set_visible(self.stations, visible)
if visible: if visible:
self._stations_apply_style() self._stations_apply_style()
"""
Sichtbarkeit des Stations-Attributs 'Name' umschalten und Labels updaten
"""
def _cbStationsName_toggled(self): def _cbStationsName_toggled(self):
checked = self.ui.cbStationsName.isChecked() checked = self.ui.cbStationsName.isChecked()
print("_cbStationsName_toggled: %s" % (checked,)) print("_cbStationsName_toggled: %s" % (checked,))
self._stations_update_labels() self._stations_update_labels()
"""
Sichtbarkeit des Stations-Attributs 'Number' umschalten und Labels updaten
"""
def _cbStationsNumber_toggled(self): def _cbStationsNumber_toggled(self):
checked = self.ui.cbStationsNumber.isChecked() checked = self.ui.cbStationsNumber.isChecked()
print("_cbStationsNumber_toggled: %s" % (checked,)) print("_cbStationsNumber_toggled: %s" % (checked,))
self._stations_update_labels() self._stations_update_labels()
"""
Sichtbarkeit des Stations-Attributs 'Agency' umschalten und Labels updaten
"""
def _cbStationsAgency_toggled(self): def _cbStationsAgency_toggled(self):
checked = self.ui.cbStationsAgency.isChecked() checked = self.ui.cbStationsAgency.isChecked()
print("_cbStationsAgency_toggled: %s" % (checked,)) print("_cbStationsAgency_toggled: %s" % (checked,))
self._stations_update_labels() self._stations_update_labels()
"""
Sichtbarkeit des Stations-Attributs 'Km' umschalten und Labels updaten
"""
def _cbStationsKm_toggled(self): def _cbStationsKm_toggled(self):
checked = self.ui.cbStationsKm.isChecked() checked = self.ui.cbStationsKm.isChecked()
print("_cbStationsKm_toggled: %s" % (checked,)) print("_cbStationsKm_toggled: %s" % (checked,))
self._stations_update_labels() self._stations_update_labels()
"""
Sichtbarkeit des Stations-Attributs 'Water' umschalten und Labels updaten
"""
def _cbStationsWater_toggled(self): def _cbStationsWater_toggled(self):
checked = self.ui.cbStationsWater.isChecked() checked = self.ui.cbStationsWater.isChecked()
print("_cbStationsWater_toggled: %s" % (checked,)) print("_cbStationsWater_toggled: %s" % (checked,))
self._stations_update_labels() self._stations_update_labels()
"""
Löscht die Stations-Layer-Referenz nachdem der Layer aus QGIS gelöscht wurde und passt die Checkbox an
"""
def _stations_disconnect(self): def _stations_disconnect(self):
print("_stations_disconnect") print("_stations_disconnect")
self.stations = None self.stations = None
self.ui.cbStationsVisible.setChecked(False) self.ui.cbStationsVisible.setChecked(False)
"""
Führt Änderungen an Stations-Labels durch
"""
def _stations_update_labels(self): def _stations_update_labels(self):
print("_stations_update_labels") print("_stations_update_labels")
if self.stations is None: if self.stations is None:
@ -205,20 +309,29 @@ class PoRunner(object):
fields = [] fields = []
if self.ui.cbStationsNumber.isChecked(): if self.ui.cbStationsNumber.isChecked():
# Attribut 'number' ausgewählt
fields.append('\'#\', "number"') fields.append('\'#\', "number"')
if self.ui.cbStationsName.isChecked(): if self.ui.cbStationsName.isChecked():
# Attribut 'shortname' ausgewählt
fields.append('"shortname"') fields.append('"shortname"')
if self.ui.cbStationsAgency.isChecked(): if self.ui.cbStationsAgency.isChecked():
# Attribut 'agency' ausgewählt
fields.append('"agency"') fields.append('"agency"')
if self.ui.cbStationsWater.isChecked(): if self.ui.cbStationsWater.isChecked():
# Attribut 'water' ausgewählt
fields.append('"water"') fields.append('"water"')
if self.ui.cbStationsKm.isChecked(): if self.ui.cbStationsKm.isChecked():
# Attribut 'km' ausgewählt
fields.append('"km", \' km\'') # 2 Teile anhängen: km, " km" fields.append('"km", \' km\'') # 2 Teile anhängen: km, " km"
self._layer_update_labels(self.stations, fields) self._layer_update_labels(self.stations, fields)
# waterlevels ------------------------------------------------------------- # waterlevels -------------------------------------------------------------
"""
Verbindet alle Pegelstand-Signale mit der UI
"""
def _waterlevels_connect_signals(self): def _waterlevels_connect_signals(self):
print("_waterlevels_connect_signals") print("_waterlevels_connect_signals")
self.ui.cbWaterlevelsVisible.toggled.connect(self._cbWaterlevelsVisible_toggled) self.ui.cbWaterlevelsVisible.toggled.connect(self._cbWaterlevelsVisible_toggled)
@ -232,28 +345,47 @@ class PoRunner(object):
self.ui.cbWaterlevelsWater.toggled.connect(self._cbWaterlevelsWater_toggled) self.ui.cbWaterlevelsWater.toggled.connect(self._cbWaterlevelsWater_toggled)
self.ui.bgWaterlevelsStyle.buttonClicked.connect(self._bgWaterlevelsStyle_clicked) self.ui.bgWaterlevelsStyle.buttonClicked.connect(self._bgWaterlevelsStyle_clicked)
"""
Wendet eingestellte Styles auf den Pegelstand-Layer an
"""
def _waterlevels_apply_style(self): def _waterlevels_apply_style(self):
button = self.ui.bgWaterlevelsStyle.checkedButton() button = self.ui.bgWaterlevelsStyle.checkedButton()
self._bgWaterlevelsStyle_clicked(button) self._bgWaterlevelsStyle_clicked(button)
"""
Behandlung eines Klicks auf einen der Pegelstand-Style-Radio-Buttons
Wendet Styles auf Pegelstand-Layer an
:param button: Geklickter Radio-Button
:type button: QRadioButton
"""
def _bgWaterlevelsStyle_clicked(self, button): def _bgWaterlevelsStyle_clicked(self, button):
print("_bgWaterlevelsStyle_clicked: %s" % (button.objectName(),)) print("_bgWaterlevelsStyle_clicked: %s" % (button.objectName(),))
if self.waterlevels is None: if self.waterlevels is None:
# Pegelstand-Layer nicht geladen → Keine Änderung
return return
if self.ui.rbWaterlevelsStyleSimple == button: if self.ui.rbWaterlevelsStylePreset == button:
# Vorgabe ausgewählt → lade aus Datei
self._layer_apply_style_from_file(self.waterlevels, WATERLEVELS_QML) self._layer_apply_style_from_file(self.waterlevels, WATERLEVELS_QML)
return return
elif self.ui.rbWaterlevelsStyleAgency == button:
if self.ui.rbWaterlevelsStyleAgency == button:
# Behörde ausgewählt
field = "agency" field = "agency"
elif self.ui.rbWaterlevelsStyleWater == button: elif self.ui.rbWaterlevelsStyleWater == button:
# Gewässer ausgewählt
field = "water" field = "water"
elif self.ui.rbWaterlevelsStyleMean == button: elif self.ui.rbWaterlevelsStyleMean == button:
# MNW, MHW ausgewählt
field = "mean" field = "mean"
elif self.ui.rbWaterlevelsStyleAbsolute == button: elif self.ui.rbWaterlevelsStyleAbsolute == button:
# NSW, HSW ausgewählt
field = "absolute" field = "absolute"
else: else:
# Programmierfehler: Unbekannter Button
print("_bgWaterlevelsStyle_clicked: Style nicht implementiert: %s" % (button.objectName(),)) print("_bgWaterlevelsStyle_clicked: Style nicht implementiert: %s" % (button.objectName(),))
self._layer_apply_style_from_file(self.waterlevels, WATERLEVELS_QML) self._layer_apply_style_from_file(self.waterlevels, WATERLEVELS_QML)
return return
@ -261,66 +393,112 @@ class PoRunner(object):
self._waterlevels_update_labels() self._waterlevels_update_labels()
self._layer_apply_style_per_category(self.waterlevels, field, WATERLEVELS_QML) self._layer_apply_style_per_category(self.waterlevels, field, WATERLEVELS_QML)
"""
Schaltet Sichtbarkeit des Pegelstand-Layers um (und erstellt ihn, falls nötig)
"""
def _cbWaterlevelsVisible_toggled(self): def _cbWaterlevelsVisible_toggled(self):
visible = self.ui.cbWaterlevelsVisible.isChecked() visible = self.ui.cbWaterlevelsVisible.isChecked()
print("_cbWaterlevelsVisible_toggled: %s" % (visible,)) print("_cbWaterlevelsVisible_toggled: %s" % (visible,))
if self.waterlevels is None and visible: if self.waterlevels is None and visible:
# Pegelstände sind noch nicht als Layer angelegt → Anlegen
reader = PoWaterlevelReaderQgs() reader = PoWaterlevelReaderQgs()
features = reader.get_features() features = reader.get_features()
self.waterlevels = self._layer_create_from_reader(reader.fields, reader.crs, features, "Pegelstände", WATERLEVELS_MAP_TIPS) self.waterlevels = self._layer_create_from_features(reader.fields, reader.crs, features, "Pegelstände", WATERLEVELS_MAP_TIPS)
self._layer_add_to_instance(self.waterlevels, self.waterlevels_disconnect) self._layer_add_to_instance(self.waterlevels, self.waterlevels_disconnect)
if self.waterlevels is not None: if self.waterlevels is not None:
# Pegelstände sind (bereits) angelegt → setze Sichtbarkeit wie gefordert
self._layer_set_visible(self.waterlevels, visible) self._layer_set_visible(self.waterlevels, visible)
if visible: if visible:
self._waterlevels_apply_style() self._waterlevels_apply_style()
"""
Sichtbarkeit des Pegelstand-Attributs 'Name' umschalten und Labels updaten
"""
def _cbWaterlevelsName_toggled(self): def _cbWaterlevelsName_toggled(self):
checked = self.ui.cbWaterlevelsName.isChecked() checked = self.ui.cbWaterlevelsName.isChecked()
print("_cbWaterlevelsName_toggled: %s" % (checked,)) print("_cbWaterlevelsName_toggled: %s" % (checked,))
self._waterlevels_update_labels() self._waterlevels_update_labels()
"""
Sichtbarkeit des Pegelstand-Attributs 'Number' umschalten und Labels updaten
"""
def _cbWaterlevelsNumber_toggled(self): def _cbWaterlevelsNumber_toggled(self):
checked = self.ui.cbWaterlevelsNumber.isChecked() checked = self.ui.cbWaterlevelsNumber.isChecked()
print("_cbWaterlevelsNumber_toggled: %s" % (checked,)) print("_cbWaterlevelsNumber_toggled: %s" % (checked,))
self._waterlevels_update_labels() self._waterlevels_update_labels()
"""
Sichtbarkeit des Pegelstand-Attributs 'Agency' umschalten und Labels updaten
"""
def _cbWaterlevelsAgency_toggled(self): def _cbWaterlevelsAgency_toggled(self):
checked = self.ui.cbWaterlevelsAgency.isChecked() checked = self.ui.cbWaterlevelsAgency.isChecked()
print("_cbWaterlevelsAgency_toggled: %s" % (checked,)) print("_cbWaterlevelsAgency_toggled: %s" % (checked,))
self._waterlevels_update_labels() self._waterlevels_update_labels()
"""
Sichtbarkeit des Pegelstand-Attributs 'Timestamp' umschalten und Labels updaten
"""
def _cbWaterlevelsTimestamp_toggled(self): def _cbWaterlevelsTimestamp_toggled(self):
checked = self.ui.cbWaterlevelsTimestamp.isChecked() checked = self.ui.cbWaterlevelsTimestamp.isChecked()
print("_cbWaterlevelsTimestamp_toggled: %s" % (checked,)) print("_cbWaterlevelsTimestamp_toggled: %s" % (checked,))
self._waterlevels_update_labels() self._waterlevels_update_labels()
"""
Sichtbarkeit des Pegelstand-Attributs 'Value' umschalten und Labels updaten
"""
def _cbWaterlevelsValue_toggled(self): def _cbWaterlevelsValue_toggled(self):
checked = self.ui.cbWaterlevelsValue.isChecked() checked = self.ui.cbWaterlevelsValue.isChecked()
print("_cbWaterlevelsValue_toggled: %s" % (checked,)) print("_cbWaterlevelsValue_toggled: %s" % (checked,))
self._waterlevels_update_labels() self._waterlevels_update_labels()
"""
Sichtbarkeit des Pegelstand-Attributs 'Mean' umschalten und Labels updaten
"""
def _cbWaterlevelsMean_toggled(self): def _cbWaterlevelsMean_toggled(self):
checked = self.ui.cbWaterlevelsMean.isChecked() checked = self.ui.cbWaterlevelsMean.isChecked()
print("_cbWaterlevelsMean_toggled: %s" % (checked,)) print("_cbWaterlevelsMean_toggled: %s" % (checked,))
self._waterlevels_update_labels() self._waterlevels_update_labels()
"""
Sichtbarkeit des Pegelstand-Attributs 'Absolute' umschalten und Labels updaten
"""
def _cbWaterlevelsAbsolute_toggled(self): def _cbWaterlevelsAbsolute_toggled(self):
checked = self.ui.cbWaterlevelsAbsolute.isChecked() checked = self.ui.cbWaterlevelsAbsolute.isChecked()
print("_cbWaterlevelsAbsolute_toggled: %s" % (checked,)) print("_cbWaterlevelsAbsolute_toggled: %s" % (checked,))
self._waterlevels_update_labels() self._waterlevels_update_labels()
"""
Sichtbarkeit des Pegelstand-Attributs 'Water' umschalten und Labels updaten
"""
def _cbWaterlevelsWater_toggled(self): def _cbWaterlevelsWater_toggled(self):
checked = self.ui.cbWaterlevelsWater.isChecked() checked = self.ui.cbWaterlevelsWater.isChecked()
print("_cbWaterlevelsWater_toggled: %s" % (checked,)) print("_cbWaterlevelsWater_toggled: %s" % (checked,))
self._waterlevels_update_labels() self._waterlevels_update_labels()
"""
Löscht die Pegelstand-Layer-Referenz nachdem der Layer aus QGIS gelöscht wurde und passt die Checkbox an
"""
def waterlevels_disconnect(self): def waterlevels_disconnect(self):
print("waterlevels_disconnect") print("waterlevels_disconnect")
self.waterlevels = None self.waterlevels = None
self.ui.cbWaterlevelsVisible.setChecked(False) self.ui.cbWaterlevelsVisible.setChecked(False)
"""
Führt Änderungen an Pegelstand-Labels durch
"""
def _waterlevels_update_labels(self): def _waterlevels_update_labels(self):
print("_waterlevels_update_labels") print("_waterlevels_update_labels")
if self.waterlevels is None: if self.waterlevels is None:
@ -328,63 +506,109 @@ class PoRunner(object):
fields = [] fields = []
if self.ui.cbWaterlevelsNumber.isChecked(): if self.ui.cbWaterlevelsNumber.isChecked():
# Attribut 'number' ausgewählt
fields.append('\'#\', "number"') fields.append('\'#\', "number"')
if self.ui.cbWaterlevelsName.isChecked(): if self.ui.cbWaterlevelsName.isChecked():
# Attribut 'shortname' ausgewählt
fields.append('"shortname"') fields.append('"shortname"')
if self.ui.cbWaterlevelsAgency.isChecked(): if self.ui.cbWaterlevelsAgency.isChecked():
# Attribut 'agency' ausgewählt
fields.append('"agency"') fields.append('"agency"')
if self.ui.cbWaterlevelsWater.isChecked(): if self.ui.cbWaterlevelsWater.isChecked():
# Attribut 'water' ausgewählt
fields.append('"water"') fields.append('"water"')
if self.ui.cbWaterlevelsTimestamp.isChecked(): if self.ui.cbWaterlevelsTimestamp.isChecked():
# Attribut 'timestamp' ausgewählt
fields.append('"timestamp"') fields.append('"timestamp"')
if self.ui.cbWaterlevelsValue.isChecked(): if self.ui.cbWaterlevelsValue.isChecked():
# Attribut 'value' ausgewählt
fields.append('"value", \' \', "unit"') # 3 Teile anhängen: value, leerzeichen, unit fields.append('"value", \' \', "unit"') # 3 Teile anhängen: value, leerzeichen, unit
if self.ui.cbWaterlevelsMean.isChecked(): if self.ui.cbWaterlevelsMean.isChecked():
# Attribut 'mean' ausgewählt
fields.append('\'MnwMhw=\', "mean"') fields.append('\'MnwMhw=\', "mean"')
if self.ui.cbWaterlevelsAbsolute.isChecked(): if self.ui.cbWaterlevelsAbsolute.isChecked():
# Attribut 'absolute' ausgewählt
fields.append('\'NswHsw=\', "absolute"') fields.append('\'NswHsw=\', "absolute"')
self._layer_update_labels(self.waterlevels, fields) self._layer_update_labels(self.waterlevels, fields)
# layers ------------------------------------------------------------------ # layers ------------------------------------------------------------------
def _layer_set_visible(self, basemap: QgsVectorLayer, visible): """
print("_layer_set_visible: %s => %s" % (basemap.name, visible)) Erzeugt einen neuen Layer aus QGIS-Features
layer_tree = QgsProject.instance().layerTreeRoot().findLayer(basemap.id()) :param fields: Anzulegende QGIS-Felder
layer_tree.setItemVisibilityChecked(visible) :type fields: list[QgsField]
:param crs: Bezugs-Koordinaten-System
:type crs: QgsCoordinateReferenceSystem
:param features: Hinzuzufügende QGIS-Features
:type features: list[QgsFeature]
:param map_tips: HTML für QGIS-Map-Tips
:type map_tips: str
"""
def _layer_create_from_features(self, fields, crs, features, title, map_tips) -> None | QgsVectorLayer:
print("_layer_create_from_features")
def _layer_create_from_reader(self, fields, crs, features, title, map_tips) -> None | QgsVectorLayer:
print("_layer_create_from_reader")
if features is None: if features is None:
# Es wurden keine Features mitgegeben → Abbruch
print("_layer_create_from_features: Keine Features mitgegeben")
return None return None
layer_path = "Point?crs=%s" % (crs.authid(),) layer_path = "Point?crs=%s" % (crs.authid(),)
print("_layer_create_from_reader: layer_path: " + layer_path) print("_layer_create_from_features: layer_path: " + layer_path)
layer = QgsVectorLayer(layer_path, title, "memory") layer = QgsVectorLayer(layer_path, title, "memory")
provider = layer.dataProvider()
if not layer.isValid():
# Fehler beim Erzeugen des Layers → Abbruch
print("_layer_create_from_features: QgsVectorLayer nicht gültig: title=%s" % (title,))
return None
# map-tips setzen # map-tips setzen
layer.setMapTipTemplate(map_tips) layer.setMapTipTemplate(map_tips)
# Felder setzen
provider = layer.dataProvider()
provider.addAttributes(fields) provider.addAttributes(fields)
layer.updateFields() layer.updateFields()
# Features zum Provider hinzufügen
provider.addFeatures(features) provider.addFeatures(features)
for error in provider.errors(): for error in provider.errors():
print("_layer_create_from_reader: Fehler beim Hinzufügen von Features: " + error) # Liste Fehler auf
print("_layer_create_from_features: Fehler beim Hinzufügen von Features: " + error)
# Über Änderungen benachrichtigen
layer.updateExtents() layer.updateExtents()
layer.commitChanges() layer.commitChanges()
if layer.isValid():
return layer return layer
return None """
Setzt die Sichtbarkeit des gegebenen Layers
:param layer: Zu behandelnder Layer
:type layer: QgsVectorLayer
:param visible: Sichtbarkeit
:type visible: bool
"""
def _layer_set_visible(self, layer: QgsVectorLayer, visible):
print("_layer_set_visible: %s => %s" % (layer.name, visible))
layer_tree = QgsProject.instance().layerTreeRoot().findLayer(layer.id())
layer_tree.setItemVisibilityChecked(visible)
"""
Fügt einen Layer der QGIS-Instanz hinzu und verbindet dessen Signale
:param layer: Zu behandelnder Layer
:type layer: QgsVectorLayer
:param disconnect: Methode die Aufgerufen wird, wenn der Layer gelöscht wird
:type disconnect: Callable[[], None]
"""
def _layer_add_to_instance(self, layer: QgsVectorLayer, disconnect: Callable[[], None]): def _layer_add_to_instance(self, layer: QgsVectorLayer, disconnect: Callable[[], None]):
print("_layer_add_to_instance") print("_layer_add_to_instance")
if layer is None: if layer is None:
# Kein Layer mitgegeben → Abbruch
print("_layer_add_to_instance: Fehler: Übergebener Layer ist None") print("_layer_add_to_instance: Fehler: Übergebener Layer ist None")
return return
@ -399,28 +623,19 @@ class PoRunner(object):
layer_tree = self.iface.layerTreeCanvasBridge().rootGroup() layer_tree = self.iface.layerTreeCanvasBridge().rootGroup()
layer_tree.insertChildNode(0, QgsLayerTreeLayer(layer)) # am oberen Ende anhängen → liegt somit über basemap layer_tree.insertChildNode(0, QgsLayerTreeLayer(layer)) # am oberen Ende anhängen → liegt somit über basemap
def _layer_selection_changed(self): """
print("_layer_selection_changed") Aktualisiert die Labels eines Layers
if self.waterlevels is None or self.waterlevels != self.iface.activeLayer(): :param layer: Zu behandelnder Layer
print("_layer_selection_changed: Aktueller Layer ist nicht unser Pegelstands-Layer") :type layer: QgsVectorLayer
return :param fields: Anzuzeigende Felder (QGIS-Label-Expressions)
:type fields: list[str]
selected = self.waterlevels.selectedFeatures() """
if len(selected) == 1:
if self.ui.slHistoryStation.count() == 0:
self._history_load_stations()
selected_shortname = selected[0].attribute("shortname")
print("_layer_selection_changed: Lade Pegelstandsverlauf zur Auswahl: %s" % (selected_shortname,))
self._historyStation_set_by_shortname(selected_shortname)
else:
print("_layer_selection_changed: Anzahl ausgewählter Elemente ist NICHT 1, lade Pegelstandsverlauf NICHT!")
def _layer_update_labels(self, layer, fields): def _layer_update_labels(self, layer, fields):
print("_layer_update_labels") print("_layer_update_labels")
labeling = QgsVectorLayerSimpleLabeling(QgsPalLayerSettings()) labeling = QgsVectorLayerSimpleLabeling(QgsPalLayerSettings())
# Feldnamen zu einem Minus-getrennten String zusammenbauen # Feldnamen zu einem Slash-getrennten QGIS-Label-Expression-String zusammenbauen
expression = ", ' / ', ".join(fields) expression = ", ' / ', ".join(fields)
settings = labeling.settings() settings = labeling.settings()
@ -432,6 +647,12 @@ class PoRunner(object):
self._layer_refresh(layer) self._layer_refresh(layer)
"""
Kümmert sich um das neuzeichnen eines Layers
:param layer: Zu behandelnder Layer
:type layer: QgsVectorLayer
"""
def _layer_refresh(self, layer): def _layer_refresh(self, layer):
print("_layerRefresh") print("_layerRefresh")
if self.iface.mapCanvas().isCachingEnabled(): if self.iface.mapCanvas().isCachingEnabled():
@ -439,35 +660,92 @@ class PoRunner(object):
else: else:
self.iface.mapCanvas().refresh() self.iface.mapCanvas().refresh()
# layers styles ----------------------------------------------------------- # layer selection ---------------------------------------------------------
def _layer_apply_style_per_category(self, layer, attribute_name, file): """
Wird aufgerufen wenn sich eine Feature-Markierung auf einem Layer geändert hat.
Öffnet das Graph-Widget mit gewünschtem Pegelverlauf,
falls nur ein Feature vom Pegelstand-Layer markiert ist.
"""
def _layer_selection_changed(self):
print("_layer_selection_changed")
if self.waterlevels is None or self.waterlevels != self.iface.activeLayer():
# Falscher Layer aktiv → Abbruch
print("_layer_selection_changed: Aktueller Layer ist nicht unser Pegelstands-Layer")
return
selected = self.waterlevels.selectedFeatures()
if len(selected) == 1:
# Juhu, nur ein Pegelstand ausgewählt → Graph laden
if self.ui.slHistoryStation.count() == 0:
# Wir haben aber noch keine Stationen geladen → machen wir zuerst
self._history_load_stations()
selected_shortname = selected[0].attribute("shortname")
print("_layer_selection_changed: Lade Pegelstandsverlauf zur Auswahl: %s" % (selected_shortname,))
self._historyStation_set_by_shortname(selected_shortname)
else:
# Leider 0 oder mehrere Pegelstände ausgewählt → Abbruch
print("_layer_selection_changed: Anzahl ausgewählter Elemente ist NICHT 1, lade Pegelstandsverlauf NICHT!")
# layer styles ------------------------------------------------------------
"""
Färbt die Features des gegebenen Layers kategorisch nach einem Attribut ein.
:param layer: Zu behandelnder Layer
:type layer: QgsVectorLayer
:param attribute_name: Name des Attributes nach dem kategorisiert werden soll
:type attribute_name: str
:param preset_fallback_file: Preset- bzw Fallback-Style-Datei (.qml)
:type preset_fallback_file: str
"""
def _layer_apply_style_per_category(self, layer, attribute_name, preset_fallback_file):
print("_layer_apply_style_per_category: Erzeuge kategorisierte Farben...") print("_layer_apply_style_per_category: Erzeuge kategorisierte Farben...")
ramp = QgsStyle().defaultStyle().colorRamp("Turbo") ramp = QgsStyle().defaultStyle().colorRamp("Turbo")
if ramp is None: if ramp is None:
self._layer_apply_style_from_file(layer, file) # Standard-Ramp 'Turbo' nicht gefunden → Fallback auf Style-Datei
self._layer_apply_style_from_file(layer, preset_fallback_file)
print("_layer_apply_style_per_category: Farbrampe nicht gefunden.") print("_layer_apply_style_per_category: Farbrampe nicht gefunden.")
return return
idx = layer.fields().indexOf(attribute_name) attribute_index = layer.fields().indexOf(attribute_name)
values = sorted(v for v in layer.uniqueValues(idx) if v is not None) if attribute_index is None or attribute_index < 0:
# Gegebenes Attribut nicht gefunden → Fallback auf Style-Datei
self._layer_apply_style_from_file(layer, preset_fallback_file)
print("_layer_apply_style_per_category: Attribut nicht gefunden: %s" % (attribute_name,))
return
cats = [] attribute_values = sorted(v for v in layer.uniqueValues(attribute_index) if v is not None)
for v in values:
sym = QgsSymbol.defaultSymbol(layer.geometryType())
if len(values) > 1:
t = values.index(v) / (len(values) - 1)
else:
t = 0.0
sym.setColor(ramp.color(t))
cats.append(QgsRendererCategory(v, sym, str(v)))
renderer = QgsCategorizedSymbolRenderer(attribute_name, cats) categories = []
for value in attribute_values:
# Hole Farbe anhand linear umgerechnetem index von der 'Color-Ramp'
symbol = QgsSymbol.defaultSymbol(layer.geometryType())
color_index = attribute_values.index(value) / (len(attribute_values) - 1)
color = ramp.color(color_index)
symbol.setColor(color)
category = QgsRendererCategory(value, symbol, str(value))
categories.append(category)
renderer = QgsCategorizedSymbolRenderer(attribute_name, categories)
layer.setRenderer(renderer) layer.setRenderer(renderer)
self._layer_refresh(layer) self._layer_refresh(layer)
def _layer_apply_style_from_file(self, layer, file): """
path = os.path.join(self.local_dir, file) Wendet die gegebene Style-Datei (.qml) auf den gegebenen Layer an
:param layer: Zu behandelnder Layer
:type layer: QgsVectorLayer
:param preset_fallback_file: Style-Datei (.qml)
:type preset_fallback_file: str
"""
def _layer_apply_style_from_file(self, layer, preset_fallback_file):
path = os.path.join(self.local_dir, preset_fallback_file)
print("_layer_apply_style_from_file: Lade Style Datei: %s" % (path,)) print("_layer_apply_style_from_file: Lade Style Datei: %s" % (path,))
res = layer.loadNamedStyle(path) res = layer.loadNamedStyle(path)
@ -477,6 +755,10 @@ class PoRunner(object):
# history signals --------------------------------------------------------- # history signals ---------------------------------------------------------
"""
Verbindet alle GraphWidget-Signale mit der UI
"""
def _history_connect_signals(self): def _history_connect_signals(self):
print("_history_connect_signals") print("_history_connect_signals")
self.ui.slHistoryStation.currentTextChanged.connect(self._slHistoryStation_changed) self.ui.slHistoryStation.currentTextChanged.connect(self._slHistoryStation_changed)
@ -484,17 +766,34 @@ class PoRunner(object):
self.ui.numHistoryDays.valueChanged.connect(self._numHistoryDays_changed) self.ui.numHistoryDays.valueChanged.connect(self._numHistoryDays_changed)
self.ui.btnHistoryGo.clicked.connect(self._history_load_graph) self.ui.btnHistoryGo.clicked.connect(self._history_load_graph)
"""
Behandelt die Stations-Änderung und Lädt den Pegelstandsverlauf-Graph neu
"""
def _slHistoryStation_changed(self): def _slHistoryStation_changed(self):
print("_slHistoryStation_changed: %s" % (self.ui.slHistoryStation.currentText(),)) print("_slHistoryStation_changed: %s" % (self.ui.slHistoryStation.currentText(),))
self._history_load_graph() self._history_load_graph()
"""
Klick auf Graph-Stations-Liste-Refresh:
Lässt die Stations-Liste für den Graphen neu laden
"""
def _btnHistoryStationsRefresh_clicked(self): def _btnHistoryStationsRefresh_clicked(self):
print("_btnHistoryStationsRefresh_clicked") print("_btnHistoryStationsRefresh_clicked")
self._history_load_stations() self._history_load_stations()
"""
Loggt lediglich die Graph-Tages-Änderung
"""
def _numHistoryDays_changed(self): def _numHistoryDays_changed(self):
print("_numHistoryDays_changed: %s" % (self.ui.numHistoryDays.value(),)) print("_numHistoryDays_changed: %s" % (self.ui.numHistoryDays.value(),))
"""
Versucht den aktuell gewünschten Pegelstandsverlauf herunterzuladen und im GraphWidget anzuzeigen.
"""
def _history_load_graph(self): def _history_load_graph(self):
print("_history_load_graph") print("_history_load_graph")
@ -513,11 +812,13 @@ class PoRunner(object):
self.graph.show() self.graph.show()
if station == '' or station is None: if station == '' or station is None:
# Keine Station ausgewählt → Abbruch
print("_history_load_graph: Fehler: Ungültige Station: %s" % (station,)) print("_history_load_graph: Fehler: Ungültige Station: %s" % (station,))
self.graph.lbHistory.setText("Bitte Station wählen...") self.graph.lbHistory.setText("Bitte Station wählen...")
return return
if days is None or days < 1 or days > 30: if days is None or days < 1 or days > 30:
# Ungültige Anzahl an Tagen ausgewählt → Abbruch
print("_history_load_graph: Fehler: Ungültige Anzahl von Tagen: %s" % (days,)) print("_history_load_graph: Fehler: Ungültige Anzahl von Tagen: %s" % (days,))
self.graph.lbHistory.setText("Bitte Tage [1, 30] wählen...") self.graph.lbHistory.setText("Bitte Tage [1, 30] wählen...")
return return
@ -526,6 +827,7 @@ class PoRunner(object):
image_data = history.download() image_data = history.download()
if image_data is None or len(image_data) == 0: if image_data is None or len(image_data) == 0:
# Keine Bild-Daten beim Herunterladen erhalten → Abbruch
print("_history_load_graph: Fehler: Keine Daten erhalten") print("_history_load_graph: Fehler: Keine Daten erhalten")
self.graph.lbHistory.setText("Fehler beim Download!") self.graph.lbHistory.setText("Fehler beim Download!")
return return
@ -537,6 +839,12 @@ class PoRunner(object):
print("_history_load_graph: Bild erfolgreich geladen") print("_history_load_graph: Bild erfolgreich geladen")
"""
Lädt die Stations-Liste für den Graphen neu.
Sperrt solange relevante UI-Elemente.
Versucht die bisher ausgewählten Station in der neuen Liste wiederzufinden.
"""
def _history_load_stations(self): def _history_load_stations(self):
print("_history_load_stations") print("_history_load_stations")
self.ui.slHistoryStation.setEnabled(False) self.ui.slHistoryStation.setEnabled(False)
@ -546,19 +854,25 @@ class PoRunner(object):
current_station = self.ui.slHistoryStation.currentText() current_station = self.ui.slHistoryStation.currentText()
print("_history_load_stations: bisherige_station=%s" % (current_station,)) print("_history_load_stations: bisherige_station=%s" % (current_station,))
# leere die bisherige Auswahlliste
self.ui.slHistoryStation.clear() self.ui.slHistoryStation.clear()
# Lade aktuelle StationsListe herunter
stations = PoStationReader().get_stations() stations = PoStationReader().get_stations()
if stations is None or len(stations) == 0: if stations is None or len(stations) == 0:
# Keine Stationen erhalte → Abbruch
print("_history_load_stations: Fehler: Keine Stationen erhalten") print("_history_load_stations: Fehler: Keine Stationen erhalten")
return return
for station in stations: for station in stations:
# Füge ein Item je Station hinzu
self.ui.slHistoryStation.addItem(station.shortname) self.ui.slHistoryStation.addItem(station.shortname)
if self._historyStation_set_by_shortname(current_station): if self._historyStation_set_by_shortname(current_station):
# Bisherige Station wiedergefunden → Wieder ausgewählt
print("_history_load_stations: Bisherige Station \"%s\" wiederhergestellt" % (current_station,)) print("_history_load_stations: Bisherige Station \"%s\" wiederhergestellt" % (current_station,))
else: else:
# Bisherige Station nicht wiedergefunden → Wähle die erste Station aus der Liste
self.ui.slHistoryStation.setCurrentIndex(0) self.ui.slHistoryStation.setCurrentIndex(0)
station = self.ui.slHistoryStation.currentText() station = self.ui.slHistoryStation.currentText()
print("_history_load_stations: Bisherige Station \"%s\" nicht wiedergefunden. Nehme erste Station: %s" % (current_station, station)) print("_history_load_stations: Bisherige Station \"%s\" nicht wiedergefunden. Nehme erste Station: %s" % (current_station, station))
@ -566,6 +880,12 @@ class PoRunner(object):
self.ui.slHistoryStation.setEnabled(True) self.ui.slHistoryStation.setEnabled(True)
self.ui.btnHistoryGo.setEnabled(True) self.ui.btnHistoryGo.setEnabled(True)
"""
Setzt eine Graph-Station anhand des gegebenen Kurznamens
:param shortname: Kurzname der Station die gesetzt werden soll
:type shortname: str
"""
def _historyStation_set_by_shortname(self, shortname): def _historyStation_set_by_shortname(self, shortname):
index = self._historyStation_get_index_by_shortname(shortname) index = self._historyStation_get_index_by_shortname(shortname)
if index is None: if index is None:
@ -573,6 +893,12 @@ class PoRunner(object):
self.ui.slHistoryStation.setCurrentIndex(index) self.ui.slHistoryStation.setCurrentIndex(index)
return True return True
"""
Sucht den Index einer Graph-Station anhand des gegebenen Kurznamens
:param shortname: Kurzname der Station die gesucht werden soll
:type shortname: str
"""
def _historyStation_get_index_by_shortname(self, shortname): def _historyStation_get_index_by_shortname(self, shortname):
print("_historyStation_get_index_by_shortname: shortname=%s" % (shortname,)) print("_historyStation_get_index_by_shortname: shortname=%s" % (shortname,))
for index in range(self.ui.slHistoryStation.count()): for index in range(self.ui.slHistoryStation.count()):