Compare commits

...
Author SHA1 Message Date
Daniel ab51838db5 Merge pull request 'test überarbeitet' (#24) from feature/Tests_erweitert into unstable
Release Plugin / release (push) Successful in 3s
Reviewed-on: #24
2026-04-22 09:22:49 +02:00
Daniel e3e858233d Merge branch 'unstable' of https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Plan41 into feature/Tests_erweitert 2026-04-22 09:21:18 +02:00
release-bot cc2d5c6200 Release 26.4.2-unstable 2026-04-22 07:07:17 +00:00
Daniel b2ee167132 QgsmaxVersion eingefügt
Release Plugin / release (push) Successful in 4s
2026-04-22 09:03:54 +02:00
Daniel 600a3c7aff test überarbeitet 2026-04-20 15:36:29 +02:00
Daniel 8f9cf4c9fb pdoc-dockstrings ergänzt 2026-04-20 14:31:49 +02:00
release-bot 18573cb3be Release 26.4.2-testing 2026-04-10 09:32:49 +00:00
Daniel 6b0d94c54e Merge pull request 'Druckfunktion nach sn_basis verschben, Bugfixes' (#22) from unstable into testing
Release Plugin / release (push) Successful in 3s
Reviewed-on: #22
2026-04-10 11:30:03 +02:00
Daniel a9f65893f0 Merge pull request 'log auf html umgestellt, meldebanner entfernt' (#21) from bug/Log_bereinigen into unstable
Reviewed-on: #21
2026-04-10 07:09:40 +02:00
Daniel dcaca4c883 log auf html umgestellt, meldebanner entfernt 2026-04-10 07:07:21 +02:00
release-bot dbd5fca40b Release 26.04.1-unstable 2026-04-08 12:39:37 +00:00
Daniel ecb1e49d8d Merge pull request 'verfahrensgebiet-Vaiable wird sauber gesetzt' (#20) from bug/Verfahrensgebiet_nicht_übernommen into unstable
Release Plugin / release (push) Successful in 4s
Reviewed-on: #20
2026-04-08 14:38:04 +02:00
Daniel ad6c7ea3e1 verfahrensgebiet-Vaiable wird sauber gesetzt 2026-04-08 14:36:30 +02:00
release-bot 1bb425e419 Release 26.3.3-unstable 2026-03-27 10:28:56 +00:00
Daniel b8f36bab7a Merge pull request 'Verfahrensgebiet-Layerauswahl nach sn_verfahrensgebiet verschoben' (#19) from refactor/Verfahrensgebiet-Auswahl_nach_sn_verfahrensgebiet into unstable
Release Plugin / release (push) Successful in 3s
Reviewed-on: #19
2026-03-27 11:26:52 +01:00
Daniel aacba1e5eb Verfahrensgebiet-Layerauswahl nach sn_verfahrensgebiet verschoben 2026-03-27 11:19:37 +01:00
release-bot 9efeff5aff Release 26.3.2-unstable 2026-03-26 20:47:58 +00:00
Daniel dd74f1e98f Merge pull request 'refactor/Druckfunktion-in-sn_basis' (#18) from refactor/Druckfunktion-in-sn_basis into unstable
Release Plugin / release (push) Successful in 3s
Reviewed-on: #18
2026-03-26 21:42:05 +01:00
Daniel 7cd71ba366 print-tab nach sn_basis verschoben 2026-03-26 20:17:45 +01:00
release-bot 927da6fe29 Release 26.3.1-testing 2026-03-25 10:44:55 +00:00
Daniel 44972dc9e0 changelog.txt aktualisiert
Release Plugin / release (push) Successful in 4s
2026-03-25 11:43:07 +01:00
17 changed files with 666 additions and 2567 deletions
+14
View File
@@ -1,3 +1,17 @@
"""sn_plan41 QGIS-Plugin für thematische Linklisten und Datenquellen.
Verwaltet Excel-basierte Link-Listen und Fachdatenabruf.
"""
def classFactory(iface):
"""Factory-Funktion für QGIS.
Wird von QGIS aufgerufen, um die Plugin-Instanz zu erstellen.
Args:
iface: QGIS-Schnittstellenobjekt
Returns:
Plan41: Instanziiertes Plugin
"""
from .main import Plan41
return Plan41(iface)
+11
View File
@@ -1,3 +1,14 @@
---
Version 26.4.2-testing:
---
Version 26.04.1-unstable:
---
Version 26.3.3-unstable:
---
Version 26.3.2-unstable:
---
Version 26.3.1-testing:
- unstable als testing veröffentlicht
---
Version 26.3.1-unstable:
- erster Test als echtes PlugIn
+6 -1
View File
@@ -1,4 +1,6 @@
# sn_plan41/main.py
"""sn_plan41/main.py Einstiegspunkt des sn_plan41 Plugins.
Orchestriert die UI und den Datenabruf für Thematische Linklisten.
"""
from qgis.utils import plugins
from sn_basis.ui.dockmanager import DockManager
@@ -8,6 +10,9 @@ from sn_basis.modules.Pruefmanager import Pruefmanager
class Plan41:
"""Einstiegspunkt des sn_plan41-Plugins.
Verwaltet die DockWidget-Initialisierung und den Datenflusss.
"""
def __init__(self, iface):
self.iface = iface
self.pruefmanager=Pruefmanager(ui_modus="qgis")
+2 -2
View File
@@ -1,12 +1,12 @@
[general]
version=26.3.1-unstable
version=26.4.2-unstable
name=LNO Sachsen | Plan41
description=Plugin zum Erzeugen der Pläne nach §38 und §41
author=Daniel Helbig, Michael Otto
homepage=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Plan41
tracker=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_SN_Plan41/issues
repository=https://entwicklung.flurneuordnung-sachsen.de/AG_QGIS/Plugin_Test_Action/src/branch/unstable/
QGISMaximumVersion=4.99
qgisMinimumVersion=3.40
experimental=true
deprecated=false
+4 -1
View File
@@ -1,4 +1,7 @@
#sn_plan41/modules/listenauswerter.py
"""sn_plan41/modules/listenauswerter.py Validierung von Linklisten.
Prüft Zeilen aus DataDict auf Vollständigkeit und Konsistenz.
"""
from typing import Any, Dict, List, Mapping, Optional, Tuple
from collections.abc import Mapping as _Mapping
# Prüfer-Typen (werden als Instanzen erwartet)
+3 -1
View File
@@ -1 +1,3 @@
#Testordner
"""sn_plan41/tests Test-Suite für das sn_plan41 Plugin.
Enthält Unit-Tests und Integrationstests für Tab-Logik und Datengrabber.
"""
+84
View File
@@ -0,0 +1,84 @@
import unittest
from typing import Any
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
from sn_plan41.modules.listenauswerter import Listenauswerter
class _DummyPruefmanager:
def __init__(self) -> None:
self.seen = []
def verarbeite(self, ergebnis: Any):
self.seen.append(ergebnis)
return ergebnis
class _DummyStilPruefer:
def __init__(self, ok: bool = True) -> None:
self.ok = ok
def pruefe(self, _path: str) -> pruef_ergebnis:
if self.ok:
return pruef_ergebnis(ok=True, meldung="ok", aktion="ok", kontext=None)
return pruef_ergebnis(ok=False, meldung="stil fehler", aktion="falsche_endung", kontext=None)
class TestListenauswerter(unittest.TestCase):
def test_validate_rows_accepts_valid_row_and_normalizes_provider(self):
manager = _DummyPruefmanager()
validator = Listenauswerter(manager, _DummyStilPruefer(ok=True))
data = {"rows": [{"ident": "A1", "Inhalt": "Test", "Link": "http://x", "Provider": "wfs"}]}
validated, processed = validator.validate_rows(data)
self.assertEqual(len(validated["rows"]), 1)
self.assertEqual(validated["rows"][0]["Provider"], "WFS")
self.assertEqual(processed, [])
def test_validate_rows_rejects_missing_required_fields(self):
manager = _DummyPruefmanager()
validator = Listenauswerter(manager, _DummyStilPruefer(ok=True))
data = {"rows": [{"ident": "", "Link": "", "Provider": ""}]}
validated, processed = validator.validate_rows(data)
self.assertEqual(validated["rows"], [])
self.assertEqual(len(processed), 1)
self.assertEqual(processed[0].aktion, "pflichtfelder_fehlen")
def test_validate_rows_sets_stildatei_none_on_style_failure(self):
manager = _DummyPruefmanager()
validator = Listenauswerter(manager, _DummyStilPruefer(ok=False))
data = {
"rows": [
{
"ident": "A1",
"Inhalt": "Thema",
"Link": "http://x",
"Provider": "WMS",
"Stildatei": "broken.qml",
}
]
}
validated, processed = validator.validate_rows(data)
self.assertEqual(len(validated["rows"]), 1)
self.assertIsNone(validated["rows"][0]["stildatei"])
self.assertEqual(len(processed), 1)
self.assertEqual(processed[0].aktion, "falsche_endung")
def test_validate_rows_handles_non_mapping_rows_edge_case(self):
manager = _DummyPruefmanager()
validator = Listenauswerter(manager, _DummyStilPruefer(ok=True))
validated, processed = validator.validate_rows({"rows": ["invalid", 123, None]})
self.assertEqual(validated, {"rows": []})
self.assertEqual(processed, [])
if __name__ == "__main__":
unittest.main()
+78 -121
View File
@@ -1,152 +1,109 @@
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from typing import Any, cast
from unittest.mock import patch
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
from sn_basis.modules.pruef_ergebnis import pruef_ergebnis
from sn_basis.functions.variable_wrapper import get_variable
from sn_plan41.ui.tab_a_logic import TabALogic
class _DummyPruefmanager:
def verarbeite(self, ergebnis):
return ergebnis
class TestTabALogic(unittest.TestCase):
# -----------------------------------------------------
# 1. Verfahrens-DB setzen und laden
# -----------------------------------------------------
def test_verfahrens_db_set_and_load(self):
logic = TabALogic()
def _make_logic(self) -> TabALogic:
return TabALogic(
pruefmanager=cast(Any, _DummyPruefmanager()),
link_pruefer=cast(Any, object()),
stil_pruefer=cast(Any, object()),
)
with TemporaryDirectory() as tmp:
db_path = Path(tmp) / "test.gpkg"
db_path.write_text("")
def test_verfahrens_db_set_and_load_roundtrip(self):
logic = self._make_logic()
logic.set_verfahrens_db("C:/tmp/v_db.gpkg")
self.assertEqual(logic.load_verfahrens_db(), "C:/tmp/v_db.gpkg")
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 = self._make_logic()
logic.set_verfahrens_db(None)
self.assertEqual(get_variable("verfahrens_db", scope="project"), "")
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()
def test_linkliste_set_and_load_existing_path(self):
logic = self._make_logic()
with TemporaryDirectory() as tmp:
link_path = Path(tmp) / "links.xlsx"
link_path.write_text("dummy")
link_path.write_text("x", encoding="utf-8")
logic.set_linkliste(str(link_path))
self.assertEqual(logic.load_linkliste(), str(link_path))
stored = get_variable("linkliste", scope="project")
self.assertEqual(stored, str(link_path))
def test_linkliste_load_returns_none_when_file_missing(self):
logic = self._make_logic()
logic.set_linkliste("C:/tmp/missing.xlsx")
self.assertIsNone(logic.load_linkliste())
loaded = logic.load_linkliste()
self.assertEqual(loaded, str(link_path))
@patch("sn_plan41.ui.tab_a_logic.get_plugin_root", return_value="C:/plugins")
@patch("sn_plan41.ui.tab_a_logic.join_path", return_value="C:/plugins/sn_plan41/assets/Linkliste.xlsx")
def test_resolve_linkliste_empty_uses_standard_and_persists(self, _mock_join_path, _mock_root):
logic = self._make_logic()
resolved = logic._resolve_linkliste(None)
# -----------------------------------------------------
# 5. Linkliste löschen
# -----------------------------------------------------
def test_linkliste_clear(self):
logic = TabALogic()
self.assertEqual(resolved, "C:/plugins/sn_plan41/assets/Linkliste.xlsx")
self.assertEqual(get_variable("linkliste", scope="project"), "C:/plugins/sn_plan41/assets/Linkliste.xlsx")
logic.set_linkliste(None)
@patch("sn_plan41.ui.tab_a_logic.Dateipruefer")
def test_resolve_linkliste_returns_none_when_pruefmanager_rejects(self, mock_dateipruefer):
logic = self._make_logic()
mock_dateipruefer.return_value.pruefe.return_value = pruef_ergebnis(
ok=False,
meldung="Ungültig",
aktion="datei_nicht_gefunden",
kontext=None,
)
stored = get_variable("linkliste", scope="project")
self.assertEqual(stored, "")
resolved = logic._resolve_linkliste("C:/tmp/missing.xlsx")
self.assertIsNone(resolved)
# -----------------------------------------------------
# 6. Layer-ID speichern
# -----------------------------------------------------
def test_verfahrensgebiet_layer_id_storage(self):
logic = TabALogic()
def test_resolve_raumfilter_without_filter_returns_none(self):
logic = self._make_logic()
result = logic._resolve_raumfilter("ohne", "C:/tmp/v_db.gpkg")
self.assertIsNone(result)
class MockLayer:
def id(self):
return "layer-123"
@patch.object(TabALogic, "_get_verfahrensgebiet_layer", return_value="layer_obj")
def test_resolve_raumfilter_verfahrensgebiet_uses_layer_lookup(self, mock_get_layer):
logic = self._make_logic()
result = logic._resolve_raumfilter("Verfahrensgebiet", "C:/tmp/v_db.gpkg")
logic.save_verfahrensgebiet_layer(MockLayer())
self.assertEqual(result, "layer_obj")
mock_get_layer.assert_called_once()
stored = get_variable("verfahrensgebiet_layer", scope="project")
self.assertEqual(stored, "layer-123")
def test_extract_datenabruf_error_maps_merges_reason_maps(self):
logic = self._make_logic()
results = [
pruef_ergebnis(
ok=False,
meldung="Head failed",
aktion="url_nicht_erreichbar",
kontext={"ident": "A1", "thema": "Dienst A", "error": "Timeout"},
),
pruef_ergebnis(
ok=False,
meldung="Fetch failed",
aktion="datenabruf",
kontext={"fehler": {"Dienst B": "HTTP 500", "Dienst A": "ignored"}},
),
]
# -----------------------------------------------------
# 7. Ungültiger Layer wird ignoriert
# -----------------------------------------------------
def test_invalid_layer_is_rejected(self):
logic = TabALogic()
merged = logic._extract_datenabruf_error_maps(results)
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, "")
self.assertEqual(merged["by_ident"]["A1"], "Timeout")
self.assertEqual(merged["by_thema"]["Dienst A"], "Timeout")
self.assertEqual(merged["by_thema"]["Dienst B"], "HTTP 500")
if __name__ == "__main__":
+111
View File
@@ -0,0 +1,111 @@
import unittest
from unittest.mock import patch
from typing import Any, cast
from sn_plan41.ui.tab_a_logic import TabALogic
class _DummyPruefmanager:
def __init__(self) -> None:
self.hinweise: list[tuple[str, str]] = []
def verarbeite(self, ergebnis):
return ergebnis
def zeige_hinweis(self, titel: str, meldung: str) -> None:
self.hinweise.append((titel, meldung))
class _DummyDateiErgebnis:
def __init__(self, ok: bool = True, kontext: str = "C:/tmp/v_db.gpkg") -> None:
self.ok = ok
self.kontext = kontext
class _DummyProgress:
def __init__(self) -> None:
self.closed = False
def set_total(self, _value: int) -> None:
pass
def set_value(self, _value: int) -> None:
pass
def set_label(self, _value: str) -> None:
pass
def close(self) -> None:
self.closed = True
class _DummyProject:
def mapLayer(self, _layer_id: str):
return None
class _DummyGrabber:
pass
class TestTabAPreconditions(unittest.TestCase):
def _make_logic(self) -> tuple[TabALogic, _DummyPruefmanager]:
pruefmanager = _DummyPruefmanager()
logic = TabALogic(
pruefmanager=cast(Any, pruefmanager),
link_pruefer=cast(Any, object()),
stil_pruefer=cast(Any, object()),
)
return logic, pruefmanager
def test_get_verfahrensgebiet_layer_shows_hint_when_layer_id_missing(self):
logic, pruefmanager = self._make_logic()
with patch.object(logic, "load_verfahrensgebiet_layer_id", return_value=None):
layer = logic._get_verfahrensgebiet_layer()
self.assertIsNone(layer)
self.assertEqual(len(pruefmanager.hinweise), 1)
self.assertIn("Verfahrensgebiet", pruefmanager.hinweise[0][0])
@patch("sn_plan41.ui.tab_a_logic.QgsProject.instance", return_value=_DummyProject())
def test_get_verfahrensgebiet_layer_shows_hint_when_layer_not_in_project(self, _mock_project_instance):
logic, pruefmanager = self._make_logic()
with patch.object(logic, "load_verfahrensgebiet_layer_id", return_value="layer_404"):
layer = logic._get_verfahrensgebiet_layer()
self.assertIsNone(layer)
self.assertEqual(len(pruefmanager.hinweise), 1)
self.assertIn("Verfahrensgebiet", pruefmanager.hinweise[0][0])
@patch("sn_plan41.ui.tab_a_logic.Dateipruefer")
def test_pipeline_aborts_and_shows_hint_when_verfahrensgebiet_missing(
self,
mock_dateipruefer,
):
logic, pruefmanager = self._make_logic()
logic.data_grabber = cast(Any, _DummyGrabber())
mock_dateipruefer.return_value.pruefe.return_value = _DummyDateiErgebnis(ok=True, kontext="C:/tmp/v_db.gpkg")
progress = _DummyProgress()
with patch("sn_plan41.ui.tab_a_logic.create_progress_dialog", return_value=progress):
with patch.object(logic, "_resolve_linkliste", return_value="C:/tmp/linkliste.xlsx"):
with patch.object(logic, "load_verfahrensgebiet_layer_id", return_value=None):
result = logic._on_run_pipeline(
source="C:/tmp/v_db.gpkg",
linkliste=None,
raumfilter="Verfahrensgebiet",
)
self.assertIsNone(result)
self.assertEqual(len(pruefmanager.hinweise), 1)
self.assertIn("Verfahrensgebiet", pruefmanager.hinweise[0][0])
self.assertTrue(progress.closed)
if __name__ == "__main__":
unittest.main()
+94 -37
View File
@@ -1,56 +1,113 @@
"""
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 unittest.mock import patch
from sn_plan41.ui.tab_a_ui import TabA #type:ignore
from sn_basis.functions.variable_wrapper import get_variable
from sn_plan41.ui.tab_a_ui import TabA
class _DummySignal:
def connect(self, _callback):
return None
class _DummyToggleButton:
def __init__(self):
self.arrow = None
self.style = ""
def setArrowType(self, arrow):
self.arrow = arrow
def setStyleSheet(self, style):
self.style = style
class _DummyWidget:
def __init__(self):
self.visible = None
def setVisible(self, visible):
self.visible = visible
class _DummyFileWidget:
def __init__(self):
self._path = ""
def setFilePath(self, path):
self._path = path
def filePath(self):
return self._path
class _DummyCombo:
def __init__(self, value="Pufferlayer"):
self._value = value
def currentText(self):
return self._value
class _DummyPruefmanager:
pass
class _DummyDataGrabber:
pass
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)
def _make_tab_without_qt(self):
tab = TabA.__new__(TabA)
tab.group_button = _DummyToggleButton()
tab.group_content = _DummyWidget()
tab.optional_button = _DummyToggleButton()
tab.optional_content = _DummyWidget()
tab.file_widget = _DummyFileWidget()
tab.linkliste_widget = _DummyFileWidget()
tab._raumfilter_combo = _DummyCombo("ohne")
tab.verfahrens_db = None
tab.lokale_linkliste = None
tab.logic = None
tab.pruefmanager = None
tab.data_grabber = None
return tab
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)
def test_tab_a_callbacks_can_be_called_without_crash(self):
tab = self._make_tab_without_qt()
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)
self.assertEqual(tab.group_content.visible, False)
self.assertEqual(tab.optional_content.visible, False)
tab._on_layer_changed(None)
def test_tab_a_raumfilter_callback_persists_value(self):
tab = self._make_tab_without_qt()
tab._on_raumfilter_changed("ohne")
self.assertEqual(get_variable("Raumfilter", scope="project"), "ohne")
@patch("sn_plan41.ui.tab_a_ui.TabALogic")
def test_set_services_initializes_logic_and_injects_data_grabber(self, mock_logic_cls):
tab = self._make_tab_without_qt()
mock_logic = mock_logic_cls.return_value
pm = _DummyPruefmanager()
dg = _DummyDataGrabber()
tab.set_services(pruefmanager=pm, data_grabber=dg)
self.assertIs(tab.pruefmanager, pm)
self.assertIs(tab.data_grabber, dg)
self.assertIs(tab.logic, mock_logic)
self.assertIs(tab.logic.data_grabber, dg)
if __name__ == "__main__":
+22 -3
View File
@@ -1,16 +1,35 @@
#sn_plan41/ui/dockwidget.py
"""sn_plan41/ui/dockwidget.py Haupt-DockWidget für sn_plan41.
Verwaltet die Tabs für Linkvalidierung, Druck und Einstellungen.
"""
from sn_basis.ui.tabs.settings_tab import SettingsTab
from sn_plan41.ui.tab_a_ui import TabA
from sn_plan41.ui.tab_b_ui import TabB
from sn_basis.ui.tabs.print_tab import PrintTab
from sn_basis.ui.base_dockwidget import BaseDockWidget
from sn_basis.functions.qt_wrapper import QTabWidget
from sn_basis.functions.message_wrapper import error
class DockWidget(BaseDockWidget):
tabs = [TabA, TabB,SettingsTab]
"""Haupt-DockWidget für sn_plan41 mit Tab-Verwaltung.
Attributes:
tabs: Liste von Tab-Klassen (TabA, PrintTab, SettingsTab)
pruefmanager: Prüfmanager-Instanz für Validierung
data_grabber: DataGrabber-Instanz für Fachdatenabruf
"""
tabs = [TabA, PrintTab, SettingsTab]
def __init__(self, parent=None, subtitle="", pruefmanager=None, data_grabber=None):
"""Initialisiert das DockWidget mit Services und Tabs.
Args:
parent: Übergeordnetes Widget
subtitle: Untertitel des DockWidgets
pruefmanager: Prüfmanager-Instanz
data_grabber: DataGrabber-Instanz
"""
super().__init__(parent, subtitle)
# Services als Attribute speichern
-1191
View File
File diff suppressed because it is too large Load Diff
+221 -53
View File
@@ -12,6 +12,7 @@ import os
import datetime
import json
import tempfile
import html
from sn_basis.functions.qgiscore_wrapper import (
QgsVectorFileWriter,
@@ -63,9 +64,88 @@ class TabALogic:
self.link_pruefer = link_pruefer
self.stil_pruefer = stil_pruefer
self.data_grabber: Optional[DataGrabber] = None
class _SilentPruefmanagerProxy:
"""Proxy, der Prüfergebnisse ohne UI-Meldungen verarbeitet."""
def __init__(self, outer: "TabALogic", fallback: Optional[Any]) -> None:
self._outer = outer
self._fallback = fallback
self.results: List[Any] = []
def verarbeite(self, ergebnis: Any) -> Any:
self.results.append(ergebnis)
ok = bool(getattr(ergebnis, "ok", False))
if not ok:
aktion = getattr(ergebnis, "aktion", "unbekannt")
meldung = getattr(ergebnis, "meldung", "")
self._outer._log(f"[SILENT-PRUEFMANAGER] {aktion}: {meldung}")
return ergebnis
def __getattr__(self, name: str) -> Any:
if self._fallback is not None:
return getattr(self._fallback, name)
raise AttributeError(name)
def _log(self, msg: str) -> None:
print(f"[TabALogic] {msg}")
def _create_silent_pruefmanager(self) -> "TabALogic._SilentPruefmanagerProxy":
return TabALogic._SilentPruefmanagerProxy(self, self.pruefmanager)
def _extract_datenabruf_error_maps(self, datenabruf_results: List[Any]) -> Dict[str, Dict[str, str]]:
by_ident: Dict[str, str] = {}
by_thema: Dict[str, str] = {}
for result in datenabruf_results or []:
if not result:
continue
context = getattr(result, "kontext", None)
if not isinstance(context, dict):
continue
if getattr(result, "aktion", None) == "url_nicht_erreichbar":
ident = str(context.get("ident") or "").strip()
thema = str(context.get("thema") or "").strip()
reason = str(context.get("error") or getattr(result, "meldung", "") or "unbekannter Fehler").strip()
if ident and reason:
by_ident[ident] = reason
if thema and reason:
by_thema[thema] = reason
if getattr(result, "aktion", None) == "datenabruf":
fehler_map = context.get("fehler")
if isinstance(fehler_map, dict):
for thema_key, value in fehler_map.items():
thema = str(thema_key or "").strip()
reason = str(value or "").strip()
if thema and reason and thema not in by_thema:
by_thema[thema] = reason
return {"by_ident": by_ident, "by_thema": by_thema}
def _zeige_verfahrensgebiet_hinweis(self) -> None:
if not self.pruefmanager:
return
show_hint = getattr(self.pruefmanager, "zeige_hinweis", None)
if callable(show_hint):
show_hint(
"Verfahrensgebiet fehlt",
"Kein gueltiger Verfahrensgebiet-Layer vorhanden. "
"Bitte zuerst in sn_verfahrensgebiet den Layer ueber 'Aus ALKIS laden' laden "
"oder im Dropdown einen gueltigen Layer auswaehlen.",
)
def _close_progress_dialog(self, progress: Optional[Any]) -> None:
if progress is None:
return
close = getattr(progress, "close", None)
if callable(close):
try:
close()
except Exception:
pass
# -------------------------------
# Verfahrens-Datenbank (Pfad-Management)
# -------------------------------
@@ -155,12 +235,6 @@ class TabALogic:
# -------------------------------
# Verfahrensgebiet-Layer
# -------------------------------
def save_verfahrensgebiet_layer(self, layer: QgsVectorLayer) -> None:
"""Speichert die Verfahrensgebiet-Layer-ID, unter Annahme, dass der Layer prevalidiert ist."""
layer_id = layer.id() if layer is not None else ""
set_variable("verfahrensgebiet_layer", layer_id or "", scope="project")
def load_verfahrensgebiet_layer_id(self) -> Optional[str]:
value = get_variable("verfahrensgebiet_layer", scope="project")
return value or None
@@ -217,6 +291,7 @@ class TabALogic:
linkliste_final = self._resolve_linkliste(linkliste)
if linkliste_final is None:
self._log("Linkliste kann nicht aufgelöst werden")
self._close_progress_dialog(progress)
return None
else:
self._log(f"Linkliste final: '{linkliste_final}'")
@@ -224,6 +299,7 @@ class TabALogic:
raumfilter_layer = self._resolve_raumfilter(raumfilter, final_pfad)
if raumfilter in ("Verfahrensgebiet", "Pufferlayer") and raumfilter_layer is None:
self._log(f"Raumfilter '{raumfilter}' nicht verfügbar")
self._close_progress_dialog(progress)
return None
# 4) Lade-Status initialisieren (funktioniert ab Bestätigung überschreiben/anhängen)
@@ -236,7 +312,16 @@ class TabALogic:
progress.set_label("Prüfe Eingaben...")
# 5) Daten aus Linkliste laden und prüfen
source_dict, grabber_summary = self.data_grabber.run(linkliste_final)
silent_pruefmanager = self._create_silent_pruefmanager()
local_data_grabber = DataGrabber(
silent_pruefmanager,
datei_pruefer_cls=self.data_grabber._datei_pruefer_cls,
link_pruefer=self.data_grabber.link_pruefer,
layer_pruefer=self.data_grabber.layer_pruefer,
stil_pruefer=self.data_grabber.stil_pruefer,
excel_importer_cls=self.data_grabber._excel_importer_cls,
)
source_dict, grabber_summary = local_data_grabber.run(linkliste_final)
self._log(f"DataGrabber: {grabber_summary.meldung} [{grabber_summary.aktion}]")
# DEBUG: detaillierter Status
@@ -249,6 +334,7 @@ class TabALogic:
if not source_dict.get("rows"):
self._log("Keine validen Linkliste-Einträge für Datenabruf")
print("[TabALogic] ... STOP: rows:", len(source_dict.get("rows", [])))
self._close_progress_dialog(progress)
return None
total_rows = len(source_dict.get("rows", []))
@@ -266,7 +352,8 @@ class TabALogic:
self._log("Warnung: DataGrabber meldet fehlerhafte Zeilen, fahre mit Validierungsdaten fort")
# 5) Datenabruf (aus validierten Zeilen)
datenabruf = Datenabruf(self.pruefmanager)
silent_pruefmanager = self._create_silent_pruefmanager()
datenabruf = Datenabruf(silent_pruefmanager)
result_dict, datenabruf_results = datenabruf.datenabruf(
result_dict=source_dict,
raumfilter=raumfilter,
@@ -296,7 +383,7 @@ class TabALogic:
progress.set_label("Fachdaten laden abgeschlossen. Bitte OK klicken, um den Dialog zu schließen.")
# 7) Log-Datei schreiben
self._write_markdown_log(final_pfad, source_dict, pipeline_context, load_summary)
self._write_html_log(final_pfad, source_dict, pipeline_context, load_summary)
print("=" * 60 + "\n")
return pipeline_context
@@ -310,7 +397,15 @@ class TabALogic:
aborted = False
if not rows:
self._log("Keine Dienste zum Laden")
return
return {
"row_count": 0,
"loaded_count": 0,
"skipped_outside": 0,
"aborted": False,
"row_stats": [],
"layer_call_status": {},
"raumfilter_name": pipeline_context.get("raumfilter_name", "unbekannt"),
}
final_pfad = pipeline_context.get("source") or ""
use_datenschreiber = bool(final_pfad)
@@ -340,7 +435,12 @@ class TabALogic:
raumfilter_extent = raumfilter_layer.extent()
temp_layers: List[Any] = []
layer_loader = LayerLoader(self.pruefmanager, stil_pruefer=self.stil_pruefer, layer_pruefer=self.link_pruefer)
silent_pruefmanager = self._create_silent_pruefmanager()
layer_loader = LayerLoader(silent_pruefmanager, stil_pruefer=self.stil_pruefer, layer_pruefer=self.link_pruefer)
error_maps = self._extract_datenabruf_error_maps(pipeline_context.get("datenabruf_results", []))
error_by_ident = error_maps.get("by_ident", {})
error_by_thema = error_maps.get("by_thema", {})
# Statistiken für Log: Raumfilter-Info pro Dienst
row_stats: List[Dict[str, Any]] = []
@@ -357,6 +457,7 @@ class TabALogic:
daten_map = (pipeline_context.get("result_dict") or {}).get("daten", {})
fetched_features = daten_map.get(ident, []) if isinstance(daten_map, dict) else []
fetched_count = len(fetched_features) if isinstance(fetched_features, list) else None
base_reason = error_by_ident.get(ident) or error_by_thema.get(thema)
if progress is not None:
progress.set_label(f"Lade Dienst {idx}/{total}: {thema}")
@@ -385,6 +486,8 @@ class TabALogic:
if not layer:
layer_call_status[thema] = "layer_nicht_ladbar"
self._log(f"[DEBUG] Layeraufruf fehlgeschlagen: thema='{thema}'")
reason = base_reason or "Layer konnte nicht geladen werden (Dienst nicht erreichbar oder Link fehlerhaft)"
self._log(f"[DEBUG] Dienst nicht geladen: {thema} | Grund: {reason}")
row_stats.append({
"dienst": thema,
"provider": provider,
@@ -394,7 +497,7 @@ class TabALogic:
"total_features": None,
"filtered_features": None,
"status": "layer_nicht_ladbar",
"raumfilter": raumfilter_name,
"reason": reason,
})
continue
@@ -448,6 +551,7 @@ class TabALogic:
layer_call_status[thema] = "raumfilter_ausserhalb"
self._log(f"Dienst {thema} ist außerhalb des Raumfilters")
skipped_outside += 1
reason = "Keine Objekte innerhalb des Raumfilters"
row_stats.append({
"dienst": thema,
"provider": provider,
@@ -457,7 +561,7 @@ class TabALogic:
"total_features": total_features,
"filtered_features": 0,
"status": "außerhalb",
"raumfilter": raumfilter_name,
"reason": reason,
})
continue
@@ -473,7 +577,7 @@ class TabALogic:
"total_features": total_features,
"filtered_features": filtered_features,
"status": "geladen",
"raumfilter": raumfilter_name,
"reason": "geladen",
})
layer_call_status[thema] = "geladen"
self._log(f"[DEBUG] Dienst geladen: thema='{thema}', provider='{provider}', filtered_features={filtered_features}")
@@ -671,75 +775,118 @@ class TabALogic:
self._log(f"[DEBUG] Fehler beim Erzeugen lokaler Featureschicht für {thema}: {exc}")
return None
def _write_markdown_log(
def _write_html_log(
self,
final_pfad: str,
source_dict: DataDict,
pipeline_context: Dict[str, Any],
load_summary: Dict[str, Any],
) -> None:
"""Schreibt den Pipeline-Log (Markdown)."""
"""Schreibt den Pipeline-Log als HTML-Datei."""
lines = [
"# Plan41 Fachdaten-Ladevorgang",
"",
f"**Datum**: {datetime.datetime.now().isoformat()}",
f"**Verfahrens-DB**: {final_pfad or 'temporär'}",
f"**Linkliste**: {pipeline_context.get('linkliste')}",
"",
"## Zusammenfassung",
"",
f"- **Zeilen gesamt**: {load_summary.get('row_count', 0)}",
f"- **Geladene Dienste**: {load_summary.get('loaded_count', 0)}",
f"- **Außerhalb Raumfilter**: {load_summary.get('skipped_outside', 0)}",
f"- **Abgebrochen**: {load_summary.get('aborted', False)}",
f"- **Raumfilter**: {load_summary.get('raumfilter_name', 'unbekannt')}",
f"- **Raumfilter-Typ**: {pipeline_context.get('raumfilter_name', 'unbekannt')}",
"",
"## Dienstliste",
"",
"| Dienst | Provider | Linkadresse | Aufrufstatus | Ergebnisstatus |",
"|---|---|---|---|---|",
"<!doctype html>",
"<html lang='de'>",
"<head>",
" <meta charset='utf-8'>",
" <title>Plan41 Fachdaten-Ladevorgang</title>",
" <style>",
" body { font-family: Arial, sans-serif; margin: 20px; color: #222; }",
" h1, h2 { margin-bottom: 8px; }",
" .meta p { margin: 4px 0; }",
" table { border-collapse: collapse; width: 100%; margin: 12px 0 24px; }",
" th, td { border: 1px solid #ccc; padding: 6px 8px; text-align: left; vertical-align: top; }",
" th { background: #f2f2f2; }",
" .ok { color: #1b5e20; font-weight: 600; }",
" .warn { color: #8d6e00; font-weight: 600; }",
" .err { color: #b71c1c; font-weight: 600; }",
" </style>",
"</head>",
"<body>",
"<h1>Plan41 Fachdaten-Ladevorgang</h1>",
"<div class='meta'>",
f" <p><strong>Datum:</strong> {datetime.datetime.now().isoformat()}</p>",
f" <p><strong>Verfahrens-DB:</strong> {final_pfad or 'temporär'}</p>",
f" <p><strong>Linkliste:</strong> {pipeline_context.get('linkliste')}</p>",
f" <p><strong>Raumfilter:</strong> {load_summary.get('raumfilter_name', 'unbekannt')}</p>",
"</div>",
"<h2>Zusammenfassung</h2>",
"<ul>",
f" <li>Zeilen gesamt: {load_summary.get('row_count', 0)}</li>",
f" <li>Geladene Dienste: {load_summary.get('loaded_count', 0)}</li>",
f" <li>Außerhalb Raumfilter: {load_summary.get('skipped_outside', 0)}</li>",
f" <li>Abgebrochen: {load_summary.get('aborted', False)}</li>",
"</ul>",
"<h2>Dienstliste</h2>",
"<table>",
"<thead><tr><th>Dienst</th><th>Provider</th><th>Linkadresse</th><th>Aufrufstatus</th></tr></thead>",
"<tbody>",
]
status_by_dienst = {
str(stat.get("dienst", "")): str(stat.get("status", "n/a"))
for stat in load_summary.get("row_stats", [])
}
aufrufstatus_by_dienst = {
str(key): str(value)
for key, value in (load_summary.get("layer_call_status", {}) or {}).items()
}
aufrufstatus_labels = {
"layeraufruf_start": "Layer-Aufruf gestartet",
"layeraufruf_ok": "Layer-Aufruf erfolgreich",
"geladen": "Geladen",
"layer_nicht_ladbar": "Nicht ladbar",
"raumfilter_ausserhalb": "Außerhalb Raumfilter",
"abbruch_vor_layeraufruf": "Abbruch vor Layer-Aufruf",
"abbruch_nach_layeraufruf": "Abbruch nach Layer-Aufruf",
"abbruch_nach_raumfilter": "Abbruch nach Raumfilter",
"nicht_aufgerufen": "Nicht aufgerufen",
}
status_labels = {
"geladen": "Geladen",
"außerhalb": "Außerhalb Raumfilter",
"layer_nicht_ladbar": "Nicht ladbar",
}
for row in source_dict.get('rows', []):
dienst = row.get('Inhalt') or row.get('ident') or ''
provider = row.get('Provider') or ''
link = row.get('Link') or ''
aufrufstatus = aufrufstatus_by_dienst.get(str(dienst), "nicht_aufgerufen")
ergebnisstatus = status_by_dienst.get(str(dienst), "n/a")
lines.append(f"| {dienst} | {provider} | {link} | {aufrufstatus} | {ergebnisstatus} |")
aufrufstatus_text = aufrufstatus_labels.get(str(aufrufstatus), str(aufrufstatus))
lines.append(
f"<tr><td>{html.escape(str(dienst))}</td><td>{html.escape(str(provider))}</td><td>{html.escape(str(link))}</td><td>{html.escape(str(aufrufstatus_text))}</td></tr>"
)
lines.extend([
"",
"## Raumfilter-Statistik",
"",
"| Dienst | Provider | Linkadresse | Datenabruf-Objekte | Gesamt-Objekte | Gefilterte Objekte | Raumfilter | Status |",
"|---|---|---|---|---|---|---|---|",
"</tbody>",
"</table>",
"<h2>Raumfilter-Statistik</h2>",
"<table>",
"<thead><tr><th>Dienst</th><th>Datenabruf-Objekte</th><th>Gesamt-Objekte</th><th>Gefilterte Objekte</th><th>Status</th></tr></thead>",
"<tbody>",
])
for stat in load_summary.get('row_stats', []):
status = str(stat.get('status', 'n/a'))
status_label = status_labels.get(status, status)
reason = str(stat.get('reason', '') or '')
status_text = status_label if not reason else f"{status_label}: {reason}"
lines.append(
f"| {stat.get('dienst', '')} | {stat.get('provider', '')} | {stat.get('link', '')} | {stat.get('datenabruf_features', 'n/a')} | {stat.get('total_features', 'n/a')} | {stat.get('filtered_features', 'n/a')} | {stat.get('raumfilter', '')} | {stat.get('status', 'n/a')} |"
f"<tr><td>{html.escape(str(stat.get('dienst', '')))}</td><td>{html.escape(str(stat.get('datenabruf_features', 'n/a')))}</td><td>{html.escape(str(stat.get('total_features', 'n/a')))}</td><td>{html.escape(str(stat.get('filtered_features', 'n/a')))}</td><td>{html.escape(status_text)}</td></tr>"
)
markdown = "\n".join(lines)
lines.extend([
"</tbody>",
"</table>",
"</body>",
"</html>",
])
html_text = "\n".join(lines)
if final_pfad:
log_dir = os.path.dirname(final_pfad)
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, "plan41_lade_log.md")
log_file = os.path.join(log_dir, "plan41_lade_log.html")
try:
with open(log_file, "w", encoding="utf-8") as fh:
fh.write(markdown)
fh.write(html_text)
self.pruefmanager.verarbeite(
pruef_ergebnis(
@@ -754,15 +901,33 @@ class TabALogic:
self.pruefmanager.verarbeite(
pruef_ergebnis(
ok=False,
meldung=f"Fehler beim Schreiben des Logle (md): {exc}",
meldung=f"Fehler beim Schreiben des Logs (html): {exc}",
aktion="log_schreiben_fehlgeschlagen",
kontext={"error": str(exc)},
)
)
warning("Lade-Log", f"Konnte Datei nicht schreiben: {exc}", duration=10)
else:
# temporärer Modus: nur anzeigen
info("Lade-Log (temporär)", markdown, duration=20)
try:
tmp_dir = os.path.join(tempfile.gettempdir(), "sn_plan41")
os.makedirs(tmp_dir, exist_ok=True)
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
log_file = os.path.join(tmp_dir, f"plan41_lade_log_{timestamp}.html")
with open(log_file, "w", encoding="utf-8") as fh:
fh.write(html_text)
self._log(f"Lade-Log (temporär) gespeichert: {log_file}")
if hasattr(os, "startfile"):
try:
os.startfile(log_file)
except Exception as exc:
self._log(f"Konnte temporäre Log-Datei nicht öffnen: {exc}")
info("Lade-Log (temporär)", f"Lade-Protokoll gespeichert: {log_file}", duration=20)
except Exception as exc:
warning("Lade-Log", f"Konnte temporäre Log-Datei nicht schreiben: {exc}", duration=10)
def _clone_layer_with_extent(self, layer: QgsVectorLayer, extent, thema: str) -> QgsVectorLayer | None:
@@ -875,15 +1040,18 @@ class TabALogic:
if not layer_id:
self._log("❌ Keine Layer-ID gespeichert")
self._zeige_verfahrensgebiet_hinweis()
return None
layer = QgsProject.instance().mapLayer(layer_id)
if not layer:
self._log("❌ Layer-ID existiert nicht im Projekt")
self._zeige_verfahrensgebiet_hinweis()
return None
if not self.is_valid_verfahrensgebiet_layer(layer):
self._log("❌ Layer ist kein gültiger Vektorlayer")
self._zeige_verfahrensgebiet_hinweis()
return None
self._log(f"Verfahrensgebiet-Layer OK: '{layer.name()}'")
+1 -45
View File
@@ -166,16 +166,7 @@ class TabA(QWidget):
self.optional_content.setVisible(False)
main_layout.addWidget(self.optional_content)
# === LAYER-AUSWAHL + RAUMFILTER ===
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.setFilters(QgsMapLayerProxyModel.VectorLayer)
self.layer_combo.layerChanged.connect(self._on_layer_changed)
main_layout.addWidget(self.layer_combo)
# === RAUMFILTER ===
main_layout.addWidget(QLabel("Raumfilter"))
self._raumfilter_combo = QComboBox(self)
self._raumfilter_combo.setToolTip("Wählt die räumliche Bezugsfläche für die Datenextraktion.")
@@ -213,18 +204,6 @@ class TabA(QWidget):
self.lokale_linkliste = link
except Exception:
pass
# Layer
try:
layer_id = get_variable("tab_a_layer_id", scope="project")
if layer_id:
layer = QgsProject.instance().mapLayer(layer_id)
if layer and self.layer_combo:
self.layer_combo.setLayer(layer)
self._pufferlayer = layer
except Exception:
pass
# Raumfilter (schon im _build_ui behandelt)
# === UI CALLBACKS ===
@@ -249,20 +228,6 @@ class TabA(QWidget):
self.lokale_linkliste = path
set_variable("tab_a_linkliste", path, scope="project")
def _on_layer_changed(self, layer) -> None:
"""Persistiert Layer-Auswahl und registriert Verfahrensgebiet."""
self._pufferlayer = layer
if not layer:
return
# UI-State speichern
set_variable("tab_a_layer_id", layer.id(), scope="project")
# 🔹 NEU: Verfahrensgebiet explizit registrieren
if self.logic:
self.logic.save_verfahrensgebiet_layer(layer)
def _on_raumfilter_changed(self, value: str) -> None:
"""Persistieret Raumfilter-Auswahl."""
@@ -296,15 +261,6 @@ class TabA(QWidget):
source = self.file_widget.filePath()
raumfilter = self._raumfilter_combo.currentText()
linkliste = self.linkliste_widget.filePath()
if self.logic and self.layer_combo:
layer = self.layer_combo.currentLayer()
else:
layer = None
if layer and layer.name() == "Verfahrensgebiet":
self.logic.save_verfahrensgebiet_layer(layer)
if self.logic:
try:
self.logic._on_run_pipeline(source, linkliste, raumfilter)
-680
View File
@@ -1,680 +0,0 @@
"""
sn_plan41/ui/tab_b_logic.py Fachlogik für Tab B (Druck)
"""
from __future__ import annotations
import math
from typing import Any
from sn_basis.functions.qt_wrapper import QVariant
from sn_basis.functions.variable_wrapper import set_variable, get_variable
from sn_basis.functions.qgiscore_wrapper import get_layer_extent
from sn_basis.functions.qgiscore_wrapper import (
QgsProject,
QgsVectorLayer,
QgsGeometry,
QgsFeature,
QgsField,
QgsVectorFileWriter,
)
from sn_basis.functions.sys_wrapper import get_plugin_root, join_path, file_exists
from sn_basis.modules.Pruefmanager import Pruefmanager
from sn_basis.modules.layerpruefer import Layerpruefer
from sn_plan41.ui.layout import Layout
KARTENNAME_VAR = "kartenname"
PLOTMASSSTAB_VAR = "plotmassstab"
VIEW_VAR = "view"
ZIELGROESSE_VAR = "zielgroesse"
FORMFAKTOR_VAR = "formfaktor"
KARTENNAME_38 = "§38"
KARTENNAME_41 = "§41"
MASSSTAB_WIE_KARTENFENSTER = "Wie Kartenfenster"
THEMA_WIE_KARTENFENSTER = "wie kartenfenster"
KARTENNAME_BY_AUSWAHL = {
KARTENNAME_38: "Planungsübersicht §38 FlurbG",
KARTENNAME_41: "Karte zum Plan über die gemeinschaftlichen und öffentlichen Anlagen (§ 41 FlurbG)",
}
PLOTMASSSTAB_BY_AUSWAHL = {
"1:5.000": "5000",
"1:10.000": "10000",
"1:15.000": "15000",
"1:20.000": "20000",
"1:25.000": "25000",
"1:50.000": "50000",
"1:100.000": "100000",
}
# Breite x Höhe in mm (Hochformat, DIN-Standard)
DIN_GROESSEN: dict[str, tuple[int, int]] = {
"DIN A0": (841, 1189),
"DIN A1": (594, 841),
"DIN A2": (420, 594),
"DIN A3": (297, 420),
"DIN A4": (210, 297),
}
DIN_STANDARD = "DIN A0"
class TabBLogic:
"""
Kapselt die Fachlogik von Tab B.
"""
def __init__(self, pruefmanager: Pruefmanager) -> None:
self.pruefmanager = pruefmanager
@staticmethod
def _wkt_rect(x_min: float, y_min: float, x_max: float, y_max: float) -> str:
return (
f"POLYGON(({x_min} {y_min}, {x_max} {y_min}, {x_max} {y_max}, "
f"{x_min} {y_max}, {x_min} {y_min}))"
)
@staticmethod
def _set_topological_editing_enabled() -> None:
project = QgsProject.instance()
set_topological = getattr(project, "setTopologicalEditing", None)
if callable(set_topological):
try:
set_topological(True)
except Exception:
pass
@staticmethod
def _apply_atlas_style(layer: Any) -> None:
style_path = join_path(get_plugin_root(), "sn_plan41", "assets", "atlasobjekte.qml")
if not file_exists(style_path):
return
try:
ok, _ = layer.loadNamedStyle(str(style_path))
if ok:
getattr(layer, "triggerRepaint", lambda: None)()
except Exception:
pass
@staticmethod
def _find_tile_grid_for_roll_atlas(
kartenbild_w_mm: float,
kartenbild_h_mm: float,
din_dims: tuple[int, int],
respect_max_sheet_size: bool = False,
) -> tuple[int, int, float, float]:
"""Bestimmt ein Atlas-Raster für Endlosrolle, das Kacheln statt Streifen erzeugt.
Ziel: Kachel-Seitenverhältnis möglichst nah am Einzelblatt-Kartenfenster
(inkl. Rändern) bei zugleich moderater Seitenanzahl.
"""
if kartenbild_w_mm <= 0 or kartenbild_h_mm <= 0:
return 1, 1, max(1.0, kartenbild_w_mm), max(1.0, kartenbild_h_mm)
# Einzelblatt-Kartenfenster für beide Orientierungen
dim_w, dim_h = float(din_dims[0]), float(din_dims[1])
orientation_candidates: list[tuple[float, float]] = [
(dim_w - 210.0, dim_h - 20.0),
(dim_h - 210.0, dim_w - 20.0),
]
best_score = math.inf
best_result: tuple[int, int, float, float] | None = None
for target_w, target_h in orientation_candidates:
if target_w <= 0 or target_h <= 0:
continue
target_aspect = target_w / target_h
px0 = max(1, int(round(kartenbild_w_mm / target_w)))
py0 = max(1, int(round(kartenbild_h_mm / target_h)))
for pages_x in range(max(1, px0 - 3), px0 + 4):
for pages_y in range(max(1, py0 - 3), py0 + 4):
tile_w = kartenbild_w_mm / pages_x
tile_h = kartenbild_h_mm / pages_y
if tile_w <= 0 or tile_h <= 0:
continue
# Im Blatt-Modus darf die resultierende Atlasseite die
# gewählte Zielgröße (inkl. Orientierung) nicht überschreiten.
if respect_max_sheet_size and (tile_w > target_w or tile_h > target_h):
continue
tile_aspect = tile_w / tile_h
# 0 bei perfekter Übereinstimmung; symmetrisch für >1/<1
aspect_error = abs(math.log(tile_aspect / target_aspect))
# Streifen bestrafen
strip_penalty = 0.0
if tile_aspect < 0.5:
strip_penalty = abs(math.log(tile_aspect / 0.5))
elif tile_aspect > 2.0:
strip_penalty = abs(math.log(tile_aspect / 2.0))
page_count = pages_x * pages_y
# Gewichtung: zuerst Formatnähe, dann Streifenvermeidung,
# danach Seitenzahl minimieren.
score = (aspect_error * 12.0) + (strip_penalty * 6.0) + (page_count * 0.20)
if score < best_score:
best_score = score
best_result = (pages_x, pages_y, tile_w, tile_h)
if best_result is None:
if respect_max_sheet_size:
fallback_candidates: list[tuple[int, int, float, float, int]] = []
for target_w, target_h in orientation_candidates:
if target_w <= 0 or target_h <= 0:
continue
pages_x = max(1, math.ceil(kartenbild_w_mm / target_w))
pages_y = max(1, math.ceil(kartenbild_h_mm / target_h))
tile_w = kartenbild_w_mm / pages_x
tile_h = kartenbild_h_mm / pages_y
fallback_candidates.append((pages_x, pages_y, tile_w, tile_h, pages_x * pages_y))
if fallback_candidates:
fallback_candidates.sort(key=lambda entry: (entry[4], abs(math.log((entry[2] / entry[3]) if entry[3] > 0 else 1.0))))
fx, fy, fw, fh, _ = fallback_candidates[0]
return fx, fy, fw, fh
return 1, 1, kartenbild_w_mm, kartenbild_h_mm
return best_result
def _create_atlasobjekte_layer(
self,
layer: Any,
extent: Any,
pages_x: int,
pages_y: int,
seite_karte_w: float,
seite_karte_h: float,
massstab_zahl: float,
) -> Any | None:
layer_crs = layer.crs() if hasattr(layer, "crs") else None
crs_authid = layer_crs.authid() if layer_crs is not None and hasattr(layer_crs, "authid") else "EPSG:25832"
atlas_layer = QgsVectorLayer(f"Polygon?crs={crs_authid}", "Atlasobjekte", "memory")
if not atlas_layer or not atlas_layer.isValid():
return None
provider = atlas_layer.dataProvider()
provider.addAttributes([
QgsField("Seitenzahl", QVariant.Int),
])
atlas_layer.updateFields()
tile_w_m = seite_karte_w * massstab_zahl / 1000.0
tile_h_m = seite_karte_h * massstab_zahl / 1000.0
x_min = extent.xMinimum()
y_min = extent.yMinimum()
x_max = extent.xMaximum()
y_max = extent.yMaximum()
seitenzahl = 1
features = []
for row_idx in range(pages_y):
tile_y_max = y_max - row_idx * tile_h_m
tile_y_min = tile_y_max - tile_h_m
for col_idx in range(pages_x):
tile_x_min = x_min + col_idx * tile_w_m
tile_x_max = tile_x_min + tile_w_m
tile_geom = QgsGeometry.fromWkt(
self._wkt_rect(tile_x_min, tile_y_min, tile_x_max, tile_y_max)
)
feat = QgsFeature(atlas_layer.fields())
feat.setGeometry(tile_geom)
feat["Seitenzahl"] = seitenzahl
features.append(feat)
seitenzahl += 1
if not features:
return None
provider.addFeatures(features)
atlas_layer.updateExtents()
self._set_topological_editing_enabled()
verfahrens_db = get_variable("verfahrens_db", scope="project") or ""
print(f"[TabBLogic] Atlasobjekte: verfahrens_db='{verfahrens_db}'")
if not verfahrens_db:
QgsProject.instance().addMapLayer(atlas_layer)
self._apply_atlas_style(atlas_layer)
print("[TabBLogic] Atlasobjekte temporär ins Projekt geladen")
return atlas_layer
opts = QgsVectorFileWriter.SaveVectorOptions()
opts.driverName = "GPKG"
opts.fileEncoding = "UTF-8"
opts.layerName = "Atlasobjekte"
if file_exists(verfahrens_db):
opts.actionOnExistingFile = QgsVectorFileWriter.CreateOrOverwriteLayer
else:
opts.actionOnExistingFile = QgsVectorFileWriter.CreateOrOverwriteFile
err_result = QgsVectorFileWriter.writeAsVectorFormatV3(
atlas_layer,
verfahrens_db,
QgsProject.instance().transformContext(),
opts,
)
# QGIS-Versionen liefern hier entweder nur WriterError
# oder ein Tupel (WriterError, msg, newPath, layerName).
err_code = err_result[0] if isinstance(err_result, tuple) else err_result
if err_code != QgsVectorFileWriter.NoError:
print(f"[TabBLogic] Atlasobjekte schreiben fehlgeschlagen: err={err_result}")
return None
for existing in QgsProject.instance().mapLayersByName("Atlasobjekte"):
try:
QgsProject.instance().removeMapLayer(existing.id())
except Exception:
pass
uri = f"{verfahrens_db}|layername=Atlasobjekte"
loaded_layer = QgsVectorLayer(uri, "Atlasobjekte", "ogr")
if not loaded_layer or not loaded_layer.isValid():
print(f"[TabBLogic] Atlasobjekte laden aus GPKG fehlgeschlagen: uri='{uri}'")
return None
QgsProject.instance().addMapLayer(loaded_layer)
self._apply_atlas_style(loaded_layer)
print("[TabBLogic] Atlasobjekte aus Verfahrens-DB geladen und gestylt")
return loaded_layer
def set_kartenname_for_auswahl(self, auswahl: str) -> None:
"""Setzt die Projektvariable ``kartenname`` anhand der Kartennamen-Auswahl."""
kartenname = KARTENNAME_BY_AUSWAHL.get(auswahl, "")
set_variable(KARTENNAME_VAR, kartenname, scope="project")
def set_plotmassstab_for_auswahl(self, auswahl: str, aktueller_massstab: float | None = None) -> None:
"""Setzt die Projektvariable ``plotmassstab`` anhand der Maßstabsauswahl."""
if auswahl == MASSSTAB_WIE_KARTENFENSTER:
if aktueller_massstab and aktueller_massstab > 0:
set_variable(PLOTMASSSTAB_VAR, str(int(round(aktueller_massstab))), scope="project")
else:
set_variable(PLOTMASSSTAB_VAR, "", scope="project")
return
value = PLOTMASSSTAB_BY_AUSWAHL.get(auswahl, "")
set_variable(PLOTMASSSTAB_VAR, value, scope="project")
def set_view_for_auswahl(self, auswahl: str) -> None:
"""Setzt die Projektvariable ``view`` auf ``aktuell`` oder den Namen des gewählten Layerthemas."""
if auswahl == THEMA_WIE_KARTENFENSTER:
set_variable(VIEW_VAR, "aktuell", scope="project")
return
set_variable(VIEW_VAR, auswahl or "", scope="project")
def set_zielgroesse_for_auswahl(self, auswahl: str) -> None:
"""Setzt ``zielgroesse`` auf den gewählten DIN-Namen."""
set_variable(ZIELGROESSE_VAR, auswahl if auswahl in DIN_GROESSEN else DIN_STANDARD, scope="project")
def set_formfaktor(self, endlosrolle: bool) -> None:
"""Setzt die Projektvariable ``formfaktor`` auf ``Endlosrolle`` oder ``Blatt``."""
set_variable(FORMFAKTOR_VAR, "Endlosrolle" if endlosrolle else "Blatt", scope="project")
# ─────────────────────────────────────────────────────────────────────────
# Pipeline: Druckvorlage_anlegen
# ─────────────────────────────────────────────────────────────────────────
def druckvorlage_anlegen(
self,
layer: object,
kartenname_auswahl: str,
massstab_auswahl: str,
zielgroesse: str,
formfaktor: bool,
) -> dict:
"""Pipeline 'Druckvorlage_anlegen'.
Prüft Parameter, berechnet Plotgröße und stellt bei Bedarf Atlas-Rückfrage.
Parameters
----------
layer:
Aktuell gewählter Verfahrensgebiet-Layer aus Tab A.
Returns
-------
dict
``ok`` (bool): Ob die Pipeline erfolgreich durchlaufen werden soll.
``switch_to_tab_a`` (bool): Ob Tab A aktiviert werden soll.
``atlas_seiten`` (int): Anzahl benötigter Seiten (1 = kein Atlas).
"""
# ─── 1. Verfahrensgebiet-Layer prüfen ─────────────────────────────
lp = Layerpruefer(layer=layer)
ergebnis = lp.pruefe()
if not ergebnis.ok:
self.pruefmanager.zeige_hinweis(
"Verfahrensgebiets-Layer angeben",
"Verfahrensgebiets-Layer angeben",
)
return {"ok": False, "switch_to_tab_a": True, "atlas_seiten": 0}
layer_id = getattr(layer, "id", lambda: "")() or ""
set_variable("verfahrensgebietslayer", layer_id, scope="project")
# ─── 2. Kartenname prüfen ─────────────────────────────────────────
if kartenname_auswahl not in (KARTENNAME_38, KARTENNAME_41):
self.pruefmanager.zeige_hinweis(
"Kartennamen wählen",
"Kartennamen wählen",
)
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
# ─── 3. Maßstab ermitteln ─────────────────────────────────────────
if massstab_auswahl == MASSSTAB_WIE_KARTENFENSTER:
massstab_str = get_variable(PLOTMASSSTAB_VAR, scope="project") or ""
try:
massstab_zahl = float(massstab_str)
except (ValueError, TypeError):
massstab_zahl = 0.0
else:
massstab_zahl = float(PLOTMASSSTAB_BY_AUSWAHL.get(massstab_auswahl, 0))
if massstab_zahl <= 0:
self.pruefmanager.zeige_hinweis(
"Maßstab fehlt",
"Kein gültiger Maßstab angegeben.",
)
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
# ─── 4. Kartenbild berechnen ──────────────────────────────────────
# Der Layer wird als metrisch projiziert (Einheit: m) vorausgesetzt,
# wie es für deutsche Planungslagen (z.B. EPSG:25832) üblich ist.
extent = get_layer_extent(layer)
if extent is None:
self.pruefmanager.zeige_hinweis(
"Fehler",
"Layer-Ausdehnung konnte nicht ermittelt werden.",
)
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
# Naturgröße (m) → Papiergröße (mm): mm = m * 1000 / massstab
kartenbild_w = extent.width() * 1000.0 / massstab_zahl
kartenbild_h = extent.height() * 1000.0 / massstab_zahl
# ─── 5. Plotgröße = Kartenbild + Randabstand (x+210 mm, y+20 mm) ──
plotgroesse_w = kartenbild_w + 210.0
plotgroesse_h = kartenbild_h + 20.0
# ─── 6. Zielgröße bestimmen ───────────────────────────────────────
din_dims = DIN_GROESSEN.get(zielgroesse, DIN_GROESSEN[DIN_STANDARD])
if formfaktor: # Endlosrolle: X-Richtung entspricht der Plotgröße
ziel_w, ziel_h = plotgroesse_w, float(min(din_dims))
else:
ziel_w, ziel_h = float(din_dims[0]), float(din_dims[1])
if ziel_w < DIN_GROESSEN["DIN A4"][0] or ziel_h < DIN_GROESSEN["DIN A4"][1]:
self.pruefmanager.zeige_hinweis(
"Blattgröße zu klein",
"Die Zielgröße darf nicht kleiner als DIN A4 sein.",
)
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
# ─── 7. Passt auf ein Blatt? -> Layout erzeugen ───────────────────
print(f"[TabBLogic] plotgroesse=({plotgroesse_w:.1f}x{plotgroesse_h:.1f}), "
f"zielgroesse=({ziel_w:.1f}x{ziel_h:.1f}), passt={plotgroesse_w <= ziel_w and plotgroesse_h <= ziel_h}")
if plotgroesse_w > ziel_w or plotgroesse_h > ziel_h:
if plotgroesse_w <= ziel_h and plotgroesse_h <= ziel_w:
ziel_w, ziel_h = ziel_h, ziel_w
if plotgroesse_w <= ziel_w and plotgroesse_h <= ziel_h:
kartenname = get_variable(KARTENNAME_VAR, scope="project") or KARTENNAME_BY_AUSWAHL.get(
kartenname_auswahl, "Vorlage"
)
thema = get_variable(VIEW_VAR, scope="project") or ""
print(f"[TabBLogic] frage_text aufrufen, default='{kartenname}', thema='{thema}'")
vorlage_name, bestaetigt = self.pruefmanager.frage_text(
"Neue Vorlage anlegen",
"Bezeichnung der Vorlage:",
default_text=kartenname,
)
print(f"[TabBLogic] frage_text Ergebnis: name='{vorlage_name}', bestaetigt={bestaetigt}")
if not bestaetigt:
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
vorlage_name = (vorlage_name or "").strip() or kartenname
print(f"[TabBLogic] Rufe Layout().create_single_page_layout auf: name='{vorlage_name}', "
f"page=({ziel_w}x{ziel_h}), map=({kartenbild_w:.1f}x{kartenbild_h:.1f}), "
f"massstab={massstab_zahl}, thema='{thema}'")
try:
Layout().create_single_page_layout(
name=vorlage_name,
page_width_mm=ziel_w,
page_height_mm=ziel_h,
map_width_mm=kartenbild_w,
map_height_mm=kartenbild_h,
extent=extent,
plotmassstab=massstab_zahl,
thema=thema,
)
print("[TabBLogic] create_single_page_layout erfolgreich abgeschlossen")
except ValueError as exc:
print(f"[TabBLogic] ValueError: {exc}")
self.pruefmanager.zeige_hinweis("Vorlage anlegen", str(exc))
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
except Exception as exc:
print(f"[TabBLogic] Exception: {exc!r}")
self.pruefmanager.zeige_hinweis("Vorlage anlegen", f"Die Vorlage konnte nicht angelegt werden: {exc}")
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
return {"ok": True, "switch_to_tab_a": False, "atlas_seiten": 1}
# ─── 8. Atlas: Anzahl Seiten berechnen ────────────────────────────
# Nutzbarer Kartenbereich pro Atlasseite (abzüglich gleichem Randabstand)
seite_karte_w = ziel_w - 210.0
seite_karte_h = ziel_h - 20.0
if seite_karte_w <= 0 or seite_karte_h <= 0:
self.pruefmanager.zeige_hinweis(
"Blattgröße zu klein",
"Die gewählte Zielgröße ist kleiner als der Mindest-Randabstand. "
"Bitte eine größere Blattgröße wählen.",
)
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
pages_x = math.ceil(kartenbild_w / seite_karte_w)
pages_y = math.ceil(kartenbild_h / seite_karte_h)
# Für Atlas in beiden Modi (Endlosrolle + Blatt): Seitenraster so wählen,
# dass Atlasobjekte näher am Einzelblattformat liegen und keine Streifen entstehen.
opt_pages_x, opt_pages_y, opt_tile_w_mm, opt_tile_h_mm = self._find_tile_grid_for_roll_atlas(
kartenbild_w_mm=kartenbild_w,
kartenbild_h_mm=kartenbild_h,
din_dims=din_dims,
respect_max_sheet_size=(not formfaktor),
)
# Endlosrolle: Nur die Breite darf wachsen, die Höhe muss innerhalb
# der gewählten Zielhöhe bleiben.
if formfaktor:
max_tile_h_mm = max(1.0, ziel_h - 20.0)
if opt_tile_h_mm > max_tile_h_mm:
required_pages_y = max(1, math.ceil(kartenbild_h / max_tile_h_mm))
opt_pages_y = max(opt_pages_y, required_pages_y)
opt_tile_h_mm = kartenbild_h / opt_pages_y
pages_x = max(1, opt_pages_x)
pages_y = max(1, opt_pages_y)
seite_karte_w = max(1.0, opt_tile_w_mm)
seite_karte_h = max(1.0, opt_tile_h_mm)
modus = "Endlosrolle" if formfaktor else "Blatt"
print(
f"[TabBLogic] {modus} Rasteroptimierung: pages_x={pages_x}, pages_y={pages_y}, "
f"tile_mm=({seite_karte_w:.1f}x{seite_karte_h:.1f}), "
f"tile_aspect={seite_karte_w / seite_karte_h:.3f}"
)
if formfaktor:
max_tile_h_mm = max(1.0, ziel_h - 20.0)
print(
f"[TabBLogic] Endlosrolle Höhenlimit: tile_h={seite_karte_h:.1f}mm, "
f"max_tile_h={max_tile_h_mm:.1f}mm, within_limit={seite_karte_h <= max_tile_h_mm}"
)
anzahl_seiten = pages_x * pages_y
ja = self.pruefmanager.frage_ja_nein(
"Ausdruck als Atlas anlegen?",
f"Für die ausgewählten Parameter sind {anzahl_seiten} Einzelseiten erforderlich.\n"
"Ausdruck als Atlas anlegen?",
default=True,
)
print(f"[TabBLogic] Atlas-Rückfrage: ja={ja}, geplante_seiten={anzahl_seiten}")
if not ja:
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": anzahl_seiten}
atlas_layer = self._create_atlasobjekte_layer(
layer=layer,
extent=extent,
pages_x=pages_x,
pages_y=pages_y,
seite_karte_w=seite_karte_w,
seite_karte_h=seite_karte_h,
massstab_zahl=massstab_zahl,
)
if atlas_layer is None:
self.pruefmanager.zeige_hinweis(
"Atlasobjekte",
"Atlasobjekte-Layer konnte nicht erzeugt werden.",
)
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
print("[TabBLogic] Atlasobjekte-Layer erfolgreich erzeugt")
try:
anzahl_seiten = int(atlas_layer.featureCount())
except Exception:
pass
# Berechne die Kachelgröße aus den Atlasobjekten.
# Für die Layout-Kartengröße muss die größte Atlas-Kachel passen,
# sonst werden größere Atlasobjekte abgeschnitten.
total_w_m = 0.0
total_h_m = 0.0
min_w_m = math.inf
min_h_m = math.inf
max_w_m = 0.0
max_h_m = 0.0
feature_count = 0
get_features = getattr(atlas_layer, "getFeatures", None)
if callable(get_features):
try:
for feat in get_features():
geom = feat.geometry() if hasattr(feat, "geometry") else None
if geom is None or geom.isEmpty():
continue
bbox = geom.boundingBox() if hasattr(geom, "boundingBox") else None
if bbox is None:
continue
feat_w_m = float(bbox.width())
feat_h_m = float(bbox.height())
total_w_m += feat_w_m
total_h_m += feat_h_m
min_w_m = min(min_w_m, feat_w_m)
min_h_m = min(min_h_m, feat_h_m)
max_w_m = max(max_w_m, feat_w_m)
max_h_m = max(max_h_m, feat_h_m)
feature_count += 1
except Exception:
pass
if feature_count > 0:
avg_tile_w_m = total_w_m / feature_count
avg_tile_h_m = total_h_m / feature_count
if not math.isfinite(min_w_m):
min_w_m = avg_tile_w_m
if not math.isfinite(min_h_m):
min_h_m = avg_tile_h_m
target_tile_w_m = max_w_m
target_tile_h_m = max_h_m
else:
avg_tile_w_m = seite_karte_w * massstab_zahl / 1000.0
avg_tile_h_m = seite_karte_h * massstab_zahl / 1000.0
min_w_m = avg_tile_w_m
min_h_m = avg_tile_h_m
max_w_m = avg_tile_w_m
max_h_m = avg_tile_h_m
target_tile_w_m = avg_tile_w_m
target_tile_h_m = avg_tile_h_m
# Konvertiere Kachelgrößen zu mm
avg_tile_w_mm = avg_tile_w_m * 1000.0 / massstab_zahl
avg_tile_h_mm = avg_tile_h_m * 1000.0 / massstab_zahl
min_tile_w_mm = min_w_m * 1000.0 / massstab_zahl
min_tile_h_mm = min_h_m * 1000.0 / massstab_zahl
max_tile_w_mm = max_w_m * 1000.0 / massstab_zahl
max_tile_h_mm = max_h_m * 1000.0 / massstab_zahl
target_tile_w_mm = target_tile_w_m * 1000.0 / massstab_zahl
target_tile_h_mm = target_tile_h_m * 1000.0 / massstab_zahl
# Layout-Kartengröße = größte Kachelgröße
atlas_map_w = max(1.0, target_tile_w_mm)
atlas_map_h = max(1.0, target_tile_h_mm)
atlas_page_w = atlas_map_w + 210.0
atlas_page_h = atlas_map_h + 20.0
# Debug: Atlasobjekte Geometrien
print(
f"[TabBLogic] Atlasobjekte Geometrien (Meter): "
f"total_w={total_w_m:.1f}m, total_h={total_h_m:.1f}m, "
f"min_w={min_w_m:.1f}m, min_h={min_h_m:.1f}m, "
f"avg_w={avg_tile_w_m:.1f}m, avg_h={avg_tile_h_m:.1f}m, "
f"max_w={max_w_m:.1f}m, max_h={max_h_m:.1f}m, features={feature_count}"
)
print(
f"[TabBLogic] Atlas layout Größen: "
f"page=(x=10, y=10, w={atlas_page_w:.1f}mm, h={atlas_page_h:.1f}mm), "
f"map=(x=10, y=10, w={atlas_map_w:.1f}mm, h={atlas_map_h:.1f}mm), "
f"kachel_min_mm=({min_tile_w_mm:.1f}x{min_tile_h_mm:.1f}), "
f"kachel_avg_mm=({avg_tile_w_mm:.1f}x{avg_tile_h_mm:.1f}), "
f"kachel_max_mm=({max_tile_w_mm:.1f}x{max_tile_h_mm:.1f})"
)
kartenname = get_variable(KARTENNAME_VAR, scope="project") or KARTENNAME_BY_AUSWAHL.get(
kartenname_auswahl, "Atlas"
)
thema = get_variable(VIEW_VAR, scope="project") or ""
vorlage_name, bestaetigt = self.pruefmanager.frage_text(
"Neue Atlasvorlage anlegen",
"Bezeichnung der Vorlage:",
default_text=f"{kartenname} Atlas",
)
print(f"[TabBLogic] Atlas frage_text Ergebnis: name='{vorlage_name}', bestaetigt={bestaetigt}")
if not bestaetigt:
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": anzahl_seiten}
vorlage_name = (vorlage_name or "").strip() or f"{kartenname} Atlas"
try:
print(
f"[TabBLogic] Rufe create_atlas_layout auf: name='{vorlage_name}', "
f"page=({atlas_page_w:.1f}x{atlas_page_h:.1f}), "
f"map=({atlas_map_w:.1f}x{atlas_map_h:.1f}), massstab={massstab_zahl}, thema='{thema}'"
)
Layout().create_atlas_layout(
name=vorlage_name,
page_width_mm=atlas_page_w,
page_height_mm=atlas_page_h,
map_width_mm=atlas_map_w,
map_height_mm=atlas_map_h,
extent=extent,
plotmassstab=massstab_zahl,
atlas_layer=atlas_layer,
thema=thema,
)
print("[TabBLogic] create_atlas_layout erfolgreich abgeschlossen")
except ValueError as exc:
self.pruefmanager.zeige_hinweis("Atlasvorlage anlegen", str(exc))
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
except Exception as exc:
self.pruefmanager.zeige_hinweis("Atlasvorlage anlegen", f"Die Atlasvorlage konnte nicht angelegt werden: {exc}")
return {"ok": False, "switch_to_tab_a": False, "atlas_seiten": 0}
return {"ok": True, "switch_to_tab_a": False, "atlas_seiten": anzahl_seiten}
-432
View File
@@ -1,432 +0,0 @@
"""
sn_plan41/ui/tab_b_ui.py UI für Tab B (Druck)
"""
from __future__ import annotations
from typing import Optional
from sn_basis.functions.qt_wrapper import (
QWidget,
QVBoxLayout,
QLabel,
QComboBox,
QCheckBox,
QHBoxLayout,
QPushButton,
)
from sn_basis.functions.qgiscore_wrapper import QgsProject
from sn_basis.functions.qgisui_wrapper import iface
from sn_basis.functions.variable_wrapper import get_variable, set_variable
# Services (werden von DockWidget injiziert)
from sn_basis.modules.Pruefmanager import Pruefmanager
from sn_basis.modules.DataGrabber import DataGrabber
from sn_plan41.ui.tab_b_logic import (
TabBLogic,
MASSSTAB_WIE_KARTENFENSTER,
PLOTMASSSTAB_BY_AUSWAHL,
THEMA_WIE_KARTENFENSTER,
DIN_GROESSEN,
DIN_STANDARD,
ZIELGROESSE_VAR,
FORMFAKTOR_VAR,
)
KARTENNAME_VAR = "tab_b_kartenname"
KARTENNAME_PLACEHOLDER = "Kartenname wählen"
KARTENNAME_38 = "§38"
KARTENNAME_41 = "§41"
MASSSTAB_VAR = "tab_b_massstab"
THEMA_VAR = "tab_b_thema"
ZIELGROESSE_UI_VAR = "tab_b_zielgroesse"
FORMFAKTOR_UI_VAR = "tab_b_formfaktor"
class TabB(QWidget):
"""
UI-Klasse für Tab B (Druck) des Plan41-Plugins.
Zuständig für:
- Auswahl des Druckthemas
- Auswahl der Druckparameter
- Start der Vorlagenanlage (Druck über QGIS-Druckfunktion)
Services (Pruefmanager, DataGrabber) werden zur Laufzeit vom DockWidget injiziert.
Alle fachlichen Prüfungen laufen über den zentralen Pruefmanager.
"""
tab_title = "Druck" #: Tab-Titel für BaseDockWidget
def __init__(self, parent: Optional[QWidget] = None):
"""
Initialisiert die UI-Struktur.
Services werden später über :meth:`set_services` injiziert.
:param parent: Parent-Widget (typischerweise DockWidget)
"""
super().__init__(parent)
# Services (werden von DockWidget gesetzt)
self.pruefmanager: Optional[Pruefmanager] = None
self.logic: Optional[TabBLogic] = None
self._kartenname_combo: Optional[QComboBox] = None
self._massstab_combo: Optional[QComboBox] = None
self._thema_combo: Optional[QComboBox] = None
self._theme_signal_connected = False
self._connected_theme_collection: object = None # Referenz für sauberes Trennen
self._zielgroesse_combo: Optional[QComboBox] = None
self._endlosrolle_cb: Optional[QCheckBox] = None
self._btn_vorlage_erstellen: Optional[QPushButton] = None
self._build_ui()
self._restore_state()
self._connect_theme_collection_signals()
self._connect_project_signals()
def set_services(self, pruefmanager: Pruefmanager, data_grabber: DataGrabber) -> None:
"""Injiziert Services vom übergeordneten DockWidget."""
_ = data_grabber
self.pruefmanager = pruefmanager
self.logic = TabBLogic(pruefmanager=self.pruefmanager)
if self._kartenname_combo:
self.logic.set_kartenname_for_auswahl(self._kartenname_combo.currentText())
if self._massstab_combo:
self.logic.set_plotmassstab_for_auswahl(
self._massstab_combo.currentText(),
self._get_current_canvas_scale(),
)
if self._thema_combo:
self.logic.set_view_for_auswahl(self._thema_combo.currentText())
if self._zielgroesse_combo:
self.logic.set_zielgroesse_for_auswahl(self._zielgroesse_combo.currentText())
if self._endlosrolle_cb:
self.logic.set_formfaktor(self._endlosrolle_cb.isChecked())
def _build_ui(self) -> None:
"""Erstellt die reduzierte UI für die Themenauswahl."""
main_layout = QVBoxLayout()
main_layout.setSpacing(4)
main_layout.setContentsMargins(4, 4, 4, 4)
kartenname_label = QLabel("Kartenname")
kartenname_label.setStyleSheet("font-weight: bold; margin-top: 6px;")
main_layout.addWidget(kartenname_label)
self._kartenname_combo = QComboBox(self)
self._kartenname_combo.addItem(KARTENNAME_PLACEHOLDER)
self._kartenname_combo.addItem(KARTENNAME_38)
self._kartenname_combo.addItem(KARTENNAME_41)
self._kartenname_combo.currentTextChanged.connect(self._on_kartenname_changed)
main_layout.addWidget(self._kartenname_combo)
massstab_label = QLabel("Maßstab")
massstab_label.setStyleSheet("font-weight: bold; margin-top: 6px;")
main_layout.addWidget(massstab_label)
self._massstab_combo = QComboBox(self)
self._massstab_combo.addItem(MASSSTAB_WIE_KARTENFENSTER)
self._massstab_combo.addItems(list(PLOTMASSSTAB_BY_AUSWAHL.keys()))
self._massstab_combo.currentTextChanged.connect(self._on_massstab_changed)
main_layout.addWidget(self._massstab_combo)
thema_label = QLabel("Thema")
thema_label.setStyleSheet("font-weight: bold; margin-top: 6px;")
main_layout.addWidget(thema_label)
self._thema_combo = QComboBox(self)
self._thema_combo.addItem(THEMA_WIE_KARTENFENSTER)
self._thema_combo.addItems(self._get_gespeicherte_themen())
self._thema_combo.currentTextChanged.connect(self._on_thema_changed)
main_layout.addWidget(self._thema_combo)
zielgroesse_label = QLabel("max. Blattgröße")
zielgroesse_label.setStyleSheet("font-weight: bold; margin-top: 6px;")
main_layout.addWidget(zielgroesse_label)
zielgroesse_row = QHBoxLayout()
zielgroesse_row.setSpacing(6)
self._zielgroesse_combo = QComboBox(self)
self._zielgroesse_combo.addItems(list(DIN_GROESSEN.keys()))
self._zielgroesse_combo.setCurrentText(DIN_STANDARD)
self._zielgroesse_combo.currentTextChanged.connect(self._on_zielgroesse_changed)
zielgroesse_row.addWidget(self._zielgroesse_combo)
self._endlosrolle_cb = QCheckBox("Endlosrolle", self)
self._endlosrolle_cb.setChecked(False)
self._endlosrolle_cb.stateChanged.connect(self._on_formfaktor_changed)
zielgroesse_row.addWidget(self._endlosrolle_cb)
main_layout.addLayout(zielgroesse_row)
self._btn_vorlage_erstellen = QPushButton("Vorlage erstellen", self)
self._btn_vorlage_erstellen.clicked.connect(self._on_vorlage_erstellen)
main_layout.addWidget(self._btn_vorlage_erstellen)
main_layout.addStretch(1)
self.setLayout(main_layout)
def _restore_state(self) -> None:
"""Stellt die gespeicherten Combobox-Zustände wieder her."""
if not self._kartenname_combo or not self._massstab_combo or not self._thema_combo:
return
if not self._zielgroesse_combo or not self._endlosrolle_cb:
return
saved_kartenname = get_variable(KARTENNAME_VAR, scope="project")
if saved_kartenname in (KARTENNAME_38, KARTENNAME_41):
self._kartenname_combo.setCurrentText(saved_kartenname)
else:
self._kartenname_combo.setCurrentText(KARTENNAME_PLACEHOLDER)
saved_massstab = get_variable(MASSSTAB_VAR, scope="project")
valid_massstaebe = [MASSSTAB_WIE_KARTENFENSTER, *PLOTMASSSTAB_BY_AUSWAHL.keys()]
if saved_massstab in valid_massstaebe:
self._massstab_combo.setCurrentText(saved_massstab)
else:
self._massstab_combo.setCurrentText(MASSSTAB_WIE_KARTENFENSTER)
aktuelle_themen = [THEMA_WIE_KARTENFENSTER, *self._get_gespeicherte_themen()]
self._thema_combo.clear()
self._thema_combo.addItems(aktuelle_themen)
saved_thema = get_variable(THEMA_VAR, scope="project")
if saved_thema in aktuelle_themen:
self._thema_combo.setCurrentText(saved_thema)
else:
self._thema_combo.setCurrentText(THEMA_WIE_KARTENFENSTER)
saved_zielgroesse = get_variable(ZIELGROESSE_UI_VAR, scope="project")
if saved_zielgroesse in DIN_GROESSEN:
self._zielgroesse_combo.setCurrentText(saved_zielgroesse)
else:
self._zielgroesse_combo.setCurrentText(DIN_STANDARD)
saved_formfaktor = get_variable(FORMFAKTOR_UI_VAR, scope="project")
self._endlosrolle_cb.setChecked(saved_formfaktor == "Endlosrolle")
def _on_kartenname_changed(self, value: str) -> None:
"""Persistiert die Kartennamen-Auswahl und setzt ``kartenname``."""
if value in (KARTENNAME_38, KARTENNAME_41):
set_variable(KARTENNAME_VAR, value, scope="project")
else:
set_variable(KARTENNAME_VAR, "", scope="project")
if self.logic:
self.logic.set_kartenname_for_auswahl(value)
def _on_massstab_changed(self, value: str) -> None:
"""Persistiert Maßstabsauswahl und setzt ``plotmassstab``."""
set_variable(MASSSTAB_VAR, value, scope="project")
if self.logic:
self.logic.set_plotmassstab_for_auswahl(value, self._get_current_canvas_scale())
def _on_thema_changed(self, value: str) -> None:
"""Persistiert die Thema-Auswahl und setzt ``view``."""
set_variable(THEMA_VAR, value, scope="project")
if self.logic:
self.logic.set_view_for_auswahl(value)
def _on_zielgroesse_changed(self, value: str) -> None:
"""Persistiert Blattgröße und setzt ``zielgroesse``."""
set_variable(ZIELGROESSE_UI_VAR, value, scope="project")
if self.logic:
self.logic.set_zielgroesse_for_auswahl(value)
def _on_formfaktor_changed(self, state: int) -> None:
"""Persistiert Endlosrolle-Zustand und setzt ``formfaktor``."""
checked = bool(state)
set_variable(FORMFAKTOR_UI_VAR, "Endlosrolle" if checked else "Blatt", scope="project")
if self.logic:
self.logic.set_formfaktor(checked)
def _on_vorlage_erstellen(self) -> None:
"""Startet die Pipeline Druckvorlage_anlegen."""
print("[TabB] _on_vorlage_erstellen aufgerufen")
if not self.logic or not self.pruefmanager:
print(f"[TabB] Abbruch: logic={self.logic}, pruefmanager={self.pruefmanager}")
return
# Layer direkt aus Tab A lesen (unabhängig von Projektvariable)
layer = None
tab_a = self._get_tab_a_widget()
print(f"[TabB] tab_a Widget: {tab_a!r}")
if tab_a is not None:
layer_combo = getattr(tab_a, "layer_combo", None)
print(f"[TabB] layer_combo: {layer_combo!r}")
if layer_combo is not None:
layer = layer_combo.currentLayer()
print(f"[TabB] layer: {layer!r} (Name: {getattr(layer, 'name', lambda: '')()})")
print(f"[TabB] Rufe druckvorlage_anlegen auf mit layer={layer!r}, "
f"kartenname={self._kartenname_combo.currentText() if self._kartenname_combo else '?'}, "
f"massstab={self._massstab_combo.currentText() if self._massstab_combo else '?'}, "
f"zielgroesse={self._zielgroesse_combo.currentText() if self._zielgroesse_combo else '?'}, "
f"formfaktor={self._endlosrolle_cb.isChecked() if self._endlosrolle_cb else '?'}")
result = self.logic.druckvorlage_anlegen(
layer=layer,
kartenname_auswahl=self._kartenname_combo.currentText() if self._kartenname_combo else "",
massstab_auswahl=self._massstab_combo.currentText() if self._massstab_combo else "",
zielgroesse=self._zielgroesse_combo.currentText() if self._zielgroesse_combo else DIN_STANDARD,
formfaktor=self._endlosrolle_cb.isChecked() if self._endlosrolle_cb else False,
)
print(f"[TabB] druckvorlage_anlegen Ergebnis: {result}")
if result.get("switch_to_tab_a"):
self._aktiviere_tab_a()
def _get_tab_widget(self):
"""Findet das übergeordnete QTabWidget anhand des ``tabBar``-Attributs."""
try:
widget = self.parent()
while widget is not None:
if hasattr(widget, "tabBar") and hasattr(widget, "setCurrentIndex"):
return widget
parent_fn = getattr(widget, "parent", None)
widget = parent_fn() if callable(parent_fn) else None
except Exception:
pass
return None
def _get_tab_a_widget(self):
"""Gibt die Tab-A-Widget-Instanz zurück (Index 0 im übergeordneten QTabWidget)."""
tab_widget = self._get_tab_widget()
if tab_widget is None:
return None
try:
return tab_widget.widget(0)
except Exception:
return None
def _aktiviere_tab_a(self) -> None:
"""Wechselt den aktiven Reiter auf Tab A im übergeordneten QTabWidget."""
tab_widget = self._get_tab_widget()
if tab_widget is not None:
try:
tab_widget.setCurrentIndex(0)
except Exception:
pass
def _connect_project_signals(self) -> None:
"""Verbindet QgsProject-Signale für Projektwechsel/-neuladen."""
project = QgsProject.instance()
for signal_name in ("readProject", "newProjectCreated", "cleared"):
signal = getattr(project, signal_name, None)
if signal is None:
continue
try:
signal.connect(self._on_project_changed)
except Exception:
pass
def _on_project_changed(self, *args) -> None:
"""Reagiert auf Projektwechsel: Signale neu binden, Combobox und State auffrischen."""
_ = args
# Alte Theme-Collection-Signals zuerst trennen
self._disconnect_theme_collection_signals()
# Neu verbinden für das jetzt geladene Projekt
self._theme_signal_connected = False
self._connect_theme_collection_signals()
# Ansicht-Liste + gespeicherten State wiederherstellen
self._restore_state()
def _disconnect_theme_collection_signals(self) -> None:
"""Trennt Signale der alten Theme-Collection sauber."""
collection = self._connected_theme_collection
if collection is not None:
for signal_name in ("mapThemesChanged", "changed", "themeChanged"):
try:
signal = getattr(collection, signal_name, None)
except RuntimeError:
break
except Exception:
continue
if signal is None:
continue
try:
signal.disconnect(self._refresh_thema_combo_live)
except Exception:
pass
self._connected_theme_collection = None
self._theme_signal_connected = False
def _connect_theme_collection_signals(self) -> None:
"""Verbindet Signale der Theme-Collection für Live-Aktualisierung der Themenliste."""
if self._theme_signal_connected:
return
try:
theme_collection = QgsProject.instance().mapThemeCollection()
except Exception:
return
if theme_collection is None:
return
connected_any = False
for signal_name in ("mapThemesChanged", "changed", "themeChanged"):
signal = getattr(theme_collection, signal_name, None)
if signal is None:
continue
try:
signal.connect(self._refresh_thema_combo_live)
connected_any = True
except Exception:
pass
if connected_any:
self._connected_theme_collection = theme_collection
self._theme_signal_connected = connected_any
def _refresh_thema_combo_live(self, *args) -> None:
"""Aktualisiert die Thema-Combobox bei Änderungen gespeicherter Layerthemen."""
_ = args
if not self._thema_combo:
return
vorherige_auswahl = self._thema_combo.currentText() or THEMA_WIE_KARTENFENSTER
eintraege = [THEMA_WIE_KARTENFENSTER, *self._get_gespeicherte_themen()]
self._thema_combo.blockSignals(True)
self._thema_combo.clear()
self._thema_combo.addItems(eintraege)
if vorherige_auswahl in eintraege:
self._thema_combo.setCurrentText(vorherige_auswahl)
else:
self._thema_combo.setCurrentText(THEMA_WIE_KARTENFENSTER)
self._thema_combo.blockSignals(False)
self._on_thema_changed(self._thema_combo.currentText())
def _get_gespeicherte_themen(self) -> list[str]:
"""Liefert die Namen der im Projekt gespeicherten Layerthemen (QgsMapThemeCollection)."""
try:
theme_collection = QgsProject.instance().mapThemeCollection()
if theme_collection is None:
return []
themes = theme_collection.mapThemes()
except Exception:
return []
namen: list[str] = []
for theme_name in themes:
name = str(theme_name or "").strip()
if name and name not in namen:
namen.append(name)
return namen
def _get_current_canvas_scale(self) -> float | None:
"""Liest den aktuellen Maßstab aus der Kartensicht."""
try:
canvas = iface.mapCanvas()
if canvas is None:
return None
scale = canvas.scale()
return float(scale) if scale else None
except Exception:
return None
+15
View File
@@ -1,9 +1,24 @@
"""sn_plan41/ui/tabs/tab_b.py Beispiel-Tab für zusätzliche Funktionen.
Ein Tab für zukünftige Erweiterungen.
"""
from qgis.PyQt.QtWidgets import QWidget, QVBoxLayout, QLabel, QTextEdit
class TabB(QWidget):
"""Beispiel-Tab für sn_plan41.
Attributes:
tab_title: Anzeigetitel des Tabs ("Tab B")
"""
tab_title = "Tab B"
def __init__(self, parent=None):
"""Initialisiert den Tab mit Beispielinhalt.
Args:
parent: Übergeordnetes Widget
"""
super().__init__(parent)
layout = QVBoxLayout()
layout.addWidget(QLabel("Plugin2 Tab B"))