diff --git a/api-app/README.md b/api-app/README.md new file mode 100644 index 0000000..a539fcd --- /dev/null +++ b/api-app/README.md @@ -0,0 +1,172 @@ +# DICOM Migration REST API + +This REST API provides an endpoint to trigger DICOM study migration for a specific Accession Number. + +## Features + +- Migrate DICOM studies by Accession Number +- Token-based authentication +- Detailed JSON responses with operation status +- Proper logging and error handling +- Automatic cleanup of DICOM files after each request + +## Requirements + +- Python 3.9 or higher +- FastAPI +- Uvicorn +- Dependencies from the main DICOM migration project + +## Installation + +1. Install the required dependencies: + +```bash +cd api-app +pip install -r requirements.txt +``` + +## Configuration + +The API uses the same configuration settings as the main DICOM migration project. + +### Authentication + +The API uses token-based authentication. Set your token using the environment variable: + +```bash +export API_TOKEN=your-secure-token +``` + +If no token is provided, a default token will be used (not recommended for production). + +## Usage + +### Starting the API Server + +```bash +# Start the server on default port 8000 +python run.py + +# Specify a custom port +python run.py --port 9000 + +# Specify a custom token +python run.py --token your-secure-token + +# Enable auto-reload for development +python run.py --reload +``` + +### API Endpoints + +#### 1. Migrate Study by Accession Number + +``` +POST /migrate/{accession_number} +``` + +**Authentication:** +- Bearer token in Authorization header + +**Response:** + +Successful migration: +```json +{ + "success": true, + "message": "Successfully migrated study for accession number: ABC12345", + "details": { + "study_uid": "1.2.826.0.1.3680043.9.7307.1.123456", + "his_integration_success": true + } +} +``` + +Failed migration: +```json +{ + "success": false, + "message": "Failed to process study: Study not found", + "details": null +} +``` + +## Example API Calls + +### Using curl + +```bash +# Migrate a study by accession number +curl -X POST "http://localhost:8000/migrate/ABC12345" \ + -H "Authorization: Bearer your-token" \ + -H "Content-Type: application/json" +``` + +### Using Python requests + +```python +import requests + +url = "http://localhost:8000/migrate/ABC12345" +headers = { + "Authorization": "Bearer your-token", + "Content-Type": "application/json" +} + +response = requests.post(url, headers=headers) +print(response.json()) +``` + +## Logs + +The API logs are stored in the `logs/api.log` file. The log includes information about: + +- API startup and shutdown +- Migration requests +- Errors during migration process +- Integration with HIS results + +## Troubleshooting + +1. **Authentication Error**: + - Make sure you're providing the correct token in the Authorization header + - Check that the token format is `Bearer your-token` + +2. **Study Not Found**: + - Verify that the accession number is correct + - Check if the PACS server is reachable + - Verify that the study exists in the source PACS + +3. **Connection Error**: + - Check PACS connectivity + - Verify network settings in `config/settings.py` + +## Integrating with External Systems + +You can call this API from other systems to trigger DICOM migrations for specific studies: + +1. Obtain the API token +2. Make a POST request to `/migrate/{accession_number}` +3. Handle the JSON response based on the `success` field + +## Technical Details + +### DICOM File Cleanup + +The API implements a robust cleanup mechanism to ensure that DICOM files are always removed after processing: + +1. **Exit Handlers**: The application registers cleanup handlers via Python's `atexit` module and signal handlers for graceful cleanup during server shutdown. + +2. **Per-Request Cleanup**: After each migration request completes (whether successful or not), the API explicitly calls the cleanup function to remove all DICOM files associated with the study. + +3. **Directory Registration**: All temporary directories created during the migration process are registered for cleanup, ensuring no files are left behind. + +This dual approach (exit handlers + explicit cleanup) ensures that: +- DICOM files are immediately removed after each API request completes +- No files are left behind even if the server crashes or is terminated abnormally +- Disk space is efficiently managed during high-volume operations + +### Authentication + +The API uses Bearer token authentication. The token is specified through: diff --git a/api-app/main.py b/api-app/main.py new file mode 100644 index 0000000..fa421a5 --- /dev/null +++ b/api-app/main.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +""" +FastAPI application for DICOM migration service. +""" +import sys +import os +import json +from typing import Dict, Any, Optional +from fastapi import FastAPI, HTTPException, Depends, status +from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials +from pydantic import BaseModel + +# Add parent directory to sys.path to allow importing from the main project +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +# Import the process module for API-specific processing +from process import process_study_by_accession +from utils.logger import setup_logger +from utils.dicom_utils import create_directory_if_not_exists +from utils.cleanup import register_exit_handlers + +# Create API logger +api_logger = setup_logger("api", "logs/api.log") + +# Initialize FastAPI app +app = FastAPI( + title="DICOM Migration API", + description="API for migrating DICOM studies by accession number", + version="1.0.0" +) + +# Security scheme for Bearer token authentication +security = HTTPBearer() + +# Environment variable for API key (use a secure method in production) +API_TOKEN = os.environ.get("API_TOKEN", "token-his2-untuk-hit-api-migrasi-clarity") + +class MigrationResponse(BaseModel): + """Response model for migration endpoints.""" + success: bool + message: str + details: Optional[Dict[str, Any]] = None + +def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)): + """ + Verify the authentication token. + + Args: + credentials: Bearer token credentials + + Returns: + bool: True if token is valid + + Raises: + HTTPException: If token is invalid + """ + if credentials.scheme != "Bearer": + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid authentication scheme. Bearer token required." + ) + + if credentials.credentials != API_TOKEN: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid token or expired token" + ) + + return True + +@app.on_event("startup") +async def startup_event(): + """Initialize resources when the API starts.""" + register_exit_handlers() + + api_logger.info("API started successfully") + +@app.get("/", response_model=MigrationResponse) +async def root(): + """ + Root endpoint to check if the API is running. + """ + return { + "success": True, + "message": "DICOM Migration API is running", + "details": { + "version": "1.0.0", + "copyright": "PT. Sadhana Abiyasa Sampoerna", + "developer": "Mario and FX Padmanto" + } + } + +@app.post("/migrate/{accession_number}", response_model=MigrationResponse) +async def migrate_study( + accession_number: str, + _: bool = Depends(verify_token) +): + """ + Migrate a single study by its accession number. + + Args: + accession_number: The accession number of the study to migrate + + Returns: + MigrationResponse: The result of the migration operation + """ + api_logger.info(f"Received migration request for accession number: {accession_number}") + + # Process the study using the specialized API processing module + result = process_study_by_accession(accession_number) + + if result['success']: + api_logger.info(f"Successfully processed migration for accession number: {accession_number}") + else: + api_logger.error(f"Failed to process migration for accession number: {accession_number}: {result['message']}") + + return result + +if __name__ == "__main__": + import uvicorn + uvicorn.run("main:app", host="0.0.0.0", port=8888, reload=True) diff --git a/api-app/process.py b/api-app/process.py new file mode 100644 index 0000000..f3095ca --- /dev/null +++ b/api-app/process.py @@ -0,0 +1,278 @@ +#!/usr/bin/env python3 +""" +Process module for DICOM migration API service. +This module contains the core processing logic for migrating DICOM studies through the API. +""" +import os +import sys +import json +import shutil +from datetime import datetime +import requests + +# Add parent directory to sys.path to allow importing from the main project +sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from config import settings +from services.dicom_finder import DicomFinder +from services.dicom_retriever import DicomRetriever +from services.dicom_sender import DicomSender +from utils.logger import setup_logger +from utils.dicom_utils import save_json_data, create_directory_if_not_exists + +# Create API process logger +process_logger = setup_logger("api_process", "logs/api_process.log") + + +def cleanup_study_directory(study_uid): + """ + Explicitly clean up the DICOM files for a study after processing. + + Args: + study_uid: Study Instance UID to clean up + + Returns: + bool: True if cleanup was successful, False otherwise + """ + study_dir = os.path.join(settings.DICOM_STORE_DIR, study_uid) + process_logger.info(f"Cleaning up DICOM files for study: {study_uid} at {study_dir}") + + try: + if os.path.exists(study_dir): + shutil.rmtree(study_dir) + process_logger.info(f"Successfully removed study directory: {study_dir}") + return True + else: + process_logger.warning(f"Study directory does not exist: {study_dir}") + return True # Return True since there's nothing to clean + except Exception as e: + process_logger.error(f"Error cleaning up study directory {study_dir}: {str(e)}") + return False + + +def process_study_by_accession(accession_number): + """ + Process a study by its accession number for the API service. + + This function: + 1. Finds the study UID based on accession number + 2. Retrieves DICOM files using C-GET + 3. Sends DICOM files to destination PACS using C-STORE + 4. Sends study information to HIS API + 5. Cleans up temporary DICOM files + + Args: + accession_number: The accession number of the study to process + + Returns: + dict: Result with success status, message, and details + """ + process_logger.info(f"Processing study with accession number: {accession_number}") + + # Initialize response data + response_data = { + "success": False, + "message": "", + "details": { + "accession_number": accession_number, + "process_start_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), + "steps_completed": [] + } + } + + try: + # STEP 1: Find study by accession number to get the Study UID + process_logger.info(f"Finding study with accession number: {accession_number}") + finder = DicomFinder() + study = finder.find_study_by_accession(accession_number) + + if not study: + process_logger.error(f"Study not found for accession number: {accession_number}") + response_data["message"] = f"Study not found for accession number: {accession_number}" + return response_data + + study_uid = study.StudyInstanceUID + patient_id = getattr(study, 'PatientID', '') + study_description = getattr(study, 'StudyDescription', '') + study_date = getattr(study, 'StudyDate', '') + study_time = getattr(study, 'StudyTime', '000000') + + response_data["details"]["study_uid"] = study_uid + response_data["details"]["patient_id"] = patient_id + response_data["details"]["study_description"] = study_description + + process_logger.info(f"Found study with UID: {study_uid} for accession number: {accession_number}") + response_data["details"]["steps_completed"].append("study_found") + + # STEP 2: Retrieve the study using C-GET + process_logger.info(f"Retrieving study: {study_uid}") + retriever = DicomRetriever() + retrieve_result = retriever.retrieve_study(study_uid, accession_number=accession_number) + + if not retrieve_result['success']: + process_logger.error(f"Failed to retrieve study {study_uid}: {retrieve_result['status']}") + cleanup_study_directory(study_uid) # Clean up any partial files + response_data["message"] = f"Failed to retrieve study: {retrieve_result['status']}" + return response_data + + process_logger.info(f"Retrieved {retrieve_result['successful_instances']} instances for study {study_uid}") + response_data["details"]["instances_retrieved"] = retrieve_result['successful_instances'] + response_data["details"]["steps_completed"].append("study_retrieved") + + # STEP 3: Send the study to destination PACS using C-STORE + process_logger.info(f"Sending study {study_uid} to destination PACS") + sender = DicomSender() + send_result = sender.send_study(os.path.join(settings.DICOM_STORE_DIR, study_uid)) + + response_data["details"]["files_sent"] = send_result['successful_sends'] + response_data["details"]["total_files"] = send_result['total_files'] + + if not send_result['success']: + process_logger.error(f"Failed to send study {study_uid}: {send_result.get('error', 'Unknown error')}") + # We continue processing even if send fails, to log the details + response_data["details"]["c_store_success"] = False + response_data["details"]["c_store_error"] = send_result.get('error', 'Unknown error') + else: + process_logger.info(f"Successfully sent {send_result['successful_sends']} of {send_result['total_files']} files to destination PACS") + response_data["details"]["c_store_success"] = True + + response_data["details"]["steps_completed"].append("study_sent") + + # STEP 4: Create detailed log for the study + process_logger.info(f"Creating detailed log for study: {study_uid}") + series_list = finder.find_series_for_study(study_uid, accession_number=accession_number) + + study_datetime = f"{study_date}{study_time}" + + study_log = { + 'Study_IUID': study_uid, + 'AccessionNumber': accession_number, + 'PatientID': patient_id, + 'StudyDescription': study_description, + 'StudyDateTime': study_datetime, + 'CstoreSuccess': send_result['success'], + 'Series': [] + } + + for series in series_list: + series_uid = series.SeriesInstanceUID + + # Get first instance for the series + instance = finder.find_first_instance_for_series(study_uid, series_uid) + + # Helper function to safely get attributes + def safe_get_attr(obj, attr_name, default=''): + """Get attribute value, ensuring None is converted to default.""" + value = getattr(obj, attr_name, default) + return default if value is None else str(value) + + series_info = { + 'Series_IUID': safe_get_attr(series, 'SeriesInstanceUID'), + 'SeriesNumber': safe_get_attr(series, 'SeriesNumber'), + 'SeriesDescription': safe_get_attr(series, 'SeriesDescription'), + 'NumberOfInstances': safe_get_attr(series, 'NumberOfSeriesRelatedInstances'), + 'SOP_IUID': safe_get_attr(instance, 'SOPInstanceUID') if instance else '' + } + + study_log['Series'].append(series_info) + + response_data["details"]["series_count"] = len(study_log['Series']) + response_data["details"]["steps_completed"].append("log_created") + + # STEP 5: Send study_log to HIS API + process_logger.info(f"Sending study data to HIS API for accession: {accession_number}") + his_url = f"http://{settings.HIS_HOST}{settings.HIS_URL}" + + try: + headers = { + 'id': 'Vmtaa2MySnRUblJTYWtKb1ZucHNNVlZVU2pSaFIwNTBZa1JDYkZaV1NtOWFSV1JIVmxkSmQxSnJUbFpTVlZwRlZsaGpPVkJSUFQwPQ==', + 'Content-Type': 'application/json' + } + response = requests.post(his_url, json=study_log, headers=headers) + + # Add response data to the log + study_log['HisApiResponse'] = { + 'StatusCode': response.status_code, + 'ResponseText': response.text, + 'Success': False, + 'Timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S') + } + + response_data["details"]["his_api_status_code"] = response.status_code + + # Parse the response + try: + response_data_his = response.json() if response.content else {} + study_log['HisApiResponse']['ResponseData'] = response_data_his + + if response.status_code == 200 and response_data_his.get('OK') == "1": + study_log['HisApiResponse']['Success'] = True + process_logger.info(f"Successfully sent JSON {accession_number} to HIS API") + response_data["details"]["his_integration_success"] = True + response_data["details"]["steps_completed"].append("his_api_success") + else: + error_msg = response_data_his.get('message', 'Unknown error') + process_logger.error(f"Failed to send JSON for {accession_number} to HIS API: {error_msg}. Study_IUID: {study_uid}") + response_data["details"]["his_integration_success"] = False + response_data["details"]["his_error"] = error_msg + response_data["details"]["steps_completed"].append("his_api_failed") + except ValueError: + study_log['HisApiResponse']['ParseError'] = "Invalid JSON response" + process_logger.error(f"Failed to parse response for {accession_number}, invalid JSON response. Status code: {response.status_code}") + response_data["details"]["his_integration_success"] = False + response_data["details"]["his_error"] = "Invalid JSON response from HIS API" + response_data["details"]["steps_completed"].append("his_api_failed") + + except Exception as e: + study_log['HisApiResponse'] = { + 'Success': False, + 'Error': str(e), + 'Timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S') + } + process_logger.error(f"Error sending study for {accession_number} to HIS API: {str(e)}") + response_data["details"]["his_integration_success"] = False + response_data["details"]["his_error"] = str(e) + response_data["details"]["steps_completed"].append("his_api_error") + + # STEP 6: Clean up DICOM files + cleanup_result = cleanup_study_directory(study_uid) + if cleanup_result: + process_logger.info(f"Successfully cleaned up DICOM files for study: {study_uid}") + response_data["details"]["cleanup_success"] = True + response_data["details"]["steps_completed"].append("cleanup_success") + else: + process_logger.warning(f"Failed to clean up DICOM files for study: {study_uid}") + response_data["details"]["cleanup_success"] = False + response_data["details"]["steps_completed"].append("cleanup_failed") + + # STEP 7: Save the study log for reference + log_dir = settings.JSON_OUTPUT_DIR + create_directory_if_not_exists(log_dir) + log_filename = f"api_{accession_number}.json" + save_json_data(study_log, log_filename, log_dir) + process_logger.info(f"Saved study log to {os.path.join(log_dir, log_filename)}") + + # Set final response based on C-STORE success (primary success metric) + response_data["success"] = send_result['success'] + if send_result['success']: + response_data["message"] = f"Berhasil migrasi file DICOM Accession Number: {accession_number}" + else: + response_data["message"] = f"Gagal migrasi file DICOM: {send_result.get('error', 'Unknown error')}" + + process_logger.info(f"Completed processing study with accession number: {accession_number}") + response_data["details"]["process_end_time"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + return response_data + + except Exception as e: + process_logger.exception(f"Unexpected error processing study {accession_number}: {str(e)}") + + # Try to clean up if we have a study_uid + if 'study_uid' in locals(): + cleanup_study_directory(study_uid) + + response_data["message"] = f"Error during migration: {str(e)}" + response_data["details"]["error"] = str(e) + response_data["details"]["process_end_time"] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + return response_data diff --git a/api-app/requirements.txt b/api-app/requirements.txt new file mode 100644 index 0000000..ae57f46 --- /dev/null +++ b/api-app/requirements.txt @@ -0,0 +1,9 @@ +fastapi>=0.95.0 +uvicorn>=0.22.0 +pydicom>=2.3.0 +pynetdicom>=2.0.0 +python-dateutil>=2.8.2 +retry==0.9.2 +requests>=2.28.0 +python-jose>=3.3.0 +python-multipart>=0.0.6 diff --git a/api-app/run.py b/api-app/run.py new file mode 100644 index 0000000..ef491ad --- /dev/null +++ b/api-app/run.py @@ -0,0 +1,41 @@ +#!/usr/bin/env python3 +""" +Script to run the FastAPI server for DICOM migration service. +""" +import os +import sys +import uvicorn +import argparse + +def parse_args(): + """Parse command-line arguments.""" + parser = argparse.ArgumentParser(description='DICOM Migration API Server') + parser.add_argument('--host', default='0.0.0.0', help='Host to bind to') + parser.add_argument('--port', type=int, default=8000, help='Port to bind to') + parser.add_argument('--token', help='API token for authentication (overrides environment variable)') + parser.add_argument('--reload', action='store_true', help='Enable auto-reload for development') + return parser.parse_args() + +def main(): + """Main function to run the API server.""" + args = parse_args() + + # Set API token if provided + if args.token: + os.environ['API_TOKEN'] = args.token + + # Check if API_TOKEN is set + if not os.environ.get('API_TOKEN'): + print("WARNING: API_TOKEN not set. Using default token 'token-his2-untuk-hit-api-migrasi-clarity'.") + print("Set environment variable API_TOKEN or use --token argument for better security.") + + print(f"Starting DICOM Migration API server on {args.host}:{args.port}") + uvicorn.run( + "main:app", + host=args.host, + port=args.port, + reload=args.reload + ) + +if __name__ == "__main__": + main() diff --git a/services/dicom_finder.py b/services/dicom_finder.py index d0cfe52..01f334c 100644 --- a/services/dicom_finder.py +++ b/services/dicom_finder.py @@ -347,4 +347,82 @@ class DicomFinder: logger.error(error_msg) raise DicomQueryError(error_msg) + return study + + @dicom_retry(exception_types=(DicomQueryError, ConnectionError)) + def find_study_by_accession(self, accession_number): + """ + Find a specific study by its AccessionNumber. + + Args: + accession_number (str): AccessionNumber to find + + Returns: + Dataset: Study dataset or None if not found + """ + logger.info(f"Finding study with AccessionNumber: {accession_number}") + + # Create query dataset + ds = Dataset() + ds.QueryRetrieveLevel = 'STUDY' + + # Set the AccessionNumber filter + ds.AccessionNumber = accession_number + + # Required fields (minimal set) + ds.StudyInstanceUID = '' + ds.StudyDate = '' + + # Additional fields we want to retrieve + ds.PatientID = '' # MedrecID + ds.StudyDescription = '' + ds.StudyTime = '' + ds.NumberOfStudyRelatedSeries = '' + + # Create association + assoc = self.ae.associate( + self.pacs_config['host'], + self.pacs_config['port'], + ae_title=self.pacs_config['aet'] + ) + + study = None + + if assoc.is_established: + try: + logger.debug("Association established, sending C-FIND request") + + # Send C-FIND request + responses = assoc.send_c_find( + ds, + StudyRootQueryRetrieveInformationModelFind + ) + + # Process responses - just get the first one + for (status, dataset) in responses: + if status and status.Status == 0xFF00: # Pending + if dataset: + study = dataset + logger.debug(f"Found study: {dataset.StudyInstanceUID} for AccessionNumber: {accession_number}") + break + elif status and status.Status != 0x0000: # Not success + logger.error(f"C-FIND error: {status}") + + if study: + logger.info(f"Found study with AccessionNumber {accession_number}, StudyUID: {study.StudyInstanceUID}") + else: + logger.warning(f"No study found with AccessionNumber {accession_number}") + + except Exception as e: + logger.error(f"Error during C-FIND by AccessionNumber: {str(e)}") + raise DicomQueryError(f"C-FIND operation failed: {str(e)}") + finally: + # Release the association + assoc.release() + logger.debug("Association released") + else: + error_msg = f"Association rejected, aborted or never connected to {self.pacs_config['aet']}" + logger.error(error_msg) + raise DicomQueryError(error_msg) + return study \ No newline at end of file diff --git a/utils/cleanup.py b/utils/cleanup.py index a5f29db..edf1f01 100644 --- a/utils/cleanup.py +++ b/utils/cleanup.py @@ -42,6 +42,36 @@ def cleanup_dicom_files(): # Clear the registry after cleanup _cleanup_dirs.clear() +def cleanup_study_directory(study_uid): + """ + Immediately remove the directory for a specific study. + + Args: + study_uid (str): Study Instance UID of the study to clean up + + Returns: + bool: True if cleanup was successful, False otherwise + """ + study_dir = os.path.join(settings.DICOM_STORE_DIR, study_uid) + + if not os.path.exists(study_dir): + logger.debug(f"Cleanup: Study directory does not exist: {study_dir}") + return True + + try: + shutil.rmtree(study_dir) + logger.info(f"Cleanup: Removed DICOM files from {study_dir}") + + # Also remove from the registry if it's there + global _cleanup_dirs + if study_dir in _cleanup_dirs: + _cleanup_dirs.remove(study_dir) + + return True + except Exception as e: + logger.error(f"Cleanup: Failed to remove DICOM files from {study_dir}: {str(e)}") + return False + def register_exit_handlers(): """ Register cleanup handlers for various exit scenarios.