mario/pydicom-migrasi-clarity
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

add: fastapi wrapped baru mau di tes di Gajah Mada

Browse Source
This commit is contained in:
mario
2025-07-11 15:01:27 +07:00
parent 1b8eeee04b
commit 853cc70edb
7 changed files with 729 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
api-app/README.md Normal file
View File

@@ -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:

121
api-app/main.py Normal file
View File

@@ -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)

278
api-app/process.py Normal file
View File

@@ -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

9
api-app/requirements.txt Normal file
View File

@@ -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

41
api-app/run.py Normal file
View File

@@ -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()

78
services/dicom_finder.py
View File

@@ -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

30
utils/cleanup.py
View File

@@ -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.