updated pipeline module

Author: Sandip117

chrisClient.py

@@ -85,7 +85,7 @@ def pacs_pull(self):
     def pacs_push(self):
         pass
 
-    def anonymize(self, dicom_dir: str, tag_struct: str, send_params: dict, pv_id: int):
+    async def anonymize(self, dicom_dir: str, tag_struct: str, send_params: dict, pv_id: int):
         """
         Run the anonymization pipeline for a given DICOM directory and push results to specified Orthanc instance
         """
@@ -105,8 +105,14 @@ def anonymize(self, dicom_dir: str, tag_struct: str, send_params: dict, pv_id: i
             }
         }
         pipe = Pipeline(self.api_base, self.token)
-        d_ret = pipe.workflow_schedule(dsdir_inst_id, "DICOM anonymization and Orthanc push 20241217",
-                               plugin_params)
+        d_ret = await pipe.run_pipeline(
+            previous_inst=dsdir_inst_id,
+            pipeline_name="DICOM anonymization and Orthanc push 20241217",
+            pipeline_params=plugin_params,
+            recipients=send_params['recipients'],
+            smtp_server=send_params['smtp_server'],
+            series_data={}
+        )
         return d_ret
 
     def run_dicomdir_plugin(self, dicom_dir: str, pv_id: int) -> int:

pipeline.py

@@ -4,7 +4,10 @@
 from requests.exceptions import RequestException, Timeout, HTTPError
 from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
 from loguru import logger
-
+import time
+import asyncio
+from urllib.parse import urlencode
+import pandas as pd
 
 def transform_plugin_data(nested_data_list: list[dict]) -> list[dict]:
     """Flatten nested plugin data into a list of dictionaries."""
@@ -107,23 +110,183 @@ def get_pipeline_id(self, name: str) -> int:
                     return field.get("value")
         return -1
 
+    def get_pipeline_total_pipings(self, pipeline_id: int) -> int:
+        """Get the total number of plugin pipings in the given pipeline."""
+        logger.info(f"Fetching pipeline plugin piping list.")
+        response = self.make_request("GET", f"/pipelines/{pipeline_id}/pipings/?limit=100")
+        return len(response)
+
+
     def get_pipeline_parameters(self, pipeline_id: int) -> list[dict]:
         """Get default parameters for a pipeline."""
         logger.info(f"Fetching default parameters for pipeline with ID: {pipeline_id}")
         response = self.make_request("GET", f"/pipelines/{pipeline_id}/parameters/?limit=1000")
         return transform_plugin_data(response)
 
-    def post_workflow(self, pipeline_id: int, previous_id: int, params: list[dict]) -> dict:
+    def get_feed_id_from_plugin_inst(self, plugin_inst: int) -> int:
+        """Get feed_id from a given plugin instance"""
+        logger.info(f"Fetching feed id for plugin instance with ID: {plugin_inst}")
+        response = self.make_request("GET", f"/plugins/instances/{plugin_inst}/")
+        for item in response:
+            for field in item.get("data", []):
+                if field.get("name") == "feed_id":
+                    return field.get("value")
+        return -1
+
+    def get_feed_details_from_id(self, feed_id: int) -> dict:
+        """Get feed details given a feed id"""
+        feed_details = {}
+
+        logger.info(f"Getting feed details for ID: {feed_id}")
+        response = self.make_request("GET", f"/{feed_id}/")
+        for item in response:
+            for field in item.get("data", []):
+                if field.get("name") == "creation_date":
+                    feed_details["date"] = field.get("value")
+                if field.get("name") == "name":
+                    feed_details["name"] = field.get("value")
+                if field.get("name") == "owner_username":
+                    feed_details["owner"] = field.get("value")
+
+        return feed_details
+
+    def post_workflow(self, pipeline_id: int, previous_id: int, params: list[dict]) -> int:
         """
         Trigger a pipeline workflow in CUBE.
         """
         payload = {
             "previous_plugin_inst_id": previous_id,
             "nodes_info": json.dumps(params)
         }
-        return self.post_request(f"/pipelines/{pipeline_id}/workflows/", json=payload)
+        response = self.post_request(f"/pipelines/{pipeline_id}/workflows/", json=payload)
+        for item in response:
+            for field in item.get("data", []):
+                if field.get("name") == "id":
+                    return field.get("value")
+        return -1
+
+    async def get_workflow_status(self, workflow_id: int) -> dict:
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, self._get_workflow_status, workflow_id)
+
+    def _get_workflow_status(self, workflow_id: int) -> dict:
+        """
+        1. Get workflow details for a given workflow id.
+        2. Check for errored jobs
+        3. return total jobs (finished + errored + cancelled)
+        """
+        finished_jobs = 0
+        errored_jobs = 0
+        cancelled_jobs = 0
+        created_jobs = 0
+        waiting_jobs = 0
+        scheduled_jobs = 0
+        started_jobs = 0
+        registering_jobs = 0
+
+        logger.info(f"Fetching workflow details for ID: {workflow_id}")
+        response = self.make_request("GET", f"/pipelines/workflows/{workflow_id}/")
+        for item in response:
+            for field in item.get("data", []):
+                if field.get("name") == "finished_jobs":
+                    finished_jobs = field.get("value")
+                if field.get("name") == "errored_jobs":
+                    errored_jobs = field.get("value")
+                if field.get("name") == "cancelled_jobs":
+                    cancelled_jobs = field.get("value")
+                if field.get("name") == "created_jobs":
+                    created_jobs = field.get("value")
+                if field.get("name") == "waiting_jobs":
+                    waiting_jobs = field.get("value")
+                if field.get("name") == "scheduled_jobs":
+                    scheduled_jobs = field.get("value")
+                if field.get("name") == "started_jobs":
+                    started_jobs = field.get("value")
+                if field.get("name") == "registering_jobs":
+                    registering_jobs = field.get("value")
+
+
+        return {
+            "finished_jobs": finished_jobs,
+            "total_jobs": finished_jobs + errored_jobs + cancelled_jobs + created_jobs + waiting_jobs + scheduled_jobs + started_jobs + registering_jobs,
+            "workflow_failed": (errored_jobs > 0)
+        }
+
+    async def monitor_pipeline(self, workflow_id, total_jobs, pv_inst, rcpts, smtp, series_data):
+        while True:
+            status = self._get_workflow_status(workflow_id)
+            if status["workflow_failed"]:
+                logger.error("Pipeline failed.")
+                self.run_notification_plugin(pv_inst, "Pipeline failed with errors", rcpts, smtp, series_data)
+                break
+            if status["finished_jobs"] >= total_jobs:
+                logger.info("Pipeline complete.")
+                break
+            if status["total_jobs"] < total_jobs:
+                self.run_notification_plugin(pv_inst, "Nodes deleted in pipeline", rcpts, smtp, series_data)
+                break
+            time.sleep(20)
 
-    def run_pipeline(self, pipeline_name: str, previous_inst: int, pipeline_params: dict):
+    def run_notification_plugin(self, pv_id: int, msg: str, rcpts: str, smtp: str, series_data: str) -> int:
+        """
+        Run the pl-notification plugin.
+        """
+        feed_id = self.get_feed_id_from_plugin_inst(pv_id)
+        feed_details = self.get_feed_details_from_id(feed_id)
+        d_series = json.loads(series_data)
+        email_content = (f"An error occurred while running anonymization pipeline on the following data: "
+                         f"\nFeed Name: {feed_details['name']}"
+                         f"\nDate: {feed_details['date']}"
+                         f"\nMRN: {d_series['PatientID']} "
+                         f"\nStudyDate: {d_series['StudyDate']}"
+                         f"\nModality: {d_series['Modality']}"
+                         f"\nSeriesDescription: {d_series['SeriesDescription']}"
+                         f"\nFolder Name: {d_series['Folder Name']}"
+                         f"\n\nKindly login to ChRIS as *{feed_details['owner']}* to access the logs for more details.")
+
+        try:
+            plugin_id = self._get_plugin_id({"name": "pl-notification", "version": "0.1.0"})
+            instance_id = self._create_plugin_instance(plugin_id, {
+                "previous_id": pv_id,
+                "content": email_content,
+                "title": msg,
+                "rcpt": rcpts,
+                "sender": "noreply@fnndsc.org",
+                "mail_server": smtp
+            })
+            return int(instance_id)
+        except Exception as ex:
+            logger.error(f"Error occurred while creating notification instance {ex}")
+
+    def _create_plugin_instance(self, plugin_id: str, params: dict):
+        """
+        Create a plugin instance and return its ID.
+        """
+        response = self.post_request( f"/plugins/{plugin_id}/instances/", json=params)
+
+        for item in response:
+            for field in item.get("data", []):
+                if field.get("name") == "id":
+                    return field.get("value")
+
+        raise RuntimeError("Plugin instance could not be scheduled.")
+
+    def _get_plugin_id(self, params: dict):
+        """
+        Fetch plugin ID by search parameters.
+        """
+        query_string = urlencode(params)
+        response = self.make_request("GET", f"/plugins/search/?{query_string}")
+
+        for item in response:
+            for field in item.get("data", []):
+                if field.get("name") == "id":
+                    return field.get("value")
+
+        raise RuntimeError(f"No plugin found with matching criteria: {params}")
+
+
+    async def run_pipeline(self, pipeline_name: str, previous_inst: int, pipeline_params: dict, recipients: str, smtp_server: str, series_data: str):
         """
         Full workflow to:
         1. Fetch pipeline ID
@@ -133,12 +296,19 @@ def run_pipeline(self, pipeline_name: str, previous_inst: int, pipeline_params:
         """
         try:
             pipeline_id = self.get_pipeline_id(pipeline_name)
+            total_jobs = self.get_pipeline_total_pipings(pipeline_id)
             default_params = self.get_pipeline_parameters(pipeline_id)
             nodes_info = compute_workflow_nodes_info(default_params, include_all_defaults=True)
             updated_params = update_plugin_parameters(nodes_info, pipeline_params)
-            workflow = self.post_workflow(pipeline_id=pipeline_id, previous_id=previous_inst, params=updated_params)
+            workflow_id = self.post_workflow(pipeline_id=pipeline_id, previous_id=previous_inst, params=updated_params)
+            #self.run_notification_plugin(previous_inst)
+
+            # Start this in the background (not awaited)
+            asyncio.create_task(self.monitor_pipeline(workflow_id, total_jobs, previous_inst, recipients, smtp_server, series_data))
+
             logger.info(f"Workflow posted successfully")
             return {"status": "Pipeline running"}
+
         except Exception as ex:
             logger.error(f"Running pipeline failed due to: {ex}")
             return {"status": "Failed", "error": str(ex)}

reg_chxr.py

@@ -13,6 +13,7 @@
 import os
 import pfdcm
 import copy
+import asyncio
 
 LOG = logger.debug
 
@@ -70,11 +71,6 @@
     default="",
     help="CUBE/ChRIS auth token"
 )
-parser.add_argument(
-    "--CUBEuser",
-    default="chris",
-    help="CUBE/ChRIS username"
-)
 parser.add_argument(
     '--inputJSONfile',
     default='',
@@ -141,6 +137,18 @@
     action="store_true",
     default=False,
 )
+parser.add_argument(
+    '--recipients',
+    default='',
+    type=str,
+    help='comma separated valid email recipient addresses'
+)
+parser.add_argument(
+    '--SMTPServer',
+    default='mailsmtp4.childrenshospital.org',
+    type=str,
+    help='valid email server'
+)
 parser.add_argument('-V', '--version', action='version',
                     version=f'%(prog)s {__version__}')
 
@@ -194,7 +202,7 @@ def main(options: Namespace, inputdir: Path, outputdir: Path):
                 raise Exception(f"Cannot verify registration for empty pacs data.")
 
             retry_table = create_hash_table(data, 5)
-            registration_errors = check_registration(options, retry_table, cube_cl)
+            registration_errors = asyncio.run(check_registration(options, retry_table, cube_cl))
 
             if registration_errors:
                 LOG(f"ERROR while running pipelines.")
@@ -240,7 +248,7 @@ def create_hash_table(retrieve_data: dict, retry: int) -> dict:
     return retry_table
 
 # Recursive method to check on registration and then run anonymization pipeline
-def check_registration(options: Namespace, retry_table: dict, client: PACSClient, contains_errors: bool=False):
+async def check_registration(options: Namespace, retry_table: dict, client: PACSClient, contains_errors: bool=False):
     # null check
     if len(retry_table) == 0:
         return contains_errors
@@ -279,21 +287,23 @@ def check_registration(options: Namespace, retry_table: dict, client: PACSClient
         if registered_series_count:
             LOG(f"Series {series_instance} successfully registered to CUBE.")
             send_params = {
-                "neuro_dcm_location": options.neuroDicomLocation,
-                "neuro_anon_location": options.neuroAnonLocation,
-                "neuro_nifti_location": options.neuroNiftiLocation,
-                "folder_name": options.folderName
+                "url": options.orthancUrl,
+                "username": options.orthancUsername,
+                "password": options.orthancPassword,
+                "aec": options.pushToRemote,
+                "recipients": options.recipients,
+                "smtp_server": options.SMTPServer
             }
             dicom_dir = client.get_pacs_files({'SeriesInstanceUID': series_instance})
 
             # create ChRIS Client Object
             cube_con = ChrisClient(options.CUBEurl, options.CUBEtoken)
-            d_ret = cube_con.anonymize(dicom_dir, send_params, options.pluginInstanceID)
+            d_ret = await cube_con.anonymize(dicom_dir, options.tagStruct, send_params, options.pluginInstanceID)
             if d_ret.get('error'):
                 contains_errors = True
         clone_retry_table.pop(series_instance)
 
-    check_registration(options, clone_retry_table, client, contains_errors)
+    await check_registration(options, clone_retry_table, client, contains_errors)
     return contains_errors
 
 

requirements.txt

@@ -1,4 +1,5 @@
 chris_plugin==0.4.0
 python-chrisclient==2.11.1
 loguru
-tenacity
+tenacity
+pandas