# Copyright (c) 2025-2026 profiler-qgis-plugin contributors.
#
#
# This file is part of profiler-qgis-plugin.
#
# profiler-qgis-plugin is free software: you can redistribute it and/or
# modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# profiler-qgis-plugin is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty
# of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with profiler-qgis-plugin. If not, see <https://www.gnu.org/licenses/>.
"""Abstract base class for performance anomaly detection meters.
Defines :class:`Meter`, which all concrete meters extend. Meters measure specific
aspects of QGIS performance, emit anomalies when thresholds are exceeded, and can
optionally record results into the profiler.
"""
import abc
from collections.abc import Callable, Generator
from contextlib import contextmanager, suppress
from functools import wraps
from typing import Any, ClassVar, NamedTuple, cast
from qgis.core import QgsApplication
from qgis.PyQt.QtCore import QObject, pyqtSignal, pyqtSlot
import qgis_profiler.utils
from qgis_profiler.profiler import ProfilerWrapper
from qgis_profiler.settings import Settings, resolve_group_name_with_cache
[docs]
class MeterContext(NamedTuple):
"""Context of a meter."""
name: str
group: str
[docs]
def with_meter_suffix(self, suffix: str) -> "MeterContext":
"""Return a new context with the given suffix appended to the name."""
return MeterContext(f"{self.name} ({suffix})", self.group)
[docs]
class MeterAnomaly(NamedTuple):
"""Anomaly detected by a meter."""
context: MeterContext
duration_seconds: float
[docs]
class Meter(QObject):
"""Abstract base class for meters to detect anomalies in QGIS performance.
Each concrete meter can be used as a decorator via the :meth:`monitor`
class method::
from qgis_profiler.meters.recovery_measurer import RecoveryMeasurer
from qgis_profiler.meters.thread_health_checker import MainThreadHealthChecker
@RecoveryMeasurer.monitor(name="Load Layers")
def load_layers():
# Recovery time is measured after this function returns
pass
@MainThreadHealthChecker.monitor(name="Heavy Processing")
def heavy_processing():
# Thread health is monitored during execution
pass
"""
__metaclass__ = abc.ABCMeta
_short_name: ClassVar[str] = ""
anomaly_detected = pyqtSignal(MeterAnomaly)
def __init__(self, supports_continuous_measurement: bool = False) -> None: # noqa: FBT001, FBT002
"""Initialize the meter with optional continuous measurement support."""
super().__init__(None)
self._default_context = MeterContext(
self.__class__.__name__, Settings.meters_group.get()
)
self._context_stack: list[MeterContext] = []
self._enabled = True
self._connected_to_profiler = False
self._supports_continuous_measuring: bool = supports_continuous_measurement
self._is_measuring: bool = False
def __del__(self) -> None:
"""Ensure cleanup when the object is garbage collected."""
self.cleanup()
[docs]
@classmethod
@abc.abstractmethod
def get(cls: type["Meter"]) -> "Meter":
"""Get a singleton instance of the meter."""
[docs]
@classmethod
def monitor( # noqa: PLR0913
cls,
function: Callable | None = None,
*,
name: str | None = None,
group: str | None = None,
name_args: list[str] | None = None,
connect_to_profiler: bool = True,
start_continuous_measuring: bool = True,
measure_after_call: bool = False,
) -> Callable:
"""Decorate a function to monitor it with this meter.
Set the meter context within a function call, start continuous measuring
if supported, and optionally measure the meter after the function call.
If the meter is disabled, nothing is done.
If you want to profile the anomalies found during the function call,
connect to profiler using the `connect_to_profiler` argument.
:param function: Provided here to support both @monitor and @monitor() syntax.
:param name: Optional name for this context. If not provided, the
name of the function being wrapped will be used.
:param group: Optional group name for the context. If not provided,
the group name is read from settings.
:param name_args: Optional list of argument names to include in the context
name. If specified, the context name will include these argument values.
:param connect_to_profiler: Optional flag to connect to meter to a profiler
if not yet connected.
:param start_continuous_measuring: Optional flag to start continuous measuring.
:param measure_after_call: Optional flag to measure the meter after the function
call. For some meters this might be expensive.
:return: A callable decorator function that wraps the given function to
set the meter context during the function call.
"""
if function is None: # @monitor() syntax
def decorator(func: Callable) -> Callable:
return cls.monitor(
function=func,
name=name,
group=group,
name_args=name_args,
connect_to_profiler=connect_to_profiler,
start_continuous_measuring=start_continuous_measuring,
measure_after_call=measure_after_call,
)
return decorator
# @monitor syntax
@wraps(function)
def wrapper(*args: Any, **kwargs: Any) -> Any:
func = cast("Callable", function)
group_name = resolve_group_name_with_cache(group)
context_name = name if name is not None else func.__name__
if name_args:
context_name += qgis_profiler.utils.parse_arguments(
func, name_args, args, kwargs
)
meter = cls.get()
if not meter.enabled:
return func(*args, **kwargs)
if connect_to_profiler and not meter.is_connected_to_profiler:
meter.connect_to_profiler()
with meter.context(context_name, group_name):
if (
start_continuous_measuring
and meter.supports_continuous_measuring
and not meter.is_measuring
):
meter.start_measuring()
try:
return func(*args, **kwargs)
finally:
QgsApplication.processEvents()
if measure_after_call:
meter.measure()
return wrapper
@property
def current_context(self) -> MeterContext:
""":return The current context of the meter."""
context = (
self._context_stack[-1] if self._context_stack else self._default_context
)
if self._short_name:
return context.with_meter_suffix(self._short_name)
return context
@property
def is_connected_to_profiler(self) -> bool:
""":return Whether the meter is connected to the profiler."""
return self._connected_to_profiler
@property
def supports_continuous_measuring(self) -> bool:
"""Return whether the meter supports continuous measurement."""
return self._supports_continuous_measuring
@property
def enabled(self) -> bool:
""":return Whether the meter is enabled."""
return self._enabled
@enabled.setter
def enabled(self, enabled: bool) -> None:
self._enabled = enabled
@property
def is_measuring(self) -> bool:
"""Return whether the meter is currently measuring."""
return self._is_measuring
[docs]
@contextmanager
def context(self, name: str, group: str) -> Generator[MeterContext, None, None]:
"""Context manager for the meter in certain context."""
self.add_context(name, group)
try:
yield self.current_context
finally:
self.pop_context()
[docs]
def add_context(self, name: str, group: str) -> None:
"""Add context to the context stack."""
self._context_stack.append(MeterContext(name, group))
[docs]
def pop_context(self) -> MeterContext | None:
"""Remove the last context from the context stack if it exists.
:return: Context or None if context stack is empty.
"""
if self._context_stack:
return self._context_stack.pop()
return None
[docs]
def connect_to_profiler(self) -> None:
"""Connect anomaly detection signal to profiler's anomaly handling.
Establish a connection between the
`anomaly_detected` signal and the `_profile_anomaly`
handler to ensure anomalies are routed correctly.
:return: None
"""
self.anomaly_detected.connect(self._profile_anomaly)
self._connected_to_profiler = True
[docs]
def measure(self) -> float | None:
"""Measure once with the meter.
Signal anomaly_detected will be emitted if applicable.
:return: Duration in seconds or None if meter is disabled.
"""
if self.enabled:
duration, anomaly_detected = self._measure()
if anomaly_detected:
self._emit_anomaly(duration)
return duration
return None
[docs]
def start_measuring(self) -> bool:
"""Start the measurement process.
:return: A boolean indicating if the measurement
process was initiated successfully.
"""
if self._supports_continuous_measuring:
self._is_measuring = True
return self._start_measuring()
return False
[docs]
def stop_measuring(self) -> None:
"""Stop the continuous measurement process if applicable."""
self._is_measuring = False
self._stop_measuring()
[docs]
def cleanup(self) -> None:
"""Cleanup the meter and stop measuring if continuous measuring is supported."""
self.stop_measuring()
with suppress(TypeError):
self.anomaly_detected.disconnect(self._profile_anomaly)
self._connected_to_profiler = False
@staticmethod
@pyqtSlot(MeterAnomaly)
def _profile_anomaly(anomaly: MeterAnomaly) -> None:
"""Profile the anomaly."""
ProfilerWrapper.get().add_record(
anomaly.context.name, anomaly.context.group, anomaly.duration_seconds
)
[docs]
@abc.abstractmethod
def reset_parameters(self) -> None:
"""Reset the meter parameters based on setting values."""
@abc.abstractmethod
def _measure(self) -> tuple[float, bool]:
"""Perform actual measurement.
:return: A tuple containing a duration in seconds and a
boolean indicating whether an anomaly was detected.
"""
def _emit_anomaly(self, duration: float) -> None:
"""Emit anomaly_detected signal."""
self.anomaly_detected.emit(MeterAnomaly(self.current_context, duration))
def _start_measuring(self) -> bool:
return False
def _stop_measuring(self) -> None:
pass