Inspect progress bars with a global handler

src/pyflask/manageNeuroconv/manage_neuroconv.py

@@ -11,12 +11,12 @@
 from shutil import copytree, rmtree
 from typing import Any, Dict, List, Optional, Union
 
-from .info import (
-    CONVERSION_SAVE_FOLDER_PATH,
-    GUIDE_ROOT_FOLDER,
-    STUB_SAVE_FOLDER_PATH,
-    announcer,
-)
+from tqdm_publisher import TQDMProgressHandler
+
+from .info import CONVERSION_SAVE_FOLDER_PATH, GUIDE_ROOT_FOLDER, STUB_SAVE_FOLDER_PATH
+from .info.sse import format_sse
+
+progress_handler = TQDMProgressHandler()
 
 EXCLUDED_RECORDING_INTERFACE_PROPERTIES = ["contact_vector", "contact_shapes", "group", "location"]
 
@@ -727,9 +727,6 @@ def convert_to_nwb(info: dict) -> str:
 
     converter = instantiate_custom_converter(resolved_source_data, info["interfaces"])
 
-    def update_conversion_progress(**kwargs):
-        announcer.announce(dict(**kwargs, nwbfile_path=nwbfile_path), "conversion_progress")
-
     # Assume all interfaces have the same conversion options for now
     available_options = converter.get_conversion_options_schema()
     options = (
@@ -915,10 +912,10 @@ def upload_project_to_dandi(
 
 # Create an events endpoint
 def listen_to_neuroconv_events():
-    messages = announcer.listen()  # returns a queue.Queue
+    messages = progress_handler.listen()  # returns a queue.Queue
     while True:
         msg = messages.get()  # blocks until a new message arrives
-        yield msg
+        yield format_sse(msg)
 
 
 def generate_dataset(input_path: str, output_path: str) -> dict:
@@ -1041,9 +1038,7 @@ def inspect_all(url, config):
 
     nwbfile_paths = list(Path(path).rglob("*.nwb"))
 
-    request_id = config.get("request_id")
-    if request_id:
-        config.pop("request_id")
+    request_id = config.pop("request_id", None)
 
     n_jobs = config.get("n_jobs", -2)  # Default to all but one CPU
     n_jobs = calculate_number_of_cpu(requested_cpu=n_jobs)
@@ -1068,7 +1063,7 @@ def inspect_all(url, config):
         # Announce directly
         def on_progress_update(message):
             message["progress_bar_id"] = request_id  # Ensure request_id matches
-            announcer.announce(
+            progress_handler.announce(
                 dict(
                     request_id=request_id,
                     **message,
@@ -1096,11 +1091,16 @@ def on_progress_update(message):
 def inspect_nwb_folder(url, payload) -> dict:
     from pickle import PicklingError
 
+    request_id = payload.get("request_id", None)
+
     from nwbinspector import load_config
     from nwbinspector.inspector_tools import format_messages, get_report_header
     from nwbinspector.nwbinspector import InspectorOutputJSONEncoder
 
     kwargs = dict(
+        n_jobs=-2,  # uses number of CPU - 1
+        progress_bar_class=progress_handler.create_progress_subscriber,
+        progress_bar_options=dict(additional_metadata=dict(request_id=request_id)),
         ignore=[
             "check_description",
             "check_data_orientation",