Add utility function to query the plugin status from napari side

napari_plugin_manager/_tests/test_qt_plugin_dialog.py

@@ -25,6 +25,7 @@
 
 from napari_plugin_manager import qt_plugin_dialog
 from napari_plugin_manager.qt_package_installer import InstallerActions
+from napari_plugin_manager.qt_plugin_dialog import ProcessStatus
 
 N_MOCKED_PLUGINS = 2
 
@@ -198,7 +199,6 @@ def set_blocked(self, plugin, blocked):
     monkeypatch.setattr(
         napari.plugins, 'plugin_manager', OldPluginManagerMock()
     )
-
     monkeypatch.setattr(importlib.metadata, 'metadata', mock_metadata)
 
     monkeypatch.setattr(npe2, 'PluginManager', PluginManagerMock())
@@ -538,3 +538,26 @@ def test_shortcut_quit(plugin_dialog, qtbot):
     )
     qtbot.wait(200)
     assert not plugin_dialog.isVisible()
+
+
+def test_query_status(plugin_dialog, monkeypatch):
+    status, description = plugin_dialog.query_status()
+    assert status == ProcessStatus.IDLE
+    assert not description
+
+    monkeypatch.setattr(
+        plugin_dialog.installer,
+        "_queue",
+        ['mock'],
+    )
+    status, description = plugin_dialog.query_status()
+    assert status == ProcessStatus.BUSY
+    assert description
+
+    monkeypatch.setattr(
+        plugin_dialog.installer,
+        "_queue",
+        ['mock', 'other-mock'],
+    )
+    assert status == ProcessStatus.BUSY
+    assert description

napari_plugin_manager/qt_plugin_dialog.py

@@ -2,7 +2,9 @@
 import importlib.metadata
 import os
 import sys
+import uuid
 import webbrowser
+from enum import auto
 from functools import partial
 from pathlib import Path
 from typing import Dict, List, Literal, NamedTuple, Optional, Sequence, Tuple
@@ -17,6 +19,7 @@
 from napari.plugins.utils import normalized_name
 from napari.settings import get_settings
 from napari.utils.misc import (
+    StringEnum,
     parse_version,
     running_as_constructor_app,
 )
@@ -57,6 +60,25 @@
 from napari_plugin_manager.qt_widgets import ClickableLabel
 from napari_plugin_manager.utils import is_conda_package
 
+try:
+    from napari.utils.processes import (
+        ProcessStatus,
+        register_process_status,
+        unregister_process_status,
+    )
+except ImportError:
+
+    class ProcessStatus(StringEnum):
+        BUSY = auto()
+        IDLE = auto()
+
+    def register_process_status(status: ProcessStatus, description: str):
+        pass
+
+    def unregister_process_status(process_status_id: uuid.UUID) -> bool:
+        pass
+
+
 # Scaling factor for each list widget item when expanding.
 CONDA = 'Conda'
 PYPI = 'PyPI'
@@ -850,6 +872,7 @@
         self._filter_texts = []
         self._filter_idxs_cache = set()
         self._filter_timer = QTimer(self)
+        self._process_status_id = None
         self.worker = None
 
         # timer to avoid triggering a filter for every keystroke
@@ -902,6 +925,19 @@
         stylesheet = get_current_stylesheet([STYLES_PATH])
         self.setStyleSheet(stylesheet)
 
+    def _register_process_status(self):
+        if self._process_status_id is not None:
+            self._unregister_process_status(self._process_status_id)
+
+        status, description = self.query_status()
+        self._process_status_id = register_process_status(status, description)
+
+    def _unregister_process_status(self):
+        if self._process_status_id is not None:
+            unregister_process_status(self._process_status_id)
+
+        self._process_status_id = None
+
     def _on_installer_start(self):
         """Updates dialog buttons and status when installing a plugin."""
         self.cancel_all_btn.setVisible(True)
@@ -910,6 +946,9 @@
         self.process_error_indicator.hide()
         self.refresh_button.setDisabled(True)
 
+        if self._process_status_id is None:
+            self._register_process_status()
+
     def _on_process_finished(self, process_finished_data: ProcessFinishedData):
         action = process_finished_data['action']
         exit_code = process_finished_data['exit_code']
@@ -972,6 +1011,8 @@
             else:
                 show_info(trans._('Plugin Manager: process completed\n'))
 
+        self._unregister_process_status()
+
     def _add_to_available(self, pkg_name):
         self._add_items_timer.stop()
         self._plugin_queue.insert(0, self._plugin_data_map[pkg_name])
@@ -1416,6 +1457,7 @@
         self._plugin_queue = []
         self._plugin_data = []
         self._plugin_data_map = {}
+        self._latest_status = None
 
         self.installed_list.clear()
         self.available_list.clear()
@@ -1447,6 +1489,21 @@
 
     # endregion - Public methods
 
+    def query_status(self) -> Tuple[ProcessStatus, str]:
+        """Return the current status of the plugin."""
+        if self.installer.hasJobs():
+            process_status = ProcessStatus.BUSY
+            description = trans._n(
+                'The plugin manager is currently busy with {n} task.',
+                'The plugin manager is currently busy with {n} tasks.',
+                n=self.installer.currentJobs(),
+            )
+        else:
+            process_status = ProcessStatus.IDLE
+            description = ''
+
+        return process_status, description
+
 
 if __name__ == "__main__":
     from qtpy.QtWidgets import QApplication