Napari PySPIM Plugin API Reference

napari_pyspim

diSPIM Processing Pipeline - A napari plugin for processing dual-view SPIM data.

Classes

DispimPipelineWidget(napari_viewer)

Bases: QWidget

Main widget containing all diSPIM processing steps.

Source code in packages/napari-pyspim/src/napari_pyspim/_main_widget.py

def __init__(self, napari_viewer):
    super().__init__()
    self.viewer = napari_viewer
    # Shared remote client across all tabs
    self.remote_client = RemoteClient()
    self.setup_ui()

Functions

setup_ui()

Set up the user interface with tabs for each processing step.

Source code in packages/napari-pyspim/src/napari_pyspim/_main_widget.py

def setup_ui(self):
    """Set up the user interface with tabs for each processing step."""
    layout = QVBoxLayout()

    # Use Preferred horizontally so napari gives a reasonable initial width,
    # but Ignored vertically so the dock height is not forced to grow.
    # The scroll area below will handle vertical overflow instead.
    self.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Preferred)

    # Create tab widget
    self.tab_widget = QTabWidget()

    # Create individual step widgets
    self.remote_connection = RemoteConnectionWidget(
        self.viewer, self.remote_client
    )
    self.roi_detection = RoiDetectionWidget(self.viewer, self.remote_client)
    self.registration = RegistrationWidget(self.viewer, self.remote_client)
    self.deconvolution = DeconvolutionWidget(self.viewer, self.remote_client)

    # Add tabs
    self.tab_widget.addTab(self.remote_connection, "0. Remote Connection")
    self.tab_widget.addTab(self.roi_detection, "1. ROI Detection")
    self.tab_widget.addTab(self.registration, "2. Registration")
    self.tab_widget.addTab(self.deconvolution, "3. Deconvolution")

    # Connect signals for data flow between steps
    self._connect_signals()

    # Wrap the tab widget in a scroll area so the napari window size
    # stays fixed and a scrollbar appears when content overflows.
    scroll = QScrollArea()
    scroll.setWidget(self.tab_widget)
    scroll.setWidgetResizable(True)
    scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)
    scroll.setVerticalScrollBarPolicy(Qt.ScrollBarPolicy.ScrollBarAsNeeded)

    layout.addWidget(scroll)
    self.setLayout(layout)

RemoteClient(parent=None)

Bases: QObject

Manages SSH connection and communication with remote compute server.

Signals

connected Emitted when the remote server is ready. disconnected Emitted when the connection is closed. error : str Emitted when an error occurs. progress : str, int Emitted for progress updates from the server (message, percentage). command_response : dict Emitted when a command response arrives from the server. The dict contains keys: request_id, success, result, error.

Source code in packages/napari-pyspim/src/napari_pyspim/_remote_client.py

def __init__(self, parent=None):  # type: ignore[no-untyped-def]
    super().__init__(parent)
    self._ssh: paramiko.SSHClient | None = None
    self._jump_ssh: paramiko.SSHClient | None = None
    self._channel: paramiko.Channel | None = None
    self._sftp: paramiko.SFTPClient | None = None
    self._sender_thread: threading.Thread | None = None
    self._receiver_thread: threading.Thread | None = None
    self._message_queue: queue.Queue[dict | None] = queue.Queue()
    self._pending_callbacks: dict[int, Callable] = {}
    self._pending_progress: dict[int, Callable] = {}
    self._request_id: int = 0
    self._lock = threading.Lock()
    self._connected = False
    self._server_capabilities: dict = {}
    # Connection params for potential reuse
    self._host: str | None = None
    self._port: int | None = None
    self._username: str | None = None
    self._key_path: str | None = None
    self._remote_script_path: str | None = None

Attributes

is_connected: bool property

Return True if SSH connection is active.

server_capabilities: dict property

Return capabilities reported by the server on startup.

Functions

connect(host: str, port: int, username: str, auth_method: str, password: str | None = None, key_path: str | None = None, key_passphrase: str | None = None, remote_venv: str | None = None, jump_host: str | None = None) -> bool

Establish SSH connection and start remote Python session.

Parameters

host : str Remote server hostname or IP. port : int SSH port (default 22). username : str SSH username. auth_method : str "password" or "key". password : str, optional SSH password (required when auth_method="password"). key_path : str, optional Path to local SSH private key (required when auth_method="key"). key_passphrase : str, optional Passphrase for encrypted private key. remote_venv : str, optional Path to the pyspim virtualenv on the remote server. jump_host : str, optional Hostname of the jump (bastion) host to route through.

Returns

bool True if connection succeeded.

Source code in packages/napari-pyspim/src/napari_pyspim/_remote_client.py

def connect(
    self,
    host: str,
    port: int,
    username: str,
    auth_method: str,
    password: str | None = None,
    key_path: str | None = None,
    key_passphrase: str | None = None,
    remote_venv: str | None = None,
    jump_host: str | None = None,
) -> bool:
    """Establish SSH connection and start remote Python session.

    Parameters
    ----------
    host : str
        Remote server hostname or IP.
    port : int
        SSH port (default 22).
    username : str
        SSH username.
    auth_method : str
        ``"password"`` or ``"key"``.
    password : str, optional
        SSH password (required when auth_method="password").
    key_path : str, optional
        Path to local SSH private key (required when auth_method="key").
    key_passphrase : str, optional
        Passphrase for encrypted private key.
    remote_venv : str, optional
        Path to the pyspim virtualenv on the remote server.
    jump_host : str, optional
        Hostname of the jump (bastion) host to route through.

    Returns
    -------
    bool
        True if connection succeeded.
    """
    try:
        self._host = host
        self._port = port
        self._username = username
        self._key_path = key_path

        # --- Build connection kwargs (reused for jump and target) ---
        connect_kwargs: dict[str, Any] = {
            "username": username,
            "allow_agent": True,
            "look_for_keys": False,  # Only use explicitly provided key
            "timeout": 30,
        }

        if auth_method == "password":
            connect_kwargs["password"] = password
        elif auth_method == "key":
            connect_kwargs["key_filename"] = key_path
            if key_passphrase:
                connect_kwargs["passphrase"] = key_passphrase

        # --- SSH connection (with optional jump host) ---
        if jump_host:
            # Step 1: Connect to the jump host
            logger.info("[CONNECT] Connecting to jump host: %s@%s:%s",
                        username, jump_host, port)
            self._jump_ssh = paramiko.SSHClient()
            self._jump_ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            self._jump_ssh.connect(
                hostname=jump_host,
                port=port,
                **connect_kwargs,
            )

            # Step 2: Open a TCP channel through the jump host to the target
            jump_transport = self._jump_ssh.get_transport()
            if jump_transport is None:
                raise ConnectionError("Failed to get transport from jump host")
            dest_addr = (host, port)
            src_addr = ("127.0.0.1", 0)
            channel = jump_transport.open_channel("direct-tcpip", dest_addr, src_addr)
            logger.info("[CONNECT] TCP tunnel established to %s:%s via %s",
                        host, port, jump_host)

            # Step 3: Connect to the target host through the tunnel
            logger.info("[CONNECT] Connecting to target host through jump: %s@%s",
                        username, host)
            self._ssh = paramiko.SSHClient()
            self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            self._ssh.connect(
                hostname=host,
                sock=channel,
                **connect_kwargs,
            )
        else:
            # Direct connection (no jump host)
            logger.info("[CONNECT] Direct connection to %s@%s:%s",
                        username, host, port)
            self._ssh = paramiko.SSHClient()
            self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            self._ssh.connect(
                hostname=host,
                port=port,
                **connect_kwargs,
            )

        # --- SFTP ---
        self._sftp = self._ssh.open_sftp()

        # --- Upload server script ---
        local_script = os.path.join(
            os.path.dirname(__file__), "_remote_server.py"
        )
        self._remote_script_path = (
            f"/tmp/pyspim_remote_server_{os.getpid()}.py"
        )
        self._sftp.put(local_script, self._remote_script_path)

        # --- Start remote Python process ---
        # Prefer the venv's python binary directly (more reliable than
        # sourcing activate which can fail in non-interactive shells).
        if remote_venv:
            python_bin = os.path.join(remote_venv, "bin", "python")
            # Derive log path: <pyspim_root>/logs/pyspim_server_<pid>.log
            # (pyspim_root = parent of the venv directory)
            pyspim_root = os.path.dirname(remote_venv)
            log_dir = os.path.join(pyspim_root, "logs")
            self._remote_log_path = os.path.join(
                log_dir, f"pyspim_server_{os.getpid()}.log"
            )
            # Ensure the logs directory exists on the remote host
            try:
                self._sftp.mkdir(log_dir)
            except IOError:
                pass  # Directory already exists
            command = (
                f"PYSPIM_LOG_PATH={self._remote_log_path} "
                f"{python_bin} {self._remote_script_path} "
                f"2>{self._remote_log_path}"
            )
        else:
            self._remote_log_path = None
            command = f"python {self._remote_script_path}"

        logger.info("[CONNECT] Remote command: %s", command)
        self._channel = self._ssh.get_transport().open_session()
        self._channel.exec_command(command)

        # --- Start sender thread ---
        self._sender_thread = threading.Thread(
            target=self._sender_loop, daemon=True
        )
        self._sender_thread.start()

        # --- Wait for ready message (before starting receiver loop) ---
        # Deliberately start _receiver_loop AFTER receiving the ready
        # message to avoid a race where two threads read from the same
        # SSH channel concurrently.
        ready = self._receive_blocking(timeout=30.0)
        if ready.get("type") != "ready":
            raise ConnectionError(
                f"Unexpected message from server: {ready}"
            )
        self._server_capabilities = ready.get("capabilities", {})

        # --- Now start the receiver thread for ongoing messages ---
        self._receiver_thread = threading.Thread(
            target=self._receiver_loop, daemon=True
        )
        self._receiver_thread.start()

        self._connected = True
        self.connected.emit()
        return True

    except Exception as e:
        logger.exception("Failed to connect to remote server")
        self._cleanup()
        self.error.emit(f"Connection failed: {e}")
        return False

disconnect_session()

Terminate remote session and close SSH connection.

Named disconnect_session to avoid conflict with QObject.disconnect().

Source code in packages/napari-pyspim/src/napari_pyspim/_remote_client.py

def disconnect_session(self):
    """Terminate remote session and close SSH connection.

    Named ``disconnect_session`` to avoid conflict with QObject.disconnect().
    """
    if not self._connected:
        return

    # Send shutdown command (non-blocking best effort)
    try:
        self._send({"request_id": 0, "command": "shutdown", "params": {}})
        time.sleep(0.5)  # Give server time to respond
    except Exception:
        pass

    self._cleanup()
    self._connected = False
    self.disconnected.emit()

send_command(command: str, params: dict, callback: Callable | None = None, progress_callback: Callable | None = None) -> int

Send a command to the remote server.

Parameters

command : str Command name (e.g. "ping", "load_deskew"). params : dict Command-specific parameters. callback : callable, optional Called with the response dict when the command completes. progress_callback : callable, optional Called with (message, percentage) for progress updates.

Returns

int The request_id assigned to this command.

Source code in packages/napari-pyspim/src/napari_pyspim/_remote_client.py

def send_command(
    self,
    command: str,
    params: dict,
    callback: Callable | None = None,
    progress_callback: Callable | None = None,
) -> int:
    """Send a command to the remote server.

    Parameters
    ----------
    command : str
        Command name (e.g. ``"ping"``, ``"load_deskew"``).
    params : dict
        Command-specific parameters.
    callback : callable, optional
        Called with the response dict when the command completes.
    progress_callback : callable, optional
        Called with ``(message, percentage)`` for progress updates.

    Returns
    -------
    int
        The request_id assigned to this command.
    """
    if not self._connected:
        raise RuntimeError("Not connected to remote server")

    with self._lock:
        self._request_id += 1
        request_id = self._request_id

    logger.info("[SEND_COMMAND] request_id=%s, command=%s, has_callback=%s",
                 request_id, command, callback is not None)
    if callback:
        self._pending_callbacks[request_id] = callback
    if progress_callback:
        self._pending_progress[request_id] = progress_callback

    message = {
        "request_id": request_id,
        "command": command,
        "params": params,
    }
    self._send(message)
    logger.info("[SEND_COMMAND] Message queued. Pending callbacks: %s",
                 list(self._pending_callbacks.keys()))
    return request_id

send_command_blocking(command: str, params: dict, timeout: float = 300.0) -> dict

Send a command and block until the response arrives.

Parameters

command : str Command name. params : dict Command-specific parameters. timeout : float Maximum seconds to wait for a response.

Returns

dict The response dict from the server.

Raises

RuntimeError If not connected or the command times out / fails.

Source code in packages/napari-pyspim/src/napari_pyspim/_remote_client.py

def send_command_blocking(
    self,
    command: str,
    params: dict,
    timeout: float = 300.0,
) -> dict:
    """Send a command and block until the response arrives.

    Parameters
    ----------
    command : str
        Command name.
    params : dict
        Command-specific parameters.
    timeout : float
        Maximum seconds to wait for a response.

    Returns
    -------
    dict
        The response dict from the server.

    Raises
    ------
    RuntimeError
        If not connected or the command times out / fails.
    """
    if not self._connected:
        raise RuntimeError("Not connected to remote server")

    result_queue: queue.Queue[dict] = queue.Queue()

    def _cb(response: dict):
        result_queue.put(response)

    request_id = self.send_command(command, params, callback=_cb)

    try:
        response = result_queue.get(timeout=timeout)
    except queue.Empty:
        raise TimeoutError(
            f"Command '{command}' (request_id={request_id}) timed out "
            f"after {timeout}s"
        )

    if not response.get("success", False):
        raise RuntimeError(
            f"Remote command '{command}' failed: {response.get('error')}"
        )

    return response.get("result", {})

list_directory(remote_path: str) -> list[dict]

List directory contents via SFTP.

Returns

list[dict] Each dict has keys: name, is_dir, size.

Source code in packages/napari-pyspim/src/napari_pyspim/_remote_client.py

def list_directory(self, remote_path: str) -> list[dict]:
    """List directory contents via SFTP.

    Returns
    -------
    list[dict]
        Each dict has keys: ``name``, ``is_dir``, ``size``.
    """
    if not self._sftp:
        raise RuntimeError("SFTP client not available")

    import stat as _stat

    entries: list[dict] = []
    for attr in self._sftp.listdir_attr(remote_path):
        mode = attr.st_mode if attr.st_mode is not None else 0
        entries.append({
            "name": attr.filename,
            "is_dir": _stat.S_ISDIR(mode),
            "size": attr.st_size,
        })
    return entries

get_sftp_client() -> paramiko.SFTPClient

Return the active SFTP client for file browser operations.

Source code in packages/napari-pyspim/src/napari_pyspim/_remote_client.py

def get_sftp_client(self) -> paramiko.SFTPClient:
    """Return the active SFTP client for file browser operations."""
    if not self._sftp:
        raise RuntimeError("Not connected")
    return self._sftp

RemoteConnectionWidget(napari_viewer, remote_client: Optional[RemoteClient] = None)

Bases: QWidget

Tab 0 widget for SSH connection management.

Parameters

napari_viewer The napari viewer instance (passed by plugin infrastructure). remote_client : RemoteClient, optional Pre-existing client to reuse. If None a new one is created.

Source code in packages/napari-pyspim/src/napari_pyspim/_remote_connection.py

def __init__(
    self,
    napari_viewer,  # type: ignore[name-defined]
    remote_client: Optional[RemoteClient] = None,
):
    super().__init__()
    self.viewer = napari_viewer
    self.client = remote_client or RemoteClient()
    self._connecting = False
    self._waiting_for_test = False  # Track if we're waiting for test connection response
    self._setup_ui()
    self._load_config()
    self._connect_signals()

SftpBrowserDialog(sftp_client: paramiko.SFTPClient, parent=None, initial_path: str = '/', title: str = 'Browse Remote Directory', select_file: bool = False, file_filter: Optional[str] = None)

Bases: QDialog

Dialog for browsing remote directories or files via SFTP.

Parameters

sftp_client : paramiko.SFTPClient Active SFTP client. parent : QWidget, optional Parent widget. initial_path : str Starting directory path. title : str Dialog title. select_file : bool If True, allow selecting individual files (double-click or Select). If False, only directories can be selected (default). file_filter : str, optional Glob-like filter pattern for file selection mode (e.g., "*.npy *.tif"). Only files matching the pattern are selectable.

Source code in packages/napari-pyspim/src/napari_pyspim/_sftp_browser.py

def __init__(
    self,
    sftp_client: paramiko.SFTPClient,
    parent=None,  # type: ignore[no-untyped-def]
    initial_path: str = "/",
    title: str = "Browse Remote Directory",
    select_file: bool = False,
    file_filter: Optional[str] = None,
):
    super().__init__(parent)
    self.sftp = sftp_client
    self.selected_path: Optional[str] = None
    self._current_path = initial_path
    self._select_file = select_file
    self._file_filter = file_filter
    self.setWindowTitle(title)
    self.setMinimumSize(600, 400)
    self._model = _DirectoryModel(self)
    self._setup_ui()
    self._navigate_to(initial_path)

Modules

tests

Modules

test_basic

Basic tests that don't require CUDA compilation.

Functions

test_manifest()

Test that the manifest can be accessed without importing the full plugin.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_basic.py

def test_manifest():
    """Test that the manifest can be accessed without importing the full plugin."""
    # Import just the manifest
    import os
    import sys

    sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

    from napari_pyspim import manifest

    assert manifest is not None
    assert manifest["name"] == "napari-pyspim"
    assert manifest["display_name"] == "diSPIM Processing Pipeline"

test_utils()

Test utility functions.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_basic.py

def test_utils():
    """Test utility functions."""
    # Import utils directly
    import os
    import sys

    sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

    from napari_pyspim._utils import format_memory_usage, format_shape_string

    # Test memory formatting
    test_array = np.zeros((100, 100, 50))
    memory_str = format_memory_usage(test_array)
    assert "MB" in memory_str or "KB" in memory_str

    # Test shape formatting
    shape_str = format_shape_string((100, 100, 50))
    assert "100×100×50" in shape_str
    assert "(Z×Y×X)" in shape_str

test_plugin

Basic tests for the napari-pyspim plugin.

Classes Functions

test_plugin_import()

Test that the plugin can be imported.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_plugin_import():
    """Test that the plugin can be imported."""
    from napari_pyspim import manifest

    assert manifest is not None
    assert manifest["name"] == "napari-pyspim"

test_widget_creation()

Test that the main widget can be created.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_widget_creation():
    """Test that the main widget can be created."""
    viewer = Viewer()
    widget = DispimPipelineWidget(viewer)
    assert widget is not None
    assert hasattr(widget, "tab_widget")
    assert widget.tab_widget.count() == 4  # 4 processing steps

test_roi_detection_widget()

Test that the ROI detection widget can be created.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_roi_detection_widget():
    """Test that the ROI detection widget can be created."""
    from napari_pyspim._roi_detection import RoiDetectionWidget

    viewer = Viewer()
    widget = RoiDetectionWidget(viewer)
    assert widget is not None
    assert hasattr(widget, "detect_button")

test_registration_widget()

Test that the registration widget can be created.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_registration_widget():
    """Test that the registration widget can be created."""
    from napari_pyspim._registration import RegistrationWidget

    viewer = Viewer()
    widget = RegistrationWidget(viewer)
    assert widget is not None
    assert hasattr(widget, "register_button")

test_deconvolution_widget()

Test that the deconvolution widget can be created.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_deconvolution_widget():
    """Test that the deconvolution widget can be created."""
    from napari_pyspim._deconvolution import DeconvolutionWidget

    viewer = Viewer()
    widget = DeconvolutionWidget(viewer)
    assert widget is not None
    assert hasattr(widget, "deconvolve_button")

test_utils()

Test utility functions.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_utils():
    """Test utility functions."""
    from napari_pyspim._utils import format_memory_usage, format_shape_string

    # Test memory formatting
    test_array = np.zeros((100, 100, 50))
    memory_str = format_memory_usage(test_array)
    assert "MB" in memory_str or "KB" in memory_str

    # Test shape formatting
    shape_str = format_shape_string((100, 100, 50))
    assert "100×100×50" in shape_str
    assert "(Z×Y×X)" in shape_str

napari_pyspim.tests.test_plugin

Basic tests for the napari-pyspim plugin.

Classes

Functions

test_plugin_import()

Test that the plugin can be imported.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_plugin_import():
    """Test that the plugin can be imported."""
    from napari_pyspim import manifest

    assert manifest is not None
    assert manifest["name"] == "napari-pyspim"

test_widget_creation()

Test that the main widget can be created.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_widget_creation():
    """Test that the main widget can be created."""
    viewer = Viewer()
    widget = DispimPipelineWidget(viewer)
    assert widget is not None
    assert hasattr(widget, "tab_widget")
    assert widget.tab_widget.count() == 4  # 4 processing steps

test_roi_detection_widget()

Test that the ROI detection widget can be created.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_roi_detection_widget():
    """Test that the ROI detection widget can be created."""
    from napari_pyspim._roi_detection import RoiDetectionWidget

    viewer = Viewer()
    widget = RoiDetectionWidget(viewer)
    assert widget is not None
    assert hasattr(widget, "detect_button")

test_registration_widget()

Test that the registration widget can be created.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_registration_widget():
    """Test that the registration widget can be created."""
    from napari_pyspim._registration import RegistrationWidget

    viewer = Viewer()
    widget = RegistrationWidget(viewer)
    assert widget is not None
    assert hasattr(widget, "register_button")

test_deconvolution_widget()

Test that the deconvolution widget can be created.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_deconvolution_widget():
    """Test that the deconvolution widget can be created."""
    from napari_pyspim._deconvolution import DeconvolutionWidget

    viewer = Viewer()
    widget = DeconvolutionWidget(viewer)
    assert widget is not None
    assert hasattr(widget, "deconvolve_button")

test_utils()

Test utility functions.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_plugin.py

def test_utils():
    """Test utility functions."""
    from napari_pyspim._utils import format_memory_usage, format_shape_string

    # Test memory formatting
    test_array = np.zeros((100, 100, 50))
    memory_str = format_memory_usage(test_array)
    assert "MB" in memory_str or "KB" in memory_str

    # Test shape formatting
    shape_str = format_shape_string((100, 100, 50))
    assert "100×100×50" in shape_str
    assert "(Z×Y×X)" in shape_str

napari_pyspim.tests.test_basic

Basic tests that don't require CUDA compilation.

Functions

test_manifest()

Test that the manifest can be accessed without importing the full plugin.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_basic.py

def test_manifest():
    """Test that the manifest can be accessed without importing the full plugin."""
    # Import just the manifest
    import os
    import sys

    sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

    from napari_pyspim import manifest

    assert manifest is not None
    assert manifest["name"] == "napari-pyspim"
    assert manifest["display_name"] == "diSPIM Processing Pipeline"

test_utils()

Test utility functions.

Source code in packages/napari-pyspim/src/napari_pyspim/tests/test_basic.py

def test_utils():
    """Test utility functions."""
    # Import utils directly
    import os
    import sys

    sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

    from napari_pyspim._utils import format_memory_usage, format_shape_string

    # Test memory formatting
    test_array = np.zeros((100, 100, 50))
    memory_str = format_memory_usage(test_array)
    assert "MB" in memory_str or "KB" in memory_str

    # Test shape formatting
    shape_str = format_shape_string((100, 100, 50))
    assert "100×100×50" in shape_str
    assert "(Z×Y×X)" in shape_str