# Copyright 2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Performance tracing helpers.
This module provides opt-in timing utilities for diagnosing slow paths.
Tracing is controlled via environment variables and can be applied with
either a context manager or a decorator.
Example usage::
from radioviz.services.perf_tracing import (
PerfTracer,
is_env_tracing_enabled,
trace_env,
)
# Context manager style
with PerfTracer(
'image_window._update_canvas',
enabled=is_env_tracing_enabled('RADIOVIZ_PERF_TRACE'),
factor=4,
):
update_canvas()
# Decorator style
@trace_env('RADIOVIZ_PERF_TRACE', name='load_image')
def load_image(path): ...
"""
from __future__ import annotations
import logging
import os
import sys
import time
from typing import Callable, Literal, Optional, TypeVar
if sys.version_info >= (3, 10):
from typing import ParamSpec
else:
from typing_extensions import ParamSpec
_DEFAULT_LOGGER_NAME = 'radioviz.perf'
P = ParamSpec('P')
R = TypeVar('R')
[docs]
def is_env_tracing_enabled(env_var: str) -> bool:
"""
Check whether tracing is enabled via the given environment variable.
:param env_var: Name of the environment variable.
:type env_var: str
:return: ``True`` when tracing is enabled, ``False`` otherwise.
:rtype: bool
"""
value = os.getenv(env_var, '')
return value.strip().lower() in {'1', 'true', 'yes', 'on'}
[docs]
def perf_logger(name: str | None = None) -> logging.Logger:
"""
Return a dedicated logger for performance tracing.
:param name: Optional logger name override.
:type name: str | None
:return: Logger instance.
:rtype: logging.Logger
"""
return logging.getLogger(name or _DEFAULT_LOGGER_NAME)
[docs]
class PerfTracer:
"""
Context manager for measuring elapsed time of a traced step.
The tracer records elapsed time and logs it when enabled.
"""
def __init__(self, name: str, *, enabled: bool = True, logger: logging.Logger | None = None, **fields: object):
"""
Initialize the tracer.
:param name: Name of the traced step.
:type name: str
:param enabled: Whether tracing is enabled for this span.
:type enabled: bool
:param logger: Logger to emit timing information.
:type logger: logging.Logger | None
:param fields: Additional structured fields for logging.
:type fields: object
"""
self._name = name
self._fields = fields
self._enabled = enabled
self._logger = logger or perf_logger()
self._start: Optional[float] = None
self._elapsed_ms: Optional[float] = None
@property
def enabled(self) -> bool:
"""
Return whether tracing is enabled for this instance.
:return: ``True`` when enabled, ``False`` otherwise.
:rtype: bool
"""
return self._enabled
@property
def elapsed_ms(self) -> Optional[float]:
"""
Return the elapsed time in milliseconds.
:return: Elapsed milliseconds or ``None`` when disabled.
:rtype: Optional[float]
"""
return self._elapsed_ms
def __enter__(self) -> 'PerfTracer':
"""
Enter the context manager.
:return: Self.
:rtype: PerfTracer
"""
if self._enabled:
self._start = time.perf_counter()
return self
def __exit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: object | None,
) -> Literal[False]:
"""
Exit the context manager and emit logs when enabled.
:param exc_type: Exception type, if any.
:type exc_type: type | None
:param exc: Exception instance, if any.
:type exc: BaseException | None
:param tb: Traceback, if any.
:type tb: object | None
:return: ``False`` to propagate exceptions.
:rtype: Literal[False]
"""
if self._enabled and self._start is not None:
self._elapsed_ms = (time.perf_counter() - self._start) * 1000.0
message = self._format_message()
self._logger.debug(message)
return False
[docs]
def trace_if(
enabled: bool,
*,
name: str | None = None,
logger: logging.Logger | None = None,
**fields: object,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
"""
Decorator factory for conditional tracing.
:param enabled: Whether tracing is enabled.
:type enabled: bool
:param name: Optional name override for the span.
:type name: str | None
:param logger: Logger to emit timing information.
:type logger: logging.Logger | None
:param fields: Additional structured fields for logging.
:type fields: object
:return: Decorator.
:rtype: Callable
"""
def _decorator(func: Callable[P, R]) -> Callable[P, R]:
span_name = name or func.__qualname__
def _wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
if not enabled:
return func(*args, **kwargs)
with PerfTracer(span_name, enabled=True, logger=logger, **fields):
return func(*args, **kwargs)
return _wrapped
return _decorator
[docs]
def trace_env(
env_var: str,
*,
name: str | None = None,
logger_name: str | None = None,
**fields: object,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
"""
Decorator factory that enables tracing via an environment variable.
:param env_var: Name of the environment variable controlling tracing.
:type env_var: str
:param name: Optional name override for the span.
:type name: str | None
:param logger_name: Optional logger name override.
:type logger_name: str | None
:param fields: Additional structured fields for logging.
:type fields: object
:return: Decorator.
:rtype: Callable
"""
enabled = is_env_tracing_enabled(env_var)
logger = perf_logger(logger_name)
return trace_if(enabled, name=name, logger=logger, **fields)