# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Image window controller module.
This module contains the controller class for managing image windows in the radioviz application.
It handles the interaction between the image data, the view, and various mouse events.
"""
from __future__ import annotations
import sys
from pathlib import Path
from typing import TYPE_CHECKING
import matplotlib.backend_bases
if sys.version_info >= (3, 11):
from typing import Any, Optional, Self
else:
from typing import Any, Optional
from typing_extensions import Self
import numpy as np
import tifffile
from cmap import Colormap
from imageio.v3 import imwrite
from PySide6.QtCore import Signal
from skimage.transform import downscale_local_mean
from superqt.utils import thread_worker
from radioviz.__about__ import __version__
from radioviz.controllers.sub_window_controller import SubWindowController, SubWindowEnum
from radioviz.models.data_model import DataSource, DataStorage
from radioviz.models.mouse_state import MouseEvent
from radioviz.services.export_services import ExportSpec
from radioviz.services.save_services import SaveSpec
from radioviz.services.spatial_calibration import DistanceUnit, select_distance_unit
from radioviz.services.tiff_metadata import (
build_tiff_extratags,
build_tiff_kwargs,
derive_tiff_metadata,
metadata_entries,
pixel_size_from_metadata,
)
from radioviz.services.workspace_manager import ImageWorkspaceSpec, WorkspaceReferenceManager
from radioviz.services.xyz_format import encode_xyz_array
from radioviz.views.image_window import ImageWindow
[docs]
class ImageWindowController(SubWindowController[ImageWindow]):
"""Controller for managing image windows in the application.
It is responsible for the business logic and for the data ownership.
This class handles the logic for displaying and interacting with image data
through the image window view. It manages mouse events, display properties,
and communication between the model and view layers.
:ivar request_canvas_update: Signal emitted to request canvas update
:ivar point_selection_mode: Signal emitted when point selection mode changes
"""
sub_window_type = SubWindowEnum.ImageWindow
request_canvas_update = Signal() # it does not send anything, because the controller has to remain the data owner
point_selection_mode = Signal(bool)
display_resolution_ready = Signal(int)
axis_label_mode_changed = Signal(str)
max_display_size = 2**11
allowed_scale_factor = (1, 2, 4, 8, 16)
def __init__(
self,
source: DataSource,
image_data: np.ndarray,
storage: Optional[DataStorage] = None,
metadata: Optional[dict[str, Any]] = None,
view: Optional[ImageWindow] = None,
) -> None:
"""Initialize the ImageWindowController.
:param source: The source information for the image
:type source: DataSource
:param image_data: The actual image data as a numpy array
:type image_data: numpy.ndarray
:param metadata: Optional metadata associated with the image
:type metadata: Optional[dict[str, Any]]
:param view: Optional view instance to associate with this controller
:type view: ImageWindow, optional
"""
super().__init__(source, storage, view)
self.path = self.storage.path
self.name = self.source.label
self.image_data = image_data
self.metadata = metadata or {}
self._resolution_levels: dict[int, np.ndarray] = {}
self._pending_resolutions: set[int] = set()
self._coarse_factor: int = 1
self.initialize_display_data()
self.vmin_init, self.vmax_init = self._set_vmin_vmax()
self.display_properties = self._set_default_display_properties()
self.save_specs = [
SaveSpec(
key='image-data',
label='Image',
filter='TIFF (*.tif *.tiff);;PNG (*.png);;XYZ (*.xyz)',
default_suffix='tif',
suffix_map={'TIFF (*.tif *.tiff)': 'tif', 'PNG (*.png)': 'png', 'XYZ (*.xyz)': 'xyz'},
writer=self.save_to_path,
)
]
self.export_specs = [
ExportSpec(
key='full-canvas',
label='Full canvas',
filter='Supported image formats (*.eps *.jpg *.jpeg *.pdf *.pgf *.png *.ps *.raw *rgba *.svg *.svgz *.tif *.tiff *.webp)',
default_suffix='png',
suffix_map={},
writer=self.export_to_path,
)
]
self.state_changed.connect(self.application_services.app_state.changed)
[docs]
def _set_vmin_vmax(self) -> tuple[int, int]:
"""Set the initial vmin and vmax values for the image display.
Determines the minimum and maximum values from the image data to establish
the initial display range. If no image data is available, defaults to a
16-bit unsigned integer range.
:return: A tuple containing (vmin, vmax) for the image display
:rtype: tuple[int, int]
"""
if self.image_data is None:
return 0, 2**16 - 1
else:
return np.min(self.image_data), np.max(self.image_data)
[docs]
def _set_default_display_properties(self) -> dict[str, Any]:
"""Set the default display properties for the image.
Initializes the display properties dictionary with default values including
vmin, vmax, colormap ('hot'), and interpolation ('none').
:return: Dictionary containing default display properties
:rtype: dict[str, Any]
"""
return {
'vmin': self.vmin_init,
'vmax': self.vmax_init,
'cmap': 'hot',
'interpolation': 'none',
'axis_label_mode': 'pixel',
}
[docs]
def set_display_properties(self, params: dict[str, Any]) -> None:
"""Update the display properties with the provided parameters.
Updates the display properties dictionary with new values. For colormap values,
converts string colormaps to matplotlib format using Colormap.to_mpl().
:param params: Dictionary containing display property updates
:type params: dict
"""
for key, value in params.items():
if key == 'cmap' and isinstance(value, str):
self.display_properties['cmap'] = Colormap(value).to_mpl()
elif key == 'axis_label_mode':
if value in {'off', 'pixel', 'distance'}:
self.display_properties['axis_label_mode'] = value
self.axis_label_mode_changed.emit(value)
else:
self.display_properties[key] = value
[docs]
def mpl_display_properties(self) -> dict[str, Any]:
"""
Return display properties suitable for Matplotlib imshow/update.
:return: Display properties filtered for Matplotlib.
:rtype: dict[str, Any]
"""
props = self.display_properties.copy()
props.pop('axis_label_mode', None)
props.pop('scalebar', None)
return props
@property
def axis_label_mode(self) -> str:
"""
Current axis label mode.
:return: Axis label mode.
:rtype: str
"""
return str(self.display_properties.get('axis_label_mode', 'pixel'))
@property
def scalebar_settings(self) -> dict[str, Any]:
"""
Current scalebar settings dictionary.
:return: Scalebar settings.
:rtype: dict[str, Any]
"""
value = self.display_properties.get('scalebar')
return value if isinstance(value, dict) else {}
[docs]
def pixel_size_m(self) -> tuple[float, float] | None:
"""
Pixel size in meters derived from metadata, if available.
:return: Tuple of pixel sizes in meters or None.
:rtype: tuple[float, float] | None
"""
return pixel_size_from_metadata(self.metadata)
[docs]
def distance_unit(self) -> DistanceUnit | None:
"""
Select a suitable distance unit based on image size and pixel size.
:return: Distance unit or None when unavailable.
:rtype: DistanceUnit | None
"""
pixel_size = self.pixel_size_m()
if pixel_size is None or self.image_data is None:
return None
pixel_x_m, pixel_y_m = pixel_size
height, width = self.image_data.shape[:2]
extent_x_m = width * pixel_x_m
extent_y_m = height * pixel_y_m
max_extent_m = max(extent_x_m, extent_y_m)
return select_distance_unit(max_extent_m)
[docs]
def initialize_display_data(self) -> None:
"""Initialize the display data with the coarsest resolution level.
Sets the initial coarse factor to the maximum allowed scale factor
and emits a signal indicating that the display resolution is ready.
"""
self._coarse_factor = self.allowed_scale_factor[-1]
self.display_resolution_ready.emit(self._coarse_factor)
[docs]
def request_resolution_level(self, factor: int) -> None:
"""Request a specific resolution level for the image.
This method snaps the requested factor to the nearest allowed scale factor,
checks if the level is already available or pending, and launches a worker
to compute the level if needed.
:param factor: The requested resolution factor
:type factor: int
"""
factor = self._snap_factor(factor)
if factor in self._resolution_levels:
self.display_resolution_ready.emit(factor)
return
if factor in self._pending_resolutions:
return
self._pending_resolutions.add(factor)
self._launch_worker(factor)
[docs]
def get_resolution_level(self, factor: int) -> np.ndarray | None:
"""Get the image data for a specific resolution level.
:param factor: The resolution factor to retrieve
:type factor: int
:return: The image data for the specified resolution level or None
:rtype: numpy.ndarray or None
"""
return self._resolution_levels.get(factor)
[docs]
def _snap_factor(self, factor: int) -> int:
"""Snap the given factor to the nearest allowed scale factor.
Finds the smallest allowed scale factor that is greater than or equal to
the provided factor. If no such factor exists, returns the largest allowed factor.
:param factor: The factor to snap
:type factor: int
:return: The snapped factor
:rtype: int
"""
for f in self.allowed_scale_factor:
if f >= factor:
return f
return self.allowed_scale_factor[-1]
[docs]
def _launch_worker(self, factor: int) -> None:
"""Launch a worker thread to compute the image at the specified resolution.
Creates a worker using the :func:`scale_image` function and connects its returned
signal to the :meth:`_on_level_ready` method.
:param factor: The resolution factor to compute
:type factor: int
"""
worker = scale_image(self.image_data, factor)
worker.returned.connect(self._on_level_ready)
worker.start()
[docs]
def _on_level_ready(self, result: tuple[int, np.ndarray]) -> None:
"""Handle the completion of a resolution level computation.
Stores the computed image data and removes the factor from pending resolutions,
then emits a signal indicating the resolution is ready.
:param result: Tuple containing (factor, image_data)
:type result: tuple[int, numpy.ndarray]
"""
factor, image = result
self._resolution_levels[factor] = image
self._pending_resolutions.remove(factor)
self.display_resolution_ready.emit(factor)
[docs]
def set_view(self, view: 'ImageWindow') -> None:
"""Set the view associated with this controller.
Connects the view signals to the appropriate controller methods.
:param view: The image window view to associate with this controller
:type view: radioviz.views.sub_window.SubWindow
:return: None
:rtype: None
"""
super().set_view(view)
# wire connections
if self.view is not None: # why? it should not happen
self.view.about_to_close.connect(self.about_to_close)
[docs]
def _press(self, ev: matplotlib.backend_bases.MouseEvent) -> None:
"""Handle mouse press events.
:param ev: Mouse event from matplotlib
:type ev: matplotlib.backend_bases.MouseEvent
"""
if self.active_tool:
radioviz_mouse_event = MouseEvent.from_mpl_mouse_event(ev)
self.active_tool.on_mouse_press(radioviz_mouse_event)
[docs]
def _move(self, ev: matplotlib.backend_bases.MouseEvent) -> None:
"""Handle mouse move events.
:param ev: Mouse event from matplotlib
:type ev: matplotlib.backend_bases.MouseEvent
"""
if self.active_tool:
radioviz_mouse_event = MouseEvent.from_mpl_mouse_event(ev)
self.active_tool.on_mouse_move(radioviz_mouse_event)
[docs]
def _context(self, pos: tuple[float, float]) -> None:
"""Handle context menu requests.
:param pos: Position where context menu was requested
:type pos: tuple
"""
self.context_menu_requested.emit(pos, self)
[docs]
def on_view_created(self) -> None:
"""Perform actions after the view has been created.
Connects the canvas update signal to the view's update method.
"""
if TYPE_CHECKING:
assert isinstance(self.view, ImageWindow)
self.request_canvas_update.connect(self.view.update_canvas)
[docs]
def default_file_name(self) -> str:
"""Return the default file name for saving the image.
:return: Default file name based on the image source label
:rtype: str
"""
return self.name
[docs]
def save_to_path(self, path: Path) -> None:
"""Save the image data to the specified path.
:param path: The file path where the image will be saved
:type path: Path
:return: None
:rtype: None
"""
self._save_image_data(path)
self.mark_saved(path)
[docs]
def get_export_specs(self) -> list[ExportSpec]:
"""Get the list of available export specifications for the image.
:return: List of export specifications or empty list if no image data exists
:rtype: list[ExportSpec]
"""
if self.image_data is None:
return []
return self.export_specs
[docs]
def export_to_path(self, path: Path) -> None:
"""Export the current view to the specified path.
:param path: The file path where the export will be saved
:type path: Path
:return: None
:rtype: None
"""
if self.view is None:
return
if isinstance(self.view, ImageWindow):
self.view.export_to_path(path)
[docs]
def can_export_as(self) -> bool:
"""Check if the current image can be exported.
:return: True if the image can be exported, False otherwise
:rtype: bool
"""
return True
[docs]
def get_save_specs(self) -> list[SaveSpec]:
"""Get the list of available save specifications for the image.
:return: List of save specifications or empty list if no image data exists
:rtype: list[SaveSpec]
"""
if self.image_data is None:
return []
return self.save_specs
[docs]
def _save_image_data(self, path: Path) -> None:
"""Save the internal image data to the specified path using imageio.
:param path: The file path where the image will be saved
:type path: Path
:return: None
:rtype: None
"""
if path.suffix.lower() == '.xyz':
path.write_bytes(encode_xyz_array(self.image_data))
self.message_requested.emit(
'Saved XYZ without DAT sidecar. Metadata will not be preserved unless a DAT file is created separately.',
'warning',
5000,
)
elif path.suffix.lower() in {'.tif', '.tiff'}:
extratags = build_tiff_extratags(
metadata=self.metadata,
derivation_note=self._describe_derivation(),
software_suffix=f'RadioViz {__version__}',
)
tiff_kwargs = build_tiff_kwargs(
metadata=self.metadata,
derivation_note=self._describe_derivation(),
software_suffix=f'RadioViz {__version__}',
)
tifffile.imwrite(path, self.image_data, metadata=None, extratags=extratags, **tiff_kwargs)
else:
imwrite(path, self.image_data)
[docs]
def to_workspace(self, include_data: bool) -> ImageWorkspaceSpec:
"""Convert the current image controller state to a workspace specification.
Creates an ImageWorkspaceSpec containing the controller's identification,
source information, storage details, display parameters, and optionally the
image data itself based on the include_data flag.
:param include_data: Flag indicating whether to include the image data in the spec
:type include_data: bool
:return: ImageWorkspaceSpec containing the controller's state
:rtype: ImageWorkspaceSpec
"""
return ImageWorkspaceSpec(
id=self.id,
source=self.source,
storage=self.storage,
display_parameters=self.display_properties,
metadata=self.metadata,
data=self.image_data if include_data else None,
)
[docs]
def _describe_derivation(self, source: DataSource | None = None) -> str | None:
"""
Describe the derivation of the current or provided data source.
:param source: Optional data source to describe; defaults to controller source.
:type source: DataSource | None
:return: Derivation description or None if not derived.
:rtype: str | None
"""
source = source or self.source
derivation = source.derivation
if derivation is None:
return None
parts: list[str] = []
if source.parent is not None:
parts.append(f'Parent={source.parent.label}')
roi = derivation.roi
geom = roi.geometry
if 'extents' in geom:
extents = geom['extents']
formatted_extents = self._format_extents(extents)
parts.append(f'ROI {roi.type} extents={formatted_extents}')
if 'angle' in geom:
parts.append(f'ROI angle={geom.get("angle")}')
if derivation.transform is not None:
params = derivation.transform.params
param_text = ', '.join(f'{k}={params[k]}' for k in sorted(params))
parts.append(f'Transform {derivation.transform.type} {param_text}'.strip())
if not parts:
return 'RadioViz derivation'
return 'RadioViz derivation: ' + '; '.join(parts)
[docs]
def from_workspace(self, spec: ImageWorkspaceSpec, context: WorkspaceReferenceManager) -> Self:
"""From Workspace placeholder-"""
raise NotImplementedError
[docs]
@thread_worker
def scale_image(input_image: np.ndarray, factor: int) -> tuple[int, np.ndarray]:
"""Scale an image to a specified factor using local mean downsampling.
:param input_image: The input image to scale
:type input_image: numpy.ndarray
:param factor: The scaling factor
:type factor: int
:return: Tuple of (factor, scaled_image)
:rtype: tuple[int, numpy.ndarray]
"""
if factor == 1:
return factor, input_image
return factor, downscale_local_mean(input_image, factor) # type: ignore[no-untyped-call]