[Python] 간단한 DICOM PACS Storage SCP 만들기

 Python을 이용하여 간단한 DICOM PACS Storage SCP를 구현해보겠습니다.

 

 필요한 Python 패키지 입니다. sqlite3, pydicom, pynetdicom 

import sqlite3
from pathlib import Path

import pydicom
from pynetdicom import AE, evt, AllStoragePresentationContexts
from pynetdicom.sop_class import Verification

 데이터셋을 다루기위해 pydicom, DICOM 통신을 위해 pynetdicom 패키지가 필요합니다. 특히 AllStoragePresentationContexts를 통해 C-STORE를 구현할 수 있습니다. 또한 C-ECHO용 SOP Class를 위해 Verification도 import 합니다.

 

 DB 초기화 입니다.

def init_db():
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()

    cur.execute("""
    CREATE TABLE IF NOT EXISTS instances (
        sop_instance_uid TEXT PRIMARY KEY,
        sop_class_uid TEXT,
        study_instance_uid TEXT,
        series_instance_uid TEXT,
        patient_id TEXT,
        patient_name TEXT,
        study_date TEXT,
        modality TEXT,
        body_part_examined TEXT,
        file_path TEXT
    )
    """)

    conn.commit()
    conn.close()

이미지와 매핑되는 SOP Instance UID를 Primary Key로 사용하여 최소한의 DICOM Tag 데이터를 테이블화했습니다. 

 

 DICOM 파일 인덱싱 

def index_file(path):
    try:
        ds = pydicom.dcmread(path, stop_before_pixels=True, force=True)

        sop_instance_uid = str(ds.SOPInstanceUID)
        sop_class_uid = str(ds.SOPClassUID)
        study_uid = str(ds.StudyInstanceUID)
        series_uid = str(ds.SeriesInstanceUID)

        patient_id = str(getattr(ds, "PatientID", ""))
        patient_name = str(getattr(ds, "PatientName", ""))
        study_date = str(getattr(ds, "StudyDate", ""))
        modality = str(getattr(ds, "Modality", ""))
        body_part = str(getattr(ds, "BodyPartExamined", ""))

        conn = sqlite3.connect(DB_PATH)
        cur = conn.cursor()

        cur.execute("""
        INSERT OR REPLACE INTO instances (
            sop_instance_uid,
            sop_class_uid,
            study_instance_uid,
            series_instance_uid,
            patient_id,
            patient_name,
            study_date,
            modality,
            body_part_examined,
            file_path
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            sop_instance_uid,
            sop_class_uid,
            study_uid,
            series_uid,
            patient_id,
            patient_name,
            study_date,
            modality,
            body_part,
            str(path)
        ))

        conn.commit()
        conn.close()

        print(f"Indexed: {path}")

    except Exception as e:
        print(f"Index failed: {path} / {e}")

 클라이언트에서 넘어온 파일을 읽을 때 stop_before_pixels 옵션을 True로 사용하여 픽셀 데이터를 제외한 메타데이터만 읽게 설정했습니다. 이를 통해 인덱싱 속도를 높이고 불필요한 메모리 사용량을 줄입니다. 읽어온 메타데이터는 SQLite DB에 저장됩니다.

 

 기존 저장 폴더 재색인

def index_existing_folder():
    for path in STORAGE_DIR.rglob("*"):
        if path.is_file():
            index_file(path)

 

 

 C-STORE 처리

def handle_store(event):
    ds = event.dataset
    ds.file_meta = event.file_meta

    sop_instance_uid = str(ds.SOPInstanceUID)
    study_uid = str(ds.StudyInstanceUID)
    series_uid = str(ds.SeriesInstanceUID)

    save_dir = STORAGE_DIR / study_uid / series_uid
    save_dir.mkdir(parents=True, exist_ok=True)

    save_path = save_dir / f"{sop_instance_uid}.dcm"

    ds.save_as(save_path, write_like_original=False)
    index_file(save_path)

    print(f"C-STORE received: {save_path}")

    return 0x0000

 이벤트 핸들러로 클라이언트에서 전달된 DICOM 데이터를 처리합니다. 이후 Study UID와 Series UID를 기준으로 저장 디렉터리를 구성해 DICOM 파일을 저장합니다. 저장 후에는 index_file()을 호출해 DB에 메타데이터를 기록합니다. 

 

 메인 호출부

def main():
    init_db()
    index_existing_folder()

    ae = AE(ae_title=AE_TITLE)

    ae.add_supported_context(Verification)

    for context in AllStoragePresentationContexts:
        ae.add_supported_context(context.abstract_syntax)

    handlers = [
        (evt.EVT_C_STORE, handle_store),
    ]

    print(f"PACS Storage SCP started")
    print(f"AE Title : {AE_TITLE}")
    print(f"Port     : {PORT}")
    print(f"Storage  : {STORAGE_DIR.resolve()}")

    ae.start_server(("0.0.0.0", PORT), block=True, evt_handlers=handlers)

 DB 초기화 및 저장 폴더 인덱싱 이후 AE를 생성하여 C-ECHO를 위한 Verification 등록을 마칩니다. 이어 C-STORE를 위한 context 등록을 한 뒤 클라이언트에서 넘어온 데이터 이벤트 핸들러를 앞서 정의한 handle_store로 설정합니다.
서버는 모든 네트워크 인터페이스에 대해 접속을 허용하며 동기 방식으로 실행됩니다. 

아래는 코드 전체입니다.

import sqlite3
from pathlib import Path

import pydicom
from pynetdicom import AE, evt, AllStoragePresentationContexts
from pynetdicom.sop_class import Verification


AE_TITLE = "XRAY_PACS"
PORT = 11112
STORAGE_DIR = Path("storage")
DB_PATH = "pacs.db"

STORAGE_DIR.mkdir(exist_ok=True)


def init_db():
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()

    cur.execute("""
    CREATE TABLE IF NOT EXISTS instances (
        sop_instance_uid TEXT PRIMARY KEY,
        sop_class_uid TEXT,
        study_instance_uid TEXT,
        series_instance_uid TEXT,
        patient_id TEXT,
        patient_name TEXT,
        study_date TEXT,
        modality TEXT,
        body_part_examined TEXT,
        file_path TEXT
    )
    """)

    conn.commit()
    conn.close()


def index_file(path):
    try:
        ds = pydicom.dcmread(path, stop_before_pixels=True, force=True)

        sop_instance_uid = str(ds.SOPInstanceUID)
        sop_class_uid = str(ds.SOPClassUID)
        study_uid = str(ds.StudyInstanceUID)
        series_uid = str(ds.SeriesInstanceUID)

        patient_id = str(getattr(ds, "PatientID", ""))
        patient_name = str(getattr(ds, "PatientName", ""))
        study_date = str(getattr(ds, "StudyDate", ""))
        modality = str(getattr(ds, "Modality", ""))
        body_part = str(getattr(ds, "BodyPartExamined", ""))

        conn = sqlite3.connect(DB_PATH)
        cur = conn.cursor()

        cur.execute("""
        INSERT OR REPLACE INTO instances (
            sop_instance_uid,
            sop_class_uid,
            study_instance_uid,
            series_instance_uid,
            patient_id,
            patient_name,
            study_date,
            modality,
            body_part_examined,
            file_path
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            sop_instance_uid,
            sop_class_uid,
            study_uid,
            series_uid,
            patient_id,
            patient_name,
            study_date,
            modality,
            body_part,
            str(path)
        ))

        conn.commit()
        conn.close()

        print(f"Indexed: {path}")

    except Exception as e:
        print(f"Index failed: {path} / {e}")


def index_existing_folder():
    for path in STORAGE_DIR.rglob("*"):
        if path.is_file():
            index_file(path)


def handle_store(event):
    ds = event.dataset
    ds.file_meta = event.file_meta

    sop_instance_uid = str(ds.SOPInstanceUID)
    study_uid = str(ds.StudyInstanceUID)
    series_uid = str(ds.SeriesInstanceUID)

    save_dir = STORAGE_DIR / study_uid / series_uid
    save_dir.mkdir(parents=True, exist_ok=True)

    save_path = save_dir / f"{sop_instance_uid}.dcm"

    ds.save_as(save_path, write_like_original=False)
    index_file(save_path)

    print(f"C-STORE received: {save_path}")

    return 0x0000


def main():
    init_db()
    index_existing_folder()

    ae = AE(ae_title=AE_TITLE)

    ae.add_supported_context(Verification)

    for context in AllStoragePresentationContexts:
        ae.add_supported_context(context.abstract_syntax)

    handlers = [
        (evt.EVT_C_STORE, handle_store),
    ]

    print(f"PACS Storage SCP started")
    print(f"AE Title : {AE_TITLE}")
    print(f"Port     : {PORT}")
    print(f"Storage  : {STORAGE_DIR.resolve()}")

    ae.start_server(("0.0.0.0", PORT), block=True, evt_handlers=handlers)


if __name__ == "__main__":
    main()