use workflows specified in the request message

Signed-off-by: Victor Chang <[email protected]>

Author: mocsharp

.github/workflows/ci.yml

@@ -14,7 +14,11 @@ name: ci
 on:
   # Triggers on pushes and on pull requests
   push:
+    paths-ignore:
+      - 'demos/**'
   pull_request:
+    paths-ignore:
+      - 'demos/**'
 
   # Allows you to run this workflow manually from the Actions tab
   workflow_dispatch:

demos/app_runner/README.md

@@ -8,7 +8,8 @@ This demo shows how an application integrates with the MONAI Deploy Informatics
 ## Requirements
 
 - MONAI Deploy Informatics Gateway 0.1.0+
-- MONAI Deploy App SDK 0.2.1+ with [dicom_series_to_image_app](https://github.com/Project-MONAI/monai-deploy-app-sdk/tree/main/examples/apps/dicom_series_to_image_app) built as `dcm-to-img:latest`
+- MONAI Deploy App SDK 0.2.1+
+  - A MAP from the [examples](https://github.com/Project-MONAI/monai-deploy-app-sdk/tree/main/examples/apps/) or BYO MAP.
 - RabbitMQ configured and running
 - MinIO configured and running
 - Python 3.7+
@@ -17,16 +18,18 @@ This demo shows how an application integrates with the MONAI Deploy Informatics
 ## Running the demo
 
 1. Install requirements specified above
-2. Install python dependencies specified in [requirements.txt](./requirements.txt)
-3. Edit `config.json` and change:
+2. Configure an AET with one or more workflows. For example, the following command would trigger the `dcm-to-img:latest` MAP.
+    ```
+    mig-cli  aet add -a DCM2PNG -w dcm-to-img:latest -v
+    ```
+3. Install python dependencies specified in [requirements.txt](./requirements.txt)
+4. Edit `config.json` and change:
    1. `endpoint`/`host`, `username`, and `password` for both storage and messaging services
    2. `bucket` where payloads are stored
-   3.  `application` to point to the correct MAP
-4. python app.py
+5. python app.py
 
-**Notes**: For MONAI Deploy App SDK 0.2.1, set `ignore_json` to false in the `config.json` file so DICOM JSON files are not downloaded.
+**Notes**: For MONAI Deploy App SDK 0.2.1, set `ignore_json` to `false` in the `config.json` file so DICOM JSON files are not downloaded.
 
-## Tips
+## Other Ideas
 
-- Try another MAP by changing `application` in the `config.json` file.
-- Instead of calling App Runner, integrate with [MIS](https://github.com/Project-MONAI/monai-deploy-app-server)
+💡 Instead of calling App Runner, integrate with [MIS](https://github.com/Project-MONAI/monai-deploy-app-server)

demos/app_runner/app.py

@@ -11,13 +11,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 import json
 import logging
 import os
-from signal import signal, SIGINT
+import re
 import sys
 from pathlib import Path
+from signal import SIGINT, signal
 from types import SimpleNamespace
 
 import pika
@@ -39,14 +39,15 @@ def _init_storage(self) -> None:
             config['endpoint'],
             config['username'],
             config['password'],
-            secure=False # DEMO purposes only!!! Make sure to use a secure connection!!!
+            secure=False  # DEMO purposes only!!! Make sure to use a secure connection!!!
         )
 
         if not self._storage_client.bucket_exists(config['bucket']):
             raise f"Bucket '{config['bucket']}' does not exist"
 
         if not os.path.exists(self._working_directory):
-            self._logger.info(f"Creating working directory {self._working_directory}")
+            self._logger.info(
+                f"Creating working directory {self._working_directory}")
             os.makedirs(self._working_directory)
 
     def _init_messaging(self) -> None:
@@ -72,31 +73,34 @@ def _load_config(self) -> None:
             self._config = json.load(f)
 
         self._working_directory = Path(self._config['working_dir'])
-        self._application = self._config['application']
 
     def _message_callback(self, ch, method, properties, body):
         correlation_id = properties.correlation_id
         self._logger.info(
             f"Message received from application={properties.app_id}. Correlation ID={correlation_id}. Delivery tag={method.delivery_tag}. Topic={method.routing_key}")
 
         request_message = json.loads(body)
-        print(" body\t%r" % (request_message))
-        print(" properties\t%r" % (properties))
+
+        if 'workflows' not in request_message or len(request_message['workflows']) == 0:
+            self._logger.error(
+                f"No applications defined in the message body, skipping.")
+            self._send_acknowledgement(method.delivery_tag)
+            return
+
+        # print(" body\t%r" % (request_message))
+        # print(" properties\t%r" % (properties))
 
         job_dir = self._working_directory / correlation_id
         job_dir_input = job_dir / "input"
         job_dir_output = job_dir / "output"
         if not os.path.exists(job_dir_input):
-            self._logger.info(f"Creating working directory for job {job_dir}")
+            self._logger.info(f"Creating input directory for job {job_dir_input}")
             os.makedirs(job_dir_input)
-        if not os.path.exists(job_dir_output):
-            self._logger.info(f"Creating working directory for job {job_dir}")
-            os.makedirs(job_dir_output)
 
         # note: in IG 0.1.1 or later, the bucket name can be found inside body.payload[]
-        bucket=self._config['storage']['bucket']
+        bucket = self._config['storage']['bucket']
         file_list = self._storage_client.list_objects(bucket, prefix=request_message['payload_id'],
-                              recursive=True)
+                                                      recursive=True)
         for file in file_list:
             if self._config['ignore_json'] and file.object_name.endswith('json'):
                 self._logger.info(f'Skipping JSON file {file.object_name}...')
@@ -114,31 +118,43 @@ def _message_callback(self, ch, method, properties, body):
                 for d in data.stream(32*1024):
                     file_data.write(d)
 
-        self._logger.info(f"Finished download payload {request_message['payload_id']}...")
-
-        argsd = {}
-        argsd['map'] = self._application
-        argsd['input'] = job_dir_input
-        argsd['output'] =job_dir_output
-        argsd['quiet'] = False
-
-
-        self._logger.info(f"Launching application {self._application}...")
-        self._logger.info(f"\tInput:\t {job_dir_input}...")
-        self._logger.info(f"\tOutput:\t {job_dir_output}...")
-        args = SimpleNamespace(**argsd)
-        try:
-            runner.main(args)
-        except:
-            e = sys.exc_info()[0]
-            self._logger.error(f'{self._application} failed with {e}.')
-        else:
+        self._logger.info(
+            f"Finished download payload {request_message['payload_id']}...")
+
+        applications = request_message['workflows']
+        for application in applications:
+            app_output = job_dir_output / re.sub(r'[^\w\-_\. ]', '-', application)
+            if not os.path.exists(app_output):
+                self._logger.info(f"Creating output directory for job {app_output}")
+                os.makedirs(app_output)
+
+            argsd = {}
+            argsd['map'] = application
+            argsd['input'] = job_dir_input
+            argsd['output'] = app_output
+            argsd['quiet'] = False
+
+            self._logger.info(f"Launching application {application}...")
             self._logger.info(f"\tInput:\t {job_dir_input}...")
-            self._logger.info(f"\tOutput:\t {job_dir_output}...")
-        finally:
-            self._logger.info(f"{self._application} completed. Sending acknowledgement...")
-            self._pika_channel.basic_ack(method.delivery_tag)
-            self._logger.info(f"{self._application} completed. Acknowledgement sent...")
+            self._logger.info(f"\tOutput:\t {app_output}...")
+            args = SimpleNamespace(**argsd)
+            try:
+                runner.main(args)
+            except:
+                e = sys.exc_info()[0]
+                self._logger.error(f'{application} failed with {e}.')
+            else:
+                self._logger.info(
+                    f"Application {application} completed successfully.")
+                self._logger.info(f"\tInput:\t {job_dir_input}")
+                self._logger.info(f"\tOutput:\t {app_output}")
+
+        self._send_acknowledgement(method.delivery_tag)
+
+    def _send_acknowledgement(self, delivery_tag):
+        self._logger.info(f"Sending acknowledgement...")
+        self._pika_channel.basic_ack(delivery_tag)
+        self._logger.info(f"Acknowledgement sent...")
 
     def run(self):
         self._logger.info('[*] Waiting for logs. To exit press CTRL+C')

demos/app_runner/config.json

@@ -14,6 +14,5 @@
         "topic": "md.workflow.request"
     },
     "working_dir": "jobs/",
-    "application": "dcm-to-img:latest",
     "ignore_json": false
 }