#!/usr/bin/env python3 """ DICOM Sender (C-STORE client) ============================== Sends DICOM files to a remote DICOM router/PACS over the network using the C-STORE service, selecting which files to send by matching Accession Number against an index CSV (the kind produced by dicom_indexer.py's 'scan' command, which has FilePath and AccessionNumber columns). This tool always works from TWO separate source files: --source-index the file/metadata index (built earlier by dicom_indexer.py) --source-accessions your list of accession numbers to send (a different file) Dependencies: pip install pydicom pynetdicom Usage: # Test connectivity to the destination router first (recommended) python dicom_sender.py echo --host 192.168.1.50 --port 104 --ae-title REMOTE_PACS # Send files whose AccessionNumber matches a list in a separate CSV python dicom_sender.py send \\ --source-index dicom_index.csv \\ --source-accessions accessions_to_send.csv \\ --host 192.168.1.50 --port 104 --ae-title REMOTE_PACS # Send files for a single accession number directly (no accession-list file needed) python dicom_sender.py send \\ --source-index dicom_index.csv \\ --accession ACC12345 \\ --host 192.168.1.50 --port 104 --ae-title REMOTE_PACS # Use a custom calling (your own) AE Title instead of the default python dicom_sender.py send ... --calling-ae-title MY_AE # Dry run: show what WOULD be sent without opening a network connection python dicom_sender.py send ... --dry-run """ import os import sys import csv import argparse import socket import time DEFAULT_CALLING_AE_TITLE = "DCM-RSABHK" # --------------------------------------------------------------------------- # Dependency check with a friendly message # --------------------------------------------------------------------------- MISSING = [] try: import pydicom except ImportError: MISSING.append("pydicom") try: from pynetdicom import AE, evt from pynetdicom.sop_class import Verification except ImportError: MISSING.append("pynetdicom") if MISSING: print("Missing required packages: " + ", ".join(MISSING)) print("Install them with:") print(f" pip install {' '.join(MISSING)}") sys.exit(1) # --------------------------------------------------------------------------- # DICOM C-STORE status code meanings (the common ones; peers may return # vendor-specific codes outside this list, which we just show as hex). # --------------------------------------------------------------------------- STATUS_MEANINGS = { 0x0000: "Success", 0x0105: "No such attribute", 0x0106: "Invalid attribute value", 0x0110: "Processing failure", 0x0117: "Invalid SOP instance", 0x0122: "SOP class not supported", 0x0124: "Not authorised", 0x0210: "Duplicate invocation", 0x0211: "Unrecognised operation", 0x0212: "Mistyped argument", 0xA700: "Out of resources", 0xA900: "Data set does not match SOP class", 0xB000: "Coercion of data elements (warning)", 0xB006: "Element discarded (warning)", 0xB007: "Data set does not match SOP class (warning)", } def describe_status(status_dataset): """Return a (code, label) tuple describing a C-STORE/C-ECHO response status.""" if status_dataset is None or "Status" not in status_dataset: return None, "No response (association/timeout/abort failure)" code = status_dataset.Status label = STATUS_MEANINGS.get(code) if label is None: # Out-of-resources and "cannot understand" are ranges, not single codes. if 0xA700 <= code <= 0xA7FF: label = "Out of resources" elif 0xC000 <= code <= 0xCFFF: label = "Cannot understand" else: label = "Unknown/vendor-specific status" return code, label def is_success(code): return code == 0x0000 # --------------------------------------------------------------------------- # Index / accession-list loading (streamed — see dicom_indexer.py's memory # notes; we apply the same approach here so large indexes don't balloon RAM) # --------------------------------------------------------------------------- def load_accession_list(path): """ Read a CSV of accession numbers to send. Accepts either: - a column literally named 'AccessionNumber' (case-insensitive), or - a single-column CSV with no recognizable header (first column used). Returns a set of accession number strings (stripped, deduplicated). """ accessions = set() with open(path, "r", newline="", encoding="utf-8") as f: reader = csv.reader(f) rows = list(reader) if not rows: return accessions header = [h.strip().lower() for h in rows[0]] col_idx = None for i, h in enumerate(header): if h in ("accessionnumber", "accession_number", "accession"): col_idx = i break if col_idx is not None: data_rows = rows[1:] else: # No recognizable header — treat every row's first column as data, # including row 0 (it wasn't actually a header). col_idx = 0 data_rows = rows for row in data_rows: if len(row) > col_idx: value = row[col_idx].strip() if value: accessions.add(value) return accessions def translate_path(path, prefix_map): """ Rewrite `path` using the first matching (from_prefix, to_prefix) pair in prefix_map, e.g. translating a host path stored in an index/database to the path it's actually mounted at inside a container. Handles the common cross-platform case where indexing ran on Windows (paths like "Z:\\some folder\\study1\\img.dcm" or "Z:/some folder/...") but the sender runs on Linux: separators are normalized to "/" and the prefix match is case-insensitive (Windows paths/drive letters are case-insensitive; this has no meaningful downside on Linux paths). prefix_map: list of (from_prefix, to_prefix) tuples, checked in order. from_prefix may use "/" or "\\" — both are accepted. Returns path unchanged if it's falsy, prefix_map is empty, or no prefix matches. Output always uses "/" separators (Linux-style), matching to_prefix's own style. """ if not path or not prefix_map: return path normalized_path = path.replace("\\", "/") for from_prefix, to_prefix in prefix_map: norm_from = from_prefix.replace("\\", "/").rstrip("/") if not norm_from: continue lower_path = normalized_path.lower() lower_from = norm_from.lower() if lower_path == lower_from: return to_prefix.rstrip("/") if lower_path.startswith(lower_from + "/"): remainder = normalized_path[len(norm_from):] # starts with "/" return to_prefix.rstrip("/") + remainder return path def find_matching_files_mongo( collection, accession_numbers, filepath_field="FilePath", accession_field="AccessionNumber", path_prefix_map=None, ): """ Query a pymongo Collection for documents whose accession_field is in accession_numbers. Yields dicts shaped like the CSV path's rows: at least {"FilePath": ..., "AccessionNumber": ...} (normalized to those exact keys regardless of what the underlying document's field names are), plus whatever other fields the document has. accession_numbers: iterable of accession number strings to match against. path_prefix_map: optional list of (from_prefix, to_prefix) tuples used to rewrite FilePath, e.g. when the path stored in Mongo is a host path but files are mounted at a different path inside a container. See translate_path(). """ accession_numbers = [a for a in accession_numbers if a] if not accession_numbers: return # $in has practical size limits on very large lists; chunk defensively. CHUNK = 1000 for i in range(0, len(accession_numbers), CHUNK): chunk = accession_numbers[i:i + CHUNK] for doc in collection.find({accession_field: {"$in": chunk}}): row = dict(doc) row["FilePath"] = translate_path(doc.get(filepath_field), path_prefix_map) row["AccessionNumber"] = doc.get(accession_field) yield row def find_matching_files(index_path, accession_numbers): """ Stream the index CSV and yield (filepath, accession_number, sop_class_uid, patient_name, series_description) for every row whose AccessionNumber is in accession_numbers. accession_numbers: a set of accession number strings to match against. """ with open(index_path, "r", newline="", encoding="utf-8") as f: reader = csv.DictReader(f) if "FilePath" not in reader.fieldnames or "AccessionNumber" not in reader.fieldnames: print( "Error: index file does not have the expected 'FilePath' and " "'AccessionNumber' columns. Make sure this is an index produced " "by dicom_indexer.py's 'scan' command.", file=sys.stderr, ) sys.exit(1) for row in reader: acc = (row.get("AccessionNumber") or "").strip() if acc in accession_numbers: yield row # --------------------------------------------------------------------------- # Subcommand: echo (connectivity test) # --------------------------------------------------------------------------- def cmd_echo(args): ae = AE(ae_title=args.calling_ae_title) ae.add_requested_context(Verification) ae.acse_timeout = args.timeout ae.dimse_timeout = args.timeout ae.network_timeout = args.timeout print( f"Sending C-ECHO to {args.ae_title}@{args.host}:{args.port} " f"(calling AE: {args.calling_ae_title})..." ) try: assoc = ae.associate(args.host, args.port, ae_title=args.ae_title) except Exception as exc: print(f"Association error: {exc}", file=sys.stderr) sys.exit(1) if not assoc.is_established: print( "FAILED: could not establish association. Check host/port/AE title " "and that the destination router is reachable and accepting " "associations from this AE.", file=sys.stderr, ) sys.exit(1) status = assoc.send_c_echo() code, label = describe_status(status) assoc.release() if code is not None and is_success(code): print(f"SUCCESS: C-ECHO accepted (status 0x{code:04X} - {label}).") else: code_str = f"0x{code:04X}" if code is not None else "N/A" print(f"FAILED: C-ECHO returned status {code_str} - {label}.", file=sys.stderr) sys.exit(1) # --------------------------------------------------------------------------- # Core send logic (shared by the CLI 'send' subcommand and any programmatic # caller, e.g. an API server). Returns a result dict rather than printing + # sys.exit()-ing, so it is safe to call from long-running processes. # --------------------------------------------------------------------------- def send_accessions( accession_numbers, host, port, ae_title, source_index=None, lookup_fn=None, calling_ae_title=DEFAULT_CALLING_AE_TITLE, timeout=30, delay=0.0, dry_run=False, verbose=True, progress_callback=None, ): """ Find files matching accession_numbers and send them via C-STORE to (host, port, ae_title). This is the reusable core used by both the CLI 'send' subcommand and any programmatic caller (e.g. a web API). Exactly one of these must be given to say *where* to look up files: source_index -> path to a CSV index (dicom_indexer.py 'scan' output) lookup_fn -> callable(accession_numbers) -> iterable of row dicts, each with at least "FilePath" and "AccessionNumber" keys. Use this to back the lookup with MongoDB or anything else instead of a CSV file — see find_matching_files_mongo() for a ready-made one. accession_numbers: iterable of accession number strings. progress_callback: optional callable(current, total, filepath, accession) invoked right before each file is sent (real sends only). verbose: if True, mirrors the original CLI's stdout/stderr messages. Returns a dict: { "found_accessions": set, "missing_accessions": set, "missing_files": [filepath, ...], "unreadable": [filepath, ...], "dry_run": bool, "sendable_count": int, "results": [(filepath, accession, code_or_None, label), ...], "error": str or None, # set on hard failures (association, etc.) } """ def log(msg, err=False): if verbose: print(msg, file=sys.stderr if err else sys.stdout) if (source_index is None) == (lookup_fn is None): raise ValueError("send_accessions() requires exactly one of source_index or lookup_fn.") accession_numbers = set(accession_numbers) result = { "found_accessions": set(), "missing_accessions": set(), "missing_files": [], "unreadable": [], "dry_run": dry_run, "sendable_count": 0, "results": [], "error": None, } index_desc = f"'{source_index}'" if source_index else "MongoDB" log(f"Looking up {len(accession_numbers)} accession number(s) in index {index_desc}...") # ---- Step 2: find matching files in the index ---- if lookup_fn is not None: matched_rows = list(lookup_fn(accession_numbers)) else: matched_rows = list(find_matching_files(source_index, accession_numbers)) if not matched_rows: log("No matching files found in the index for the given accession number(s).") result["missing_accessions"] = set(accession_numbers) return result found_accessions = {r["AccessionNumber"] for r in matched_rows} missing_accessions = accession_numbers - found_accessions result["found_accessions"] = found_accessions result["missing_accessions"] = missing_accessions log(f"Found {len(matched_rows)} file(s) across {len(found_accessions)} accession number(s).") if missing_accessions: log( f"Warning: {len(missing_accessions)} accession number(s) had no matching " f"file in the index: {', '.join(sorted(missing_accessions)[:10])}" + (" ..." if len(missing_accessions) > 10 else ""), err=True, ) # ---- Step 3: verify files exist on disk ---- sendable = [] missing_files = [] for row in matched_rows: filepath = row["FilePath"] if os.path.isfile(filepath): sendable.append(row) else: missing_files.append(filepath) result["missing_files"] = missing_files if missing_files: log( f"Warning: {len(missing_files)} file(s) listed in the index no longer " f"exist on disk and will be skipped:", err=True, ) for fp in missing_files[:10]: log(f" - {fp}", err=True) if len(missing_files) > 10: log(f" ... and {len(missing_files) - 10} more", err=True) result["sendable_count"] = len(sendable) if not sendable: log("No sendable files remain after checking disk. Aborting.") return result if dry_run: log("\n--- DRY RUN: the following files would be sent ---") for row in sendable: log(f" [{row.get('AccessionNumber','')}] {row['FilePath']}") log(f"\nTotal: {len(sendable)} file(s). No network connection was made.") result["results"] = [ (row["FilePath"], row.get("AccessionNumber", ""), None, "Dry run - not sent") for row in sendable ] return result # ---- Step 4: read each file's SOP Class UID / Transfer Syntax UID to # build presentation contexts (max 128 per association) ---- log("Reading SOP Class / Transfer Syntax info from files to build presentation contexts...") file_info = [] # (filepath, row, sop_class_uid, transfer_syntax_uid) unreadable = [] for row in sendable: filepath = row["FilePath"] try: ds_meta = pydicom.dcmread(filepath, stop_before_pixels=True, force=True) sop_class = getattr(ds_meta, "SOPClassUID", None) ts = ds_meta.file_meta.TransferSyntaxUID if hasattr(ds_meta, "file_meta") else None if sop_class is None: unreadable.append(filepath) continue file_info.append((filepath, row, str(sop_class), str(ts) if ts else None)) except Exception: unreadable.append(filepath) result["unreadable"] = unreadable if unreadable: log( f"Warning: {len(unreadable)} file(s) could not be read as DICOM and " f"will be skipped:", err=True, ) for fp in unreadable[:10]: log(f" - {fp}", err=True) if not file_info: log("No readable DICOM files remain. Aborting.") return result # Build the minimal set of presentation contexts needed: one context per # DISTINCT (SOP Class UID, Transfer Syntax UID) pair actually used by the # files being sent. We deliberately do NOT bundle multiple transfer # syntaxes into a single context, because the peer only accepts ONE # transfer syntax per context — if a SOP Class has some files in (say) # uncompressed Explicit VR LE and others in JPEG Lossless, bundling them # into one context risks the peer accepting only one of the two, and # files using the other transfer syntax would still fail to send. A # separate context per pair guarantees every transfer syntax actually # present gets its own negotiation slot. # # This also fixes the root cause of "No presentation context ... has # been accepted by the peer with ''" errors: previously # contexts were requested using pynetdicom's DEFAULT transfer syntax # list (uncompressed only), so compressed files (JPEG Lossless, JPEG # 2000, RLE, etc.) never had a matching context at all. seen_pairs = set() context_pairs = [] # list of (sop_class, transfer_syntax_or_None) for _, _, sop_class, ts in file_info: pair = (sop_class, ts) if pair not in seen_pairs: seen_pairs.add(pair) context_pairs.append(pair) if len(context_pairs) > 128: log( f"Error: files span {len(context_pairs)} distinct SOP " f"Class/Transfer Syntax combinations, which exceeds the 128 " f"presentation contexts allowed per association. Split the " f"send into smaller batches.", err=True, ) result["error"] = "too_many_presentation_contexts" return result # ---- Step 5: build AE, request contexts, associate ---- ae = AE(ae_title=calling_ae_title) for sop_class, ts in context_pairs: if ts: ae.add_requested_context(sop_class, ts) else: # Transfer syntax couldn't be determined for this file # (unexpected/corrupt file_meta) — fall back to pynetdicom's # default uncompressed transfer syntax list rather than failing. ae.add_requested_context(sop_class) ae.acse_timeout = timeout ae.dimse_timeout = timeout ae.network_timeout = timeout distinct_sop_classes = {p[0] for p in context_pairs} log( f"\nConnecting to {ae_title}@{host}:{port} " f"(calling AE: {calling_ae_title}) requesting " f"{len(context_pairs)} presentation context(s) covering " f"{len(distinct_sop_classes)} SOP Class(es)..." ) try: assoc = ae.associate(host, port, ae_title=ae_title) except Exception as exc: log(f"Association error: {exc}", err=True) result["error"] = f"association_error: {exc}" return result if not assoc.is_established: log( "FAILED: could not establish association. Check host/port/AE title, " "that the destination accepts the SOP classes being sent, and that " "it allows associations from this calling AE title.", err=True, ) result["error"] = "association_not_established" return result # ---- Step 6: send each file via C-STORE ---- send_results = [] # (filepath, accession, code, label) total = len(file_info) try: for i, (filepath, row, sop_class, ts) in enumerate(file_info, start=1): accession = row.get("AccessionNumber", "") if progress_callback: progress_callback(i, total, filepath, accession) if verbose: sys.stdout.write(f"\rSending {i}/{total} (accession {accession})...") sys.stdout.flush() try: status = assoc.send_c_store(filepath) except Exception as exc: send_results.append((filepath, accession, None, f"Exception: {exc}")) continue code, label = describe_status(status) send_results.append((filepath, accession, code, label)) if delay: time.sleep(delay) finally: assoc.release() if verbose: print() # newline after progress line result["results"] = send_results # ---- Step 7: summarize ---- succeeded = [r for r in send_results if r[2] is not None and is_success(r[2])] failed = [r for r in send_results if r not in succeeded] log(f"\n--- Send Summary ---") log(f"Total attempted: {len(send_results)}") log(f"Succeeded: {len(succeeded)}") log(f"Failed: {len(failed)}") if failed: log("\nFailed files:") for filepath, accession, code, label in failed: code_str = f"0x{code:04X}" if code is not None else "N/A" log(f" [{accession}] {filepath}") log(f" -> status {code_str}: {label}") return result def write_report_csv(path, results): """Write the (filepath, accession, code, label) results list to a CSV report.""" with open(path, "w", newline="", encoding="utf-8") as f: writer = csv.writer(f) writer.writerow(["FilePath", "AccessionNumber", "StatusCode", "StatusLabel"]) for filepath, accession, code, label in results: code_str = f"0x{code:04X}" if code is not None else "" writer.writerow([filepath, accession, code_str, label]) # --------------------------------------------------------------------------- # Subcommand: send (thin CLI wrapper around send_accessions()) # --------------------------------------------------------------------------- def cmd_send(args): # ---- Step 1: resolve the set of accession numbers to send ---- # Two independent source files are used here: # args.source_index -> which files exist + their metadata # args.source_accessions -> which accession numbers to actually send accession_numbers = set() if args.source_accessions: if not os.path.isfile(args.source_accessions): print(f"Error: accession source file not found: {args.source_accessions}", file=sys.stderr) sys.exit(1) accession_numbers |= load_accession_list(args.source_accessions) if args.accession: accession_numbers |= {a.strip() for a in args.accession if a.strip()} if not accession_numbers: print( "Error: no accession numbers given. Use --source-accessions and/or " "one or more --accession .", file=sys.stderr, ) sys.exit(1) if not os.path.isfile(args.source_index): print(f"Error: index source file not found: {args.source_index}", file=sys.stderr) sys.exit(1) result = send_accessions( source_index=args.source_index, accession_numbers=accession_numbers, host=args.host, port=args.port, ae_title=args.ae_title, calling_ae_title=args.calling_ae_title, timeout=args.timeout, delay=args.delay, dry_run=args.dry_run, verbose=True, ) if result.get("error"): sys.exit(1) if result["dry_run"] or result["sendable_count"] == 0: return send_results = result["results"] failed = [r for r in send_results if r[2] is None or not is_success(r[2])] if args.report: write_report_csv(args.report, send_results) print(f"\nDetailed report saved to: {args.report}") if failed: sys.exit(1) # --------------------------------------------------------------------------- # Argument parsing # --------------------------------------------------------------------------- def build_parser(): parser = argparse.ArgumentParser( prog="dicom_sender.py", description=( "Send DICOM files to a remote DICOM router/PACS via C-STORE, " "selecting files by Accession Number from a dicom_indexer.py index." ), ) subparsers = parser.add_subparsers(dest="command", required=True) # Shared connection arguments def add_connection_args(p): p.add_argument("--host", required=True, help="Destination IP address or hostname.") p.add_argument("--port", required=True, type=int, help="Destination port number.") p.add_argument( "--ae-title", required=True, dest="ae_title", help="Destination (called) AE Title.", ) p.add_argument( "--calling-ae-title", dest="calling_ae_title", default=DEFAULT_CALLING_AE_TITLE, help=f"Your own (calling) AE Title (default: {DEFAULT_CALLING_AE_TITLE}).", ) p.add_argument( "--timeout", type=int, default=30, help="Network/association/DIMSE timeout in seconds (default: 30).", ) # echo p_echo = subparsers.add_parser("echo", help="Test connectivity to a destination via C-ECHO.") add_connection_args(p_echo) p_echo.set_defaults(func=cmd_echo) # send p_send = subparsers.add_parser("send", help="Send DICOM files matching accession number(s) via C-STORE.") add_connection_args(p_send) p_send.add_argument( "--source-index", "--index", dest="source_index", required=True, help=( "SOURCE FILE 1: path to the index CSV produced by dicom_indexer.py's " "'scan' command (needs FilePath + AccessionNumber columns). This is " "the catalog of files that exist and where they are on disk." ), ) p_send.add_argument( "--source-accessions", "--accessions", dest="source_accessions", help=( "SOURCE FILE 2: path to a separate CSV file containing the accession " "numbers you want to send (column named AccessionNumber, or a plain " "single-column list with no header). This is the work order — which " "of the files cataloged in --source-index should actually be sent." ), ) p_send.add_argument( "--accession", action="append", help="A single accession number to send. Can be given multiple times.", ) p_send.add_argument( "--dry-run", action="store_true", help="Show which files would be sent without making any network connection.", ) p_send.add_argument( "--delay", type=float, default=0.0, help="Delay in seconds between sending each file (default: 0, no delay).", ) p_send.add_argument( "-o", "--report", help="Save a detailed per-file CSV report (FilePath, AccessionNumber, StatusCode, StatusLabel) to this path.", ) p_send.set_defaults(func=cmd_send) return parser def main(): parser = build_parser() args = parser.parse_args() args.func(args) if __name__ == "__main__": main()