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

Merge branch 'main' of https://devone.aplikasi.web.id/gitea/mario/pydicom-migrasi-clarity into main

Browse Source
This commit is contained in:
mario
2025-07-11 15:16:11 +07:00
parent 3850aa103d 1b8eeee04b
commit 54f29f2a88
6 changed files with 524 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

143
README.md
View File

@@ -1,3 +1,144 @@
# pydicom-migrasi-clarity
Untuk migrasi PACS Clarity ke ABPACS dengan cara: FindSCU by date, GetSCU for each StudyIUID found, StoreSCU to ABPACS, then send API to HIS2 to fill pacs_result_series
## Tujuan Proyek
Aplikasi ini dibuat untuk melakukan migrasi data PACS dari sistem Clarity ke ABPACS dengan proses sebagai berikut:
1. Melakukan pencarian studi DICOM menggunakan FindSCU berdasarkan tanggal
2. Mengambil data DICOM lengkap menggunakan GetSCU untuk setiap StudyIUID yang ditemukan
3. Mengirim data DICOM ke ABPACS menggunakan StoreSCU
4. Mengirim data ke API HIS2 untuk mengisi pacs_result_series
5. Mencatat seluruh proses dalam log terstruktur
## Alur Kerja Detail
Aplikasi ini menggunakan pendekatan berbasis Python murni dengan bantuan library pynetdicom. Berikut tahapan detail proses:
1. **Pencarian (FindSCU)**:
- Mencari studi berdasarkan rentang tanggal (StudyDate)
- Mengekstrak metadata penting seperti StudyInstanceUID, AccessionNumber, PatientID
- Menyimpan hasil pencarian dalam format JSON
2. **Pengambilan (GetSCU)**:
- Mengunduh seluruh data DICOM untuk setiap StudyInstanceUID yang ditemukan
- Menyimpan data DICOM dalam struktur direktori terorganisir
- Data disimpan sementara dan dihapus otomatis setelah pengiriman
3. **Pengiriman (StoreSCU)**:
- Mengirim data DICOM yang diunduh ke server ABPACS
- Melacak status pengiriman dan menangani error atau kegagalan
- Menghapus data DICOM setelah pengiriman berhasil
4. **Integrasi HIS**:
- Mengirim metadata DICOM ke API HIS2
- Memastikan informasi studi dan series tersimpan di sistem HIS
- Menyimpan log hasil komunikasi dengan sistem HIS
## Persyaratan
- Python 3.9
- Pustaka Python utama:
- pydicom
- pynetdicom
- requests
- python-dateutil
- retry
- Akses jaringan ke server PACS sumber dan tujuan
## Cara Mengkloning Repositori
```bash
git clone https://devone.aplikasi.web.id/gitea/mario/pydicom-migrasi-clarity.git
cd pydicom-migrasi-clarity
```
## Instalasi dengan Virtual Environment
### 1. Membuat Virtual Environment
```bash
# Untuk Linux
python3.9 -m venv venv
# Untuk Windows
python -m venv venv
```
### 2. Mengaktifkan Virtual Environment
```bash
# Untuk Linux
source venv/bin/activate
# Untuk windows
venv/Scripts/activate
```
### 3. Menginstall Dependensi
```bash
pip install -r requirements.txt
```
## Konfigurasi PACS
**Sesuaikan `SOURCE_PACS` dan `DESTINATION_PACS` pada `config/settings.py`**
## Cara Menjalankan Aplikasi
### Aktifkan venv
```bash
source /home/pacs/pydicom-migrasi-clarity/venv/bin/activate
```
### Contoh Penggunaan Umum
Untuk migrasi data harian baru:
```bash
python main.py process --start-date 20250507 --end-date 20250507
```
### Perintah pendukung yang Tersedia
Program ini mendukung beberapa mode operasi:
1. **Process** - Menjalankan alur kerja lengkap (pencarian, pengambilan, pengiriman):
```bash
python main.py process --start-date 20250501 --end-date 20250502
```
2. **Process-Study** - Menjalankan alur kerja lengkap untuk studi tertentu:
```bash
python main.py process-study --study-uid 1.2.826.1.3680043.9.5282.150415.30338.202504010001
```
3. **Find-Studies** - Hanya mencari studi berdasarkan rentang tanggal:
```bash
python main.py find-studies --start-date 20250501 --end-date 20250502
```
4. **Get-Study** - Mengambil studi tertentu berdasarkan StudyInstanceUID:
```bash
python main.py get-study --study-uid 1.2.826.1.3680043.9.5282.150415.30338.202504010001
```
5. **Send-Study** - Mengirim studi tertentu ke PACS tujuan:
```bash
python main.py send-study --study-uid 1.2.826.1.3680043.9.5282.150415.30338.202504010001
```
### Parameter Umum
* `--start-date`: Tanggal awal pencarian dalam format YYYYMMDD
* `--end-date`: Tanggal akhir pencarian dalam format YYYYMMDD
* `--study-uid`: StudyInstanceUID untuk operasi pada satu studi
* `--series-uid`: SeriesInstanceUID untuk operasi pada satu seri
* `--skip-existing`: Lewati studi yang sudah memiliki log (untuk melanjutkan proses yang terhenti)
## Struktur Direktori
- `config/`: Berisi file konfigurasi aplikasi
- `logs/`: Menyimpan log operasi dan hasil pengiriman
- `output/`: Tempat menyimpan hasil DICOM dan file JSON
- `services/`: Modul-modul untuk operasi DICOM (finder, retriever, sender)
- `utils/`: Fungsi-fungsi pembantu dan utilitas
## Troubleshooting
Jika menemui masalah, silakan periksa file log di direktori `logs/` untuk informasi lebih detail.
Untuk menonaktifkan virtual environment setelah selesai:
```bash
deactivate
```

10
config/settings.py
View File

@@ -8,14 +8,14 @@ SOURCE_PORT = 8888 # Our port
# Source PACS Configuration (where to query/get data from)
SOURCE_PACS = {
"host": "192.168.22.3",
"host": "192.168.2.30",
"port": 11112,
"aet": "ABPACS"
"aet": "FREEDOM"
}
# Destination PACS Configuration (where to send data to)
DESTINATION_PACS = {
"host": "152.42.173.210",
"host": "192.168.1.29",
"port": 11112,
"aet": "ABPACS"
}
@@ -40,5 +40,5 @@ JSON_OUTPUT_DIR = "output/json"
DICOM_STORE_DIR = "output/dicom"
# HIS Configuration
HIS_HOST = "localhost:8787"
HIS_URL = "/result_series"
HIS_HOST = "192.168.1.1:80"
HIS_URL = "/poapi/api_migrasiclarity.php?name=migrasi_data_pacs"

332
main.py
View File

@@ -15,6 +15,7 @@ from utils.cleanup import register_exit_handlers, register_cleanup_dir
from services.dicom_finder import DicomFinder
from services.dicom_retriever import DicomRetriever
from services.dicom_sender import DicomSender
import signal
def parse_args():
"""Parse command line arguments."""
@@ -63,6 +64,11 @@ def parse_args():
send_file_parser = subparsers.add_parser('send-file', help='Send a single DICOM file to destination PACS')
send_file_parser.add_argument('--file-path', required=True, help='Path to DICOM file to send')
# Add to the parse_args() function after the 'process' command
process_study_parser = subparsers.add_parser('process-study', help='Process full workflow for a specific study by StudyInstanceUID')
process_study_parser.add_argument('--study-uid', required=True, help='StudyInstanceUID to process')
process_study_parser.add_argument('--log-dir', help='Directory to save JSON logs (default: JSON_OUTPUT_DIR)')
# Process command - full workflow: find, get, and send DICOM data for a date range
process_parser = subparsers.add_parser('process', help='Process full workflow (find, get, send) for a date range')
process_parser.add_argument('--start-date', required=True, help='Start date in YYYYMMDD format')
@@ -357,6 +363,150 @@ def send_file(args):
logger.error(f"Error sending file: {str(e)}")
sys.exit(1)
def process_single_study(study_uid, accession_number=None, log_dir=None):
"""
Process a single study: retrieve it, send to destination PACS,
and create detailed JSON logs.
Args:
study_uid: Study Instance UID
accession_number: Optional accession number
log_dir: Directory to save logs
Returns:
dict: Result information
"""
log_dir = log_dir or settings.JSON_OUTPUT_DIR
# Create services
finder = DicomFinder()
retriever = DicomRetriever()
sender = DicomSender()
# First, get the study metadata with a query if not provided
if not accession_number:
study_metadata = finder.find_study_by_uid(study_uid)
if not study_metadata:
logger.error(f"Study not found: {study_uid}")
return {'success': False, 'error': 'Study not found'}
accession_number = getattr(study_metadata, 'AccessionNumber', '')
else:
study_metadata = None
logger.info(f"Processing study: {accession_number} with Study_IUID ({study_uid})")
# STEP 1: Retrieve the study
retrieve_result = retriever.retrieve_study(study_uid, accession_number=accession_number)
if not retrieve_result['success']:
logger.error(f"Failed to retrieve study {study_uid}: {retrieve_result['status']}")
return {'success': False, 'error': f"Failed to retrieve study: {retrieve_result['status']}"}
logger.info(f"Retrieved {retrieve_result['successful_instances']} instances for study {study_uid}")
# STEP 2: Send the study to destination PACS
send_result = sender.send_study(os.path.join(settings.DICOM_STORE_DIR, study_uid))
if not send_result['success']:
logger.error(f"Failed to send study {study_uid}: {send_result.get('error', 'Unknown error')}")
logger.info(f"Sent {send_result['successful_sends']} of {send_result['total_files']} files to destination PACS")
# STEP 3: Find all series and create detailed logs
series_list = finder.find_series_for_study(study_uid, accession_number=accession_number)
# Get metadata from study_metadata if available, otherwise from the first element of study list
if study_metadata:
study_date = getattr(study_metadata, 'StudyDate', '')
study_time = getattr(study_metadata, 'StudyTime', '000000')
patient_id = getattr(study_metadata, 'PatientID', '')
study_description = getattr(study_metadata, 'StudyDescription', '')
else:
# We need to query for it again
study_details = finder.find_study_by_uid(study_uid)
study_date = getattr(study_details, 'StudyDate', '') if study_details else ''
study_time = getattr(study_details, 'StudyTime', '000000') if study_details else '000000'
patient_id = getattr(study_details, 'PatientID', '') if study_details else ''
study_description = getattr(study_details, 'StudyDescription', '') if study_details else ''
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)
# Kalau null atau None, forced ke ''
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)
# STEP 4: Send study_log to HIS API
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')
}
# Safer JSON parsing
try:
response_data = response.json() if response.content else {}
study_log['HisApiResponse']['ResponseData'] = response_data
if response.status_code == 200 and response_data.get('OK') == "1":
study_log['HisApiResponse']['Success'] = True
logger.info(f"Successfully sent JSON {accession_number} to HIS API")
return {'success': True, 'study_log': study_log, 'his_success': True}
else:
error_msg = response_data.get('message', 'Unknown error')
logger.error(f"Failed to send JSON for {accession_number} to HIS API: {error_msg}. Study_IUID: {study_uid}")
return {'success': True, 'study_log': study_log, 'his_success': False}
except ValueError:
study_log['HisApiResponse']['ParseError'] = "Invalid JSON response"
logger.error(f"Failed to parse response for {accession_number}, invalid JSON response. Status code: {response.status_code}")
return {'success': True, 'study_log': study_log, 'his_success': False}
except Exception as e:
study_log['HisApiResponse'] = {
'Success': False,
'Error': str(e),
'Timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
logger.error(f"Error sending study for {accession_number} to HIS API: {str(e)}")
return {'success': True, 'study_log': study_log, 'his_success': False, 'error': str(e)}
def process_workflow(args):
"""
Process the full workflow: find studies by date range, retrieve them, and send to destination PACS.
@@ -387,6 +537,7 @@ def process_workflow(args):
# Create directory for logs
create_directory_if_not_exists(log_dir)
create_directory_if_not_exists("logs")
# STEP 1: Find studies in the date range
try:
@@ -406,12 +557,30 @@ def process_workflow(args):
save_json_data(studies_summary, f"studies_{start_date}_to_{end_date}.json", log_dir)
# STEP 2, 3, 4: For each study, retrieve, send, and log details
his_log_filename = f"sendtohis_{start_date}_{end_date}.json"
his_log_path = os.path.join("logs", his_log_filename)
his_log = []
his_fail_log_filename = f"fail_sendtohis_{start_date}_{end_date}.json"
his_fail_log_path = os.path.join("logs", his_fail_log_filename)
his_fail_log = []
his_log_filename = f"sendtohis_{start_date}_{end_date}.json"
his_fail_log_filename = f"fail_sendtohis_{start_date}_{end_date}.json"
# Register signal handler for SIGINT
original_sigint_handler = signal.getsignal(signal.SIGINT)
def sigint_handler(sig, frame):
logger.warning("Process interrupted by user. Saving logs before exiting...")
# Save HIS logs
if his_log:
save_json_data(his_log, his_log_filename, "logs")
logger.info(f"Saved {len(his_log)} successful logs to {his_log_filename}")
if his_fail_log:
save_json_data(his_fail_log, his_fail_log_filename, "logs")
logger.info(f"Saved {len(his_fail_log)} failed logs to {his_fail_log_filename}")
# Restore original signal handler and raise KeyboardInterrupt
signal.signal(signal.SIGINT, original_sigint_handler)
sys.exit(1)
# Set up our custom signal handler
signal.signal(signal.SIGINT, sigint_handler)
for i, study in enumerate(studies):
accession_number = study.AccessionNumber
@@ -424,77 +593,35 @@ def process_workflow(args):
logger.info(f"Skipping study {study_uid} - log file already exists")
continue
# STEP 2: Retrieve the study
try:
retrieve_result = retriever.retrieve_study(study_uid, accession_number=accession_number)
if not retrieve_result['success']:
logger.error(f"Failed to retrieve study {study_uid}: {retrieve_result['status']}")
result = process_single_study(study_uid, accession_number, log_dir)
if not result['success']:
logger.error(f"Failed to process study {study_uid}: {result.get('error', 'Unknown error')}")
continue
logger.info(f"Retrieved {retrieve_result['successful_instances']} instances for study {study_uid}")
# Record results for HIS logs
if result.get('his_success', False):
his_log.append(result['study_log'])
else:
his_fail_log.append(result['study_log'])
# STEP 3: Send the study to destination PACS
send_result = sender.send_study(os.path.join(settings.DICOM_STORE_DIR, study_uid))
if not send_result['success']:
logger.error(f"Failed to send study {study_uid}: {send_result.get('error', 'Unknown error')}")
logger.info(f"Sent {send_result['successful_sends']} of {send_result['total_files']} files to destination PACS")
# STEP 4: Find all series and create detailed logs
series_list = finder.find_series_for_study(study_uid, accession_number=accession_number)
study_date = getattr(study, 'StudyDate', '')
study_time = getattr(study, 'StudyTime', '000000') # Default to midnight if StudyTime is not available
study_datetime = f"{study_date}{study_time}"
study_log = {
'Study_IUID': study_uid,
'AccessionNumber': getattr(study, 'AccessionNumber', ''),
'PatientID': getattr(study, 'PatientID', ''), # MedrecID
'StudyDescription': getattr(study, 'StudyDescription', ''),
'StudyDateTime': study_datetime,
'CstoreSuccess': send_result['success'],
'Series': []
}
for series in series_list:
series_uid = series.SeriesInstanceUID
# Save progress incrementally every 5 studies
if (i + 1) % 5 == 0:
if his_log:
save_json_data(his_log, his_log_filename, "logs")
if his_fail_log:
save_json_data(his_fail_log, his_fail_log_filename, "logs")
logger.info(f"Saved progress after processing {i+1} studies")
# Get first instance for the series
instance = finder.find_first_instance_for_series(study_uid, series_uid)
series_info = {
'Series_IUID': getattr(series, 'SeriesInstanceUID', ''),
'SeriesNumber': getattr(series, 'SeriesNumber', ''),
'SeriesDescription': getattr(series, 'SeriesDescription', ''),
'NumberOfInstances': getattr(series, 'NumberOfSeriesRelatedInstances', ''),
'SOP_IUID': getattr(instance, 'SOPInstanceUID', '') if instance else ''
}
study_log['Series'].append(series_info)
# Save the detailed log
# save_json_data(study_log, f"{study_uid}.json", log_dir)
# logger.info(f"Created detailed log for study {study_uid}")
# STEP 5: Send study_log to HIS API
his_url = f"http://{settings.HIS_HOST}{settings.HIS_URL}"
try:
response = requests.post(his_url, json=study_log)
if response.status_code == 200 and response.json().get('status') == "OK":
his_log.append(study_log)
logger.info(f"Successfully sent JSON {accession_number} to HIS API")
else:
his_fail_log.append(study_log)
logger.error(f"Failed to send JSON for {accession_number} to HIS API: {response.msg}. Study_IUID: {study_uid}")
except Exception as e:
logger.error(f"Error sending study for {accession_number} to HIS API: {str(e)}")
except Exception as e:
logger.error(f"Error processing study {accession_number}: {str(e)}")
continue
# Save HIS log
# Restore original signal handler
signal.signal(signal.SIGINT, original_sigint_handler)
# Final Save HIS logs
if his_log:
save_json_data(his_log, his_log_filename, "logs")
if his_fail_log:
@@ -506,6 +633,77 @@ def process_workflow(args):
logger.error(f"Error during workflow processing: {str(e)}")
sys.exit(1)
def process_workflow_by_study(args):
"""
Process the full workflow for a specific study.
Args:
args: Command line arguments
"""
study_uid = args.study_uid
log_dir = args.log_dir or settings.JSON_OUTPUT_DIR
logger.info(f"Starting full workflow process for study: {study_uid}")
# Create directory for logs
create_directory_if_not_exists(log_dir)
create_directory_if_not_exists("logs")
# Register signal handler for SIGINT
original_sigint_handler = signal.getsignal(signal.SIGINT)
# Create empty containers for logs
his_log = []
his_fail_log = []
his_log_filename = f"sendtohis_study_{study_uid}.json"
his_fail_log_filename = f"fail_sendtohis_study_{study_uid}.json"
def sigint_handler(sig, frame):
logger.warning("Process interrupted by user. Saving logs before exiting...")
# Save individual study log
if 'result' in locals() and result.get('success', False):
save_json_data(result['study_log'], f"{study_uid}.json", log_dir)
# Save HIS logs
if result.get('his_success', False):
save_json_data([result['study_log']], his_log_filename, "logs")
logger.info(f"Saved successful log to {his_log_filename}")
else:
save_json_data([result['study_log']], his_fail_log_filename, "logs")
logger.info(f"Saved failed log to {his_fail_log_filename}")
# Restore original signal handler and exit
signal.signal(signal.SIGINT, original_sigint_handler)
sys.exit(1)
# Set up our custom signal handler
signal.signal(signal.SIGINT, sigint_handler)
try:
result = process_single_study(study_uid, log_dir=log_dir)
if not result['success']:
logger.error(f"Failed to process study {study_uid}: {result.get('error', 'Unknown error')}")
sys.exit(1)
# Save individual study log
save_json_data(result['study_log'], f"{study_uid}.json", log_dir)
# Save logs
if result.get('his_success', False):
save_json_data([result['study_log']], his_log_filename, "logs")
else:
save_json_data([result['study_log']], his_fail_log_filename, "logs")
logger.info(f"Completed processing study: {study_uid}")
except Exception as e:
logger.error(f"Error during study processing: {str(e)}")
sys.exit(1)
finally:
# Restore original signal handler
signal.signal(signal.SIGINT, original_sigint_handler)
def main():
"""Main function."""
# Register cleanup handlers for graceful exit
@@ -544,6 +742,8 @@ def main():
send_file(args)
elif args.command == 'process':
process_workflow(args)
elif args.command == 'process-study':
process_workflow_by_study(args)
else:
logger.error("No command specified. Use --help for options.")
sys.exit(1)

80
services/dicom_finder.py
View File

@@ -269,4 +269,82 @@ class DicomFinder:
logger.error(error_msg)
raise DicomQueryError(error_msg)
return first_instance
return first_instance
@dicom_retry(exception_types=(DicomQueryError, ConnectionError))
def find_study_by_uid(self, study_instance_uid):
"""
Find a specific study by its StudyInstanceUID.
Args:
study_instance_uid (str): StudyInstanceUID to find
Returns:
Dataset: Study dataset or None if not found
"""
logger.info(f"Finding study with UID: {study_instance_uid}")
# Create query dataset
ds = Dataset()
ds.QueryRetrieveLevel = 'STUDY'
# Set the StudyInstanceUID filter
ds.StudyInstanceUID = study_instance_uid
# Required fields (minimal set)
ds.StudyDate = ''
# Additional fields we want to retrieve
ds.AccessionNumber = ''
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}")
break
elif status and status.Status != 0x0000: # Not success
logger.error(f"C-FIND error: {status}")
if study:
logger.info(f"Found study {study_instance_uid}")
else:
logger.warning(f"No study found with UID {study_instance_uid}")
except Exception as e:
logger.error(f"Error during C-FIND: {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

1
services/dicom_retriever.py
View File

@@ -39,6 +39,7 @@ class DicomRetriever:
# Add storage presentation contexts (needed for receiving the images)
# for context in StoragePresentationContexts:
# self.ae.add_requested_context(context.abstract_syntax)
storage_uids = [
'1.2.840.10008.5.1.4.1.1.1', # CR Storage
'1.2.840.10008.5.1.4.1.1.1.1', # Digital X-Ray Image Storage

35
services/dicom_sender.py
View File

@@ -6,11 +6,15 @@ import glob
import pydicom
import shutil
from pydicom.dataset import Dataset
from pynetdicom import AE, StoragePresentationContexts, evt
from pynetdicom import AE, StoragePresentationContexts, evt, build_role
from config import settings
from utils.logger import main_logger as logger
from utils.error_handler import DicomStoreError, dicom_retry
from utils.dicom_utils import create_directory_if_not_exists
from pynetdicom.sop_class import (
StudyRootQueryRetrieveInformationModelGet,
PatientRootQueryRetrieveInformationModelGet
)
class DicomSender:
"""
@@ -27,9 +31,32 @@ class DicomSender:
self.pacs_config = pacs_config or settings.DESTINATION_PACS
self.ae = AE(ae_title=settings.SOURCE_AET)
# Add storage presentation contexts (all standard transfer syntaxes for each SOP class)
for context in StoragePresentationContexts:
self.ae.add_requested_context(context.abstract_syntax)
# Add the Query/Retrieve SOP classes
self.ae.add_requested_context(StudyRootQueryRetrieveInformationModelGet)
self.ae.add_requested_context(PatientRootQueryRetrieveInformationModelGet)
# Add storage presentation contexts (needed for receiving the images)
# for context in StoragePresentationContexts:
# self.ae.add_requested_context(context.abstract_syntax)
storage_uids = [
'1.2.840.10008.5.1.4.1.1.1', # CR Storage
'1.2.840.10008.5.1.4.1.1.1.1', # Digital X-Ray Image Storage
'1.2.840.10008.5.1.4.1.1.2', # CT Image Storage
'1.2.840.10008.5.1.4.1.1.4', # MR Image Storage
'1.2.840.10008.5.1.4.1.1.7', # Secondary Capture Image Storage
'1.2.840.10008.5.1.4.1.1.6.1', # Ultrasound Image Storage
'1.2.840.10008.5.1.4.1.1.128', # PET Image Storage
'1.2.840.10008.5.1.4.1.1.20', # Nuclear Medicine Image Storage
'1.2.840.10008.5.1.4.1.1.9.1.1', # 12-lead ECG Waveform Storage
'1.2.840.10008.5.1.4.1.1.9.1.2', # General ECG Waveform Storage
]
self.ext_neg = []
for uid in storage_uids:
self.ae.add_requested_context(uid)
role = build_role(uid, scp_role=True)
self.ext_neg.append(role)
logger.info(f"DicomSender initialized with destination PACS: {self.pacs_config['aet']}@{self.pacs_config['host']}:{self.pacs_config['port']}")