im_anca/dicom-index-send
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first

Browse Source
This commit is contained in:
Anca 2026-06-24 17:24:11 +07:00
commit 64d81804bf
5 changed files with 81282 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

599
app.py Normal file
View File

@ -0,0 +1,599 @@
#!/usr/bin/env python3
"""
DICOM Folder Indexer (CLI)
============================
Recursively scans a folder for DICOM files, extracts key metadata
(patient, study, series, instance level), builds an index, lets you
search/filter that index, and export it to CSV or Excel.
Dependencies:
pip install pydicom
pip install openpyxl # only needed if you use --xlsx or an .xlsx index
Usage:
# Build an index from a folder and save it
python dicom_indexer.py scan /path/to/dicom_folder -o index.csv
# Build an index and also save as Excel
python dicom_indexer.py scan /path/to/dicom_folder -o index.csv --xlsx index.xlsx
# Search an existing index
python dicom_indexer.py search index.csv --patient "Doe" --modality CT
# Search across all fields for a keyword
python dicom_indexer.py search index.csv -q "chest"
# List unique patients/studies/series in an index
python dicom_indexer.py summary index.csv
"""
import os
import sys
import csv
import argparse
import traceback
from datetime import datetime
# ---------------------------------------------------------------------------
# Dependency check with a friendly message.
# Note: pandas is no longer required — all CSV/Excel handling below streams
# rows directly via the csv module and openpyxl, which keeps memory usage
# low regardless of folder/index size. openpyxl is only needed if you use
# --xlsx (scan) or an .xlsx index file (search/summary); it's checked
# lazily where it's actually used, not at startup.
# ---------------------------------------------------------------------------
MISSING = []
try:
import pydicom
except ImportError:
MISSING.append("pydicom")
if MISSING:
print("Missing required packages: " + ", ".join(MISSING))
print("Install them with:")
print(f" pip install {' '.join(MISSING)}")
sys.exit(1)
# ---------------------------------------------------------------------------
# Metadata fields to extract from each DICOM file.
# Format: (DataFrame column name, DICOM keyword)
# Add/remove entries here to customize what gets indexed.
# ---------------------------------------------------------------------------
DICOM_FIELDS = [
("PatientName", "PatientName"),
("PatientID", "PatientID"),
("PatientBirthDate", "PatientBirthDate"),
("PatientSex", "PatientSex"),
("PatientAge", "PatientAge"),
("StudyDate", "StudyDate"),
("StudyTime", "StudyTime"),
("StudyDescription", "StudyDescription"),
("StudyInstanceUID", "StudyInstanceUID"),
("AccessionNumber", "AccessionNumber"),
("Modality", "Modality"),
("SeriesDescription", "SeriesDescription"),
("SeriesNumber", "SeriesNumber"),
("SeriesInstanceUID", "SeriesInstanceUID"),
("InstanceNumber", "InstanceNumber"),
("SOPInstanceUID", "SOPInstanceUID"),
("Manufacturer", "Manufacturer"),
("ManufacturerModelName", "ManufacturerModelName"),
("InstitutionName", "InstitutionName"),
("BodyPartExamined", "BodyPartExamined"),
("Rows", "Rows"),
("Columns", "Columns"),
("SliceThickness", "SliceThickness"),
]
# Maps CLI search flags -> DataFrame column name
SEARCH_FLAG_TO_COLUMN = {
"patient": "PatientName",
"patient_id": "PatientID",
"study_date": "StudyDate",
"study": "StudyDescription",
"modality": "Modality",
"series": "SeriesDescription",
"body_part": "BodyPartExamined",
"accession": "AccessionNumber",
}
# ---------------------------------------------------------------------------
# Extension-based fast skip.
#
# DICOM files often have NO extension at all (common with PACS/modality
# exports), or use .dcm/.dicom/.ima/.img. Because a missing extension is
# normal and valid, we do NOT filter to an allow-list of DICOM extensions
# only — that would silently skip real DICOM files.
#
# Instead we skip files whose extension is UNAMBIGUOUSLY something else
# (images, documents, archives, executables, etc.) without even opening
# them. Everything else (no extension, .dcm, or anything unrecognized)
# still goes through extract_metadata() as before, so correctness is
# preserved while common junk/sidecar files are skipped cheaply.
# ---------------------------------------------------------------------------
SKIP_EXTENSIONS = {
# common non-DICOM image formats
".jpg", ".jpeg", ".png", ".gif", ".bmp", ".tif", ".tiff", ".webp", ".ico", ".svg",
# documents / text / data
".txt", ".md", ".csv", ".json", ".xml", ".html", ".htm", ".pdf", ".doc", ".docx",
".xls", ".xlsx", ".ppt", ".pptx", ".log", ".ini", ".yaml", ".yml",
# archives
".zip", ".rar", ".7z", ".tar", ".gz", ".bz2", ".xz",
# executables / scripts / libraries
".exe", ".dll", ".so", ".bat", ".sh", ".py", ".js", ".jar",
# media
".mp3", ".mp4", ".avi", ".mov", ".wav",
# misc OS/editor cruft
".ds_store", ".db", ".tmp", ".bak", ".lnk", ".url",
}
# Used only when --strict-extension is passed: files must have one of these
# extensions (or no extension) to be opened at all.
DICOM_EXTENSIONS = {".dcm", ".dicom", ".dic", ".ima", ".img", ".dcm30"}
def format_dicom_date(value):
"""Convert DICOM DA format (YYYYMMDD) to YYYY-MM-DD for readability."""
if not value:
return ""
value = str(value)
if len(value) == 8 and value.isdigit():
try:
return datetime.strptime(value, "%Y%m%d").strftime("%Y-%m-%d")
except ValueError:
return value
return value
def extract_metadata(filepath):
"""
Read a single file with pydicom and return a dict of extracted fields,
or None if the file is not a valid DICOM file.
"""
try:
ds = pydicom.dcmread(filepath, stop_before_pixels=True, force=False)
except Exception:
# Try once more with force=True in case the file is missing the
# standard preamble but is still valid DICOM (common with exported
# files that strip the 128-byte header).
try:
ds = pydicom.dcmread(filepath, stop_before_pixels=True, force=True)
if "SOPClassUID" not in ds and "Modality" not in ds:
return None
except Exception:
return None
row = {"FilePath": filepath, "FileName": os.path.basename(filepath)}
for col_name, keyword in DICOM_FIELDS:
try:
value = getattr(ds, keyword, "")
except Exception:
value = ""
if value is None:
value = ""
value = str(value).strip()
if keyword in ("StudyDate", "PatientBirthDate"):
value = format_dicom_date(value)
row[col_name] = value
try:
size_bytes = os.path.getsize(filepath)
except OSError:
size_bytes = 0
row["FileSizeKB"] = round(size_bytes / 1024, 1)
return row
def scan_folder(root_folder, show_progress=True, strict_extension=False):
"""
Walk root_folder recursively and yield a metadata dict for every valid
DICOM file found, one at a time.
This is a generator (not a list-returning function) so that callers can
write each row to disk as it's produced, instead of holding every row
in memory at once. Memory use stays roughly constant regardless of how
many files are in the folder.
Extension filtering (speed optimization):
- Default: files with an extension that's unambiguously NOT DICOM
(.jpg, .txt, .zip, etc. — see SKIP_EXTENSIONS) are skipped without
being opened. Files with no extension, a .dcm-style extension, or
anything unrecognized are still opened and checked, since DICOM
files commonly have no extension at all.
- strict_extension=True: only files with no extension or a
.dcm/.dicom/.ima/.img-style extension (see DICOM_EXTENSIONS) are
opened. Faster, but will silently skip real DICOM files saved
with an unusual extension.
Note: total file count for progress percentage isn't known up front
(we never pre-list the whole tree into memory), so progress is reported
as a running count instead of a percentage.
"""
count_seen = 0
count_found = 0
count_skipped = 0
for dirpath, _dirnames, filenames in os.walk(root_folder):
for fn in filenames:
filepath = os.path.join(dirpath, fn)
count_seen += 1
ext = os.path.splitext(fn)[1].lower()
if strict_extension:
should_skip = ext != "" and ext not in DICOM_EXTENSIONS
else:
should_skip = ext in SKIP_EXTENSIONS
if should_skip:
count_skipped += 1
else:
row = extract_metadata(filepath)
if row is not None:
count_found += 1
yield row
if show_progress and count_seen % 25 == 0:
sys.stdout.write(
f"\rScanning... {count_seen} files checked — {count_found} DICOM found "
f"({count_skipped} skipped by extension)"
)
sys.stdout.flush()
if show_progress and count_seen:
sys.stdout.write(
f"\rScanning... {count_seen} files checked — {count_found} DICOM found "
f"({count_skipped} skipped by extension)\n"
)
sys.stdout.flush()
# ---------------------------------------------------------------------------
# Subcommand: scan
# ---------------------------------------------------------------------------
def cmd_scan(args):
folder = args.folder
if not os.path.isdir(folder):
print(f"Error: '{folder}' is not a valid folder.", file=sys.stderr)
sys.exit(1)
print(f"Scanning '{folder}' for DICOM files...")
# Column order for the CSV (FilePath/FileName first, then the
# standard DICOM_FIELDS, then file size last).
fieldnames = ["FilePath", "FileName"] + [c for c, _ in DICOM_FIELDS] + ["FileSizeKB"]
out_csv = args.output
row_count = 0
modality_counts = {}
patient_ids = set()
study_uids = set()
series_uids = set()
try:
with open(out_csv, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for row in scan_folder(folder, show_progress=not args.quiet, strict_extension=args.strict_extension):
writer.writerow(row)
row_count += 1
# Track summary stats incrementally (O(1) memory) instead
# of loading everything back into a DataFrame afterward.
modality_counts[row.get("Modality", "")] = modality_counts.get(row.get("Modality", ""), 0) + 1
if row.get("PatientID"):
patient_ids.add(row["PatientID"])
if row.get("StudyInstanceUID"):
study_uids.add(row["StudyInstanceUID"])
if row.get("SeriesInstanceUID"):
series_uids.add(row["SeriesInstanceUID"])
except Exception as exc:
print(f"Error during scan: {exc}", file=sys.stderr)
if args.verbose:
traceback.print_exc()
sys.exit(1)
if row_count == 0:
os.remove(out_csv) # don't leave a header-only file behind
print("No valid DICOM files found in this folder.")
sys.exit(0)
print(f"Indexed {row_count} DICOM files.")
print(f"CSV index saved to: {out_csv}")
if args.xlsx:
# Excel format can't be streamed row-by-row as easily as CSV, so we
# read the CSV back in chunks and write it out to .xlsx. This keeps
# peak memory bounded by chunk size rather than the full dataset,
# at the cost of a second pass over the data.
_csv_to_xlsx(out_csv, args.xlsx)
print(f"Excel index saved to: {args.xlsx}")
if not args.quiet:
print("\n--- Index Summary ---")
print(f"Total files indexed: {row_count}")
print(f"Unique patients: {len(patient_ids)}")
print(f"Unique studies: {len(study_uids)}")
print(f"Unique series: {len(series_uids)}")
print("Modalities:")
for modality, count in sorted(modality_counts.items(), key=lambda kv: -kv[1]):
label = modality if modality else "(unknown)"
print(f" {label:<10} {count}")
print()
def _csv_to_xlsx(csv_path, xlsx_path, chunksize=20000):
"""
Convert a CSV file to .xlsx without ever loading the entire CSV into
memory at once. Uses openpyxl's write-only mode, which streams rows
directly to disk instead of building the whole workbook in RAM.
"""
try:
from openpyxl import Workbook
except ImportError:
print("Error: openpyxl is required for --xlsx export. Install it with:", file=sys.stderr)
print(" pip install openpyxl", file=sys.stderr)
sys.exit(1)
wb = Workbook(write_only=True)
ws = wb.create_sheet("DICOM Index")
with open(csv_path, "r", newline="", encoding="utf-8") as f:
reader = csv.reader(f)
for row in reader:
ws.append(row)
wb.save(xlsx_path)
# ---------------------------------------------------------------------------
# Subcommand: search
# ---------------------------------------------------------------------------
DISPLAY_LIMIT = 200 # cap rows kept in memory for screen display
def cmd_search(args):
field_filters = {}
for flag, column in SEARCH_FLAG_TO_COLUMN.items():
value = getattr(args, flag, None)
if value:
field_filters[column] = value.lower()
query = args.query.lower() if args.query else None
match_count = 0
display_rows = [] # bounded to DISPLAY_LIMIT regardless of result size
fieldnames = None
writer = None
out_f = None
try:
if args.output:
out_f = open(args.output, "w", newline="", encoding="utf-8")
for row in _iter_index_rows(args.index):
if not _row_matches(row, field_filters, query):
continue
match_count += 1
if out_f is not None:
if writer is None:
fieldnames = list(row.keys())
writer = csv.DictWriter(out_f, fieldnames=fieldnames)
writer.writeheader()
writer.writerow(row)
if len(display_rows) < DISPLAY_LIMIT:
display_rows.append(row)
finally:
if out_f is not None:
out_f.close()
print(f"{match_count} matching record(s) found.\n")
if match_count == 0:
if args.output and os.path.exists(args.output):
os.remove(args.output) # no header-only leftover file
return
display_columns = args.columns.split(",") if args.columns else [
"PatientName", "PatientID", "StudyDate", "StudyDescription",
"Modality", "SeriesDescription", "SeriesNumber", "InstanceNumber",
"FileName",
]
display_columns = [c for c in display_columns if c in display_rows[0]]
_print_table(display_rows, display_columns)
if match_count > len(display_rows):
print(f"\n... showing first {len(display_rows)} of {match_count} matches. Use -o to save all results.")
if args.output:
print(f"\nFiltered results ({match_count} rows) saved to: {args.output}")
def _row_matches(row, field_filters, query):
"""Check a single CSV row (dict) against field filters and free-text query."""
for column, needle in field_filters.items():
haystack = str(row.get(column, "")).lower()
if needle not in haystack:
return False
if query:
if not any(query in str(v).lower() for v in row.values()):
return False
return True
# ---------------------------------------------------------------------------
# Subcommand: summary
# ---------------------------------------------------------------------------
def cmd_summary(args):
total = 0
patient_ids = set()
study_uids = set()
series_uids = set()
modality_counts = {}
group_counts = {} # (PatientName, PatientID, StudyDescription, SeriesDescription) -> count
for row in _iter_index_rows(args.index):
total += 1
if row.get("PatientID"):
patient_ids.add(row["PatientID"])
if row.get("StudyInstanceUID"):
study_uids.add(row["StudyInstanceUID"])
if row.get("SeriesInstanceUID"):
series_uids.add(row["SeriesInstanceUID"])
modality = row.get("Modality", "")
modality_counts[modality] = modality_counts.get(modality, 0) + 1
key = (
row.get("PatientName", ""), row.get("PatientID", ""),
row.get("StudyDescription", ""), row.get("SeriesDescription", ""),
)
group_counts[key] = group_counts.get(key, 0) + 1
if total == 0:
print("Index is empty.")
return
print("\n--- Index Summary ---")
print(f"Total files indexed: {total}")
print(f"Unique patients: {len(patient_ids)}")
print(f"Unique studies: {len(study_uids)}")
print(f"Unique series: {len(series_uids)}")
print("Modalities:")
for modality, count in sorted(modality_counts.items(), key=lambda kv: -kv[1]):
label = modality if modality else "(unknown)"
print(f" {label:<10} {count}")
print("\nPatients / Studies / Series:")
for (pname, pid, study, series), count in group_counts.items():
print(f" {pname} | {pid} | {study} | {series} -> {count} file(s)")
print()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _iter_index_rows(path):
"""
Yield rows (as dicts) from an index file one at a time, without ever
loading the whole file into memory. Supports both .csv and .xlsx.
"""
if not os.path.isfile(path):
print(f"Error: index file '{path}' not found. Run 'scan' first.", file=sys.stderr)
sys.exit(1)
try:
if path.lower().endswith(".xlsx"):
try:
from openpyxl import load_workbook
except ImportError:
print("Error: openpyxl is required to read .xlsx index files. Install it with:", file=sys.stderr)
print(" pip install openpyxl", file=sys.stderr)
sys.exit(1)
# read_only=True streams rows from disk instead of loading the
# whole sheet into memory.
wb = load_workbook(path, read_only=True, data_only=True)
ws = wb.active
rows_iter = ws.iter_rows(values_only=True)
header = [str(h) if h is not None else "" for h in next(rows_iter)]
for raw_row in rows_iter:
yield {
header[i]: ("" if raw_row[i] is None else str(raw_row[i]))
for i in range(len(header))
}
wb.close()
else:
with open(path, "r", newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
yield {k: (v if v is not None else "") for k, v in row.items()}
except SystemExit:
raise
except Exception as exc:
print(f"Error reading index file: {exc}", file=sys.stderr)
sys.exit(1)
def _print_table(rows, columns):
"""Print a list of dicts as a simple aligned text table for the given columns."""
if not rows:
return
col_widths = {
col: min(max(len(col), max(len(str(r.get(col, ""))) for r in rows)), 40)
for col in columns
}
header = " ".join(col.ljust(col_widths[col]) for col in columns)
print(header)
print("-" * len(header))
for row in rows:
line = " ".join(
str(row.get(col, ""))[: col_widths[col]].ljust(col_widths[col])
for col in columns
)
print(line)
# ---------------------------------------------------------------------------
# Argument parsing
# ---------------------------------------------------------------------------
def build_parser():
parser = argparse.ArgumentParser(
prog="dicom_indexer.py",
description="Index DICOM folders by metadata and search the resulting index.",
)
subparsers = parser.add_subparsers(dest="command", required=True)
# scan
p_scan = subparsers.add_parser("scan", help="Scan a folder and build a DICOM metadata index.")
p_scan.add_argument("folder", help="Path to the folder containing DICOM files (scanned recursively).")
p_scan.add_argument("-o", "--output", default="dicom_index.csv", help="Output CSV path (default: dicom_index.csv).")
p_scan.add_argument("--xlsx", help="Also save the index as an Excel (.xlsx) file at this path.")
p_scan.add_argument("-q", "--quiet", action="store_true", help="Suppress progress output and summary.")
p_scan.add_argument("--verbose", action="store_true", help="Show full error tracebacks on failure.")
p_scan.add_argument(
"--strict-extension", action="store_true",
help=(
"Only attempt to read files with a .dcm/.dicom/.ima/.img extension "
"(or no extension). Fastest option, but will silently skip DICOM "
"files saved with an unusual extension. Off by default."
),
)
p_scan.set_defaults(func=cmd_scan)
# search
p_search = subparsers.add_parser("search", help="Search/filter an existing DICOM index.")
p_search.add_argument("index", help="Path to the index CSV/XLSX file (created by 'scan').")
p_search.add_argument("-q", "--query", help="Free-text search across all fields.")
p_search.add_argument("--patient", help="Filter by patient name (substring match).")
p_search.add_argument("--patient-id", dest="patient_id", help="Filter by patient ID (substring match).")
p_search.add_argument("--study", help="Filter by study description (substring match).")
p_search.add_argument("--study-date", dest="study_date", help="Filter by study date, e.g. 2024-01-15 (substring match).")
p_search.add_argument("--modality", help="Filter by modality, e.g. CT, MR, US.")
p_search.add_argument("--series", help="Filter by series description (substring match).")
p_search.add_argument("--body-part", dest="body_part", help="Filter by body part examined.")
p_search.add_argument("--accession", help="Filter by accession number.")
p_search.add_argument("--columns", help="Comma-separated list of columns to display (default: a sensible subset).")
p_search.add_argument("-o", "--output", help="Save filtered results to this CSV path.")
p_search.set_defaults(func=cmd_search)
# summary
p_summary = subparsers.add_parser("summary", help="Show summary statistics for an existing index.")
p_summary.add_argument("index", help="Path to the index CSV/XLSX file (created by 'scan').")
p_summary.set_defaults(func=cmd_summary)
return parser
def main():
parser = build_parser()
args = parser.parse_args()
args.func(args)
if __name__ == "__main__":
main()

549
dicom_sender.py Normal file
View File

@ -0,0 +1,549 @@
#!/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 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)
# ---------------------------------------------------------------------------
# Subcommand: send
# ---------------------------------------------------------------------------
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 <csv> and/or "
"one or more --accession <value>.",
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)
print(f"Looking up {len(accession_numbers)} accession number(s) in index '{args.source_index}'...")
# ---- Step 2: find matching files in the index ----
matched_rows = list(find_matching_files(args.source_index, accession_numbers))
if not matched_rows:
print("No matching files found in the index for the given accession number(s).")
sys.exit(0)
found_accessions = {r["AccessionNumber"] for r in matched_rows}
missing_accessions = accession_numbers - found_accessions
print(f"Found {len(matched_rows)} file(s) across {len(found_accessions)} accession number(s).")
if missing_accessions:
print(
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 ""),
file=sys.stderr,
)
# ---- Step 3: verify files exist on disk, and (for real sends) preload
# each file's SOP Class UID + Transfer Syntax UID so we can build
# the minimal set of presentation contexts to request ----
sendable = []
missing_files = []
for row in matched_rows:
filepath = row["FilePath"]
if os.path.isfile(filepath):
sendable.append(row)
else:
missing_files.append(filepath)
if missing_files:
print(
f"Warning: {len(missing_files)} file(s) listed in the index no longer "
f"exist on disk and will be skipped:",
file=sys.stderr,
)
for fp in missing_files[:10]:
print(f" - {fp}", file=sys.stderr)
if len(missing_files) > 10:
print(f" ... and {len(missing_files) - 10} more", file=sys.stderr)
if not sendable:
print("No sendable files remain after checking disk. Aborting.")
sys.exit(1)
if args.dry_run:
print("\n--- DRY RUN: the following files would be sent ---")
for row in sendable:
print(f" [{row.get('AccessionNumber','')}] {row['FilePath']}")
print(f"\nTotal: {len(sendable)} file(s). No network connection was made.")
return
# ---- Step 4: read each file's SOP Class UID / Transfer Syntax UID to
# build presentation contexts (max 128 per association) ----
print("Reading SOP Class / Transfer Syntax info from files to build presentation contexts...")
file_info = [] # (filepath, row, sop_class_uid, transfer_syntax_uid) or None on read failure
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 as exc:
unreadable.append(filepath)
if unreadable:
print(
f"Warning: {len(unreadable)} file(s) could not be read as DICOM and "
f"will be skipped:",
file=sys.stderr,
)
for fp in unreadable[:10]:
print(f" - {fp}", file=sys.stderr)
if not file_info:
print("No readable DICOM files remain. Aborting.")
sys.exit(1)
# 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 '<transfer syntax>'" 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:
print(
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.",
file=sys.stderr,
)
sys.exit(1)
# ---- Step 5: build AE, request contexts, associate ----
ae = AE(ae_title=args.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 = args.timeout
ae.dimse_timeout = args.timeout
ae.network_timeout = args.timeout
distinct_sop_classes = {p[0] for p in context_pairs}
print(
f"\nConnecting to {args.ae_title}@{args.host}:{args.port} "
f"(calling AE: {args.calling_ae_title}) requesting "
f"{len(context_pairs)} presentation context(s) covering "
f"{len(distinct_sop_classes)} SOP Class(es)..."
)
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, "
"that the destination accepts the SOP classes being sent, and that "
"it allows associations from this calling AE title.",
file=sys.stderr,
)
sys.exit(1)
# ---- Step 6: send each file via C-STORE ----
results = [] # (filepath, accession, code, label)
try:
for i, (filepath, row, sop_class, ts) in enumerate(file_info, start=1):
accession = row.get("AccessionNumber", "")
sys.stdout.write(
f"\rSending {i}/{len(file_info)} (accession {accession})..."
)
sys.stdout.flush()
try:
status = assoc.send_c_store(filepath)
except Exception as exc:
results.append((filepath, accession, None, f"Exception: {exc}"))
continue
code, label = describe_status(status)
results.append((filepath, accession, code, label))
if args.delay:
time.sleep(args.delay)
finally:
assoc.release()
print() # newline after progress line
# ---- Step 7: report results ----
succeeded = [r for r in results if r[2] is not None and is_success(r[2])]
failed = [r for r in results if r not in succeeded]
print(f"\n--- Send Summary ---")
print(f"Total attempted: {len(results)}")
print(f"Succeeded: {len(succeeded)}")
print(f"Failed: {len(failed)}")
if failed:
print("\nFailed files:")
for filepath, accession, code, label in failed:
code_str = f"0x{code:04X}" if code is not None else "N/A"
print(f" [{accession}] {filepath}")
print(f" -> status {code_str}: {label}")
if args.report:
with open(args.report, "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])
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()

BIN
index-all.csv Normal file
View File

Binary file not shown.
Can't render this file because it is too large.

77730
index.csv Normal file
View File

File diff suppressed because it is too large Load Diff

2404
send.csv Normal file
View File

File diff suppressed because it is too large Load Diff