Compare commits

..

3 Commits

Author SHA1 Message Date
80bbeeabbf docstring FIX 2025-09-29 16:26:42 +02:00
29c2608d05 nach PegelonlineDockWidgetGraph.load ausgelagert 2025-09-29 15:43:36 +02:00
1c9c863d1d kleinere clean-ups 2025-09-29 15:33:40 +02:00
12 changed files with 416 additions and 372 deletions

2
.gitignore vendored
View File

@ -1,3 +1,5 @@
/.venv/
/docs/
/basemap/
/resources.py
/*.zip

View File

@ -175,7 +175,7 @@
<item>
<widget class="QGroupBox" name="groupBox_2">
<property name="title">
<string>Styles:</string>
<string>Stile:</string>
</property>
<layout class="QHBoxLayout" name="horizontalLayout_8">
<property name="leftMargin">
@ -420,7 +420,7 @@
</size>
</property>
<property name="text">
<string>Neu Laden</string>
<string>Neu laden</string>
</property>
</widget>
</item>
@ -611,7 +611,7 @@
<item>
<widget class="QCheckBox" name="cbWaterlevelsValue">
<property name="text">
<string>Aktueller Wert</string>
<string>Aktueller Pegel</string>
</property>
</widget>
</item>
@ -637,7 +637,7 @@
<item>
<widget class="QGroupBox" name="groupBox">
<property name="title">
<string>Styles:</string>
<string>Stile:</string>
</property>
<layout class="QHBoxLayout" name="horizontalLayout_7">
<property name="spacing">

View File

@ -24,9 +24,11 @@
import os
from qgis.PyQt import QtWidgets, uic
from qgis.PyQt import QtWidgets, uic, QtGui
from qgis.PyQt.QtCore import pyqtSignal
from .po_modules.po_graph_reader import PoGraphReader
FORM_CLASS, _ = uic.loadUiType(os.path.join(os.path.dirname(__file__), 'pegelonline_dockwidget_graph.ui'))
@ -34,15 +36,51 @@ class PegelonlineDockWidgetGraph(QtWidgets.QDockWidget, FORM_CLASS):
closingPlugin = pyqtSignal()
def __init__(self, parent=None):
"""Constructor."""
super(PegelonlineDockWidgetGraph, self).__init__(parent)
# Set up the user interface from Designer.
# After setupUI you can access any designer object by doing
# self.<objectname>, and you can use autoconnect slots - see
# http://doc.qt.io/qt-5/designer-using-a-ui-file.html
# #widgets-and-dialogs-with-auto-connect
self.setupUi(self)
def closeEvent(self, event):
self.closingPlugin.emit()
event.accept()
def load(self, station, days):
"""
Versucht den aktuell gewünschten Pegelstandsverlauf herunterzuladen und im GraphWidget anzuzeigen.
:param station: Der 'Kurzname' der gewünschten Station
:type station: str
:param days: Anzahl der gewünschten vergangenen Tage
:type days: int
"""
print("PegelonlineDockWidgetGraph::load: station=%s days=%s" % (station, days))
self.lbGraph.clear()
self.setWindowTitle("%s / %d Tag(e)" % (station, days))
self.show()
if station == '' or station is None:
# Keine Station ausgewählt → Abbruch
print("PegelonlineDockWidgetGraph::load: Fehler: Ungültige Station: %s" % (station,))
self.lbGraph.setText("Bitte Station wählen...")
return
if days is None or days < 1 or days > 30:
# Ungültige Anzahl an Tagen ausgewählt → Abbruch
print("PegelonlineDockWidgetGraph::load: Fehler: Ungültige Anzahl von Tagen: %s" % (days,))
self.lbGraph.setText("Bitte Tage [1, 30] wählen...")
return
graph = PoGraphReader(station, days)
image_data = graph.download()
if image_data is None or len(image_data) == 0:
# Keine Bild-Daten beim Herunterladen erhalten → Abbruch
print("PegelonlineDockWidgetGraph::load: Fehler: Keine Daten erhalten")
self.lbGraph.setText("Fehler beim Download!")
return
pixmap = QtGui.QPixmap()
pixmap.loadFromData(image_data)
self.lbGraph.setPixmap(pixmap)
self.lbGraph.resize(pixmap.width(), pixmap.height())
print("PegelonlineDockWidgetGraph::load: Bild erfolgreich geladen")

View File

@ -7,13 +7,24 @@ from .urlreader import UrlReader
class PoGraphReader(UrlReader):
def __init__(self, station: str, days: int):
"""
Initialisiert die Super-Klasse mit einer URL aus gegebener Station und Anzahl Tage
:param station: Kurzname der gewünschten Station
:type station: str
:param days: gewünschte Anzahl an Tagen
:type days: int
"""
super().__init__(poBaseURL + 'stations/%s/W/measurements.png?start=P%dD' % (quote(station), days))
def download(self):
"""
Versucht die Grafik über die super.get_data_response Methode herunterzuladen
"""
print("PoGraphReader::download: Lade Bild herunter...")
image_data = self.getDataResponse()
image_data = self.get_data_response()
if image_data is None or len(image_data) == 0:
# Keine Daten erhalten → Abbruch
print("PoGraphReader::download: Fehler: Keine Daten erhalten")
return None

View File

@ -4,6 +4,11 @@ from qgis._core import QgsFeature, QgsGeometry, QgsPointXY
class PoStation(object):
def __init__(self, json):
"""
Nimmt JSON-Daten einer Pegelonline-Station entgegen und füllt die entsprechenden Felder
:param json: Json-Daten einer Station
:type json: dict
"""
self.longitude = json['longitude'] if 'longitude' in json else None
self.latitude = json['latitude'] if 'latitude' in json else None
self.uuid = json['uuid']
@ -15,7 +20,14 @@ class PoStation(object):
self.water = json['water']['longname']
def new_feature(self, fields) -> None | QgsFeature:
"""
Erzeugt ein QgsFeature mit Attributen dieser Station oder None im Fehlerfall
:param fields: QgsField-Liste um das Feature zu befüllen
:type fields: list[QgsField]
:return: None | QgsFeature
"""
if self.longitude is None or self.latitude is None:
# Keine Koordinaten → Abbruch
print("PoStation::new_feature: WARN: Station hat fehlende Koordinaten: %s" % (self.shortname,))
return None

View File

@ -6,21 +6,31 @@ from .urlreader import UrlReader
class PoStationReader(UrlReader):
def __init__(self):
"""
Initialisiert die Super-Klasse mit URL für die Stationen
"""
super().__init__(poBaseURL + 'stations.json')
def get_stations(self):
"""
Fragt die Liste aller Stationen via UrlReader.get_json_response ab und macht PoStation daraus oder None im Fehlerfall
:return: list[PoStation] | None
"""
print("PoStationReader::get_stations: Lade Stationen herunter...")
stations_json = self.getJsonResponse()
stations_json = self.get_json_response()
if stations_json is None or len(stations_json) == 0:
# Keine Stationen erhalten → Abbruch
print("PoStationReader::get_stations: Keine Stationen erhalten")
return None
stations = []
for station_json in stations_json:
try:
# Versuche eine Station zu erstellen
stations.append(PoStation(station_json))
except Exception as e:
# Fehler → Überspringe diese Station
print("PoStationReader::get_stations: Fehler: error=%s, json=%s" % (e, station_json))
print("PoStationReader::get_stations: %d Stationen erhalten" % (len(stations),))

View File

@ -2,17 +2,24 @@ from PyQt5.QtCore import QVariant
from qgis._core import QgsCoordinateReferenceSystem
from qgis.core import QgsFields, QgsField
from .po_stations import PoStationReader
from .po_station_reader import PoStationReader
class PoStationReaderQgs(PoStationReader):
def __init__(self):
"""
Initialisiert die Super-Klasse und das Koordinaten-System der Features
"""
super().__init__()
self.fields = None
self.crs = QgsCoordinateReferenceSystem(4326, QgsCoordinateReferenceSystem.EpsgCrsId)
def get_features(self):
"""
Erzeugt Features aus allen Stationen von super.get_stations() oder None im Fehlerfall
:return: list[QgsFeature] | None
"""
print("PoStationReaderQgs::get_features: Erzeuge Features...")
self.fields = QgsFields()
@ -28,12 +35,14 @@ class PoStationReaderQgs(PoStationReader):
stations = self.get_stations()
if stations is None or len(stations) == 0:
# Keine Stationen erhalten → Abbruch
print("PoStationReaderQgs::get_features: Fehler: Keine Stationen erhalten")
return None
for station in stations:
feature = station.new_feature(self.fields)
if feature is not None:
# Feature-Erzeugung erfolgreich, füge zu Liste hinzu
features.append(feature)
print("PoStationReaderQgs::get_features: %d Features erzeugt" % (len(features),))

View File

@ -4,6 +4,11 @@ from qgis._core import QgsFeature, QgsGeometry, QgsPointXY
class PoWaterlevel(object):
def __init__(self, json):
"""
Nimmt JSON-Daten eines Pegelonline-Waterlevels entgegen und füllt die entsprechenden Felder
:param json: Json-Daten einer Waterlevel
:type json: dict
"""
self.longitude = json['longitude'] if 'longitude' in json else None
self.latitude = json['latitude'] if 'latitude' in json else None
self.uuid = json['uuid']
@ -18,8 +23,15 @@ class PoWaterlevel(object):
self.water = json['water']['longname']
def new_feature(self, fields):
"""
Erzeugt ein QgsFeature mit Attributen dieses Waterlevels oder None im Fehlerfall
:param fields: QgsField-Liste um das Feature zu befüllen
:type fields: list[QgsField]
:return: None | QgsFeature
"""
if self.longitude is None or self.latitude is None:
print("PoWaterlevel::new_feature: WARN: Station hat fehlende Koordinaten: %s" % (self.shortname,))
# Keine Koordinaten → Abbruch
print("PoWaterlevel::new_feature: WARN: Waterlevel hat fehlende Koordinaten: %s" % (self.shortname,))
return None
feature = QgsFeature(fields)

View File

@ -6,20 +6,33 @@ from .urlreader import UrlReader
class PoWaterlevelReader(UrlReader):
def __init__(self):
"""
Initialisiert die Super-Klasse mit URL für die Waterlevel
"""
super().__init__(poBaseURL + 'stations.json?timeseries=W&includeTimeseries=true&includeCurrentMeasurement=true')
def get_waterlevels(self):
"""
Fragt die Liste aller Waterlevels via UrlReader.get_json_response ab und macht PoWaterlevel daraus oder None im Fehlerfall
:return: list[PoWaterlevel] | None
"""
print("PoWaterlevelReader::get_waterlevels: Lade Pegelstände herunter...")
stations_json = self.getJsonResponse()
if stations_json is None or len(stations_json) == 0:
waterlevels_json = self.get_json_response()
if waterlevels_json is None or len(waterlevels_json) == 0:
# Keine Waterlevels erhalten → Abbruch
print("PoWaterlevelReader::get_waterlevels: FEHLER: Keine Pegelstände erhalten")
return None
stations = []
for station_json in stations_json:
stations.append(PoWaterlevel(station_json))
waterlevels = []
for waterlevel_json in waterlevels_json:
try:
# Versuche eine Waterlevel zu erstellen
waterlevels.append(PoWaterlevel(waterlevel_json))
except Exception as e:
# Fehler → Überspringe diesen Waterlevel
print("PoWaterlevelReader::get_waterlevels: Fehler: error=%s, json=%s" % (e, waterlevel_json))
print("PoWaterlevelReader::get_waterlevels: %d Pegelstände erhalten" % (len(stations),))
print("PoWaterlevelReader::get_waterlevels: %d Pegelstände erhalten" % (len(waterlevels),))
return stations
return waterlevels

View File

@ -8,11 +8,18 @@ from .po_waterlevel_reader import PoWaterlevelReader
class PoWaterlevelReaderQgs(PoWaterlevelReader):
def __init__(self):
"""
Initialisiert die Super-Klasse und das Koordinaten-System der Features
"""
super().__init__()
self.fields = None
self.crs = QgsCoordinateReferenceSystem(4326, QgsCoordinateReferenceSystem.EpsgCrsId)
def get_features(self):
"""
Erzeugt Features aus allen Waterlevels von super.get_waterlevels() oder None im Fehlerfall
:return: list[QgsFeature] | None
"""
print("PoWaterlevelReaderQgs::get_features: Erzeuge Features...")
self.fields = QgsFields()
@ -29,12 +36,14 @@ class PoWaterlevelReaderQgs(PoWaterlevelReader):
features = []
waterlevels = self.get_waterlevels()
if waterlevels is None or len(waterlevels) == 0:
# Kein Waterlevel erhalten → Abbruch
print("PoWaterlevelReaderQgs::get_features: Fehler: Keine Pegelstände erhalten")
return None
for waterlevel in waterlevels:
feature = waterlevel.new_feature(self.fields)
if feature is not None:
# Feature-Erzeugung erfolgreich, füge zu Liste hinzu
features.append(feature)
print("PoWaterlevelReaderQgs::get_features: %d Features erzeugt" % (len(features),))

View File

@ -15,105 +15,114 @@ class UrlReader(object):
"""
self.url = _url
def openUrl(self):
def open_url(self):
"""
Öffnet eine URL-Verbindung, fragt GZIP-Kompression an und gibt das Response-Objekt zurück
:return: Response-Objekt oder None im Fehlerfall
"""
print("openURL: url: \"%s\"" % (self.url,))
print("open_url: url: \"%s\"" % (self.url,))
try:
# Versuche Verbindung zu öffnen
request = Request(self.url)
request.add_header('Accept-Encoding', 'gzip')
response = urlopen(request)
print("openURL: Verbindung hergestellt")
print("open_url: Verbindung hergestellt")
return response
except URLError as e: # auch HTTPError
print("openURL: FEHLER: " + str(e))
print("open_url: FEHLER: " + str(e))
return None # Fehler
return None
def getDataResponse(self):
def get_data_response(self):
"""
Benutzt openUrl und gibt die (entpackten) Daten zurück.
Benutzt open_url und gibt die (entpackten) Daten zurück.
:return: (entpackte) Daten oder None im Fehlerfall
"""
print("getDataResponse: url: \"%s\"" % (self.url,))
response = self.openUrl()
print("get_data_response: url: \"%s\"" % (self.url,))
response = self.open_url()
if response is None:
print("getDataResponse: FEHLER: Kein Response-Objekt erhalten")
#FEHLER: Kein Response-Objekt erhalten
print("get_data_response: FEHLER: Kein Response-Objekt erhalten")
return None
try:
if response.headers['Content-Encoding'] == 'gzip':
print("getDataResponse: Empfange GZIP Daten...")
# Empfange GZIP Daten...
print("get_data_response: Empfange GZIP Daten...")
daten = GzipFile(fileobj=response).read()
else:
print("getDataResponse: Empfange unkomprimierte Daten...")
# Empfange unkomprimierte Daten...
print("get_data_response: Empfange unkomprimierte Daten...")
daten = response.read()
print("getDataResponse: Daten empfangen")
print("get_data_response: Daten empfangen")
return daten
except OSError as e:
print("getDataResponse: FEHLER: " + str(e))
print("get_data_response: FEHLER: " + str(e))
return None # Kein Erfolg
return None
def getJsonResponse(self):
def get_json_response(self):
"""
Benutzt getDataResponse zum Herunterladen, interpretiert die Daten als JSON und gibt das Ergebnis zurück.
Benutzt get_data_response zum Herunterladen, interpretiert die Daten als JSON und gibt das Ergebnis zurück.
:return: Geparste JSON Daten oder None im Fehlerfall
"""
print("getJsonResponse: url=" + self.url)
daten = self.getDataResponse()
print("get_json_response: url=" + self.url)
daten = self.get_data_response()
if daten is None:
print("getJsonResponse: FEHLER: Keine Daten erhalten")
# FEHLER: Keine Daten erhalten
print("get_json_response: FEHLER: Keine Daten erhalten")
return None
try:
print("getJsonResponse: Lese JSON...")
# Versuche JSON zu lesen
print("get_json_response: Lese JSON...")
parsed = json.loads(daten)
print("getJsonResponse: JSON gelesen")
print("get_json_response: JSON gelesen")
return parsed
except ValueError as e: # JSONDecodeError
print("getJsonResponse: ValueError: " + str(e))
print("get_json_response: ValueError: " + str(e))
except TypeError as e: # JSONDecodeError
print("getJsonResponse: TypeError: " + str(e))
print("get_json_response: TypeError: " + str(e))
return None # Kein Erfolg
def _dateiname_von_url(self):
"""
Extrahiert den Dateinamen aus einer der URL
"""
result = urlparse(self.url)
dateiname = basename(result.path)
return dateiname
def getFileResponse(self, pfad):
def get_file_response(self, pfad):
"""
Benutzt getDataResponse zum Herunterladen, schreibt die Daten in eine Datei und gibt ihren Pfad zurück (gegebenes Verzeichnis + basename des URL-Pfades).
Benutzt get_data_response zum Herunterladen, schreibt die Daten in eine Datei und gibt ihren Pfad zurück (gegebenes Verzeichnis + basename des URL-Pfades).
:param pfad: Verzeichnis in dem die Datei gespeichert werden soll.
:return: Pfad der erzeugten Datei oder None im Fehlerfall
"""
print("getFileResponse: url: \"%s\"" % (self.url,))
print("getFileResponse: pfad: \"%s\"" % (pfad,))
print("get_file_response: url: \"%s\"" % (self.url,))
print("get_file_response: pfad: \"%s\"" % (pfad,))
daten = self.getDataResponse()
daten = self.get_data_response()
if daten is None:
print("getFileResponse: FEHLER: Keine Daten erhalten")
print("get_file_response: FEHLER: Keine Daten erhalten")
return None
dateiname = self._dateiname_von_url()
print("getFileResponse: dateiname: \"%s\"" % (dateiname,))
print("get_file_response: dateiname: \"%s\"" % (dateiname,))
dateipfad = join(pfad, dateiname)
print("getFileResponse: dateipfad: \"%s\"" % (dateipfad,))
print("get_file_response: dateipfad: \"%s\"" % (dateipfad,))
try:
print("getFileResponse: Schreibe Datei...")
print("get_file_response: Schreibe Datei...")
with open(dateipfad, 'wb') as datei:
datei.write(daten)
print("getFileResponse: Datei geschrieben")
print("get_file_response: Datei geschrieben")
return dateipfad
except OSError as e:
@ -124,4 +133,4 @@ class UrlReader(object):
if __name__ == '__main__':
url = "https://ia800302.us.archive.org/8/items/BennyGoodmanQuartetAndTrio/BodySoul-BennyGoodmanGeneKrupaTeddyWilsoncarnegieHall1938_64kb.mp3"
print(UrlReader(url).getFileResponse(""))
print(UrlReader(url).get_file_response(""))

View File

@ -1,16 +1,14 @@
import os.path
from typing import Callable
from PyQt5 import QtGui
from PyQt5.QtWidgets import QAction, QCheckBox
from qgis._core import QgsVectorLayer, QgsProject, QgsLayerTreeLayer, QgsPalLayerSettings, QgsVectorLayerSimpleLabeling, QgsStyle, QgsSymbol, QgsRendererCategory, QgsCategorizedSymbolRenderer
from .map_tips import WATERLEVELS_MAP_TIPS, STATIONS_MAP_TIPS, BASEMAP_MAP_TIPS
from .pegelonline_dockwidget import PegelonlineDockWidget
from .pegelonline_dockwidget_graph import PegelonlineDockWidgetGraph
from .po_modules.po_graph_reader import PoGraphReader
from .po_modules.po_stations import PoStationReader
from .po_modules.po_stations_qgs import PoStationReaderQgs
from .po_modules.po_station_reader import PoStationReader
from .po_modules.po_stations_reader_qgs import PoStationReaderQgs
from .po_modules.po_waterlevels_reader_qgs import PoWaterlevelReaderQgs
STATIONS_QML = "styles/stations.qml"
@ -19,6 +17,8 @@ WATERLEVELS_QML = "styles/waterlevels.qml"
class PoRunner(object):
def __init__(self, ui: PegelonlineDockWidget, graph: PegelonlineDockWidgetGraph, iface):
"""
Initialisiert ein neues PoRunner-Objekt:
- Initialisiert Objekt-Variablen
@ -31,8 +31,6 @@ class PoRunner(object):
:param iface: QGIS-Interface
:type iface: QgisInterface
"""
def __init__(self, ui: PegelonlineDockWidget, graph: PegelonlineDockWidgetGraph, iface):
# Widget-Referenzen
self.ui = ui
self.graph = graph
@ -68,20 +66,19 @@ class PoRunner(object):
# basemap -----------------------------------------------------------------
def _basemap_create(self, path, name, disconnect: Callable[[], None], map_tips, checkbox) -> None | QgsVectorLayer:
"""
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
:param path: Pfad zu den 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
:param map_tips: HTML Code für die QGIS-Map-Tipps
:param checkbox: Sichtbarkeit-Checkbox um sie via Signal upzudaten
:type checkbox: QCheckBox
"""
def _basemap_create(self, path, name, disconnect: Callable[[], None], map_tips, checkbox) -> None | QgsVectorLayer:
print("_basemap_create: %s" % (name,))
path = os.path.join(self.local_dir, "basemap", path)
basemap = QgsVectorLayer(path, name, "ogr")
@ -109,20 +106,18 @@ class PoRunner(object):
return basemap
def _basemap_connect_signals(self):
"""
Verbindet alle Basemap-Signale mit der UI
"""
def _basemap_connect_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):
"""
Behandelt die Sichbarkeitsänderung der Flüsse durch die UI-Checkbox
"""
def _cbBasemapLines_toggled(self):
checked = self.ui.cbBasemapLines.isChecked()
print("_cbBasemapLines_toggled: %s" % (checked,))
@ -141,11 +136,10 @@ class PoRunner(object):
self._layer_set_visible(self.lines, checked)
self._layer_refresh(self.lines)
def _cbBasemapAreas_toggled(self):
"""
Behandelt die Sichtbarkeitsänderung der Flächen durch die UI-Checkbox
"""
def _cbBasemapAreas_toggled(self):
checked = self.ui.cbBasemapAreas.isChecked()
print("_cbBasemapAreas_toggled: %s" % (checked,))
@ -164,31 +158,28 @@ class PoRunner(object):
self._layer_set_visible(self.areas, checked)
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):
"""
Löscht die Flüsse-Layer-Referenz, nachdem der Layer aus QGIS gelöscht wurde und passt die Checkbox an
"""
print("_basemap_disconnect_lines")
self.lines = None
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):
"""
Löscht die Flächen-Layer-Referenz, nachdem der Layer aus QGIS gelöscht wurde und passt die Checkbox an
"""
print("_basemap_disconnect_areas")
self.areas = None
self.ui.cbBasemapAreas.setChecked(False)
# stations ----------------------------------------------------------------
def _stations_connect_signals(self):
"""
Verbindet alle Stations-Signale mit der UI
"""
def _stations_connect_signals(self):
print("_connect_stations_signals")
self.ui.cbStationsVisible.toggled.connect(self._cbStationsVisible_toggled)
self.ui.cbStationsName.toggled.connect(self._cbStationsName_toggled)
@ -198,22 +189,20 @@ class PoRunner(object):
self.ui.cbStationsWater.toggled.connect(self._cbStationsWater_toggled)
self.ui.bgStationsStyle.buttonClicked.connect(self._bgStationsStyle_clicked)
def _stations_apply_style(self):
"""
Wendet eingestellte Styles auf den Stations-Layer an
"""
def _stations_apply_style(self):
button = self.ui.bgStationsStyle.checkedButton()
self._bgStationsStyle_clicked(button)
def _bgStationsStyle_clicked(self, button):
"""
Behandlung eines Klicks auf einen der Stations-Style-Radio-Buttons
Wendet Styles auf Stations-Layer an
:param button: Geklickter Radio-Button
:param button: geklickter Radio-Button
:type button: QRadioButton
"""
def _bgStationsStyle_clicked(self, button):
print("_bgStationsStyle_clicked: %s" % (button.objectName(),))
if self.stations is None:
@ -241,11 +230,10 @@ class PoRunner(object):
self.stations.setMapTipTemplate(STATIONS_MAP_TIPS)
self._layer_apply_style_per_category(self.stations, field, STATIONS_QML)
def _cbStationsVisible_toggled(self):
"""
Schaltet Sichtbarkeit des Stations-Layers um (und erstellt ihn, falls nötig)
"""
def _cbStationsVisible_toggled(self):
visible = self.ui.cbStationsVisible.isChecked()
print("_cbStationsVisible_toggled: %s" % (visible,))
@ -262,65 +250,58 @@ class PoRunner(object):
if visible:
self._stations_apply_style()
def _cbStationsName_toggled(self):
"""
Sichtbarkeit des Stations-Attributs 'Name' umschalten und Labels updaten
"""
def _cbStationsName_toggled(self):
checked = self.ui.cbStationsName.isChecked()
print("_cbStationsName_toggled: %s" % (checked,))
self._stations_update_labels()
def _cbStationsNumber_toggled(self):
"""
Sichtbarkeit des Stations-Attributs 'Number' umschalten und Labels updaten
"""
def _cbStationsNumber_toggled(self):
checked = self.ui.cbStationsNumber.isChecked()
print("_cbStationsNumber_toggled: %s" % (checked,))
self._stations_update_labels()
def _cbStationsAgency_toggled(self):
"""
Sichtbarkeit des Stations-Attributs 'Agency' umschalten und Labels updaten
"""
def _cbStationsAgency_toggled(self):
checked = self.ui.cbStationsAgency.isChecked()
print("_cbStationsAgency_toggled: %s" % (checked,))
self._stations_update_labels()
def _cbStationsKm_toggled(self):
"""
Sichtbarkeit des Stations-Attributs 'Km' umschalten und Labels updaten
"""
def _cbStationsKm_toggled(self):
checked = self.ui.cbStationsKm.isChecked()
print("_cbStationsKm_toggled: %s" % (checked,))
self._stations_update_labels()
def _cbStationsWater_toggled(self):
"""
Sichtbarkeit des Stations-Attributs 'Water' umschalten und Labels updaten
"""
def _cbStationsWater_toggled(self):
checked = self.ui.cbStationsWater.isChecked()
print("_cbStationsWater_toggled: %s" % (checked,))
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):
"""
Löscht die Stations-Layer-Referenz, nachdem der Layer aus QGIS gelöscht wurde und passt die Checkbox an
"""
print("_stations_disconnect")
self.stations = None
self.ui.cbStationsVisible.setChecked(False)
def _stations_update_labels(self):
"""
Führt Änderungen an Stations-Labels durch
"""
def _stations_update_labels(self):
print("_stations_update_labels")
if self.stations is None:
return
@ -346,11 +327,10 @@ class PoRunner(object):
# waterlevels -------------------------------------------------------------
def _waterlevels_connect_signals(self):
"""
Verbindet alle Pegelstand-Signale mit der UI
"""
def _waterlevels_connect_signals(self):
print("_waterlevels_connect_signals")
self.ui.cbWaterlevelsVisible.toggled.connect(self._cbWaterlevelsVisible_toggled)
self.ui.cbWaterlevelsName.toggled.connect(self._cbWaterlevelsName_toggled)
@ -363,22 +343,20 @@ class PoRunner(object):
self.ui.cbWaterlevelsWater.toggled.connect(self._cbWaterlevelsWater_toggled)
self.ui.bgWaterlevelsStyle.buttonClicked.connect(self._bgWaterlevelsStyle_clicked)
def _waterlevels_apply_style(self):
"""
Wendet eingestellte Styles auf den Pegelstand-Layer an
"""
def _waterlevels_apply_style(self):
button = self.ui.bgWaterlevelsStyle.checkedButton()
self._bgWaterlevelsStyle_clicked(button)
def _bgWaterlevelsStyle_clicked(self, button):
"""
Behandlung eines Klicks auf einen der Pegelstand-Style-Radio-Buttons
Wendet Styles auf Pegelstand-Layer an
:param button: Geklickter Radio-Button
:param button: geklickter Radio-Button
:type button: QRadioButton
"""
def _bgWaterlevelsStyle_clicked(self, button):
print("_bgWaterlevelsStyle_clicked: %s" % (button.objectName(),))
if self.waterlevels is None:
@ -412,11 +390,10 @@ class PoRunner(object):
self.waterlevels.setMapTipTemplate(WATERLEVELS_MAP_TIPS)
self._layer_apply_style_per_category(self.waterlevels, field, WATERLEVELS_QML)
def _cbWaterlevelsVisible_toggled(self):
"""
Schaltet Sichtbarkeit des Pegelstand-Layers um (und erstellt ihn, falls nötig)
"""
def _cbWaterlevelsVisible_toggled(self):
visible = self.ui.cbWaterlevelsVisible.isChecked()
print("_cbWaterlevelsVisible_toggled: %s" % (visible,))
@ -433,92 +410,82 @@ class PoRunner(object):
if visible:
self._waterlevels_apply_style()
def _cbWaterlevelsName_toggled(self):
"""
Sichtbarkeit des Pegelstand-Attributs 'Name' umschalten und Labels updaten
"""
def _cbWaterlevelsName_toggled(self):
checked = self.ui.cbWaterlevelsName.isChecked()
print("_cbWaterlevelsName_toggled: %s" % (checked,))
self._waterlevels_update_labels()
def _cbWaterlevelsNumber_toggled(self):
"""
Sichtbarkeit des Pegelstand-Attributs 'Number' umschalten und Labels updaten
"""
def _cbWaterlevelsNumber_toggled(self):
checked = self.ui.cbWaterlevelsNumber.isChecked()
print("_cbWaterlevelsNumber_toggled: %s" % (checked,))
self._waterlevels_update_labels()
def _cbWaterlevelsAgency_toggled(self):
"""
Sichtbarkeit des Pegelstand-Attributs 'Agency' umschalten und Labels updaten
"""
def _cbWaterlevelsAgency_toggled(self):
checked = self.ui.cbWaterlevelsAgency.isChecked()
print("_cbWaterlevelsAgency_toggled: %s" % (checked,))
self._waterlevels_update_labels()
def _cbWaterlevelsTimestamp_toggled(self):
"""
Sichtbarkeit des Pegelstand-Attributs 'Timestamp' umschalten und Labels updaten
"""
def _cbWaterlevelsTimestamp_toggled(self):
checked = self.ui.cbWaterlevelsTimestamp.isChecked()
print("_cbWaterlevelsTimestamp_toggled: %s" % (checked,))
self._waterlevels_update_labels()
def _cbWaterlevelsValue_toggled(self):
"""
Sichtbarkeit des Pegelstand-Attributs 'Value' umschalten und Labels updaten
"""
def _cbWaterlevelsValue_toggled(self):
checked = self.ui.cbWaterlevelsValue.isChecked()
print("_cbWaterlevelsValue_toggled: %s" % (checked,))
self._waterlevels_update_labels()
def _cbWaterlevelsMean_toggled(self):
"""
Sichtbarkeit des Pegelstand-Attributs 'Mean' umschalten und Labels updaten
"""
def _cbWaterlevelsMean_toggled(self):
checked = self.ui.cbWaterlevelsMean.isChecked()
print("_cbWaterlevelsMean_toggled: %s" % (checked,))
self._waterlevels_update_labels()
def _cbWaterlevelsAbsolute_toggled(self):
"""
Sichtbarkeit des Pegelstand-Attributs 'Absolute' umschalten und Labels updaten
"""
def _cbWaterlevelsAbsolute_toggled(self):
checked = self.ui.cbWaterlevelsAbsolute.isChecked()
print("_cbWaterlevelsAbsolute_toggled: %s" % (checked,))
self._waterlevels_update_labels()
def _cbWaterlevelsWater_toggled(self):
"""
Sichtbarkeit des Pegelstand-Attributs 'Water' umschalten und Labels updaten
"""
def _cbWaterlevelsWater_toggled(self):
checked = self.ui.cbWaterlevelsWater.isChecked()
print("_cbWaterlevelsWater_toggled: %s" % (checked,))
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):
"""
Löscht die Pegelstand-Layer-Referenz, nachdem der Layer aus QGIS gelöscht wurde und passt die Checkbox an
"""
print("waterlevels_disconnect")
self.waterlevels = None
self.ui.cbWaterlevelsVisible.setChecked(False)
def _waterlevels_update_labels(self):
"""
Führt Änderungen an Pegelstand-Labels durch
"""
def _waterlevels_update_labels(self):
print("_waterlevels_update_labels")
if self.waterlevels is None:
return
@ -553,6 +520,7 @@ class PoRunner(object):
# layers ------------------------------------------------------------------
def _layer_create_from_features(self, fields, crs, features, title) -> None | QgsVectorLayer:
"""
Erzeugt einen neuen Layer aus QGIS-Features
:param fields: Anzulegende QGIS-Felder
@ -561,11 +529,7 @@ class PoRunner(object):
: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) -> None | QgsVectorLayer:
print("_layer_create_from_features")
if features is None:
@ -600,30 +564,28 @@ class PoRunner(object):
return layer
def _layer_set_visible(self, layer: QgsVectorLayer, visible):
"""
Setzt die Sichtbarkeit des gegebenen Layers
:param layer: Zu behandelnder Layer
: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" % (visible,))
layer_tree = QgsProject.instance().layerTreeRoot().findLayer(layer.id())
layer_tree.setItemVisibilityChecked(visible)
def _layer_add_to_instance(self, layer: QgsVectorLayer, disconnect: Callable[[], None], checkbox: QCheckBox):
"""
Fügt einen Layer der QGIS-Instanz hinzu und verbindet dessen Signale
:param layer: Zu behandelnder Layer
:param layer: zu behandelnder Layer
:type layer: QgsVectorLayer
:param disconnect: Methode die Aufgerufen wird, wenn der Layer gelöscht wird
:type disconnect: Callable[[], None]
:param checkbox: Sichtbarkeit-Checkbox um sie via Signal upzudaten
:type checkbox: QCheckBox
"""
def _layer_add_to_instance(self, layer: QgsVectorLayer, disconnect: Callable[[], None], checkbox: QCheckBox):
print("_layer_add_to_instance")
if layer is None:
# Kein Layer mitgegeben → Abbruch
@ -644,48 +606,45 @@ class PoRunner(object):
# Signal zur Erkennung von Sichtbarkeitsänderungen verbinden
self._connect_layer_list_visibility_signal(layer, checkbox)
def _connect_layer_list_visibility_signal(self, layer, checkbox):
"""
Verbinde das Layer-Sichtbarkeit-Signal der QGIS-Layer-Liste mit unserer Layer-Referenz und unserer Checkbox
:param layer: Zu behandelnder Layer
:param layer: zu behandelnder Layer
:type layer: QgsVectorLayer
:param checkbox: Sichtbarkeit-Checkbox um sie via Signal upzudaten
:type checkbox: QCheckBox
"""
def _connect_layer_list_visibility_signal(self, layer, checkbox):
root = QgsProject.instance().layerTreeRoot()
node = root.findLayer(layer.id())
if node:
# Node existiert → Signal verbinden
node.visibilityChanged.connect(lambda: self._layer_set_visible_includingCheckbox(layer, node.isVisible(), checkbox))
print("_layer_add_to_instance: Layer-Sichtbarkeits-Signal verbunden!")
print("_layer_add_to_instance: Layer-Sichtbarkeit-Signal verbunden!")
else:
# Node nicht gefunden → ignorieren (Programmierfehler)
print("_layer_add_to_instance: Fehler: Node im Tree nicht gefunden!")
def _layer_set_visible_includingCheckbox(self, layer, visible: bool, checkbox: QCheckBox):
"""
Setzt die Sichtbarkeit des gegebenen Layers (inklusive gegebener Checkbox)
:param layer: Zu behandelnder Layer
:param layer: zu behandelnder Layer
:type layer: QgsVectorLayer
:param visible: Sichtbarkeit
:type visible: bool
:param checkbox: Sichtbarkeit-Checkbox um sie via Signal upzudaten
:type checkbox: QCheckBox
"""
def _layer_set_visible_includingCheckbox(self, layer, visible: bool, checkbox: QCheckBox):
checkbox.setChecked(visible)
self._layer_set_visible(layer, visible)
def _layer_update_labels(self, layer, fields):
"""
Aktualisiert die Labels eines Layers
:param layer: Zu behandelnder Layer
:param layer: zu behandelnder Layer
:type layer: QgsVectorLayer
:param fields: Anzuzeigende Felder (QGIS-Label-Expressions)
:type fields: list[str]
"""
def _layer_update_labels(self, layer, fields):
print("_layer_update_labels")
labeling = QgsVectorLayerSimpleLabeling(QgsPalLayerSettings())
@ -701,13 +660,12 @@ class PoRunner(object):
self._layer_refresh(layer)
def _layer_refresh(self, layer):
"""
Kümmert sich um das neuzeichnen eines Layers
:param layer: Zu behandelnder Layer
:param layer: zu behandelnder Layer
:type layer: QgsVectorLayer
"""
def _layer_refresh(self, layer):
print("_layerRefresh")
if self.iface.mapCanvas().isCachingEnabled():
layer.triggerRepaint()
@ -716,13 +674,12 @@ class PoRunner(object):
# layer selection ---------------------------------------------------------
def _layer_selection_changed(self):
"""
Wird aufgerufen wenn sich eine Feature-Markierung auf einem Layer geändert hat.
Öffnet das Graph-Widget mit gewünschtem Pegelverlauf,
Wird aufgerufen, wenn sich eine Feature-Markierung auf einem Layer geändert hat.
Öffnet das Graf-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():
@ -747,17 +704,16 @@ class PoRunner(object):
# layer styles ------------------------------------------------------------
def _layer_apply_style_per_category(self, layer, attribute_name, preset_fallback_file):
"""
Färbt die Features des gegebenen Layers kategorisch nach einem Attribut ein.
:param layer: Zu behandelnder Layer
: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...")
ramp = QgsStyle().defaultStyle().colorRamp("Turbo")
@ -790,15 +746,14 @@ class PoRunner(object):
layer.setRenderer(renderer)
self._layer_refresh(layer)
def _layer_apply_style_from_file(self, layer, preset_fallback_file):
"""
Wendet die gegebene Style-Datei (.qml) auf den gegebenen Layer an
:param layer: Zu behandelnder Layer
: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,))
@ -809,46 +764,41 @@ class PoRunner(object):
# graph signals ---------------------------------------------------------
"""
Verbindet alle GraphWidget-Signale mit der UI
"""
def _graph_connect_signals(self):
"""
Verbindet alle GrafWidget-Signale mit der UI
"""
print("_graph_connect_signals")
self.ui.slGraphStation.currentTextChanged.connect(self._slGraphStation_changed)
self.ui.btnGraphStationsRefresh.clicked.connect(self._btnGraphStationsRefresh_clicked)
self.ui.numGraphDays.valueChanged.connect(self._numGraphDays_changed)
self.ui.btnGraphLoad.clicked.connect(self._graph_load_graph)
"""
Behandelt die Stations-Änderung und Lädt den Pegelstandsverlauf-Graph neu
"""
def _slGraphStation_changed(self):
"""
Behandelt die Stations-Änderung und lädt den Pegelstandsverlauf-Graf neu
"""
print("_slGraphStation_changed: %s" % (self.ui.slGraphStation.currentText(),))
self._graph_load_graph()
def _btnGraphStationsRefresh_clicked(self):
"""
Klick auf Graph-Stations-Liste-Refresh:
Klick auf Graf-Stations-Liste-Refresh:
Lässt die Stations-Liste für den Graphen neu laden
"""
def _btnGraphStationsRefresh_clicked(self):
print("_btnGraphStationsRefresh_clicked")
self._graph_load_stations()
"""
Loggt lediglich die Graph-Tages-Änderung
"""
def _numGraphDays_changed(self):
"""
Loggt lediglich die Graf-Tages-Änderung
"""
print("_numGraphDays_changed: %s" % (self.ui.numGraphDays.value(),))
"""
Versucht den aktuell gewünschten Pegelstandsverlauf herunterzuladen und im GraphWidget anzuzeigen.
"""
def _graph_load_graph(self):
"""
Versucht den aktuell gewünschten Pegelstandsverlauf herunterzuladen und im GrafWidget anzuzeigen.
"""
print("_graph_load_graph")
if not self.ui.slGraphStation.isEnabled():
@ -859,47 +809,14 @@ class PoRunner(object):
station = self.ui.slGraphStation.currentText()
days = self.ui.numGraphDays.value()
print("_graph_load_graph: station=%s days=%s" % (station, days))
self.graph.lbGraph.clear()
self.graph.setWindowTitle("%s / %d Tag(e)" % (station, days))
self.graph.show()
if station == '' or station is None:
# Keine Station ausgewählt → Abbruch
print("_graph_load_graph: Fehler: Ungültige Station: %s" % (station,))
self.graph.lbGraph.setText("Bitte Station wählen...")
return
if days is None or days < 1 or days > 30:
# Ungültige Anzahl an Tagen ausgewählt → Abbruch
print("_graph_load_graph: Fehler: Ungültige Anzahl von Tagen: %s" % (days,))
self.graph.lbGraph.setText("Bitte Tage [1, 30] wählen...")
return
graph = PoGraphReader(station, days)
image_data = graph.download()
if image_data is None or len(image_data) == 0:
# Keine Bild-Daten beim Herunterladen erhalten → Abbruch
print("_graph_load_graph: Fehler: Keine Daten erhalten")
self.graph.lbGraph.setText("Fehler beim Download!")
return
pixmap = QtGui.QPixmap()
pixmap.loadFromData(image_data)
self.graph.lbGraph.setPixmap(pixmap)
self.graph.lbGraph.resize(pixmap.width(), pixmap.height())
print("_graph_load_graph: Bild erfolgreich geladen")
self.graph.load(station, days)
def _graph_load_stations(self):
"""
Lädt die Stations-Liste für den Graphen neu.
Sperrt so lange relevante UI-Elemente.
Versucht die bisher ausgewählten Station in der neuen Liste wiederzufinden.
"""
def _graph_load_stations(self):
print("_graph_load_stations")
self.ui.slGraphStation.setEnabled(False)
self.ui.btnGraphLoad.setEnabled(False)
@ -934,31 +851,33 @@ class PoRunner(object):
self.ui.slGraphStation.setEnabled(True)
self.ui.btnGraphLoad.setEnabled(True)
def _graphStation_set_by_shortname(self, shortname):
"""
Setzt eine Graph-Station anhand des gegebenen Kurznamens
Setzt eine Graf-Station anhand des gegebenen Kurznamens
:param shortname: Kurzname der Station die gesetzt werden soll
:type shortname: str
"""
def _graphStation_set_by_shortname(self, shortname):
index = self._graphStation_get_index_by_shortname(shortname)
if index is None:
# Station nicht gefunden → Abbruch
return False
self.ui.slGraphStation.setCurrentIndex(index)
return True
def _graphStation_get_index_by_shortname(self, shortname):
"""
Sucht den Index einer Graph-Station anhand des gegebenen Kurznamens
Sucht den Index einer Graf-Station anhand des gegebenen Kurznamens
:param shortname: Kurzname der Station die gesucht werden soll
:type shortname: str
"""
def _graphStation_get_index_by_shortname(self, shortname):
print("_graphStation_get_index_by_shortname: shortname=%s" % (shortname,))
for index in range(self.ui.slGraphStation.count()):
text = self.ui.slGraphStation.itemText(index)
if shortname == text:
# Station nicht gefunden → Abbruch
print("_graphStation_get_index_by_shortname: index=%d" % (index,))
return index
print("_graphStation_get_index_by_shortname: Nicht gefunden")
return None