355 lines
12 KiB
Python
355 lines
12 KiB
Python
#!/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) |