10 Commits
main ... dev

13 changed files with 1247 additions and 30 deletions

BIN
assets/Linkliste.xlsx Normal file

Binary file not shown.

58
doc/Datenbank_ERD.md Normal file
View File

@@ -0,0 +1,58 @@
```mermaid
erDiagram
tbl_akteure{
Int4 fid PK
varchar Bezeichnung
varchar(6) vkz "Verfahren, für das der Akteur relevant ist"
}
tbl_konten{
varchar(3) kontonr
varchar bezeichnung
}
tbl_ausbauart{
int4 id PK
varchar Ausbauart_text
int4 preis
varchar(3) tbe_nr
varchar(3) Ausbauart_nr "Nr. der Ausbauart zur einfacheren Referenz"
}
tbl_Massnahme{
int4 MnNr
int4 Abschnitte FK
}
p41_Massnahmen_linie{
int4 id PK
geom Geometrie
varchar mnnr "Berechnet aus mn_konto und lfd_nr"
varchar mnname
int4 ausbauart FK "ref:tbl_ausbauart.id"
varchar(3) mn_konto FK "ref: tbl_konten.kontonr"
varchar(2) lfd_nr
varchar(1) tbe
bool umsetzung
int4 unterhalt_bisher FK "ref: tbl_akteure.id"
int4 unterhalt_zukuenftig FK "ref: tbl_akteure.id"
int4 bautraeger FK "ref: tbl_akteure.id"
int4 kostentraeger FK "ref: tbl_akteure.id"
int4 Planungsjahr
int4 baujahr
int4 Pflege_Anfang
int4 Pflege_Ende
float8 fahrbahnbreite
float8 gesamtbreite
varchar vorgesehene_regelungen
varchar bemerkungen
int4 laenge
int4 flaeche
varchar foerdersatz
bool ingenieurbauwerk
varchar bildpfad
bool fertiggestellt
int4 ausbauart_nr
bool plangenehmigt
}
tbl_konten ||--o{ p41_Massnahmen_linie : verwendet
tbl_akteure ||--o{ p41_Massnahmen_linie : verwendet
tbl_ausbauart ||--o{ p41_Massnahmen_linie : verwendet
```

39
main.py
View File

@@ -1,11 +1,17 @@
# sn_plan41/main.py
from qgis.utils import plugins from qgis.utils import plugins
from sn_basis.ui.dockmanager import DockManager from sn_basis.ui.dockmanager import DockManager
from .ui.dockwidget import DockWidget from sn_plan41.ui.dockwidget import DockWidget
from sn_basis.modules.DataGrabber import DataGrabber
from sn_basis.modules.Pruefmanager import Pruefmanager
class Plan41: class Plan41:
def __init__(self, iface): def __init__(self, iface):
self.iface = iface self.iface = iface
self.pruefmanager=Pruefmanager(ui_modus="qgis")
self.data_grabber=DataGrabber(pruefmanager=self.pruefmanager)
self.action = None self.action = None
self.dockwidget = None self.dockwidget = None
@@ -15,14 +21,17 @@ class Plan41:
def initGui(self): def initGui(self):
basis = plugins.get("sn_basis") basis = plugins.get("sn_basis")
if basis and basis.ui: if not basis or not getattr(basis, "ui", None):
self.action = basis.ui.add_action( return
self.plugin_name,
self.run, self.action = basis.ui.add_action(
tooltip=f"Öffnet {self.plugin_name}", self.plugin_name,
priority=20 self.run,
) tooltip=f"Öffnet {self.plugin_name}",
basis.ui.finalize_menu_and_toolbar() priority=20,
)
basis.ui.finalize_menu_and_toolbar()
print("Plan41/sn_Basis:initGui called")
def unload(self): def unload(self):
if self.dockwidget: if self.dockwidget:
@@ -32,13 +41,17 @@ class Plan41:
if self.action: if self.action:
basis = plugins.get("sn_basis") basis = plugins.get("sn_basis")
if basis and basis.ui: if basis and getattr(basis, "ui", None):
# Action aus Menü und Toolbar entfernen
basis.ui.remove_action(self.action) basis.ui.remove_action(self.action)
self.action = None self.action = None
def run(self): def run(self):
self.dockwidget = DockWidget(self.iface.mainWindow(), subtitle=self.plugin_name) self.dockwidget = DockWidget(
self.iface.mainWindow(),
subtitle=self.plugin_name,
pruefmanager=self.pruefmanager,
data_grabber=self.data_grabber)
self.dockwidget.setObjectName(self.dock_name) self.dockwidget.setObjectName(self.dock_name)
# Action-Referenz im Dock speichern # Action-Referenz im Dock speichern
@@ -48,5 +61,5 @@ class Plan41:
# Toolbar-Button als aktiv markieren # Toolbar-Button als aktiv markieren
basis = plugins.get("sn_basis") basis = plugins.get("sn_basis")
if basis and basis.ui: if basis and getattr(basis, "ui", None):
basis.ui.set_active_plugin(self.action) basis.ui.set_active_plugin(self.action)

137
modules/listenauswerter.py Normal file
View File

@@ -0,0 +1,137 @@
#sn_plan41/modules/listenauswerter.py
from typing import Any, Dict, List, Mapping, Optional, Tuple
from collections.abc import Mapping as _Mapping
# Prüfer-Typen (werden als Instanzen erwartet)
from sn_basis.modules.Pruefmanager import Pruefmanager # type: ignore
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
from sn_basis.modules.stilpruefer import Stilpruefer # type: ignore
class Listenauswerter:
"""
Validiert Zeilen aus einem DataDict, das vom DataGrabber stammt.
Erwartet wird die Struktur::
{"rows": [ {attr}, ... ]}
Die Linkprüfung entfällt vollständig, da der DataGrabber nur gültige
Links liefert. Diese Methode prüft ausschließlich die Konsistenz der
Zeilen mit dem erwarteten Datenschema und führt optional eine
Stilprüfung durch.
"""
def __init__(self, pruefmanager, stil_pruefer):
""" Parameters
----------
pruefmanager: Instanz des Pruefmanagers, der pruef_ergebnis verarbeitet.
stil_pruefer: Instanz des Stilpruefers, der Stildateien prüft.
"""
self.pruefmanager = pruefmanager
self.stil_pruefer = stil_pruefer
def validate_rows(
self,
data_dict: Dict[str, List[Mapping[str, Any]]]
) -> Tuple[Dict[str, List[Mapping[str, Any]]], List[Any]]:
"""
Validiert die Zeilen aus ``data_dict`` anhand des erwarteten Schemas.
Erwartete Felder pro Zeile
--------------------------
Pflichtfelder:
- ``ident``: eindeutige Kennung
- ``Link``: bereits geprüfter Link (vom DataGrabber garantiert gültig)
- ``Provider``: Datenquelle (wird in Großbuchstaben normalisiert)
Optionale Felder:
- ``Inhalt``: thematische Beschreibung
- ``Stildatei``: Pfad zur Stildatei (falls vorhanden)
Verhalten
---------
- Zeilen, denen Pflichtfelder fehlen oder deren Werte leer sind,
werden verworfen.
- ``Provider`` wird in Großbuchstaben normalisiert.
- Wenn ``Stildatei`` vorhanden ist, wird sie durch
``self.stil_pruefer.pruefe(...)`` geprüft.
- Bei OK bleibt der Wert erhalten.
- Bei nicht OK wird ``Stildatei`` auf ``None`` gesetzt und das
verarbeitete Prüfergebnis gesammelt.
- Alle Prüfergebnisse werden durch ``self.pruefmanager.verarbeite(...)``
geleitet und in der Rückgabe gesammelt.
Rückgabe
--------
Tuple[Dict[str, List[Mapping[str, Any]]]], List[Any]]
- ``valid_data_dict``: enthält nur Zeilen, die dem Schema entsprechen
- ``processed_results``: Liste der verarbeiteten Prüfergebnisse
Hinweise
--------
- Diese Methode führt **keine Linkprüfung** durch.
- Die Verantwortung für die Linkvalidität liegt vollständig beim DataGrabber.
- Die Methode verändert die Zeilen nur minimal (ProviderNormalisierung,
Stildatei ggf. auf ``None``).
"""
processed_results: List[Any] = []
valid_rows: List[Mapping[str, Any]] = []
# Grundstruktur prüfen
if not isinstance(data_dict, dict):
return {"rows": []}, processed_results
rows = data_dict.get("rows", [])
if not isinstance(rows, (list, tuple)):
return {"rows": []}, processed_results
for raw in rows:
# Sicherstellen, dass raw ein Mapping ist
if not isinstance(raw, _Mapping):
continue
ident = raw.get("ident")
inhalt = raw.get("Inhalt")
link = raw.get("Link")
stildatei = raw.get("Stildatei")
provider = raw.get("Provider")
# Pflichtfelder prüfen
if not ident or not link or not provider:
# Fehler dokumentieren
pe = pruef_ergebnis(
ok=False,
meldung="Pflichtfelder fehlen oder sind leer",
aktion="pflichtfelder_fehlen",
kontext=raw,
)
processed_results.append(self.pruefmanager.verarbeite(pe))
continue
# Provider normalisieren
provider_norm = str(provider).upper()
# Stildatei prüfen (falls vorhanden)
if stildatei:
pe_stil = self.stil_pruefer.pruefe(stildatei)
processed_stil = self.pruefmanager.verarbeite(pe_stil)
if not getattr(processed_stil, "ok", False):
processed_results.append(processed_stil)
stildatei_value: Optional[str] = None
else:
stildatei_value = stildatei
else:
stildatei_value = None
# Validierte Zeile zusammenbauen
validated_row = {
"ident": ident,
"Inhalt": inhalt,
"Link": link,
"Stildatei": stildatei_value,
"Provider": provider_norm,
}
valid_rows.append(validated_row)
result_dict = {"rows": valid_rows}
return result_dict, processed_results

1
tests/__init__.py Normal file
View File

@@ -0,0 +1 @@
#Testordner

148
tests/run_tests.py Normal file
View File

@@ -0,0 +1,148 @@
"""
sn_plan41/test/run_tests.py
Zentraler Test-Runner für sn_plan41.
Wrapper-konform, QGIS-unabhängig, CI- und IDE-fähig.
"""
import unittest
import datetime
import inspect
import os
import sys
from pathlib import Path
# ---------------------------------------------------------
# Plugin-Roots bestimmen
# ---------------------------------------------------------
THIS_FILE = Path(__file__).resolve()
# .../plugins/sn_plan41
SN_PLAN41_ROOT = THIS_FILE.parents[1]
# .../plugins/sn_basis
SN_BASIS_ROOT = SN_PLAN41_ROOT.parent / "sn_basis"
# ---------------------------------------------------------
# sys.path Bootstrap
# ---------------------------------------------------------
for path in (SN_PLAN41_ROOT, SN_BASIS_ROOT):
path_str = str(path)
if path_str not in sys.path:
sys.path.insert(0, path_str)
# ---------------------------------------------------------
# Farben
# ---------------------------------------------------------
RED = "\033[91m"
YELLOW = "\033[93m"
GREEN = "\033[92m"
CYAN = "\033[96m"
MAGENTA = "\033[95m"
RESET = "\033[0m"
GLOBAL_TEST_COUNTER = 0
# ---------------------------------------------------------
# Farbige TestResult-Klasse
# ---------------------------------------------------------
class ColoredTestResult(unittest.TextTestResult):
_last_test_class = None
def startTest(self, test):
global GLOBAL_TEST_COUNTER
GLOBAL_TEST_COUNTER += 1
self.stream.write(f"{CYAN}[Test {GLOBAL_TEST_COUNTER}]{RESET}\n")
super().startTest(test)
def startTestClass(self, test):
cls = test.__class__
file = inspect.getfile(cls)
filename = os.path.basename(file)
self.stream.write(
f"\n{MAGENTA}{'=' * 70}\n"
f"Starte Testklasse: {filename}{cls.__name__}\n"
f"{'=' * 70}{RESET}\n"
)
def addError(self, test, err):
super().addError(test, err)
self.stream.write(f"{RED}ERROR{RESET}\n")
def addFailure(self, test, err):
super().addFailure(test, err)
self.stream.write(f"{RED}FAILURE{RESET}\n")
def addSkip(self, test, reason):
super().addSkip(test, reason)
self.stream.write(f"{YELLOW}SKIPPED{RESET}: {reason}\n")
def addSuccess(self, test):
super().addSuccess(test)
self.stream.write(f"{GREEN}OK{RESET}\n")
# ---------------------------------------------------------
# Farbiger TestRunner
# ---------------------------------------------------------
class ColoredTestRunner(unittest.TextTestRunner):
def _makeResult(self):
result = ColoredTestResult(
self.stream,
self.descriptions,
self.verbosity,
)
original_start_test = result.startTest
def patched_start_test(test):
if result._last_test_class != test.__class__:
result.startTestClass(test)
result._last_test_class = test.__class__
original_start_test(test)
result.startTest = patched_start_test
return result
# ---------------------------------------------------------
# Testlauf starten
# ---------------------------------------------------------
def main():
print("\n" + "=" * 70)
print(
f"{CYAN}Testlauf gestartet am: "
f"{datetime.datetime.now():%Y-%m-%d %H:%M:%S}{RESET}"
)
print("=" * 70 + "\n")
loader = unittest.TestLoader()
TEST_ROOT = SN_PLAN41_ROOT / "tests"
suite = loader.discover(
start_dir=str(TEST_ROOT),
pattern="test_*.py",
top_level_dir=str(SN_PLAN41_ROOT.parent),
)
runner = ColoredTestRunner(verbosity=2)
result = runner.run(suite)
return 0 if result.wasSuccessful() else 1
if __name__ == "__main__":
raise SystemExit(main())

View File

@@ -0,0 +1,9 @@
@echo off
SET OSGEO4W_ROOT=D:\QGISQT5
call %OSGEO4W_ROOT%\bin\o4w_env.bat
set QGIS_PREFIX_PATH=%OSGEO4W_ROOT%\apps\qgis
set PYTHONPATH=%QGIS_PREFIX_PATH%\python;%PYTHONPATH%
set PATH=%OSGEO4W_ROOT%\bin;%QGIS_PREFIX_PATH%\bin;%PATH%
REM Neue Eingabeaufforderung starten und Python-Skript ausführen
start cmd /k "python run_tests.py"

153
tests/test_tab_a_logic.py Normal file
View File

@@ -0,0 +1,153 @@
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from sn_plan41.ui.tab_a_logic import TabALogic # type: ignore
from sn_basis.functions.variable_wrapper import get_variable # type: ignore
from sn_basis.functions.sys_wrapper import file_exists # type: ignore
class TestTabALogic(unittest.TestCase):
# -----------------------------------------------------
# 1. Verfahrens-DB setzen und laden
# -----------------------------------------------------
def test_verfahrens_db_set_and_load(self):
logic = TabALogic()
with TemporaryDirectory() as tmp:
db_path = Path(tmp) / "test.gpkg"
db_path.write_text("")
logic.set_verfahrens_db(str(db_path))
stored = get_variable("verfahrens_db", scope="project")
self.assertEqual(stored, str(db_path))
loaded = logic.load_verfahrens_db()
self.assertEqual(loaded, str(db_path))
# -----------------------------------------------------
# 2. Verfahrens-DB löschen
# -----------------------------------------------------
def test_verfahrens_db_clear(self):
logic = TabALogic()
logic.set_verfahrens_db(None)
stored = get_variable("verfahrens_db", scope="project")
self.assertEqual(stored, "")
# -----------------------------------------------------
# 3. Neue Verfahrens-DB anlegen
# -----------------------------------------------------
def test_create_new_verfahrens_db(self):
logic = TabALogic()
with TemporaryDirectory() as tmp:
db_path = Path(tmp) / "neu.gpkg"
result = logic.create_new_verfahrens_db(str(db_path))
self.assertTrue(result)
self.assertTrue(file_exists(db_path))
stored = get_variable("verfahrens_db", scope="project")
self.assertEqual(stored, str(db_path))
def test_create_new_verfahrens_db_with_none_path(self):
logic = TabALogic()
result = logic.create_new_verfahrens_db(None)
self.assertFalse(result)
# -----------------------------------------------------
# 4. Linkliste setzen und laden
# -----------------------------------------------------
def test_linkliste_set_and_load(self):
logic = TabALogic()
with TemporaryDirectory() as tmp:
link_path = Path(tmp) / "links.xlsx"
link_path.write_text("dummy")
logic.set_linkliste(str(link_path))
stored = get_variable("linkliste", scope="project")
self.assertEqual(stored, str(link_path))
loaded = logic.load_linkliste()
self.assertEqual(loaded, str(link_path))
# -----------------------------------------------------
# 5. Linkliste löschen
# -----------------------------------------------------
def test_linkliste_clear(self):
logic = TabALogic()
logic.set_linkliste(None)
stored = get_variable("linkliste", scope="project")
self.assertEqual(stored, "")
# -----------------------------------------------------
# 6. Layer-ID speichern
# -----------------------------------------------------
def test_verfahrensgebiet_layer_id_storage(self):
logic = TabALogic()
class MockLayer:
def id(self):
return "layer-123"
logic.save_verfahrensgebiet_layer(MockLayer())
stored = get_variable("verfahrensgebiet_layer", scope="project")
self.assertEqual(stored, "layer-123")
# -----------------------------------------------------
# 7. Ungültiger Layer wird ignoriert
# -----------------------------------------------------
def test_invalid_layer_is_rejected(self):
logic = TabALogic()
class InvalidLayer:
pass
logic.save_verfahrensgebiet_layer(InvalidLayer())
stored = get_variable("verfahrensgebiet_layer", scope="project")
self.assertEqual(stored, "")
#-----------------------------------------------------
# 8. Layer-ID wirft Exception
#----------------------------------------------------
def test_layer_id_raises_exception(self):
logic = TabALogic()
class BadLayer:
def id(self):
raise RuntimeError("boom")
logic.save_verfahrensgebiet_layer(BadLayer())
stored = get_variable("verfahrensgebiet_layer", scope="project")
self.assertEqual(stored, "")
# -----------------------------------------------------
# 11. Layer ID wird leer zurückgegeben
# -----------------------------------------------------
def test_layer_id_returns_empty(self):
logic = TabALogic()
class EmptyLayer:
def id(self):
return ""
logic.save_verfahrensgebiet_layer(EmptyLayer())
stored = get_variable("verfahrensgebiet_layer", scope="project")
self.assertEqual(stored, "")
if __name__ == "__main__":
unittest.main()

57
tests/test_tab_a_ui.py Normal file
View File

@@ -0,0 +1,57 @@
"""
Smoke-Tests für TabA UI (sn_plan41/ui/tab_a_ui.py)
Ziel:
- UI kann erstellt werden
- Callbacks crashen nicht
- Keine Qt-Verhaltensprüfung
"""
import unittest
from sn_plan41.ui.tab_a_ui import TabA #type:ignore
class TestTabAUI(unittest.TestCase):
# -----------------------------------------------------
# 1. UI kann erstellt werden
# -----------------------------------------------------
def test_tab_a_ui_can_be_created(self):
tab = TabA(parent=None,build_ui=False)
self.assertIsNotNone(tab)
self.assertEqual(tab.tab_title, "Daten")
# -----------------------------------------------------
# 2. Toggle-Callbacks crashen nicht
# -----------------------------------------------------
def test_tab_a_toggle_callbacks_do_not_crash(self):
tab = TabA(parent=None,build_ui=False)
tab._toggle_group(True)
tab._toggle_group(False)
tab._toggle_optional(True)
tab._toggle_optional(False)
# -----------------------------------------------------
# 3. Datei-Callbacks akzeptieren leere Eingaben
# -----------------------------------------------------
def test_tab_a_file_callbacks_accept_empty_input(self):
tab = TabA(parent=None,build_ui=False)
tab._on_verfahrens_db_changed("")
tab._on_linkliste_changed("")
# -----------------------------------------------------
# 4. Layer-Callback akzeptiert None
# -----------------------------------------------------
def test_tab_a_layer_callback_accepts_none(self):
tab = TabA(parent=None,build_ui=False)
tab._on_layer_changed(None)
if __name__ == "__main__":
unittest.main()

View File

@@ -1,7 +1,7 @@
from sn_basis.ui.tabs.settings_tab import SettingsTab from sn_basis.ui.tabs.settings_tab import SettingsTab
from sn_plan41.ui.tabs.tab_a import TabA from sn_plan41.ui.tab_a_ui import TabA
from sn_plan41.ui.tabs.tab_b import TabB #from sn_plan41.ui.tabs.tab_b import TabB
from sn_basis.ui.base_dockwidget import BaseDockWidget from sn_basis.ui.base_dockwidget import BaseDockWidget
class DockWidget(BaseDockWidget): class DockWidget(BaseDockWidget):
tabs = [TabA, TabB, SettingsTab] tabs = [TabA, SettingsTab]

158
ui/tab_a_logic.py Normal file
View File

@@ -0,0 +1,158 @@
"""
sn_plan41/ui/tab_a_logic.py Fachlogik für Tab A (Daten)
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, Tuple
from collections.abc import Mapping as _Mapping
from sn_basis.functions.variable_wrapper import ( # type: ignore
get_variable,
set_variable,
)
from sn_basis.functions.sys_wrapper import ( # type: ignore
file_exists,
write_text,
)
from sn_basis.functions.ly_existence_wrapper import layer_exists # type: ignore
from sn_basis.functions.ly_metadata_wrapper import get_layer_type # type: ignore
# Prüfer-Typen (werden als Instanzen erwartet)
from sn_basis.modules.Pruefmanager import Pruefmanager # type: ignore
from sn_basis.modules.linkpruefer import Linkpruefer # type: ignore
from sn_basis.modules.stilpruefer import Stilpruefer # type: ignore
# Typalias für Klarheit
Row = Dict[str, Any]
DataDict = Dict[str, List[Row]]
class TabALogic:
"""
Kapselt die komplette Logik von Tab A:
- Verfahrens-Datenbank
- optionale Linkliste
- Verfahrensgebiet-Layer
Diese Klasse erwartet beim Erzeugen Instanzen von Pruefmanager, Linkpruefer
und Stilpruefer. Die validate_and_filter_rows-Methode verwendet diese
Instanzen über self.
"""
def __init__(self, pruefmanager: Pruefmanager, link_pruefer: Linkpruefer, stil_pruefer: Stilpruefer) -> None:
"""
:param pruefmanager: Instanz des Pruefmanagers (für verarbeite/report)
:param link_pruefer: Instanz des Linkpruefers (mit methode pruefe)
:param stil_pruefer: Instanz des Stilpruefers (mit methode pruefe)
"""
self.pruefmanager = pruefmanager
self.link_pruefer = link_pruefer
self.stil_pruefer = stil_pruefer
# -------------------------------
# Verfahrens-Datenbank
# -------------------------------
def load_verfahrens_db(self) -> Optional[str]:
"""
Lädt die gespeicherte Verfahrens-Datenbank.
"""
path = get_variable("verfahrens_db", scope="project")
if path and file_exists(path):
return path
return None
def set_verfahrens_db(self, path: Optional[str]) -> None:
"""
Speichert oder löscht die Verfahrens-Datenbank.
"""
if path:
set_variable("verfahrens_db", path, scope="project")
else:
set_variable("verfahrens_db", "", scope="project")
def create_new_verfahrens_db(self, path: str) -> bool:
"""
Legt eine neue leere GPKG-Datei an.
"""
if not path:
return False
try:
write_text(path, "")
except Exception:
return False
self.set_verfahrens_db(path)
return True
# -------------------------------
# Lokale Linkliste
# -------------------------------
def load_linkliste(self) -> Optional[str]:
"""
Lädt die gespeicherte lokale Linkliste.
"""
path = get_variable("linkliste", scope="project")
if path and file_exists(path):
return path
return None
def set_linkliste(self, path: Optional[str]) -> None:
"""
Speichert oder löscht die lokale Linkliste.
"""
if path:
set_variable("linkliste", path, scope="project")
else:
set_variable("linkliste", "", scope="project")
# -------------------------------
# Verfahrensgebiet-Layer
# -------------------------------
def save_verfahrensgebiet_layer(self, layer) -> None:
"""
Speichert die ID des Verfahrensgebiet-Layers.
Ungültige Layer werden ignoriert.
"""
if layer is None:
set_variable("verfahrensgebiet_layer", "", scope="project")
return
if not hasattr(layer, "id") or not callable(layer.id):
set_variable("verfahrensgebiet_layer", "", scope="project")
return
try:
layer_id = layer.id()
except Exception:
set_variable("verfahrensgebiet_layer", "", scope="project")
return
if not layer_id:
set_variable("verfahrensgebiet_layer", "", scope="project")
return
set_variable("verfahrensgebiet_layer", layer_id, scope="project")
def load_verfahrensgebiet_layer_id(self) -> Optional[str]:
"""
Lädt die gespeicherte Layer-ID.
"""
value = get_variable("verfahrensgebiet_layer", scope="project")
return value or None
def is_valid_verfahrensgebiet_layer(self, layer) -> bool:
"""
Prüft, ob ein Layer als Verfahrensgebiet geeignet ist.
"""
if not layer_exists(layer):
return False
layer_type = get_layer_type(layer)
return layer_type == "vector"

495
ui/tab_a_ui.py Normal file
View File

@@ -0,0 +1,495 @@
# sn_plan41/ui/tab_a_ui.py UI für Tab A (Daten)
from __future__ import annotations
from typing import Optional
from sn_basis.functions.qt_wrapper import (
QWidget,
QVBoxLayout,
QLabel,
QPushButton,
QToolButton,
QFileDialog,
QMessageBox,
ToolButtonTextBesideIcon,
ArrowDown,
ArrowRight,
SizePolicyPreferred,
SizePolicyMaximum,
QComboBox,
)
from sn_basis.functions.qgisui_wrapper import QgsFileWidget, QgsMapLayerComboBox
from sn_basis.functions.qgiscore_wrapper import QgsProject, QgsMapLayerProxyModel
from sn_basis.functions.variable_wrapper import get_variable, set_variable
from sn_plan41.ui.tab_a_logic import TabALogic
# PrüfWorkflow / DataGrabber (werden zur Laufzeit vom Pruefmanager/Pruefern verwendet)
from sn_basis.modules.Dateipruefer import Dateipruefer
from sn_basis.modules.Pruefmanager import Pruefmanager
from sn_basis.modules.DataGrabber import DataGrabber
from sn_basis.modules.linkpruefer import Linkpruefer
from sn_basis.modules.stilpruefer import Stilpruefer
from sn_basis.modules.Datenschreiber import Datenschreiber
# Raumfilter-Optionen
RAUMFILTER_VAR = "Raumfilter"
RAUMFILTER_OPTIONS = ("Verfahrensgebiet", "Pufferlayer", "ohne")
RAUMFILTER_DEFAULT = "Pufferlayer"
pm = Pruefmanager(ui_modus="qgis")
lp = Linkpruefer()
sp = Stilpruefer()
class TabA(QWidget):
"""
UI-Klasse für Tab A (Daten).
Diese bereinigte Version enthält ausschließlich UI-Elemente und
einfache, nicht-validierende Callback-Handler. Alle fachlichen Prüfungen
und Fehlerbehandlungen werden zur Laufzeit vom Pruefmanager und den Prüfern
übernommen.
"""
tab_title = "Daten"
def __init__(self, parent=None, pruefmanager=None, link_pruefer=None, stil_pruefer=None, build_ui=True):
super().__init__(parent)
self.parent = parent
self.tab_title = "Daten"
# Logik-Adapter (TabALogic verwaltet persistente Projektvariablen)
self.logic = TabALogic(pruefmanager=pruefmanager, link_pruefer=link_pruefer, stil_pruefer=stil_pruefer)
# Prüfmanager-Instanz (UI-Modus wird zur Laufzeit vom Pruefmanager gehandhabt)
self.pruefmanager = Pruefmanager(ui_modus="qgis")
# DataGrabber-Instanz (synchroner Aufruf; Prüfungen übernimmt Pruefmanager/Pruefer)
# Hinweis: DataGrabber erwartet ggf. Prüfer-Objekte; hier werden sie nicht übergeben,
# da TabALogic / Pruefmanager diese zur Laufzeit bereitstellen können.
self.data_grabber = DataGrabber(pruefmanager=self.pruefmanager)
# Platzhalter, die vom Plugin oder Nutzer gesetzt werden können
self._attributes_list = [] # optionale Attributliste (z. B. Excel-Import)
self._pufferlayer = None # optionaler Layer (Verfahrensgebiet)
self.verfahrens_db: Optional[str] = None
self.lokale_linkliste: Optional[str] = None
# UI-Widget-Referenz für Raumfilter
self._raumfilter_combo: Optional[QComboBox] = None
if build_ui:
self._build_ui()
self._restore_state()
# ---------------------------------------------------------
# UI-Aufbau
# ---------------------------------------------------------
def _build_ui(self) -> None:
main_layout = QVBoxLayout()
main_layout.setSpacing(4)
main_layout.setContentsMargins(4, 4, 4, 4)
# Verfahrens-Datenbank Gruppe
self.group_button = QToolButton()
self.group_button.setText("Verfahrens-Datenbank")
self.group_button.setCheckable(True)
self.group_button.setChecked(True)
self.group_button.setToolButtonStyle(ToolButtonTextBesideIcon)
self.group_button.setArrowType(ArrowDown)
self.group_button.setStyleSheet("font-weight: bold;")
self.group_button.toggled.connect(self._toggle_group)
main_layout.addWidget(self.group_button)
self.group_content = QWidget()
self.group_content.setSizePolicy(SizePolicyPreferred, SizePolicyMaximum)
group_layout = QVBoxLayout()
group_layout.setSpacing(2)
group_layout.setContentsMargins(10, 4, 4, 4)
group_layout.addWidget(QLabel("bestehende Datei auswählen"))
self.file_widget = QgsFileWidget()
self.file_widget.setStorageMode(QgsFileWidget.GetFile)
self.file_widget.setFilter("Geopackage (*.gpkg)")
self.file_widget.fileChanged.connect(self._on_verfahrens_db_changed)
group_layout.addWidget(self.file_widget)
group_layout.addWidget(QLabel("-oder-"))
self.btn_new = QPushButton("Neue Verfahrens-DB anlegen")
self.btn_new.clicked.connect(self._create_new_gpkg)
group_layout.addWidget(self.btn_new)
self.group_content.setLayout(group_layout)
main_layout.addWidget(self.group_content)
# Optionale Linkliste
self.optional_button = QToolButton()
self.optional_button.setText("Optional: Lokale Linkliste")
self.optional_button.setCheckable(True)
self.optional_button.setChecked(False)
self.optional_button.setToolButtonStyle(ToolButtonTextBesideIcon)
self.optional_button.setArrowType(ArrowRight)
self.optional_button.setStyleSheet("font-weight: bold; margin-top: 6px;")
self.optional_button.toggled.connect(self._toggle_optional)
main_layout.addWidget(self.optional_button)
self.optional_content = QWidget()
self.optional_content.setSizePolicy(SizePolicyPreferred, SizePolicyMaximum)
optional_layout = QVBoxLayout()
optional_layout.setSpacing(2)
optional_layout.setContentsMargins(10, 4, 4, 20)
optional_layout.addWidget(QLabel("(frei lassen für globale Linkliste)"))
self.linkliste_widget = QgsFileWidget()
self.linkliste_widget.setStorageMode(QgsFileWidget.GetFile)
self.linkliste_widget.setFilter("Excelliste (*.xlsx)")
self.linkliste_widget.fileChanged.connect(self._on_linkliste_changed)
optional_layout.addWidget(self.linkliste_widget)
self.optional_content.setLayout(optional_layout)
self.optional_content.setVisible(False)
main_layout.addWidget(self.optional_content)
# Layer-Auswahl
layer_label = QLabel("Verfahrensgebiet-Layer auswählen")
layer_label.setStyleSheet("font-weight: bold; margin-top: 6px;")
main_layout.addWidget(layer_label)
self.layer_combo = QgsMapLayerComboBox()
self.layer_combo.setSizePolicy(SizePolicyPreferred, SizePolicyMaximum)
self.layer_combo.setFilters(QgsMapLayerProxyModel.VectorLayer)
self.layer_combo.layerChanged.connect(self._on_layer_changed)
main_layout.addWidget(self.layer_combo)
# Raumfilter-Label + ComboBox (unterhalb der Layer-Auswahl)
main_layout.addWidget(QLabel("Raumfilter"))
self._raumfilter_combo = QComboBox(self)
# Fülle Optionen (Wrapper stellt addItems bereit)
try:
self._raumfilter_combo.addItems(list(RAUMFILTER_OPTIONS))
except Exception:
# fallback: iterativ hinzufügen, falls Wrapper andere API hat
for opt in RAUMFILTER_OPTIONS:
if hasattr(self._raumfilter_combo, "addItem"):
self._raumfilter_combo.addItem(opt)
# Initialisiere Auswahl aus Projekt-Variable oder Default
stored = get_variable(RAUMFILTER_VAR, scope="project")
if isinstance(stored, str) and stored in RAUMFILTER_OPTIONS:
try:
self._raumfilter_combo.setCurrentText(stored)
except Exception:
try:
idx = self._raumfilter_combo.findText(stored)
if idx is not None and idx >= 0:
self._raumfilter_combo.setCurrentIndex(idx)
except Exception:
pass
else:
try:
self._raumfilter_combo.setCurrentText(RAUMFILTER_DEFAULT)
except Exception:
try:
idx = self._raumfilter_combo.findText(RAUMFILTER_DEFAULT)
if idx is not None and idx >= 0:
self._raumfilter_combo.setCurrentIndex(idx)
except Exception:
pass
# persistiere Default, falls noch kein Wert gesetzt
if not stored:
set_variable(RAUMFILTER_VAR, RAUMFILTER_DEFAULT, scope="project")
# Signal: bei Änderung Variable setzen
try:
self._raumfilter_combo.currentTextChanged.connect(self._on_raumfilter_changed)
except Exception:
try:
self._raumfilter_combo.current_text_changed.connect(self._on_raumfilter_changed)
except Exception:
pass
main_layout.addWidget(self._raumfilter_combo)
# Neuer Button direkt unterhalb der Raumfilter-Combo: "Fachdaten laden"
self.btn_pipeline = QPushButton("Fachdaten laden")
self.btn_pipeline.setToolTip("Starte Pipeline: Linkliste → DataGrabber → Datenschreiber → Log")
self.btn_pipeline.clicked.connect(self._on_run_pipeline)
main_layout.addWidget(self.btn_pipeline)
# (Optional) bestehender Button weiter unten für alternative Platzierung
self.btn_load = QPushButton("Fachdaten laden (alt)")
self.btn_load.clicked.connect(self._on_load_fachdaten)
main_layout.addWidget(self.btn_load)
main_layout.addStretch(1)
self.setLayout(main_layout)
# ---------------------------------------------------------
# State Restore (UI-Wiederherstellung ohne Prüfungen)
# ---------------------------------------------------------
def _restore_state(self) -> None:
db = self.logic.load_verfahrens_db()
if db:
self.verfahrens_db = db
try:
self.file_widget.setFilePath(db)
except Exception:
pass
self._update_group_color()
link = self.logic.load_linkliste()
if link:
self.lokale_linkliste = link
try:
self.linkliste_widget.setFilePath(link)
except Exception:
pass
layer_id = self.logic.load_verfahrensgebiet_layer_id()
if layer_id:
layer = QgsProject.instance().mapLayer(layer_id)
if layer:
self.layer_combo.setLayer(layer)
# Raumfilter aus Variable wiederherstellen (falls Combo existiert)
try:
stored = get_variable(RAUMFILTER_VAR, scope="project")
if stored and self._raumfilter_combo is not None:
try:
self._raumfilter_combo.setCurrentText(stored)
except Exception:
idx = self._raumfilter_combo.findText(stored)
if idx is not None and idx >= 0:
self._raumfilter_combo.setCurrentIndex(idx)
except Exception:
pass
# ---------------------------------------------------------
# UI-Callbacks (ohne Prüfungen / Exceptions)
# ---------------------------------------------------------
def _toggle_group(self, checked: bool) -> None:
self.group_button.setArrowType(ArrowDown if checked else ArrowRight)
self.group_content.setVisible(checked)
def _toggle_optional(self, checked: bool) -> None:
self.optional_button.setArrowType(ArrowDown if checked else ArrowRight)
self.optional_content.setVisible(checked)
def _on_verfahrens_db_changed(self, path: str) -> None:
self.verfahrens_db = path
self.logic.set_verfahrens_db(path)
self._update_group_color()
def _on_linkliste_changed(self, path: str) -> None:
self.lokale_linkliste = path
self.logic.set_linkliste(path)
def _on_layer_changed(self, layer) -> None:
self.logic.save_verfahrensgebiet_layer(layer)
self._pufferlayer = layer
def _create_new_gpkg(self) -> None:
file_path, _ = QFileDialog.getSaveFileName(
self,
"Neue Verfahrens-Datenbank anlegen",
"",
"Geopackage (*.gpkg)",
)
if not file_path:
return
if not file_path.lower().endswith(".gpkg"):
file_path += ".gpkg"
# Delegation an TabALogic; TabALogic / Pruefmanager übernehmen Prüfungen
self.logic.create_new_verfahrens_db(file_path)
self.verfahrens_db = file_path
try:
self.file_widget.setFilePath(file_path)
except Exception:
pass
self._update_group_color()
def _on_load_fachdaten(self) -> None:
"""
Bestehender, kompakter Handler für 'Fachdaten laden'.
Führt Dateiprüfung und DataGrabber.run aus (wie zuvor).
"""
pfad = self.file_widget.filePath()
# Dateipruefer wird zur Laufzeit verwendet; hier nur der Aufruf
pruefer = Dateipruefer(pfad=pfad, temporaer_erlaubt=True)
ergebnis = pruefer.pruefe()
ergebnis = self.pruefmanager.verarbeite(ergebnis)
zielpfad = None
if ergebnis.kontext is not None:
try:
zielpfad = str(ergebnis.kontext)
except Exception:
zielpfad = ergebnis.kontext
# DataGrabber.run wird wie bisher aufgerufen; Signatur kann variieren.
# Wir übergeben die bekannten Parameter; DataGrabber ist verantwortlich,
# die Linkliste intern zu verwenden (z. B. aus TabALogic oder über Argumente).
try:
self.data_grabber.run(
attributes_list=self._attributes_list,
pufferlayer=self._pufferlayer,
zielpfad=zielpfad,
temporaer=(ergebnis.aktion == "temporaer_erzeugen"),
temporaer_erlaubt=True,
)
except Exception:
# Fehler werden vom Pruefmanager / DataGrabber protokolliert
pass
def _on_run_pipeline(self) -> None:
"""
Neuer, vollständiger Pipeline-Handler, der:
- Dateiprüfung (Verfahrens-DB)
- DataGrabber-Ausführung (mit Linkliste)
- Datenschreiber (schreiben, laden)
- Logschreiber (Log-Datei)
ausführt und Ergebnisse über den Pruefmanager protokolliert.
"""
# 1) Verfahrens-DB prüfen / ermitteln
pfad = self.file_widget.filePath()
pruefer = Dateipruefer(pfad=pfad, temporaer_erlaubt=True)
ergebnis = pruefer.pruefe()
ergebnis = self.pruefmanager.verarbeite(ergebnis)
zielpfad = None
if ergebnis.kontext is not None:
try:
zielpfad = str(ergebnis.kontext)
except Exception:
zielpfad = ergebnis.kontext
if not zielpfad:
# Falls kein Zielpfad ermittelt werden konnte, protokollieren und abbrechen
pe_err = pruef_ergebnis(
ok=False,
meldung="Kein gültiger Speicherort für Verfahrens-DB ermittelt; Pipeline abgebrochen.",
aktion="kein_dateipfad",
kontext={},
)
self.pruefmanager.verarbeite(pe_err)
return
# 2) DataGrabber ausführen
# Erwartung: DataGrabber.run gibt (daten_dict, processed_results) zurück.
# Falls die konkrete Implementierung anders ist, passt dieser Aufruf entsprechend an.
try:
run_result = self.data_grabber.run(
attributes_list=self._attributes_list,
pufferlayer=self._pufferlayer,
zielpfad=zielpfad,
temporaer=(ergebnis.aktion == "temporaer_erzeugen"),
temporaer_erlaubt=True,
)
except Exception as exc:
pe_err = pruef_ergebnis(
ok=False,
meldung=f"DataGrabber-Fehler: {exc}",
aktion="datenabruf",
kontext={},
)
self.pruefmanager.verarbeite(pe_err)
return
# Normalisiere Rückgabe: unterstütze sowohl None, einzelnes dict oder Tuple
daten_dict = {}
processed_results = []
if isinstance(run_result, tuple) and len(run_result) >= 2:
daten_dict, processed_results = run_result[0], run_result[1]
elif isinstance(run_result, dict) and "daten" in run_result:
daten_dict = run_result
# processed_results bleiben leer oder werden vom DataGrabber intern protokolliert
else:
# Wenn run() nichts zurückgibt, versuchen wir, auf DataGrabber intern gespeicherte Ergebnisse zuzugreifen
daten_dict = getattr(self.data_grabber, "last_daten_dict", {}) or {}
processed_results = getattr(self.data_grabber, "last_processed_results", []) or []
# 3) Datenschreiber: Daten in GPKG schreiben
try:
ds = Datenschreiber(pruefmanager=self.pruefmanager, gpkg_path=zielpfad)
layer_infos = ds.schreibe_Daten(daten_dict=daten_dict, processed_results=processed_results, speicherort=zielpfad)
except Exception as exc:
pe_err = pruef_ergebnis(
ok=False,
meldung=f"Fehler beim Schreiben der Daten: {exc}",
aktion="save_exception",
kontext={},
)
self.pruefmanager.verarbeite(pe_err)
return
# 4) Layer laden und Stile anwenden
try:
ds.lade_Layer(layer_infos)
except Exception as exc:
pe_warn = pruef_ergebnis(
ok=True,
meldung=f"Fehler beim Laden der Layer: {exc}",
aktion="layer_nicht_gefunden",
kontext={},
)
self.pruefmanager.verarbeite(pe_warn)
# 5) Log schreiben
try:
log_path = ds.schreibe_log(processed_results=processed_results, speicherort=zielpfad)
# Optional: zeige Erfolgsmeldung
try:
QMessageBox.information(self, "Pipeline abgeschlossen", f"Pipeline erfolgreich abgeschlossen.\nLog: {log_path}")
except Exception:
pass
except Exception as exc:
pe_warn = pruef_ergebnis(
ok=True,
meldung=f"Log konnte nicht geschrieben werden: {exc}",
aktion="standarddatei_vorschlagen",
kontext={},
)
self.pruefmanager.verarbeite(pe_warn)
# ---------------------------------------------------------
# Raumfilter Callback
# ---------------------------------------------------------
def _on_raumfilter_changed(self, value: str) -> None:
# Persistiere Auswahl in Projekt-Variable; Prüfungen übernimmt die Laufzeitlogik
set_variable(RAUMFILTER_VAR, value, scope="project")
# ---------------------------------------------------------
# UI-Helfer
# ---------------------------------------------------------
def _prompt_user_to_select_file(self) -> None:
fname, _ = QFileDialog.getOpenFileName(
self,
"Verfahrens-DB auswählen",
"",
"Geopackage (*.gpkg)",
)
if fname:
try:
self.file_widget.setFilePath(fname)
except Exception:
try:
self.file_widget.setFileName(fname)
except Exception:
self.file_widget.setProperty("filePath", fname)
self.verfahrens_db = fname
self.logic.set_verfahrens_db(fname)
self._update_group_color()
def _update_group_color(self) -> None:
if self.verfahrens_db:
self.group_button.setStyleSheet("font-weight: bold;")
else:
self.group_button.setStyleSheet("")

View File

@@ -1,12 +0,0 @@
from qgis.PyQt.QtWidgets import QWidget, QVBoxLayout, QLabel, QLineEdit
class TabA(QWidget):
tab_title = "Tab A"
def __init__(self, parent=None):
super().__init__(parent)
layout = QVBoxLayout()
layout.addWidget(QLabel("Plugin2 Tab A"))
layout.addWidget(QLineEdit("Feld A1"))
layout.addWidget(QLineEdit("Feld A2"))
self.setLayout(layout)