Compare commits
5 Commits
master
...
production
| Author | SHA1 | Date | |
|---|---|---|---|
| 030f67ddc3 | |||
|
|
1845c8069b | ||
|
|
a8a1b62f31 | ||
|
|
56c07edbbd | ||
|
|
15c319eb9e |
8
.env
Normal file
8
.env
Normal file
@ -0,0 +1,8 @@
|
||||
# Copy this file to .env and adjust as needed.
|
||||
# The indexer reads these to connect to MongoDB, so you don't have to
|
||||
# pass --mongo-uri/--db/--collection on every command.
|
||||
|
||||
MONGO_URI=mongodb://root:AtHEntRyPrOchite@172.16.44.35:27017/
|
||||
MONGO_DB=rsabhk_simrs
|
||||
MONGO_COLLECTION=dicom-index
|
||||
PATH_MAPPINGS_FILE=/app/config.json
|
||||
BIN
__pycache__/dicom_sender.cpython-310.pyc
Normal file
BIN
__pycache__/dicom_sender.cpython-310.pyc
Normal file
Binary file not shown.
BIN
__pycache__/sender.cpython-310.pyc
Normal file
BIN
__pycache__/sender.cpython-310.pyc
Normal file
Binary file not shown.
355
api.py
Normal file
355
api.py
Normal file
@ -0,0 +1,355 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
DICOM Sender API Server
|
||||
=======================
|
||||
A small REST API that wraps dicom_sender.py so a client can trigger sends by
|
||||
Accession Number (single or as a group) without needing DICOM tooling itself.
|
||||
|
||||
Destination PACS/router details and the MongoDB index connection are fixed
|
||||
in config.json on the server. Clients only ever supply accession numbers.
|
||||
|
||||
Sends run as background jobs (a single C-STORE association can take a while
|
||||
for large studies), and the client polls for status/results using the
|
||||
returned job_id.
|
||||
|
||||
Run:
|
||||
pip install fastapi uvicorn pydicom pynetdicom
|
||||
python3 api_server.py
|
||||
# or: uvicorn api_server:app --host 0.0.0.0 --port 8000
|
||||
|
||||
Endpoints:
|
||||
GET /health liveness check
|
||||
GET /config sanitized view of active server config
|
||||
POST /echo synchronous C-ECHO connectivity test
|
||||
POST /jobs queue a send job -> {job_id}
|
||||
GET /jobs list recent jobs (most recent first)
|
||||
GET /jobs/{job_id} full status/results for one job
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import uuid
|
||||
import threading
|
||||
from collections import OrderedDict
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime, timezone
|
||||
from typing import List, Optional
|
||||
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from pydantic import BaseModel, Field, model_validator
|
||||
from pymongo import MongoClient
|
||||
|
||||
import sender as sender
|
||||
|
||||
LOGGING_CONFIG = {
|
||||
"version": 1,
|
||||
"disable_existing_loggers": False,
|
||||
"formatters": {
|
||||
"default": {
|
||||
"()": "uvicorn.logging.DefaultFormatter",
|
||||
"fmt": "%(asctime)s %(levelprefix)s %(message)s",
|
||||
"datefmt": "%Y-%m-%d %H:%M:%S",
|
||||
},
|
||||
"access": {
|
||||
"()": "uvicorn.logging.AccessFormatter",
|
||||
"fmt": '%(asctime)s %(levelprefix)s %(client_addr)s - "%(request_line)s" %(status_code)s',
|
||||
"datefmt": "%Y-%m-%d %H:%M:%S",
|
||||
},
|
||||
},
|
||||
"handlers": {
|
||||
"default": {"formatter": "default", "class": "logging.StreamHandler", "stream": "ext://sys.stderr"},
|
||||
"access": {"formatter": "access", "class": "logging.StreamHandler", "stream": "ext://sys.stdout"},
|
||||
},
|
||||
"loggers": {
|
||||
"uvicorn": {"handlers": ["default"], "level": "INFO"},
|
||||
"uvicorn.error": {"level": "INFO"},
|
||||
"uvicorn.access": {"handlers": ["access"], "level": "INFO", "propagate": False},
|
||||
},
|
||||
}
|
||||
|
||||
CONFIG_PATH = os.environ.get("DICOM_API_CONFIG", os.path.join(os.path.dirname(__file__), "config.json"))
|
||||
|
||||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||||
CONFIG = json.load(f)
|
||||
|
||||
DEST = CONFIG["destination"]
|
||||
SEND_DELAY = CONFIG.get("send_delay", 0.0)
|
||||
MAX_CONCURRENT_JOBS = CONFIG.get("max_concurrent_jobs", 1)
|
||||
JOB_HISTORY_LIMIT = CONFIG.get("job_history_limit", 500)
|
||||
|
||||
INDEX_SOURCE = CONFIG["index_source"]
|
||||
if INDEX_SOURCE.get("type") != "mongodb":
|
||||
raise RuntimeError(
|
||||
f"Unsupported index_source.type: {INDEX_SOURCE.get('type')!r}. This server build only supports 'mongodb'."
|
||||
)
|
||||
|
||||
# The URI can carry credentials, so prefer an env var over the checked-in
|
||||
# config.json when both are present (handy for docker-compose + .env).
|
||||
_MONGO_URI = os.environ.get("MONGO_URI") or INDEX_SOURCE.get("uri")
|
||||
if not _MONGO_URI:
|
||||
raise RuntimeError("No MongoDB URI configured: set index_source.uri in config.json or the MONGO_URI env var.")
|
||||
|
||||
_mongo_client = MongoClient(_MONGO_URI)
|
||||
_mongo_collection = _mongo_client[INDEX_SOURCE["database"]][INDEX_SOURCE["collection"]]
|
||||
_FILEPATH_FIELD = INDEX_SOURCE.get("filepath_field", "FilePath")
|
||||
_ACCESSION_FIELD = INDEX_SOURCE.get("accession_field", "AccessionNumber")
|
||||
|
||||
# Host path -> container path translation, e.g. Mongo stores
|
||||
# "/mnt/pacs_data/study1/img.dcm" (the path on the machine that indexed the
|
||||
# files) but this container has that same folder mounted at "/data".
|
||||
_PATH_PREFIX_MAP = [(m["from"], m["to"]) for m in CONFIG.get("path_mappings", [])]
|
||||
|
||||
|
||||
def _mongo_lookup(accession_numbers):
|
||||
return sender.find_matching_files_mongo(
|
||||
_mongo_collection,
|
||||
accession_numbers,
|
||||
filepath_field=_FILEPATH_FIELD,
|
||||
accession_field=_ACCESSION_FIELD,
|
||||
path_prefix_map=_PATH_PREFIX_MAP,
|
||||
)
|
||||
|
||||
|
||||
app = FastAPI(title="DICOM Sender API", version="1.0")
|
||||
|
||||
_executor = ThreadPoolExecutor(max_workers=MAX_CONCURRENT_JOBS)
|
||||
_jobs_lock = threading.Lock()
|
||||
_jobs = OrderedDict() # job_id -> job dict, insertion order = creation order
|
||||
|
||||
|
||||
def _now():
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
def _trim_history():
|
||||
# Only drop finished jobs (completed/error), oldest first, keep queued/running always.
|
||||
while len(_jobs) > JOB_HISTORY_LIMIT:
|
||||
for jid, job in _jobs.items():
|
||||
if job["status"] in ("completed", "error"):
|
||||
del _jobs[jid]
|
||||
break
|
||||
else:
|
||||
break # nothing removable
|
||||
|
||||
|
||||
def _serializable_result(result):
|
||||
"""Convert send_accessions()'s result dict (which contains sets) to JSON-safe types."""
|
||||
if result is None:
|
||||
return None
|
||||
out = dict(result)
|
||||
out["found_accessions"] = sorted(result.get("found_accessions", set()))
|
||||
out["missing_accessions"] = sorted(result.get("missing_accessions", set()))
|
||||
out["results"] = [
|
||||
{
|
||||
"filepath": fp,
|
||||
"accession": acc,
|
||||
"status_code": (f"0x{code:04X}" if code is not None else None),
|
||||
"status_label": label,
|
||||
"success": bool(code is not None and sender.is_success(code)),
|
||||
}
|
||||
for fp, acc, code, label in result.get("results", [])
|
||||
]
|
||||
return out
|
||||
|
||||
|
||||
def _run_job(job_id):
|
||||
with _jobs_lock:
|
||||
job = _jobs[job_id]
|
||||
job["status"] = "running"
|
||||
job["started_at"] = _now()
|
||||
accessions = job["requested_accessions"]
|
||||
dry_run = job["dry_run"]
|
||||
|
||||
def progress_cb(current, total, filepath, accession):
|
||||
with _jobs_lock:
|
||||
_jobs[job_id]["progress"] = {"current": current, "total": total}
|
||||
|
||||
try:
|
||||
result = sender.send_accessions(
|
||||
accession_numbers=accessions,
|
||||
lookup_fn=_mongo_lookup,
|
||||
host=DEST["host"],
|
||||
port=DEST["port"],
|
||||
ae_title=DEST["ae_title"],
|
||||
calling_ae_title=DEST.get("calling_ae_title", sender.DEFAULT_CALLING_AE_TITLE),
|
||||
timeout=DEST.get("timeout", 30),
|
||||
delay=SEND_DELAY,
|
||||
dry_run=dry_run,
|
||||
verbose=False,
|
||||
progress_callback=progress_cb,
|
||||
)
|
||||
except Exception as exc:
|
||||
with _jobs_lock:
|
||||
job = _jobs[job_id]
|
||||
job["status"] = "error"
|
||||
job["error"] = f"Unhandled exception: {exc}"
|
||||
job["finished_at"] = _now()
|
||||
return
|
||||
|
||||
with _jobs_lock:
|
||||
job = _jobs[job_id]
|
||||
job["result"] = _serializable_result(result)
|
||||
job["error"] = result.get("error")
|
||||
job["status"] = "error" if result.get("error") else "completed"
|
||||
job["finished_at"] = _now()
|
||||
_trim_history()
|
||||
|
||||
|
||||
class SendRequest(BaseModel):
|
||||
accessions: Optional[List[str]] = Field(
|
||||
default=None, description="List of accession numbers to send as a group."
|
||||
)
|
||||
accession: Optional[str] = Field(
|
||||
default=None, description="A single accession number to send (alternative to 'accessions')."
|
||||
)
|
||||
dry_run: bool = Field(
|
||||
default=False, description="If true, resolve and report matching files without sending."
|
||||
)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _require_at_least_one(self):
|
||||
if not self.accessions and not self.accession:
|
||||
raise ValueError("Provide 'accession' (single) or 'accessions' (list).")
|
||||
return self
|
||||
|
||||
def all_accessions(self) -> List[str]:
|
||||
out = list(self.accessions or [])
|
||||
if self.accession:
|
||||
out.append(self.accession)
|
||||
# de-dupe, strip, drop empties, preserve order
|
||||
seen = set()
|
||||
cleaned = []
|
||||
for a in out:
|
||||
a = (a or "").strip()
|
||||
if a and a not in seen:
|
||||
seen.add(a)
|
||||
cleaned.append(a)
|
||||
return cleaned
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
def health():
|
||||
return {"status": "ok", "time": _now()}
|
||||
|
||||
|
||||
@app.get("/config")
|
||||
def get_config():
|
||||
try:
|
||||
_mongo_client.admin.command("ping")
|
||||
mongo_reachable = True
|
||||
except Exception:
|
||||
mongo_reachable = False
|
||||
|
||||
return {
|
||||
"index_source_reachable": mongo_reachable,
|
||||
"destination": {
|
||||
"host": DEST["host"],
|
||||
"port": DEST["port"],
|
||||
"ae_title": DEST["ae_title"],
|
||||
"calling_ae_title": DEST.get("calling_ae_title", sender.DEFAULT_CALLING_AE_TITLE),
|
||||
"timeout": DEST.get("timeout", 30),
|
||||
},
|
||||
"index_source": {
|
||||
"type": "mongodb",
|
||||
"database": INDEX_SOURCE["database"],
|
||||
"collection": INDEX_SOURCE["collection"],
|
||||
},
|
||||
"path_mappings": [{"from": f, "to": t} for f, t in _PATH_PREFIX_MAP],
|
||||
"max_concurrent_jobs": MAX_CONCURRENT_JOBS,
|
||||
}
|
||||
|
||||
|
||||
@app.post("/echo")
|
||||
def echo():
|
||||
"""Synchronous C-ECHO connectivity test against the configured destination."""
|
||||
from pynetdicom import AE
|
||||
from pynetdicom.sop_class import Verification
|
||||
|
||||
ae = AE(ae_title=DEST.get("calling_ae_title", sender.DEFAULT_CALLING_AE_TITLE))
|
||||
ae.add_requested_context(Verification)
|
||||
timeout = DEST.get("timeout", 30)
|
||||
ae.acse_timeout = timeout
|
||||
ae.dimse_timeout = timeout
|
||||
ae.network_timeout = timeout
|
||||
|
||||
try:
|
||||
assoc = ae.associate(DEST["host"], DEST["port"], ae_title=DEST["ae_title"])
|
||||
except Exception as exc:
|
||||
raise HTTPException(status_code=502, detail=f"Association error: {exc}")
|
||||
|
||||
if not assoc.is_established:
|
||||
raise HTTPException(status_code=502, detail="Could not establish association.")
|
||||
|
||||
status = assoc.send_c_echo()
|
||||
code, label = sender.describe_status(status)
|
||||
assoc.release()
|
||||
|
||||
success = code is not None and sender.is_success(code)
|
||||
return {
|
||||
"success": success,
|
||||
"status_code": (f"0x{code:04X}" if code is not None else None),
|
||||
"status_label": label,
|
||||
}
|
||||
|
||||
|
||||
@app.post("/jobs", status_code=202)
|
||||
def create_job(req: SendRequest):
|
||||
try:
|
||||
_mongo_client.admin.command("ping")
|
||||
except Exception as exc:
|
||||
raise HTTPException(status_code=500, detail=f"Cannot reach MongoDB index source: {exc}")
|
||||
|
||||
accessions = req.all_accessions()
|
||||
job_id = str(uuid.uuid4())
|
||||
job = {
|
||||
"job_id": job_id,
|
||||
"status": "queued",
|
||||
"dry_run": req.dry_run,
|
||||
"requested_accessions": accessions,
|
||||
"created_at": _now(),
|
||||
"started_at": None,
|
||||
"finished_at": None,
|
||||
"progress": {"current": 0, "total": 0},
|
||||
"result": None,
|
||||
"error": None,
|
||||
}
|
||||
with _jobs_lock:
|
||||
_jobs[job_id] = job
|
||||
|
||||
_executor.submit(_run_job, job_id)
|
||||
|
||||
return {"job_id": job_id, "status": "queued", "status_url": f"/jobs/{job_id}"}
|
||||
|
||||
|
||||
@app.get("/jobs")
|
||||
def list_jobs(limit: int = 50):
|
||||
with _jobs_lock:
|
||||
items = list(_jobs.values())[::-1][:limit]
|
||||
return [
|
||||
{
|
||||
"job_id": j["job_id"],
|
||||
"status": j["status"],
|
||||
"dry_run": j["dry_run"],
|
||||
"requested_count": len(j["requested_accessions"]),
|
||||
"created_at": j["created_at"],
|
||||
"finished_at": j["finished_at"],
|
||||
}
|
||||
for j in items
|
||||
]
|
||||
|
||||
|
||||
@app.get("/jobs/{job_id}")
|
||||
def get_job(job_id: str):
|
||||
with _jobs_lock:
|
||||
job = _jobs.get(job_id)
|
||||
if job is None:
|
||||
raise HTTPException(status_code=404, detail="Job not found.")
|
||||
return dict(job)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
|
||||
uvicorn.run(app, host=CONFIG["server"]["host"], port=CONFIG["server"]["port"], log_config=LOGGING_CONFIG)
|
||||
599
app.py
599
app.py
@ -1,599 +0,0 @@
|
||||
#!/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()
|
||||
29
config.json
Normal file
29
config.json
Normal file
@ -0,0 +1,29 @@
|
||||
{
|
||||
"destination": {
|
||||
"host": "172.16.201.100",
|
||||
"port": 11112,
|
||||
"ae_title": "DCMROUTER",
|
||||
"calling_ae_title": "DCM-RSABHK",
|
||||
"timeout": 30
|
||||
},
|
||||
"index_source": {
|
||||
"type": "mongodb",
|
||||
"uri": "mongodb://root:AtHEntRyPrOchite@172.16.44.35:27017/",
|
||||
"database": "rsabhk_simrs",
|
||||
"collection": "dicom-index",
|
||||
"filepath_field": "FilePath",
|
||||
"accession_field": "AccessionNumber"
|
||||
},
|
||||
"path_mappings": [
|
||||
{"from": "/mnt/pacs_data", "to": "/data"},
|
||||
{"from": "Z:/2026", "to": "Z:/2026"},
|
||||
{"from": "/mnt/data/2026", "to": "Z:/2026"}
|
||||
],
|
||||
"send_delay": 0.0,
|
||||
"max_concurrent_jobs": 1000,
|
||||
"job_history_limit": 500,
|
||||
"server": {
|
||||
"host": "0.0.0.0",
|
||||
"port": 8000
|
||||
}
|
||||
}
|
||||
1288
dicom_indexer.py
Normal file
1288
dicom_indexer.py
Normal file
File diff suppressed because it is too large
Load Diff
BIN
index-all.csv
BIN
index-all.csv
Binary file not shown.
|
Can't render this file because it is too large.
|
@ -10,4 +10,7 @@
|
||||
|
||||
pydicom>=3.0.2
|
||||
openpyxl>=3.1.5
|
||||
pynetdicom>=3.0.4
|
||||
pynetdicom>=3.0.4
|
||||
fastapi
|
||||
uvicorn[standard]
|
||||
pymongo
|
||||
@ -155,6 +155,79 @@ def load_accession_list(path):
|
||||
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,
|
||||
@ -223,7 +296,307 @@ def cmd_echo(args):
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Subcommand: send
|
||||
# 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 '<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:
|
||||
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 ----
|
||||
@ -251,212 +624,30 @@ def cmd_send(args):
|
||||
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)..."
|
||||
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,
|
||||
)
|
||||
|
||||
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)
|
||||
if result.get("error"):
|
||||
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)
|
||||
if result["dry_run"] or result["sendable_count"] == 0:
|
||||
return
|
||||
|
||||
# ---- 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}")
|
||||
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:
|
||||
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])
|
||||
write_report_csv(args.report, send_results)
|
||||
print(f"\nDetailed report saved to: {args.report}")
|
||||
|
||||
if failed:
|
||||
Loading…
x
Reference in New Issue
Block a user