"""The plugin's Processing provider and its GUI-only default-refresh hookups (SPEC §5)."""
from __future__ import annotations
from typing import TYPE_CHECKING, override
from qgis.core import QgsProcessingProvider, QgsProject
from qgis.PyQt.QtCore import QMetaObject, QObject, QTimer
from stratified_packager.__about__ import __title__, __version__
from stratified_packager.identity import PLUGIN_SLUG, plugin_icon
from stratified_packager.toolbelt.logging import QgisLoggerWrapper
from .algorithm import StratifiedPackagerAlgorithm
from .params import LAYER_VARIABLE_PROPERTY_KEYS
if TYPE_CHECKING:
from collections.abc import Iterable
from qgis.core import QgsMapLayer
from qgis.PyQt.QtGui import QIcon
log = QgisLoggerWrapper.get_logger(__name__)
[docs]
class StratifiedPackagerProvider(QgsProcessingProvider):
"""
Registers the plugin's single algorithm and keeps its dynamic defaults fresh.
The default values shown in the Processing dialog are computed in
:meth:`.algorithm.StratifiedPackagerAlgorithm.initAlgorithm`
from the active project's variables, the plugin settings and the layer ``exclude``
flags (SPEC §5). To keep that prefill in step with edits to the project,
:meth:`connect_project_signals` wires the relevant :class:`~qgis.core.QgsProject` signals
to :meth:`~qgis.core.QgsProcessingProvider.refreshAlgorithms` (which rebuilds every
algorithm instance). Those hookups are made **only from a GUI session**: connecting
them headless and refreshing from them mid-run segfaults ``qgis_process`` (SPEC
§5), so :meth:`connect_project_signals` is called from
:meth:`~stratified_packager.main.StratifiedPackager.initGui` and never from
:meth:`~stratified_packager.main.StratifiedPackager.initProcessing`.
"""
[docs]
@override
def __init__(self) -> None:
"""Initialize the provider with no signal hookups (those are GUI-only)."""
super().__init__()
self._refresh_timer: QTimer | None = None
"""Single-shot coalescing timer; created lazily by :meth:`connect_project_signals`."""
self._project_connections: list[QMetaObject.Connection] = []
"""Tokens of the project-level signal connections, for tidy disconnection."""
self._layer_connections: dict[str, QMetaObject.Connection] = {}
"""Per-layer ``customPropertyChanged`` connection tokens, keyed by layer id."""
[docs]
@override
def loadAlgorithms(self) -> None:
"""Load algorithms belonging to this provider."""
# This is the shallowest plugin frame QGIS invokes, so a failure to register
# is logged here rather than propagated into the Processing framework.
alg = StratifiedPackagerAlgorithm()
if not self.addAlgorithm(alg):
log.error(self.tr("Failed to register the %s algorithm."), alg.displayName())
# ------------------------------------------------------------------
# Dynamic-default refresh hookups (SPEC §5; GUI sessions only)
# ------------------------------------------------------------------
[docs]
def connect_project_signals(self) -> None:
"""
Wire project edits to a coalesced algorithm refresh (SPEC §5).
Idempotent. Call from ``initGui`` only — never headless (SPEC §5). Connects
the project's ``readProject`` / ``cleared`` / ``customVariablesChanged`` and
``layersAdded`` / ``layersWillBeRemoved`` signals, plus every current layer's
``customPropertyChanged``, each routed through a single-shot 0 ms timer so the
burst of signals QGIS emits while loading a project collapses into one refresh
(SPEC §5).
"""
if self._refresh_timer is not None:
return
project = QgsProject.instance()
if project is None:
log.warning(self.tr("No project available; default-refresh signals not connected."))
return
timer = QTimer(self)
timer.setSingleShot(True)
timer.setInterval(0)
timer.timeout.connect(self._refresh_algorithms)
self._refresh_timer = timer
self._project_connections = [
project.readProject.connect(self._schedule_refresh),
project.cleared.connect(self._schedule_refresh),
project.customVariablesChanged.connect(self._schedule_refresh),
project.layersAdded.connect(self._on_layers_added),
project.layersWillBeRemoved.connect(self._on_layers_will_be_removed),
]
self._connect_layers(project.mapLayers().values())
[docs]
def disconnect_project_signals(self) -> None:
"""
Reverse :meth:`connect_project_signals` (called from ``unload``).
Disconnects the project-level and per-layer connections, then stops and
disposes the coalescing timer, leaving the provider safe to re-register.
"""
for connection in self._project_connections:
QObject.disconnect(connection)
self._project_connections.clear()
for connection in self._layer_connections.values():
QObject.disconnect(connection)
self._layer_connections.clear()
if self._refresh_timer is not None:
self._refresh_timer.stop()
self._refresh_timer.deleteLater()
self._refresh_timer = None
[docs]
def _connect_layers(self, layers: Iterable[QgsMapLayer]) -> None:
"""
Connect each layer's ``customPropertyChanged`` to the refresh scheduler.
:param layers: Layers to start tracking (already-tracked ids are skipped).
"""
for layer in layers:
layer_id = layer.id()
if layer_id in self._layer_connections:
continue
self._layer_connections[layer_id] = layer.customPropertyChanged.connect(
self._on_layer_property_changed
)
[docs]
def _on_layers_added(self, layers: Iterable[QgsMapLayer]) -> None:
"""
Track newly added layers and schedule a refresh (SPEC §5).
:param layers: The layers QGIS just added to the project.
"""
self._connect_layers(layers)
self._schedule_refresh()
[docs]
def _on_layers_will_be_removed(self, layers: Iterable[QgsMapLayer | str]) -> None:
"""
Stop tracking layers about to be removed and schedule a refresh (SPEC §5).
Tolerates both overloads of ``layersWillBeRemoved`` (layer ids or layer objects).
:param layers: The layers (or their ids) QGIS is about to remove.
"""
for item in layers:
layer_id = item if isinstance(item, str) else item.id()
connection = self._layer_connections.pop(layer_id, None)
if connection is not None:
QObject.disconnect(connection)
self._schedule_refresh()
[docs]
def _on_layer_property_changed(self, key: str) -> None:
"""
Schedule a refresh when a *layer-variable* custom property changes (SPEC §5).
``setLayerVariable`` fires ``customPropertyChanged`` twice (keys ``variableNames``
and ``variableValues``); the 0 ms timer debounces them, and filtering on those
keys avoids spurious refreshes from unrelated custom-property writes (e.g. styles).
:param key: The custom-property key that changed.
"""
if key in LAYER_VARIABLE_PROPERTY_KEYS:
self._schedule_refresh()
[docs]
def _schedule_refresh(self) -> None:
"""
Restart the single-shot timer, coalescing a burst of signals into one refresh.
Connected to argument-bearing signals too (e.g. ``readProject``); PyQt drops the
extra payloads when the slot takes none.
"""
if self._refresh_timer is not None:
self._refresh_timer.start()
[docs]
def _refresh_algorithms(self) -> None:
"""Rebuild the provider's algorithms so dialog defaults pick up project edits."""
self.refreshAlgorithms()
# ------------------------------------------------------------------
# Identity
# ------------------------------------------------------------------
[docs]
@override
def id(self) -> str:
"""
Return the unique provider id string.
:return: provider ID
"""
return PLUGIN_SLUG
[docs]
@override
def name(self) -> str:
"""
Human-friendly provider name, used to describe the provider within the GUI.
:return: provider name
"""
return __title__
[docs]
@override
def icon(self) -> QIcon:
"""
Icon used for the provider inside the Processing toolbox.
:return: provider icon
"""
return plugin_icon()
[docs]
@override
def versionInfo(self) -> str:
"""
Version information for the provider.
:return: version
"""
return __version__