678 lines
25 KiB
Python
678 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
DICOM Folder Indexer (CLI) — MongoDB edition
|
|
=============================================
|
|
Recursively scans a folder for DICOM files, extracts key metadata
|
|
(patient, study, series, instance level), and indexes it into MongoDB.
|
|
Files whose path is already present in the index are skipped on
|
|
subsequent scans, so you can re-run `scan` on a growing folder and only
|
|
new files will be read/indexed.
|
|
|
|
Dependencies:
|
|
pip install pydicom pymongo python-dotenv
|
|
|
|
MongoDB connection config:
|
|
Connection settings are read from a .env file (in the current directory,
|
|
or pointed to with --env-file) instead of being passed as CLI flags.
|
|
Create a .env file like this (see .env.example):
|
|
|
|
MONGO_URI=mongodb://localhost:27017
|
|
MONGO_DB=dicom_index
|
|
MONGO_COLLECTION=files
|
|
|
|
CLI flags --mongo-uri/--db/--collection are still available and take
|
|
priority over the .env file if you ever need a one-off override.
|
|
|
|
Usage:
|
|
# Build/update an index from a folder (uses .env for Mongo connection)
|
|
python dicom_indexer.py scan /path/to/dicom_folder
|
|
|
|
# Force re-reading files that are already in the index (e.g. metadata
|
|
# extraction logic changed, or files were modified in place)
|
|
python dicom_indexer.py scan /path/to/dicom_folder --rescan
|
|
|
|
# Search the index
|
|
python dicom_indexer.py search --patient "Doe" --modality CT
|
|
|
|
# Search across all fields for a keyword
|
|
python dicom_indexer.py search -q "chest"
|
|
|
|
# Show summary statistics for the index
|
|
python dicom_indexer.py summary
|
|
|
|
# Export the current index to CSV
|
|
python dicom_indexer.py export -o index.csv
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import csv
|
|
import argparse
|
|
import traceback
|
|
from datetime import datetime
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dependency check with a friendly message.
|
|
# ---------------------------------------------------------------------------
|
|
MISSING = []
|
|
try:
|
|
import pydicom
|
|
except ImportError:
|
|
MISSING.append("pydicom")
|
|
|
|
try:
|
|
from pymongo import MongoClient, UpdateOne
|
|
from pymongo.errors import PyMongoError
|
|
except ImportError:
|
|
MISSING.append("pymongo")
|
|
|
|
if MISSING:
|
|
print("Missing required packages: " + ", ".join(MISSING))
|
|
print("Install them with:")
|
|
print(f" pip install {' '.join(MISSING)}")
|
|
sys.exit(1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# .env file loading.
|
|
#
|
|
# Uses python-dotenv if it's installed (handles quoting, comments, export
|
|
# prefixes, etc. correctly). If it's not installed, falls back to a small
|
|
# manual KEY=VALUE parser so the script still works without the extra
|
|
# dependency — just with fewer edge cases covered.
|
|
# ---------------------------------------------------------------------------
|
|
def load_env_file(path):
|
|
"""Load KEY=VALUE pairs from an env file into os.environ (without
|
|
overriding variables already set in the real environment)."""
|
|
if not os.path.isfile(path):
|
|
return False
|
|
|
|
try:
|
|
from dotenv import load_dotenv
|
|
load_dotenv(path, override=False)
|
|
return True
|
|
except ImportError:
|
|
pass
|
|
|
|
# Minimal fallback parser: KEY=VALUE per line, '#' comments, optional
|
|
# surrounding quotes, blank lines ignored.
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _, value = line.partition("=")
|
|
key = key.strip()
|
|
value = value.strip().strip('"').strip("'")
|
|
if key and key not in os.environ:
|
|
os.environ[key] = value
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Metadata fields to extract from each DICOM file.
|
|
# Format: (Mongo field 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 -> Mongo field name
|
|
SEARCH_FLAG_TO_COLUMN = {
|
|
"patient": "PatientName",
|
|
"patient_id": "PatientID",
|
|
"study_date": "StudyDate",
|
|
"study": "StudyDescription",
|
|
"modality": "Modality",
|
|
"series": "SeriesDescription",
|
|
"body_part": "BodyPartExamined",
|
|
"accession": "AccessionNumber",
|
|
}
|
|
|
|
# Columns actually stored in every document (used for CSV export / display)
|
|
ALL_FIELDS = ["FilePath", "FileName"] + [c for c, _ in DICOM_FIELDS] + ["FileSizeKB", "IndexedAt"]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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"}
|
|
|
|
DEFAULT_MONGO_URI = "mongodb://root:AtHEntRyPrOchite@172.16.44.35:27017/"
|
|
DEFAULT_DB_NAME = "rsabhk_simrs"
|
|
DEFAULT_COLLECTION_NAME = "dicom-index"
|
|
|
|
|
|
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
|
|
|
|
filepath_abs = os.path.abspath(filepath)
|
|
row = {"FilePath": filepath_abs, "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)
|
|
row["IndexedAt"] = datetime.utcnow().isoformat()
|
|
|
|
return row
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MongoDB helpers
|
|
# ---------------------------------------------------------------------------
|
|
def get_collection(mongo_uri, db_name, collection_name):
|
|
"""Connect to MongoDB and return the target collection, with indexes set up."""
|
|
try:
|
|
client = MongoClient(mongo_uri, serverSelectionTimeoutMS=5000)
|
|
client.admin.command("ping")
|
|
except PyMongoError as exc:
|
|
print(f"Error: could not connect to MongoDB at '{mongo_uri}': {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
coll = client[db_name][collection_name]
|
|
|
|
# FilePath is unique so re-scanning the same file upserts instead of
|
|
# duplicating, and so we can pre-load "already indexed" paths cheaply.
|
|
coll.create_index("FilePath", unique=True)
|
|
coll.create_index("PatientID")
|
|
coll.create_index("StudyInstanceUID")
|
|
coll.create_index("SeriesInstanceUID")
|
|
coll.create_index("SOPInstanceUID")
|
|
coll.create_index("Modality")
|
|
|
|
return coll
|
|
|
|
|
|
def scan_folder(root_folder, already_indexed, show_progress=True, strict_extension=False):
|
|
"""
|
|
Walk root_folder recursively and yield a metadata dict for every valid,
|
|
not-yet-indexed DICOM file found, one at a time.
|
|
|
|
`already_indexed` is a set of absolute FilePaths already present in the
|
|
Mongo collection; files whose path is in this set are skipped without
|
|
even being opened. Pass an empty set to force re-reading everything.
|
|
|
|
This is a generator so callers can write each row to Mongo as it's
|
|
produced, instead of holding every row in memory at once.
|
|
"""
|
|
count_seen = 0
|
|
count_found = 0
|
|
count_skipped_ext = 0
|
|
count_skipped_indexed = 0
|
|
for dirpath, _dirnames, filenames in os.walk(root_folder):
|
|
for fn in filenames:
|
|
filepath = os.path.join(dirpath, fn)
|
|
filepath_abs = os.path.abspath(filepath)
|
|
count_seen += 1
|
|
|
|
if filepath_abs in already_indexed:
|
|
count_skipped_indexed += 1
|
|
else:
|
|
ext = os.path.splitext(fn)[1].lower()
|
|
if strict_extension:
|
|
should_skip_ext = ext != "" and ext not in DICOM_EXTENSIONS
|
|
else:
|
|
should_skip_ext = ext in SKIP_EXTENSIONS
|
|
|
|
if should_skip_ext:
|
|
count_skipped_ext += 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} new DICOM found "
|
|
f"({count_skipped_indexed} already indexed, {count_skipped_ext} skipped by extension)"
|
|
)
|
|
sys.stdout.flush()
|
|
|
|
if show_progress and count_seen:
|
|
sys.stdout.write(
|
|
f"\rScanning... {count_seen} files checked — {count_found} new DICOM found "
|
|
f"({count_skipped_indexed} already indexed, {count_skipped_ext} skipped by extension)\n"
|
|
)
|
|
sys.stdout.flush()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: scan
|
|
# ---------------------------------------------------------------------------
|
|
BULK_WRITE_BATCH_SIZE = 500
|
|
|
|
|
|
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)
|
|
|
|
coll = get_collection(args.mongo_uri, args.db, args.collection)
|
|
|
|
if args.rescan:
|
|
already_indexed = set()
|
|
print("--rescan given: re-reading all files, even ones already in the index.")
|
|
else:
|
|
print("Loading list of already-indexed files...")
|
|
already_indexed = {doc["FilePath"] for doc in coll.find({}, {"FilePath": 1, "_id": 0})}
|
|
print(f"Found {len(already_indexed)} file(s) already in the index.")
|
|
|
|
print(f"Scanning '{folder}' for DICOM files...")
|
|
|
|
row_count = 0
|
|
modality_counts = {}
|
|
patient_ids = set()
|
|
study_uids = set()
|
|
series_uids = set()
|
|
|
|
batch = []
|
|
|
|
def flush_batch():
|
|
if not batch:
|
|
return
|
|
try:
|
|
coll.bulk_write(batch, ordered=False)
|
|
except PyMongoError as exc:
|
|
print(f"\nWarning: a batch write to MongoDB failed: {exc}", file=sys.stderr)
|
|
batch.clear()
|
|
|
|
try:
|
|
for row in scan_folder(
|
|
folder,
|
|
already_indexed,
|
|
show_progress=not args.quiet,
|
|
strict_extension=args.strict_extension,
|
|
):
|
|
batch.append(UpdateOne({"FilePath": row["FilePath"]}, {"$set": row}, upsert=True))
|
|
row_count += 1
|
|
|
|
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"])
|
|
|
|
if len(batch) >= BULK_WRITE_BATCH_SIZE:
|
|
flush_batch()
|
|
|
|
flush_batch()
|
|
|
|
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:
|
|
print("No new DICOM files found (nothing to add to the index).")
|
|
sys.exit(0)
|
|
|
|
print(f"Indexed {row_count} new DICOM file(s) into MongoDB "
|
|
f"({args.db}.{args.collection} at {args.mongo_uri}).")
|
|
|
|
if not args.quiet:
|
|
print("\n--- New Files This Scan ---")
|
|
print(f"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()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: search
|
|
# ---------------------------------------------------------------------------
|
|
DISPLAY_LIMIT = 200 # cap rows kept in memory for screen display
|
|
|
|
|
|
def cmd_search(args):
|
|
coll = get_collection(args.mongo_uri, args.db, args.collection)
|
|
|
|
mongo_filter = {}
|
|
for flag, column in SEARCH_FLAG_TO_COLUMN.items():
|
|
value = getattr(args, flag, None)
|
|
if value:
|
|
mongo_filter[column] = {"$regex": value, "$options": "i"}
|
|
|
|
if args.query:
|
|
searchable_columns = [c for c, _ in DICOM_FIELDS] + ["FileName", "FilePath"]
|
|
mongo_filter["$or"] = [
|
|
{col: {"$regex": args.query, "$options": "i"}} for col in searchable_columns
|
|
]
|
|
|
|
match_count = coll.count_documents(mongo_filter)
|
|
print(f"{match_count} matching record(s) found.\n")
|
|
|
|
if match_count == 0:
|
|
return
|
|
|
|
cursor = coll.find(mongo_filter, {"_id": 0})
|
|
|
|
display_rows = []
|
|
out_f = None
|
|
writer = None
|
|
try:
|
|
if args.output:
|
|
out_f = open(args.output, "w", newline="", encoding="utf-8")
|
|
writer = csv.DictWriter(out_f, fieldnames=ALL_FIELDS)
|
|
writer.writeheader()
|
|
|
|
for row in cursor:
|
|
if writer is not None:
|
|
writer.writerow({k: row.get(k, "") for k in ALL_FIELDS})
|
|
if len(display_rows) < DISPLAY_LIMIT:
|
|
display_rows.append(row)
|
|
finally:
|
|
if out_f is not None:
|
|
out_f.close()
|
|
|
|
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}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: summary
|
|
# ---------------------------------------------------------------------------
|
|
def cmd_summary(args):
|
|
coll = get_collection(args.mongo_uri, args.db, args.collection)
|
|
|
|
total = coll.count_documents({})
|
|
if total == 0:
|
|
print("Index is empty.")
|
|
return
|
|
|
|
patient_count = len(coll.distinct("PatientID", {"PatientID": {"$ne": ""}}))
|
|
study_count = len(coll.distinct("StudyInstanceUID", {"StudyInstanceUID": {"$ne": ""}}))
|
|
series_count = len(coll.distinct("SeriesInstanceUID", {"SeriesInstanceUID": {"$ne": ""}}))
|
|
|
|
modality_pipeline = [
|
|
{"$group": {"_id": "$Modality", "count": {"$sum": 1}}},
|
|
{"$sort": {"count": -1}},
|
|
]
|
|
modality_counts = list(coll.aggregate(modality_pipeline))
|
|
|
|
group_pipeline = [
|
|
{"$group": {
|
|
"_id": {
|
|
"PatientName": "$PatientName", "PatientID": "$PatientID",
|
|
"StudyDescription": "$StudyDescription", "SeriesDescription": "$SeriesDescription",
|
|
},
|
|
"count": {"$sum": 1},
|
|
}},
|
|
]
|
|
group_counts = list(coll.aggregate(group_pipeline))
|
|
|
|
print("\n--- Index Summary ---")
|
|
print(f"Total files indexed: {total}")
|
|
print(f"Unique patients: {patient_count}")
|
|
print(f"Unique studies: {study_count}")
|
|
print(f"Unique series: {series_count}")
|
|
print("Modalities:")
|
|
for entry in modality_counts:
|
|
label = entry["_id"] if entry["_id"] else "(unknown)"
|
|
print(f" {label:<10} {entry['count']}")
|
|
|
|
print("\nPatients / Studies / Series:")
|
|
for entry in group_counts:
|
|
g = entry["_id"]
|
|
print(f" {g.get('PatientName','')} | {g.get('PatientID','')} | "
|
|
f"{g.get('StudyDescription','')} | {g.get('SeriesDescription','')} -> {entry['count']} file(s)")
|
|
print()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Subcommand: export
|
|
# ---------------------------------------------------------------------------
|
|
def cmd_export(args):
|
|
coll = get_collection(args.mongo_uri, args.db, args.collection)
|
|
|
|
total = coll.count_documents({})
|
|
if total == 0:
|
|
print("Index is empty, nothing to export.")
|
|
return
|
|
|
|
with open(args.output, "w", newline="", encoding="utf-8") as f:
|
|
writer = csv.DictWriter(f, fieldnames=ALL_FIELDS)
|
|
writer.writeheader()
|
|
for row in coll.find({}, {"_id": 0}):
|
|
writer.writerow({k: row.get(k, "") for k in ALL_FIELDS})
|
|
|
|
print(f"Exported {total} row(s) to: {args.output}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
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 _add_mongo_args(p):
|
|
p.add_argument("--env-file", default=".env",
|
|
help="Path to a .env file with MONGO_URI/MONGO_DB/MONGO_COLLECTION (default: .env).")
|
|
p.add_argument("--mongo-uri", default=None,
|
|
help=f"MongoDB connection URI. Overrides .env / $MONGO_URI. (default: {DEFAULT_MONGO_URI})")
|
|
p.add_argument("--db", default=None,
|
|
help=f"MongoDB database name. Overrides .env / $MONGO_DB. (default: {DEFAULT_DB_NAME})")
|
|
p.add_argument("--collection", default=None,
|
|
help=f"MongoDB collection name. Overrides .env / $MONGO_COLLECTION. (default: {DEFAULT_COLLECTION_NAME})")
|
|
|
|
|
|
def resolve_mongo_args(args):
|
|
"""
|
|
Fill in args.mongo_uri/db/collection from (in priority order):
|
|
1. CLI flags, if explicitly given
|
|
2. The .env file (args.env_file)
|
|
3. Real environment variables
|
|
4. Hard-coded defaults
|
|
"""
|
|
load_env_file(args.env_file)
|
|
|
|
if args.mongo_uri is None:
|
|
args.mongo_uri = os.environ.get("MONGO_URI", DEFAULT_MONGO_URI)
|
|
if args.db is None:
|
|
args.db = os.environ.get("MONGO_DB", DEFAULT_DB_NAME)
|
|
if args.collection is None:
|
|
args.collection = os.environ.get("MONGO_COLLECTION", DEFAULT_COLLECTION_NAME)
|
|
|
|
|
|
def build_parser():
|
|
parser = argparse.ArgumentParser(
|
|
prog="dicom_indexer.py",
|
|
description="Index DICOM folders by metadata into MongoDB, 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 index new DICOM files into MongoDB.")
|
|
p_scan.add_argument("folder", help="Path to the folder containing DICOM files (scanned recursively).")
|
|
_add_mongo_args(p_scan)
|
|
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(
|
|
"--rescan", action="store_true",
|
|
help="Re-read and re-index files that are already in the index (normally they're skipped).",
|
|
)
|
|
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 the DICOM index in MongoDB.")
|
|
_add_mongo_args(p_search)
|
|
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 the MongoDB index.")
|
|
_add_mongo_args(p_summary)
|
|
p_summary.set_defaults(func=cmd_summary)
|
|
|
|
# export
|
|
p_export = subparsers.add_parser("export", help="Export the MongoDB index to a CSV file.")
|
|
_add_mongo_args(p_export)
|
|
p_export.add_argument("-o", "--output", default="dicom_index.csv", help="Output CSV path (default: dicom_index.csv).")
|
|
p_export.set_defaults(func=cmd_export)
|
|
|
|
return parser
|
|
|
|
|
|
def main():
|
|
parser = build_parser()
|
|
args = parser.parse_args()
|
|
resolve_mongo_args(args)
|
|
args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|