Add Conversion Progress Bars

pyflask/apis/neuroconv.py

@@ -15,6 +15,7 @@
     get_source_schema,
     get_metadata_schema,
     convert_to_nwb,
+    convert_all_to_nwb,
     validate_metadata,
     listen_to_neuroconv_events,
     inspect_nwb_file,
@@ -110,7 +111,12 @@ class Convert(Resource):
     @neuroconv_api.doc(responses={200: "Success", 400: "Bad Request", 500: "Internal server error"})
     def post(self):
         try:
-            return convert_to_nwb(neuroconv_api.payload)
+            has_files = "files" in neuroconv_api.payload
+            if has_files:
+                url = f"{request.url_root}neuroconv/announce"
+                return convert_all_to_nwb(url, **neuroconv_api.payload)
+            else:
+                return convert_to_nwb(neuroconv_api.payload)
 
         except Exception as exception:
             if notBadRequestException(exception):

pyflask/manageNeuroconv/__init__.py

@@ -6,6 +6,7 @@
     get_source_schema,
     get_metadata_schema,
     convert_to_nwb,
+    convert_all_to_nwb,
     validate_metadata,
     upload_project_to_dandi,
     upload_folder_to_dandi,

pyflask/manageNeuroconv/manage_neuroconv.py

@@ -667,6 +667,12 @@ def get_interface_alignment(info: dict) -> dict:
 def convert_to_nwb(info: dict) -> str:
     """Function used to convert the source data to NWB format using the specified metadata."""
 
+    from tqdm_publisher import TQDMProgressSubscriber
+    import requests
+
+    url = info.get("url", None)
+    request_id = info.get("request_id", None)
+
     nwbfile_path = Path(info["nwbfile_path"])
     custom_output_directory = info.get("output_folder")
     project_name = info.get("project_name")
@@ -695,14 +701,26 @@ 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")
+        update_dict = dict(request_id=request_id, **kwargs)
+        if (url) or not run_stub_test:
+            requests.post(url=url, json=update_dict)
+        else:
+            announcer.announce(update_dict)
+
+    progress_bar_options = dict(
+        mininterval=0,
+        on_progress_update=update_conversion_progress,
+    )
 
     # Assume all interfaces have the same conversion options for now
     available_options = converter.get_conversion_options_schema()
     options = (
         {
             interface: (
-                {"stub_test": info["stub_test"]}  # , "iter_opts": {"report_hook": update_conversion_progress}}
+                {
+                    "stub_test": info["stub_test"],
+                    # "iterator_opts": dict( display_progress=True, progress_bar_class=TQDMProgressSubscriber, progress_bar_options=progress_bar_options )
+                }
                 if available_options.get("properties").get(interface).get("properties", {}).get("stub_test")
                 else {}
             )
@@ -788,6 +806,58 @@ def update_conversion_progress(**kwargs):
     return dict(file=str(resolved_output_path))
 
 
+def convert_all_to_nwb(
+    url: str,
+    files: List[dict],
+    request_id: Optional[str],
+    max_workers: int = 1,
+) -> List[str]:
+
+    from tqdm_publisher import TQDMProgressSubscriber
+    from concurrent.futures import ProcessPoolExecutor, as_completed
+
+    def on_progress_update(message):
+        message["progress_bar_id"] = request_id  # Ensure request_id matches
+        announcer.announce(
+            dict(
+                request_id=request_id,
+                **message,
+            )
+        )
+
+    futures = []
+    file_paths = []
+
+    with ProcessPoolExecutor(max_workers=max_workers) as executor:
+
+        for file_info in files:
+
+            futures.append(
+                executor.submit(
+                    convert_to_nwb,
+                    dict(
+                        url=url,
+                        request_id=request_id,
+                        **file_info,
+                    ),
+                )
+            )
+
+        inspection_iterable = TQDMProgressSubscriber(
+            iterable=as_completed(futures),
+            desc="Total files converted",
+            total=len(futures),
+            mininterval=0,
+            on_progress_update=on_progress_update,
+        )
+
+        for future in inspection_iterable:
+            output_filepath = future.result()
+            file_paths.append(output_filepath)
+
+        return file_paths
+
+
 def upload_multiple_filesystem_objects_to_dandi(**kwargs):
     tmp_folder_path = _aggregate_symlinks_in_new_directory(kwargs["filesystem_paths"], "upload")
     innerKwargs = {**kwargs}

src/renderer/src/stories/pages/Page.js

@@ -1,5 +1,5 @@
 import { LitElement, html } from "lit";
-import { runConversion } from "./guided-mode/options/utils.js";
+import { run } from "./guided-mode/options/utils.js";
 import { get, save } from "../../progress/index.js";
 import { dismissNotification, isStorybook, notify } from "../../dependencies/globals.js";
 import { randomizeElements, mapSessions, merge } from "./utils.js";
@@ -154,18 +154,10 @@ export class Page extends LitElement {
 
         const results = {};
 
-        const isMultiple = toRun.length > 1;
-
         const swalOpts = await createProgressPopup({ title: `Running conversion`, ...options });
-        const { close: closeProgressPopup, elements } = swalOpts;
-
-        elements.container.insertAdjacentHTML(
-            "beforeend",
-            `<small><small><b>Note:</b> This may take a while to complete...</small></small><hr style="margin-bottom: 0;">`
-        );
+        const { close: closeProgressPopup } = swalOpts;
 
-        let completed = 0;
-        elements.progress.format = { n: completed, total: toRun.length };
+        const fileConfiguration = [];
 
         for (let info of toRun) {
             const { subject, session, globalState = this.info.globalState } = info;
@@ -184,43 +176,54 @@ export class Page extends LitElement {
                 source_data: merge(SourceData, sourceDataCopy),
             };
 
-            const result = await runConversion(
-                {
-                    output_folder: conversionOptions.stub_test ? undefined : conversion_output_folder,
-                    project_name: name,
-                    nwbfile_path: file,
-                    overwrite: true, // We assume override is true because the native NWB file dialog will not allow the user to select an existing file (unless they approve the overwrite)
-                    ...sessionInfo, // source_data and metadata are passed in here
-                    ...conversionOptions, // Any additional conversion options override the defaults
-
-                    interfaces: globalState.interfaces,
-                },
-                swalOpts
-            ).catch((error) => {
-                let message = error.message;
-
-                if (message.includes("The user aborted a request.")) {
-                    this.notify("Conversion was cancelled.", "warning");
-                    throw error;
-                }
+            const payload = {
+                output_folder: conversionOptions.stub_test ? undefined : conversion_output_folder,
+                project_name: name,
+                nwbfile_path: file,
+                overwrite: true, // We assume override is true because the native NWB file dialog will not allow the user to select an existing file (unless they approve the overwrite)
+                ...sessionInfo, // source_data and metadata are passed in here
+                ...conversionOptions, // Any additional conversion options override the defaults
 
-                this.notify(message, "error");
-                closeProgressPopup();
-                throw error;
-            });
+                interfaces: globalState.interfaces,
+            };
 
-            completed++;
-            if (isMultiple) {
-                const progressInfo = { n: completed, total: toRun.length };
-                elements.progress.format = progressInfo;
+            fileConfiguration.push(payload);
+        }
+
+        const conversionResults = await run(
+            `convert`,
+            {
+                files: fileConfiguration,
+                max_workers: 2, // TODO: Make this configurable and confirm default value
+                request_id: swalOpts.id,
+            },
+            {
+                title: "Running the conversion",
+                onError: () => "Conversion failed with current metadata. Please try again.",
+                ...swalOpts,
             }
+        ).catch(async (error) => {
+            let message = error.message;
 
+            if (message.includes("The user aborted a request.")) {
+                this.notify("Conversion was cancelled.", "warning");
+                throw error;
+            }
+
+            this.notify(message, "error");
+            await closeProgressPopup();
+            throw error;
+        });
+
+        conversionResults.forEach((info) => {
+            const { file } = info;
+            const fileName = file.split("/").pop();
+            const [subject, session] = fileName.match(/sub-(.+)_ses-(.+)\.nwb/).slice(1);
             const subRef = results[subject] ?? (results[subject] = {});
-            subRef[session] = result;
-        }
+            subRef[session] = info;
+        });
 
-        closeProgressPopup();
-        elements.container.style.textAlign = ""; // Clear style update
+        await closeProgressPopup();
 
         return results;
     }

src/renderer/src/stories/pages/guided-mode/options/GuidedInspectorPage.js

@@ -162,15 +162,15 @@ export class GuidedInspectorPage extends Page {
                             "inspect_folder",
                             { path, ...options, request_id: swalOpts.id },
                             swalOpts
-                        ).catch((error) => {
+                        ).catch(async (error) => {
                             this.notify(error.message, "error");
-                            closeProgressPopup();
+                            await closeProgressPopup();
                             return null;
                         });
 
                         if (!result) return "Failed to generate inspector report.";
 
-                        closeProgressPopup();
+                        await closeProgressPopup();
 
                         this.report = globalState.preview.inspector = {
                             ...result,

src/renderer/src/stories/pages/inspect/InspectPage.js

@@ -29,13 +29,15 @@ export class InspectPage extends Page {
 
         const { close: closeProgressPopup } = swalOpts;
 
-        const result = await run("inspect", { request_id: swalOpts.id, paths, ...kwargs }, swalOpts).catch((error) => {
-            this.notify(error.message, "error");
-            closeProgressPopup();
-            throw error;
-        });
-
-        closeProgressPopup();
+        const result = await run("inspect", { request_id: swalOpts.id, paths, ...kwargs }, swalOpts).catch(
+            async (error) => {
+                this.notify(error.message, "error");
+                await closeProgressPopup();
+                throw error;
+            }
+        );
+
+        await closeProgressPopup();
 
         if (typeof result === "string") return result;
 

src/renderer/src/stories/pages/utils.js

@@ -35,7 +35,7 @@ export const sanitize = (item, condition = isPrivate) => {
             if (condition(k, value)) delete item[k];
             else sanitize(value, condition);
         }
-    }
+    } else if (Array.isArray(item)) item.forEach((value) => sanitize(value, condition));
 
     return item;
 };

src/renderer/src/stories/utils/progress.js

@@ -60,7 +60,6 @@ export const createProgressPopup = async (options, tqdmCallback) => {
     const onProgressMessage = ({ data }) => {
         const parsed = JSON.parse(data);
         const { request_id, ...update } = parsed;
-        console.warn("parsed", parsed);
 
         if (request_id && request_id !== id) return;
         lastUpdate = Date.now();
@@ -74,13 +73,16 @@ export const createProgressPopup = async (options, tqdmCallback) => {
 
     progressHandler.addEventListener("message", onProgressMessage);
 
-    const close = () => {
+    const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
+
+    const close = async () => {
         if (lastUpdate) {
             // const timeSinceLastUpdate = now - lastUpdate;
             const animationLeft = 1000; // ProgressBar.animationDuration - timeSinceLastUpdate; // Add 100ms to ensure the animation has time to complete
-            if (animationLeft) setTimeout(() => popup.close(), animationLeft);
-            else popup.close();
-        } else popup.close();
+            if (animationLeft) await sleep(animationLeft);
+        }
+
+        popup.close();
 
         progressHandler.removeEventListener("message", onProgressMessage);
     };