added notification mechanism

Author: Sandip117

README.md

@@ -9,7 +9,6 @@ _ds_ plugin which takes in ...  as input files and
 creates ... as output files.
 
 ## Abstract
-
 ...
 
 ## Installation

base_client.py

@@ -10,7 +10,7 @@ def pacs_pull(self):
         pass
 
     @abstractmethod
-    def anonymize(self, dicom_dir: str,  send_params: dict, pv_id: int):
+    def anonymize(self, dicom_dir: str,  send_params: dict, pv_id: int, series_dat: str):
         pass
 
     @abstractmethod

chrisClient.py

@@ -87,10 +87,13 @@ def pacs_pull(self):
     def pacs_push(self):
         pass  # Placeholder for PACS push implementation
 
-    def anonymize(self, dicom_dir: str, send_params: dict, pv_id: int):
+    async def anonymize(self, dicom_dir: str, send_params: dict, pv_id: int, series_data: str):
         """
         Run the anonymization pipeline for a given DICOM directory and push results to specified neuro locations.
         """
+        d_series = json.loads(series_data)
+        d_series['Folder Name'] = send_params['folder_name']
+        d_series['SeriesDescription'] = dicom_dir.split('/')[-1]
         dsdir_inst_id = self.run_dicomdir_plugin(dicom_dir, pv_id)
 
         plugin_params = {
@@ -121,10 +124,13 @@ def anonymize(self, dicom_dir: str, send_params: dict, pv_id: int):
         }
 
         pipe = Pipeline(self.api_base, self.token)
-        d_ret = pipe.run_pipeline(
+        d_ret = await pipe.run_pipeline(
             previous_inst=dsdir_inst_id,
             pipeline_name="DICOM anonymization, niftii conversion, and push to neuro tree v20250326",
-            pipeline_params=plugin_params
+            pipeline_params=plugin_params,
+            recipients=send_params['recipients'],
+            smtp_server=send_params['smtp_server'],
+            series_data=json.dumps(d_series)
         )
         return d_ret
 

dy_regiFlow.py

@@ -13,6 +13,7 @@
 import os
 import pfdcm
 import copy
+import asyncio
 
 LOG = logger.debug
 
@@ -27,7 +28,7 @@
 logger.remove()
 logger.add(sys.stderr, format=logger_format)
 
-__version__ = '1.0.9'
+__version__ = '1.1.0'
 
 DISPLAY_TITLE = r"""
        _           _                          _______ _               
@@ -126,6 +127,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__}')
 
@@ -179,7 +192,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.")
@@ -221,10 +234,14 @@ def create_hash_table(retrieve_data: dict, retry: int) -> dict:
         retry_table[series["SeriesInstanceUID"]]["SeriesInstanceUID"] = series["SeriesInstanceUID"]
         retry_table[series["SeriesInstanceUID"]]["StudyInstanceUID"] = series["StudyInstanceUID"]
         retry_table[series["SeriesInstanceUID"]]["AccessionNumber"] = series["AccessionNumber"]
+        retry_table[series["SeriesInstanceUID"]]["PatientID"] = series["PatientID"]
+        retry_table[series["SeriesInstanceUID"]]["StudyDate"] = series["StudyDate"]
+        retry_table[series["SeriesInstanceUID"]]["Modality"] = series["Modality"]
+
     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
@@ -266,18 +283,21 @@ def check_registration(options: Namespace, retry_table: dict, client: PACSClient
                 "neuro_dcm_location": options.neuroDicomLocation,
                 "neuro_anon_location": options.neuroAnonLocation,
                 "neuro_nifti_location": options.neuroNiftiLocation,
-                "folder_name": options.folderName
+                "folder_name": options.folderName,
+                "recipients": options.recipients,
+                "smtp_server": options.SMTPServer
             }
             dicom_dir = client.get_pacs_files({'SeriesInstanceUID': series_instance})
+            series_data = json.dumps(retry_table[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, send_params, options.pluginInstanceID, series_data)
             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
 
 if __name__ == '__main__':

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,152 @@ 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 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_notification_plugin(self, pv_id: int, msg: str, rcpts: str, smtp: str, series_data: str) -> int:
+        """
+        Run the pl-notification plugin.
+        """
+        d_series = json.loads(series_data)
+        email_content = (f"An error occurred while running anonymization pipeline on the following data: "
+                         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 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": "pipeline-error",
+                "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}")
+
 
-    def run_pipeline(self, pipeline_name: str, previous_inst: int, pipeline_params: dict):
+    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 +265,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)}

requirements.txt

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