Source code for radioviz.tools.tool_api

#  Copyright 2025–2026 European Union
#  Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
#  SPDX-License-Identifier: EUPL-1.2
"""
Tool API module for RadioViz application.

This module provides the core API for implementing tools within the RadioViz application.
It defines the base classes and interfaces that tools can use to interact with the application
environment, manage sessions, and access various services such as window management,
overlay rendering, and color handling.

The main components include:
- BaseToolSession for managing tool execution lifecycle
- ToolContext for providing application-facing APIs to tools
"""

from __future__ import annotations

from abc import abstractmethod
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Callable, Generic, Literal, TypeVar, Union
from uuid import UUID, uuid4

from PySide6.QtCore import QObject, Signal

from radioviz.controllers.image_window_controller import ImageWindowController
from radioviz.controllers.sub_window_controller import T_SubWindowController
from radioviz.models.overlays import OverlayRole, RendererFn
from radioviz.services.application_services import ApplicationServices
from radioviz.services.color_service import ColorService
from radioviz.services.window_factory import SubWindowProtocol
from radioviz.typing.tool_types import ToolSessionProtocol

if TYPE_CHECKING:
    from radioviz.controllers.sub_window_controller import SubWindowController
    from radioviz.tools.base_controller import ToolController

T_Tool_Controller = TypeVar('T_Tool_Controller', bound='ToolController[Any, Any]')
T_Session = TypeVar('T_Session', bound='BaseToolSession[Any, Any]')




@dataclass
class ToolSessionResult:
    """
    Data class representing the result of a tool session execution.

    This class encapsulates the session identifier and the payload returned
    by a tool session upon completion. It serves as a container for
    communication between tool sessions and their callers.

    :ivar session_id: Unique identifier for the tool session
    :vartype session_id: UUID
    :ivar payload: Result data returned by the tool session
    :vartype payload: Any
    """

    session_id: UUID
    payload: Any





class BaseToolSession(QObject, Generic[T_Tool_Controller, T_SubWindowController]):
    """
    Base class for tool execution sessions.

    This class manages the lifecycle of a tool session, including starting,
    canceling, and finishing operations. It provides a common interface for
    tool implementations to handle their execution state.

    :ivar finished: Signal emitted when the session finishes successfully
    :ivar canceled: Signal emitted when the session is canceled
    """

    finished = Signal()
    canceled = Signal()
    returned = Signal(ToolSessionResult)

    def __init__(
        self,
        tool_controller: T_Tool_Controller,
        window_controller: T_SubWindowController,
        parent: QObject | None = None,
    ) -> None:
        """
        Initialize a new tool session.

        :param tool_controller: The controller managing this tool session
        :type tool_controller: ToolController
        :param window_controller: The window controller associated with this session
        :type window_controller: SubWindowController[Any]
        :param parent: Parent QObject for Qt ownership management
        :type parent: QObject, optional
        """
        super().__init__(parent)
        self.tool_controller = tool_controller
        self.window_controller = window_controller
        self.session_id: UUID = uuid4()
        self._active = False
        self._payload = None



    def start(self) -> None:
        """
        Start the tool session.

        If the session is already active, this method does nothing.
        Otherwise, it sets the session as active and calls the on_start method.
        """
        if self._active:
            return
        self._active = True
        self.on_start()




    def cancel(self, reason: str = '') -> None:
        """
        Cancel the tool session.

        If the session is not active, this method does nothing.
        Otherwise, it sets the session as inactive, calls the on_cancel method,
        and emits the canceled signal.

        :param reason: Reason for cancellation
        :type reason: str
        """
        if not self._active:
            return
        self._active = False
        self.on_cancel(reason)
        self.canceled.emit()




    def finish(self) -> None:
        """
        Finish the tool session.

        If the session is not active, this method does nothing.
        Otherwise, it sets the session as inactive, calls the on_finish method,
        and emits the finished signal.
        """
        if not self._active:
            return
        self._active = False
        self.on_finish()
        self.finished.emit()
        self.returned.emit(ToolSessionResult(self.session_id, self._payload))




    def assign_payload(self, payload: Any) -> None:
        """
        Assign a payload to the tool session.

        This method stores the provided payload which will be included in the
        session result when the session finishes.

        :param payload: The data to be stored as the session's result payload
        :type payload: Any
        """
        self._payload = payload




    def is_active(self) -> bool:
        """
        Check if the session is currently active.

        :return: True if the session is active, False otherwise
        :rtype: bool
        """
        return self._active




    @abstractmethod
    def on_start(self) -> None:
        """
        Handle session start event.

        This method must be implemented by subclasses to define what happens
        when a session starts.
        """
        pass




    @abstractmethod
    def on_cancel(self, reason: str) -> None:
        """
        Handle session cancellation event.

        This method must be implemented by subclasses to define what happens
        when a session is cancelled.

        :param reason: Reason for cancellation
        :type reason: str
        """
        pass




    @abstractmethod
    def on_finish(self) -> None:
        """
        Handle session finish event.

        This method must be implemented by subclasses to define what happens
        when a session finishes.
        """
        pass






class WorkspaceRestoreContext(QObject):
    """
    Context manager for tracking tool sessions during workspace restoration.

    This class coordinates the restoration of multiple tool sessions by tracking
    their completion status. It ensures that a restoration process is only considered
    complete when all registered sessions have finished executing.

    It also acts as a container for running sessions avoiding a session to be garbage collected
    when its variable goes out of scope.

    :ivar finished: Signal emitted when all tracked sessions have completed
    """

    finished = Signal()

    def __init__(self) -> None:
        """
        Initialize the workspace restore context.
        """
        super().__init__()
        self._pending = 0
        self._sessions: dict[UUID, ToolSessionProtocol] = {}



    def register_session(self, session: ToolSessionProtocol) -> None:
        """
        Register a new tool session for tracking.

        This method adds a session to the tracking list and increments the pending
        count. The session will be monitored until it completes.

        :param session: The tool session to register
        :type session: ToolSessionProtocol
        """
        self._pending += 1
        self._sessions[session.session_id] = session




    def _on_session_finished(self, session_id: UUID) -> None:
        """
        Internal handler for session completion events.

        This method removes a completed session from tracking and decrements the
        pending count. When all sessions are complete, it emits the finished signal.

        :param session_id: The unique identifier of the completed session
        :type session_id: UUID
        """
        if session_id in self._sessions:
            del self._sessions[session_id]
            self._pending -= 1
            if self._pending == 0:
                self.finished.emit()






class ToolContext(QObject):
    """
    Application-facing API exposed to tools.

    This class provides a comprehensive API for tools to interact with the
    RadioViz application environment. It offers access to window management,
    overlay systems, color services, and other application-level features.

    :ivar message_requested: Signal emitted when a message needs to be shown
    """

    message_requested = Signal(str, str, int)  # text, level, timeout

    def __init__(self) -> None:
        """
        Initialize the tool context with application services.
        """
        super().__init__()
        self.application_services = ApplicationServices()
        self._window_manager = self.application_services.window_manager
        self._window_factory = self.application_services.window_factory
        self._overlay_factory = self.application_services.overlay_factory
        self._color_service = self.application_services.color_service
        self._workspace_reference_manger = self.application_services.workspace_reference_manager
        self._tool_controllers: list['ToolController[Any, Any]'] = []
        self._active_tool_controller: ToolController[Any, Any] | None = None

        # forward window-manager signal
        self._window_manager.active_changed.connect(self._on_active_changed)
        self._active_callbacks: list[Callable[[SubWindowController[Any] | None], None]] = []

    @property
    def active_tool_controller(self) -> ToolController[Any, Any] | None:
        """
        Return the currently active tool controller.

        :return: Active tool controller or ``None`` when no tool is active.
        :rtype: ToolController | None
        """
        return self._active_tool_controller



    def request_tool_activation(self, controller: 'ToolController[Any, Any]') -> None:
        """
        Request activation of the given tool controller.

        If another tool is active, it is deactivated before the new controller
        becomes active. This enforces a single active tool at any time.

        :param controller: Tool controller requesting activation.
        :type controller: ToolController
        """
        if self._active_tool_controller is not None:
            self._active_tool_controller.deactivate()
        self._active_tool_controller = controller




    def clear_active_tool(self, controller: 'ToolController[Any, Any]') -> None:
        """
        Clear the active tool if it matches the given controller.

        :param controller: Tool controller to clear.
        :type controller: ToolController
        """
        if self._active_tool_controller is controller:
            self._active_tool_controller = None




    def activate_image(self, window_controller: 'SubWindowController[Any]') -> None:
        """
        Activate a specific image window.

        :param window_controller: The window controller to activate
        :type window_controller: SubWindowController[Any]
        """
        self._window_manager.active_window = window_controller




    def active_image(self) -> Union['SubWindowController[Any]', None]:
        """
        Get the currently active window controller.

        :return: The active window controller or None if no window is active
        :rtype: Union[SubWindowController[Any], None]
        """
        return self._window_manager.active_window




    def on_active_image_changed(self, callback: Callable[[SubWindowController[Any] | None], None]) -> None:
        """
        Register a callback for active image change events.

        :param callback: Function to call when active image changes
        :type callback: Callable
        """
        self._active_callbacks.append(callback)




    def _on_active_changed(self, image_controller: 'SubWindowController[Any] | None') -> None:
        """
        Internal handler for active window changes.

        :param image_controller: The new active window controller
        :type image_controller: SubWindowController[Any]
        """
        for cb in self._active_callbacks:
            cb(image_controller)




    def all_images(self) -> list['SubWindowController[Any]']:
        """
        Get a list of all available image windows.

        :return: List of all image window controllers
        :rtype: list[SubWindowController[Any]]
        """
        return list(self._window_manager.window_list)




    def on_list_changed(self, callback: Callable[[], None]) -> None:
        """
        Register a callback for window list change events.

        :param callback: Function to call when window list changes
        :type callback: Callable
        """
        self._window_manager.list_changed.connect(callback)




    def request_redraw(self) -> None:
        """
        Request a redraw of the active image window.

        This method triggers a canvas update on the currently active image window.
        """
        image = self.active_image()
        if isinstance(image, ImageWindowController):
            image.request_canvas_update.emit()




    def show_message(self, text: str, level: Literal['info', 'warning', 'error'], timeout_ms: int = 3000) -> None:
        """
        Show a message in the application UI.

        :param text: The message text to display
        :type text: str
        :param level: The severity level of the message
        :type level: Literal['info', 'warning', 'error']
        :param timeout_ms: Timeout in milliseconds before message disappears
        :type timeout_ms: int
        """
        self.message_requested.emit(text, level, timeout_ms)




    def get_color_service(self) -> ColorService:
        """
        Get the application's color service.

        :return: The color service instance
        :rtype: ColorService
        """
        return self._color_service




    def register_new_window_type(self, kind: str, view_cls: type[SubWindowProtocol], overwrite: bool = False) -> None:
        """
        Register a new window type with the application.

        :param kind: The identifier for the new window type
        :type kind: str
        :param view_cls: The window class to register
        :type view_cls: type[SubWindowProtocol]
        :param overwrite: Whether to overwrite existing registration
        :type overwrite: bool
        """
        self._window_factory.register(kind, view_cls, overwrite=overwrite)




    def register_new_overlay(self, overlay_type: str, overlay_role: OverlayRole, renderer: RendererFn) -> None:
        """
        Register a new overlay type with the application.

        :param overlay_type: The identifier for the new overlay type
        :type overlay_type: str
        :param overlay_role: The role this overlay plays
        :type overlay_role: OverlayRole
        :param renderer: The renderer function for this overlay
        :type renderer: RendererFn
        """
        self._overlay_factory.register(overlay_type, overlay_role, renderer)




    def register_workspace_reference(self, obj_id: UUID, obj: object) -> None:
        """
        Register an object with a unique identifier in the workspace reference manager.

        This method allows tools to store objects with a specific ID so they can be
        retrieved later using the ID. This is useful for maintaining references to
        objects across different tool sessions or operations.

        :param obj_id: The unique identifier for the object
        :type obj_id: UUID
        :param obj: The object to register
        :type obj: object
        """
        self._workspace_reference_manger.register(obj_id, obj)




    def get_reference_by_id(self, obj_id: UUID) -> object:
        """
        Retrieve an object by its unique identifier from the workspace reference manager.

        This method fetches an object that was previously registered with the given ID.
        If no object is found with the specified ID, this method will raise a KeyError.

        :param obj_id: The unique identifier of the object to retrieve
        :type obj_id: UUID
        :return: The registered object
        :rtype: object
        """
        return self._workspace_reference_manger.resolve(obj_id)




    def register_tool_controller(self, controller: 'ToolController[Any, Any]') -> None:
        """
        Register a tool controller in this context.

        :param controller: Tool controller instance
        :type controller: ToolController
        """
        self._tool_controllers.append(controller)




    def clear_tool_controllers(self) -> None:
        """
        Clear all registered tool controllers.
        """
        self._tool_controllers.clear()




    def iter_tool_controllers(self) -> tuple['ToolController[Any, Any]', ...]:
        """
        Return all registered tool controllers.

        :return: A tuple of registered controllers
        :rtype: tuple[ToolController, ...]
        """
        return tuple(self._tool_controllers)




    def collect_dependency_counts(self, image_ids: list[UUID]) -> dict[UUID, dict[str, int]]:
        """
        Collect dependency counts for the specified image IDs.

        :param image_ids: Image controller UUIDs to check
        :type image_ids: list[UUID]
        :return: Mapping of image_id to per-tool dependency counts
        :rtype: dict[UUID, dict[str, int]]
        """
        counts: dict[UUID, dict[str, int]] = {image_id: {} for image_id in image_ids}
        if not image_ids:
            return counts

        by_id = {window.id: window for window in self._window_manager.window_list}
        for controller in self.iter_tool_controllers():
            for image_id in image_ids:
                window = by_id.get(image_id)
                if window is None:
                    continue
                count_fn = getattr(controller, 'count_dependencies_for_image', None)
                if count_fn is None:
                    continue
                count = count_fn(window)
                if count:
                    counts[image_id][controller.tool_id] = count

        return counts