npx skills add https://github.com/aurabx/skills --skill 'DICOMweb Protocol'为 DICOMweb 标准(WADO-RS、STOW-RS、QIDO-RS)生成正确的 HTTP 客户端代码。处理那些让开发者感到棘手的部分:多部分 MIME 编码、DICOM 特定的内容类型、查询参数语法以及批量数据检索。可与任何符合 DICOMweb 标准的服务器配合使用。
requests(或使用 httpx 进行异步操作)DICOMweb 为医学影像提供三种 RESTful 服务:
| 服务 | 用途 | HTTP 方法 | 路径模式 |
|---|---|---|---|
| QIDO-RS | 查询(搜索) | GET | /studies, /series, |
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
/instances| WADO-RS | 检索(下载) | GET | /studies/{uid}, .../series/{uid}, .../instances/{uid} |
| STOW-RS | 存储(上传) | POST | /studies, /studies/{uid} |
{base_url}/studies # 所有研究
{base_url}/studies/{StudyInstanceUID} # 单个研究
{base_url}/studies/{StudyInstanceUID}/series # 研究中的序列
{base_url}/studies/{StudyInstanceUID}/series/{SeriesInstanceUID}
{base_url}/studies/{StudyInstanceUID}/series/{SeriesInstanceUID}/instances
{base_url}/studies/{StudyInstanceUID}/series/{SeriesInstanceUID}/instances/{SOPInstanceUID}
根据 DICOM 属性搜索研究、序列或实例。
import requests
BASE_URL = "https://your-dicomweb-server.com/dicomweb"
HEADERS = {"Accept": "application/dicom+json"}
# 根据需要添加认证头:
# HEADERS["Authorization"] = "Bearer {token}"
# 搜索所有 CT 研究
response = requests.get(
f"{BASE_URL}/studies",
headers=HEADERS,
params={"ModalitiesInStudy": "CT"},
)
studies = response.json()
# 按患者姓名搜索(支持通配符)
response = requests.get(
f"{BASE_URL}/studies",
headers=HEADERS,
params={"PatientName": "Smith*"},
)
# 按日期范围搜索
response = requests.get(
f"{BASE_URL}/studies",
headers=HEADERS,
params={"StudyDate": "20250101-20250131"},
)
# 在特定研究中搜索序列
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/series",
headers=HEADERS,
params={"Modality": "CT"},
)
QIDO-RS 使用 DICOM 标签关键字作为查询参数:
| 参数 | 示例 | 描述 |
|---|---|---|
PatientName | Smith* | 使用 * 进行通配符搜索 |
PatientID | 12345 | 精确匹配 |
StudyDate | 20250115 | 单个日期 |
StudyDate | 20250101-20250131 | 日期范围 |
ModalitiesInStudy | CT | 按模态过滤 |
StudyInstanceUID | 1.2.3... | 精确 UID 匹配 |
AccessionNumber | ACC001 | RIS 登记号 |
StudyDescription | *CHEST* | 描述中的通配符 |
limit | 25 | 每页最大结果数 |
offset | 50 | 跳过前 N 个结果 |
includefield | all | 返回所有字段(或指定标签) |
[
{
"00080020": { "vr": "DA", "Value": ["20250115"] },
"00080060": { "vr": "CS", "Value": ["CT"] },
"00080090": { "vr": "PN", "Value": [{ "Alphabetic": "Smith^John" }] },
"0008103E": { "vr": "LO", "Value": ["CT CHEST W CONTRAST"] },
"00100010": { "vr": "PN", "Value": [{ "Alphabetic": "Doe^Jane" }] },
"00100020": { "vr": "LO", "Value": ["PAT001"] },
"0020000D": { "vr": "UI", "Value": ["1.2.840.113619..."] }
}
]
def parse_dicom_json_value(element: dict):
"""从 DICOM JSON 元素中提取值。"""
if "Value" not in element:
return None
values = element["Value"]
vr = element.get("vr", "")
if vr == "PN":
# 人员姓名具有嵌套结构
return values[0].get("Alphabetic", "") if values else ""
elif len(values) == 1:
return values[0]
else:
return values
def get_tag_value(result: dict, tag: str):
"""从 QIDO-RS 结果中获取标签值。
Args:
result: 单个 QIDO-RS 结果字典
tag: 标签,为 8 字符十六进制字符串,例如,'00100010' 对应 PatientName
"""
element = result.get(tag, {})
return parse_dicom_json_value(element)
# 常见标签十六进制代码
PATIENT_NAME = "00100010"
PATIENT_ID = "00100020"
STUDY_DATE = "00080020"
MODALITY = "00080060"
STUDY_UID = "0020000D"
SERIES_UID = "0020000E"
SOP_UID = "00080018"
STUDY_DESCRIPTION = "00081030"
SERIES_DESCRIPTION = "0008103E"
NUM_INSTANCES = "00201208"
# 用法
for study in studies:
patient = get_tag_value(study, PATIENT_NAME)
date = get_tag_value(study, STUDY_DATE)
modality = get_tag_value(study, MODALITY)
print(f"{patient} | {date} | {modality}")
def paginate_qido(base_url: str, path: str, params: dict = None,
headers: dict = None, page_size: int = 50):
"""对 QIDO-RS 结果进行分页。"""
params = params or {}
params["limit"] = page_size
offset = 0
while True:
params["offset"] = offset
response = requests.get(f"{base_url}/{path}",
headers=headers, params=params)
results = response.json()
if not results:
break
yield from results
offset += len(results)
if len(results) < page_size:
break
# 用法
for study in paginate_qido(BASE_URL, "studies",
params={"ModalitiesInStudy": "CT"},
headers=HEADERS):
print(get_tag_value(study, STUDY_UID))
下载 DICOM 实例、元数据或渲染后的图像。
# 获取研究中所有实例的元数据(无像素数据)
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/metadata",
headers={"Accept": "application/dicom+json"},
)
metadata = response.json() # 实例元数据字典列表
import re
def retrieve_instance(base_url: str, study_uid: str, series_uid: str,
sop_uid: str, headers: dict = None) -> bytes:
"""以字节形式检索单个 DICOM 实例。"""
url = (f"{base_url}/studies/{study_uid}/series/{series_uid}"
f"/instances/{sop_uid}")
h = {**(headers or {}), "Accept": "application/dicom"}
response = requests.get(url, headers=h)
response.raise_for_status()
return response.content
def retrieve_study_multipart(base_url: str, study_uid: str,
headers: dict = None) -> list[bytes]:
"""以多部分响应形式检索研究中的所有实例。"""
url = f"{base_url}/studies/{study_uid}"
h = {
**(headers or {}),
"Accept": 'multipart/related; type="application/dicom"',
}
response = requests.get(url, headers=h, stream=True)
response.raise_for_status()
# 解析多部分响应
content_type = response.headers["Content-Type"]
return parse_multipart_dicom(response.content, content_type)
def parse_multipart_dicom(content: bytes, content_type: str) -> list[bytes]:
"""将 multipart/related 响应解析为单独的 DICOM 部分。"""
# 从 content-type 中提取边界
boundary_match = re.search(r'boundary="?([^";]+)"?', content_type)
if not boundary_match:
raise ValueError("Content-Type 中未找到边界")
boundary = boundary_match.group(1).encode()
parts = content.split(b"--" + boundary)
dicom_parts = []
for part in parts:
# 跳过前导码和后记
part = part.strip()
if not part or part == b"--":
continue
# 找到分隔头部和主体的空行
header_end = part.find(b"\r\n\r\n")
if header_end == -1:
header_end = part.find(b"\n\n")
if header_end == -1:
continue
body = part[header_end + 2:]
else:
body = part[header_end + 4:]
if body:
dicom_parts.append(body)
return dicom_parts
# 获取实例的渲染 PNG 图像
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/series/{series_uid}"
f"/instances/{sop_uid}/rendered",
headers={"Accept": "image/png"},
params={
"window": "40,400", # 中心,宽度(用于 CT)
},
)
with open("output.png", "wb") as f:
f.write(response.content)
# 获取渲染后的 JPEG 缩略图
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/series/{series_uid}"
f"/instances/{sop_uid}/rendered",
headers={"Accept": "image/jpeg"},
params={
"viewport": "256,256", # 最大宽度,高度
"quality": "80",
},
)
# 检索多帧实例的第 1 帧(1 起始索引)
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/series/{series_uid}"
f"/instances/{sop_uid}/frames/1",
headers={"Accept": "application/dicom"},
)
将 DICOM 实例上传到 DICOMweb 服务器。
from pathlib import Path
import uuid
def stow_rs_upload(base_url: str, dicom_files: list[str],
study_uid: str = None, headers: dict = None) -> dict:
"""
通过 STOW-RS 上传 DICOM 文件。
Args:
base_url: DICOMweb 基础 URL
dicom_files: DICOM 文件路径列表
study_uid: 可选的 StudyInstanceUID(附加到 URL)
headers: 可选的头部信息(例如,认证)
"""
boundary = f"boundary-{uuid.uuid4().hex}"
# 构建 URL
url = f"{base_url}/studies"
if study_uid:
url = f"{url}/{study_uid}"
# 构建多部分主体
body = b""
for filepath in dicom_files:
data = Path(filepath).read_bytes()
body += f"--{boundary}\r\n".encode()
body += b"Content-Type: application/dicom\r\n"
body += b"\r\n"
body += data
body += b"\r\n"
body += f"--{boundary}--\r\n".encode()
# 设置头部
h = {
**(headers or {}),
"Content-Type": (
f'multipart/related; type="application/dicom"; '
f"boundary={boundary}"
),
"Accept": "application/dicom+json",
}
response = requests.post(url, headers=h, data=body)
response.raise_for_status()
return response.json() if response.content else {}
# 用法
result = stow_rs_upload(
base_url="https://your-server.com/dicomweb",
dicom_files=["image1.dcm", "image2.dcm", "image3.dcm"],
headers={"Authorization": "Bearer {token}"},
)
成功的 STOW-RS 响应包含已存储实例的列表:
{
"00081190": {
"vr": "UR",
"Value": ["https://server.com/dicomweb/studies/1.2.3..."]
},
"00081198": {
"vr": "SQ",
"Value": []
},
"00081199": {
"vr": "SQ",
"Value": [
{
"00081150": { "vr": "UI", "Value": ["1.2.840.10008.5.1.4.1.1.2"] },
"00081155": { "vr": "UI", "Value": ["1.2.3.4.5..."] },
"00081190": { "vr": "UR", "Value": ["https://server.com/..."] }
}
]
}
}
| 标签 | 含义 |
|---|---|
00081190 | 检索 URL |
00081198 | 失败的 SOP 序列(空 = 全部成功) |
00081199 | 引用的 SOP 序列(成功存储的实例) |
def stow_rs_batch(base_url: str, dicom_files: list[str],
batch_size: int = 50, headers: dict = None):
"""分批上传 DICOM 文件以避免请求大小限制。"""
for i in range(0, len(dicom_files), batch_size):
batch = dicom_files[i:i + batch_size]
print(f"上传批次 {i // batch_size + 1} "
f"({len(batch)} 个文件)...")
result = stow_rs_upload(base_url, batch, headers=headers)
failed = result.get("00081198", {}).get("Value", [])
if failed:
print(f" 警告:{len(failed)} 个文件失败")
else:
print(f" 成功:{len(batch)} 个文件已存储")
STOW-RS 也支持上传 DICOM JSON 以及批量数据 URI,而不是原始的 DICOM 文件。这种方式不太常见,但对于编程式创建实例很有用:
import json
def stow_rs_json(base_url: str, dicom_json: list[dict],
headers: dict = None) -> dict:
"""以 DICOM JSON 形式上传实例(无需二进制 DICOM 文件)。"""
boundary = f"boundary-{uuid.uuid4().hex}"
url = f"{base_url}/studies"
body = f"--{boundary}\r\n".encode()
body += b"Content-Type: application/dicom+json\r\n\r\n"
body += json.dumps(dicom_json).encode()
body += b"\r\n"
body += f"--{boundary}--\r\n".encode()
h = {
**(headers or {}),
"Content-Type": (
f'multipart/related; type="application/dicom+json"; '
f"boundary={boundary}"
),
"Accept": "application/dicom+json",
}
response = requests.post(url, headers=h, data=body)
response.raise_for_status()
return response.json() if response.content else {}
import requests
from pathlib import Path
from typing import Optional
import uuid
import json
class DICOMwebClient:
"""用于 DICOMweb QIDO-RS、WADO-RS 和 STOW-RS 的客户端。"""
def __init__(self, base_url: str, auth_token: str = None,
api_key: str = None):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
if auth_token:
self.session.headers["Authorization"] = f"Bearer {auth_token}"
elif api_key:
self.session.headers["Authorization"] = f"Bearer {api_key}"
# --- QIDO-RS ---
def search_studies(self, **params) -> list[dict]:
"""搜索研究。将 DICOM 关键字作为关键字参数传递。"""
return self._qido("studies", params)
def search_series(self, study_uid: str = None, **params) -> list[dict]:
path = f"studies/{study_uid}/series" if study_uid else "series"
return self._qido(path, params)
def search_instances(self, study_uid: str = None,
series_uid: str = None, **params) -> list[dict]:
if study_uid and series_uid:
path = f"studies/{study_uid}/series/{series_uid}/instances"
elif study_uid:
path = f"studies/{study_uid}/instances"
else:
path = "instances"
return self._qido(path, params)
def _qido(self, path: str, params: dict) -> list[dict]:
response = self.session.get(
f"{self.base_url}/{path}",
headers={"Accept": "application/dicom+json"},
params=params,
)
response.raise_for_status()
return response.json() if response.content else []
# --- WADO-RS ---
def retrieve_metadata(self, study_uid: str,
series_uid: str = None,
sop_uid: str = None) -> list[dict]:
"""检索实例元数据(无像素数据)。"""
path = self._build_path(study_uid, series_uid, sop_uid)
response = self.session.get(
f"{self.base_url}/{path}/metadata",
headers={"Accept": "application/dicom+json"},
)
response.raise_for_status()
return response.json()
def retrieve_instance(self, study_uid: str, series_uid: str,
sop_uid: str) -> bytes:
"""以字节形式检索单个 DICOM 实例。"""
path = self._build_path(study_uid, series_uid, sop_uid)
response = self.session.get(
f"{self.base_url}/{path}",
headers={"Accept": "application/dicom"},
)
response.raise_for_status()
return response.content
def retrieve_rendered(self, study_uid: str, series_uid: str,
sop_uid: str, media_type: str = "image/png",
window: str = None,
viewport: str = None) -> bytes:
"""检索渲染后的图像(PNG/JPEG)。"""
path = self._build_path(study_uid, series_uid, sop_uid)
params = {}
if window:
params["window"] = window
if viewport:
params["viewport"] = viewport
response = self.session.get(
f"{self.base_url}/{path}/rendered",
headers={"Accept": media_type},
params=params,
)
response.raise_for_status()
return response.content
# --- STOW-RS ---
def store(self, dicom_files: list[str],
study_uid: str = None) -> dict:
"""通过 STOW-RS 上传 DICOM 文件。"""
boundary = f"boundary-{uuid.uuid4().hex}"
url = f"{self.base_url}/studies"
if study_uid:
url = f"{url}/{study_uid}"
body = b""
for filepath in dicom_files:
data = Path(filepath).read_bytes()
body += f"--{boundary}\r\n".encode()
body += b"Content-Type: application/dicom\r\n\r\n"
body += data
body += b"\r\n"
body += f"--{boundary}--\r\n".encode()
response = self.session.post(
url,
headers={
"Content-Type": (
f'multipart/related; type="application/dicom"; '
f"boundary={boundary}"
),
"Accept": "application/dicom+json",
},
data=body,
)
response.raise_for_status()
return response.json() if response.content else {}
# --- Helpers ---
def _build_path(self, study_uid: str, series_uid: str = None,
sop_uid: str = None) -> str:
path = f"studies/{study_uid}"
if series_uid:
path = f"{path}/series/{series_uid}"
if sop_uid:
path = f"{path}/instances/{sop_uid}"
return path
# Usage
client = DICOMwebClient(
base_url="https://your-server.com/dicomweb",
auth_token="your-token",
)
# Search
studies = client.search_studies(PatientName="Smith*", ModalitiesInStudy="CT")
# Retrieve metadata
metadata = client.retrieve_metadata(study_uid="1.2.3...")
# Download a rendered image
png = client.retrieve_rendered(
study_uid="1.2.3...",
series_uid="1.2.3...",
sop_uid="1.2.3...",
window="40,400",
)
Path("output.png").write_bytes(png)
# Upload
result = client.store(["image1.dcm", "image2.dcm"])
"Plugins": ["libOrthancDicomWeb.so"]http://localhost:8042/dicom-weborthanc:orthanc)http://localhost:8080/dcm4chee-arc/aets/DCM4CHEE/rshttps://healthcare.googleapis.com/v1/projects/{project}/locations/{location}/datasets/{dataset}/dicomStores/{store}/dicomWeb| 操作 | 请求 Accept/Content-Type | 响应 Content-Type |
|---|---|---|
| QIDO-RS | Accept: application/dicom+json | application/dicom+json |
| WADO-RS(元数据) | Accept: application/dicom+json | application/dicom+json |
| WADO-RS(实例) | Accept: application/dicom | application/dicom 或 multipart/related |
| WADO-RS(渲染) | Accept: image/png 或 image/jpeg | image/png 或 image/jpeg |
| STOW-RS(DICOM) | Content-Type: multipart/related; type="application/dicom" | application/dicom+json |
| STOW-RS(JSON) | Content-Type: multipart/related; type="application/dicom+json" | application/dicom+json |
type= 参数:multipart/related; type="application/dicom"; boundary=xxx。缺少 type 参数会导致许多服务器拒绝请求。"00100010" 而不是 "PatientName"。请使用查找表或 DICOM 标准浏览器。{"Alphabetic": "Family^Given"},而不是纯字符串。* 作为通配符(不是 % 或 ?)。大小写敏感性因服务器而异。-:StudyDate=20250101-20250131,而不是 StudyDate>=20250101&StudyDate<=20250131。limit/offset,另一些使用 Link 头。请查看您服务器的文档。每周安装次数
0
代码仓库
GitHub 星标数
6
首次出现
1970年1月1日
安全审计
Generates correct HTTP client code for the DICOMweb standard (WADO-RS, STOW-RS, QIDO-RS). Handles the parts that trip up developers: multipart MIME encoding, DICOM-specific content types, query parameter syntax, and bulk data retrieval. Works with any DICOMweb-compliant server.
requests (or httpx for async)DICOMweb provides three RESTful services for medical imaging:
| Service | Purpose | HTTP Methods | Path Pattern |
|---|---|---|---|
| QIDO-RS | Query (search) | GET | /studies, /series, /instances |
| WADO-RS | Retrieve (download) | GET | /studies/{uid}, .../series/{uid}, .../instances/{uid} |
| STOW-RS | Store (upload) | POST | /studies, /studies/{uid} |
{base_url}/studies # All studies
{base_url}/studies/{StudyInstanceUID} # One study
{base_url}/studies/{StudyInstanceUID}/series # Series in study
{base_url}/studies/{StudyInstanceUID}/series/{SeriesInstanceUID}
{base_url}/studies/{StudyInstanceUID}/series/{SeriesInstanceUID}/instances
{base_url}/studies/{StudyInstanceUID}/series/{SeriesInstanceUID}/instances/{SOPInstanceUID}
Search for studies, series, or instances by DICOM attributes.
import requests
BASE_URL = "https://your-dicomweb-server.com/dicomweb"
HEADERS = {"Accept": "application/dicom+json"}
# Add auth headers as needed:
# HEADERS["Authorization"] = "Bearer {token}"
# Search for all CT studies
response = requests.get(
f"{BASE_URL}/studies",
headers=HEADERS,
params={"ModalitiesInStudy": "CT"},
)
studies = response.json()
# Search for studies by patient name (wildcard supported)
response = requests.get(
f"{BASE_URL}/studies",
headers=HEADERS,
params={"PatientName": "Smith*"},
)
# Search for studies by date range
response = requests.get(
f"{BASE_URL}/studies",
headers=HEADERS,
params={"StudyDate": "20250101-20250131"},
)
# Search for series within a specific study
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/series",
headers=HEADERS,
params={"Modality": "CT"},
)
QIDO-RS uses DICOM tag keywords as query parameters:
| Parameter | Example | Description |
|---|---|---|
PatientName | Smith* | Wildcard search with * |
PatientID | 12345 | Exact match |
StudyDate | 20250115 | Single date |
[
{
"00080020": { "vr": "DA", "Value": ["20250115"] },
"00080060": { "vr": "CS", "Value": ["CT"] },
"00080090": { "vr": "PN", "Value": [{ "Alphabetic": "Smith^John" }] },
"0008103E": { "vr": "LO", "Value": ["CT CHEST W CONTRAST"] },
"00100010": { "vr": "PN", "Value": [{ "Alphabetic": "Doe^Jane" }] },
"00100020": { "vr": "LO", "Value": ["PAT001"] },
"0020000D": { "vr": "UI", "Value": ["1.2.840.113619..."] }
}
]
def parse_dicom_json_value(element: dict):
"""Extract the value from a DICOM JSON element."""
if "Value" not in element:
return None
values = element["Value"]
vr = element.get("vr", "")
if vr == "PN":
# Person Name has nested structure
return values[0].get("Alphabetic", "") if values else ""
elif len(values) == 1:
return values[0]
else:
return values
def get_tag_value(result: dict, tag: str):
"""Get a tag value from a QIDO-RS result.
Args:
result: A single QIDO-RS result dict
tag: Tag as 8-char hex string, e.g., '00100010' for PatientName
"""
element = result.get(tag, {})
return parse_dicom_json_value(element)
# Common tag hex codes
PATIENT_NAME = "00100010"
PATIENT_ID = "00100020"
STUDY_DATE = "00080020"
MODALITY = "00080060"
STUDY_UID = "0020000D"
SERIES_UID = "0020000E"
SOP_UID = "00080018"
STUDY_DESCRIPTION = "00081030"
SERIES_DESCRIPTION = "0008103E"
NUM_INSTANCES = "00201208"
# Usage
for study in studies:
patient = get_tag_value(study, PATIENT_NAME)
date = get_tag_value(study, STUDY_DATE)
modality = get_tag_value(study, MODALITY)
print(f"{patient} | {date} | {modality}")
def paginate_qido(base_url: str, path: str, params: dict = None,
headers: dict = None, page_size: int = 50):
"""Paginate through QIDO-RS results."""
params = params or {}
params["limit"] = page_size
offset = 0
while True:
params["offset"] = offset
response = requests.get(f"{base_url}/{path}",
headers=headers, params=params)
results = response.json()
if not results:
break
yield from results
offset += len(results)
if len(results) < page_size:
break
# Usage
for study in paginate_qido(BASE_URL, "studies",
params={"ModalitiesInStudy": "CT"},
headers=HEADERS):
print(get_tag_value(study, STUDY_UID))
Download DICOM instances, metadata, or rendered images.
# Get metadata for all instances in a study (no pixel data)
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/metadata",
headers={"Accept": "application/dicom+json"},
)
metadata = response.json() # List of instance metadata dicts
import re
def retrieve_instance(base_url: str, study_uid: str, series_uid: str,
sop_uid: str, headers: dict = None) -> bytes:
"""Retrieve a single DICOM instance as bytes."""
url = (f"{base_url}/studies/{study_uid}/series/{series_uid}"
f"/instances/{sop_uid}")
h = {**(headers or {}), "Accept": "application/dicom"}
response = requests.get(url, headers=h)
response.raise_for_status()
return response.content
def retrieve_study_multipart(base_url: str, study_uid: str,
headers: dict = None) -> list[bytes]:
"""Retrieve all instances in a study as a multipart response."""
url = f"{base_url}/studies/{study_uid}"
h = {
**(headers or {}),
"Accept": 'multipart/related; type="application/dicom"',
}
response = requests.get(url, headers=h, stream=True)
response.raise_for_status()
# Parse multipart response
content_type = response.headers["Content-Type"]
return parse_multipart_dicom(response.content, content_type)
def parse_multipart_dicom(content: bytes, content_type: str) -> list[bytes]:
"""Parse a multipart/related response into individual DICOM parts."""
# Extract boundary from content-type
boundary_match = re.search(r'boundary="?([^";]+)"?', content_type)
if not boundary_match:
raise ValueError("No boundary found in Content-Type")
boundary = boundary_match.group(1).encode()
parts = content.split(b"--" + boundary)
dicom_parts = []
for part in parts:
# Skip preamble and epilogue
part = part.strip()
if not part or part == b"--":
continue
# Find the blank line separating headers from body
header_end = part.find(b"\r\n\r\n")
if header_end == -1:
header_end = part.find(b"\n\n")
if header_end == -1:
continue
body = part[header_end + 2:]
else:
body = part[header_end + 4:]
if body:
dicom_parts.append(body)
return dicom_parts
# Get a rendered PNG of an instance
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/series/{series_uid}"
f"/instances/{sop_uid}/rendered",
headers={"Accept": "image/png"},
params={
"window": "40,400", # center,width (for CT)
},
)
with open("output.png", "wb") as f:
f.write(response.content)
# Get a rendered JPEG thumbnail
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/series/{series_uid}"
f"/instances/{sop_uid}/rendered",
headers={"Accept": "image/jpeg"},
params={
"viewport": "256,256", # max width, height
"quality": "80",
},
)
# Retrieve frame 1 of a multi-frame instance (1-indexed)
response = requests.get(
f"{BASE_URL}/studies/{study_uid}/series/{series_uid}"
f"/instances/{sop_uid}/frames/1",
headers={"Accept": "application/dicom"},
)
Upload DICOM instances to a DICOMweb server.
from pathlib import Path
import uuid
def stow_rs_upload(base_url: str, dicom_files: list[str],
study_uid: str = None, headers: dict = None) -> dict:
"""
Upload DICOM files via STOW-RS.
Args:
base_url: DICOMweb base URL
dicom_files: List of paths to DICOM files
study_uid: Optional StudyInstanceUID (appended to URL)
headers: Optional headers (e.g., auth)
"""
boundary = f"boundary-{uuid.uuid4().hex}"
# Build URL
url = f"{base_url}/studies"
if study_uid:
url = f"{url}/{study_uid}"
# Build multipart body
body = b""
for filepath in dicom_files:
data = Path(filepath).read_bytes()
body += f"--{boundary}\r\n".encode()
body += b"Content-Type: application/dicom\r\n"
body += b"\r\n"
body += data
body += b"\r\n"
body += f"--{boundary}--\r\n".encode()
# Set headers
h = {
**(headers or {}),
"Content-Type": (
f'multipart/related; type="application/dicom"; '
f"boundary={boundary}"
),
"Accept": "application/dicom+json",
}
response = requests.post(url, headers=h, data=body)
response.raise_for_status()
return response.json() if response.content else {}
# Usage
result = stow_rs_upload(
base_url="https://your-server.com/dicomweb",
dicom_files=["image1.dcm", "image2.dcm", "image3.dcm"],
headers={"Authorization": "Bearer {token}"},
)
A successful STOW-RS response includes a list of stored instances:
{
"00081190": {
"vr": "UR",
"Value": ["https://server.com/dicomweb/studies/1.2.3..."]
},
"00081198": {
"vr": "SQ",
"Value": []
},
"00081199": {
"vr": "SQ",
"Value": [
{
"00081150": { "vr": "UI", "Value": ["1.2.840.10008.5.1.4.1.1.2"] },
"00081155": { "vr": "UI", "Value": ["1.2.3.4.5..."] },
"00081190": { "vr": "UR", "Value": ["https://server.com/..."] }
}
]
}
}
| Tag | Meaning |
|---|---|
00081190 | Retrieve URL |
00081198 | Failed SOP Sequence (empty = all succeeded) |
00081199 | Referenced SOP Sequence (successfully stored instances) |
def stow_rs_batch(base_url: str, dicom_files: list[str],
batch_size: int = 50, headers: dict = None):
"""Upload DICOM files in batches to avoid request size limits."""
for i in range(0, len(dicom_files), batch_size):
batch = dicom_files[i:i + batch_size]
print(f"Uploading batch {i // batch_size + 1} "
f"({len(batch)} files)...")
result = stow_rs_upload(base_url, batch, headers=headers)
failed = result.get("00081198", {}).get("Value", [])
if failed:
print(f" WARNING: {len(failed)} files failed")
else:
print(f" OK: {len(batch)} files stored")
STOW-RS also supports uploading DICOM JSON with bulk data URIs instead of raw DICOM files. This is less common but useful for programmatic instance creation:
import json
def stow_rs_json(base_url: str, dicom_json: list[dict],
headers: dict = None) -> dict:
"""Upload instances as DICOM JSON (no binary DICOM files needed)."""
boundary = f"boundary-{uuid.uuid4().hex}"
url = f"{base_url}/studies"
body = f"--{boundary}\r\n".encode()
body += b"Content-Type: application/dicom+json\r\n\r\n"
body += json.dumps(dicom_json).encode()
body += b"\r\n"
body += f"--{boundary}--\r\n".encode()
h = {
**(headers or {}),
"Content-Type": (
f'multipart/related; type="application/dicom+json"; '
f"boundary={boundary}"
),
"Accept": "application/dicom+json",
}
response = requests.post(url, headers=h, data=body)
response.raise_for_status()
return response.json() if response.content else {}
import requests
from pathlib import Path
from typing import Optional
import uuid
import json
class DICOMwebClient:
"""Client for DICOMweb QIDO-RS, WADO-RS, and STOW-RS."""
def __init__(self, base_url: str, auth_token: str = None,
api_key: str = None):
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
if auth_token:
self.session.headers["Authorization"] = f"Bearer {auth_token}"
elif api_key:
self.session.headers["Authorization"] = f"Bearer {api_key}"
# --- QIDO-RS ---
def search_studies(self, **params) -> list[dict]:
"""Search for studies. Pass DICOM keywords as keyword args."""
return self._qido("studies", params)
def search_series(self, study_uid: str = None, **params) -> list[dict]:
path = f"studies/{study_uid}/series" if study_uid else "series"
return self._qido(path, params)
def search_instances(self, study_uid: str = None,
series_uid: str = None, **params) -> list[dict]:
if study_uid and series_uid:
path = f"studies/{study_uid}/series/{series_uid}/instances"
elif study_uid:
path = f"studies/{study_uid}/instances"
else:
path = "instances"
return self._qido(path, params)
def _qido(self, path: str, params: dict) -> list[dict]:
response = self.session.get(
f"{self.base_url}/{path}",
headers={"Accept": "application/dicom+json"},
params=params,
)
response.raise_for_status()
return response.json() if response.content else []
# --- WADO-RS ---
def retrieve_metadata(self, study_uid: str,
series_uid: str = None,
sop_uid: str = None) -> list[dict]:
"""Retrieve instance metadata (no pixel data)."""
path = self._build_path(study_uid, series_uid, sop_uid)
response = self.session.get(
f"{self.base_url}/{path}/metadata",
headers={"Accept": "application/dicom+json"},
)
response.raise_for_status()
return response.json()
def retrieve_instance(self, study_uid: str, series_uid: str,
sop_uid: str) -> bytes:
"""Retrieve a single DICOM instance as bytes."""
path = self._build_path(study_uid, series_uid, sop_uid)
response = self.session.get(
f"{self.base_url}/{path}",
headers={"Accept": "application/dicom"},
)
response.raise_for_status()
return response.content
def retrieve_rendered(self, study_uid: str, series_uid: str,
sop_uid: str, media_type: str = "image/png",
window: str = None,
viewport: str = None) -> bytes:
"""Retrieve a rendered image (PNG/JPEG)."""
path = self._build_path(study_uid, series_uid, sop_uid)
params = {}
if window:
params["window"] = window
if viewport:
params["viewport"] = viewport
response = self.session.get(
f"{self.base_url}/{path}/rendered",
headers={"Accept": media_type},
params=params,
)
response.raise_for_status()
return response.content
# --- STOW-RS ---
def store(self, dicom_files: list[str],
study_uid: str = None) -> dict:
"""Upload DICOM files via STOW-RS."""
boundary = f"boundary-{uuid.uuid4().hex}"
url = f"{self.base_url}/studies"
if study_uid:
url = f"{url}/{study_uid}"
body = b""
for filepath in dicom_files:
data = Path(filepath).read_bytes()
body += f"--{boundary}\r\n".encode()
body += b"Content-Type: application/dicom\r\n\r\n"
body += data
body += b"\r\n"
body += f"--{boundary}--\r\n".encode()
response = self.session.post(
url,
headers={
"Content-Type": (
f'multipart/related; type="application/dicom"; '
f"boundary={boundary}"
),
"Accept": "application/dicom+json",
},
data=body,
)
response.raise_for_status()
return response.json() if response.content else {}
# --- Helpers ---
def _build_path(self, study_uid: str, series_uid: str = None,
sop_uid: str = None) -> str:
path = f"studies/{study_uid}"
if series_uid:
path = f"{path}/series/{series_uid}"
if sop_uid:
path = f"{path}/instances/{sop_uid}"
return path
# Usage
client = DICOMwebClient(
base_url="https://your-server.com/dicomweb",
auth_token="your-token",
)
# Search
studies = client.search_studies(PatientName="Smith*", ModalitiesInStudy="CT")
# Retrieve metadata
metadata = client.retrieve_metadata(study_uid="1.2.3...")
# Download a rendered image
png = client.retrieve_rendered(
study_uid="1.2.3...",
series_uid="1.2.3...",
sop_uid="1.2.3...",
window="40,400",
)
Path("output.png").write_bytes(png)
# Upload
result = client.store(["image1.dcm", "image2.dcm"])
"Plugins": ["libOrthancDicomWeb.so"]http://localhost:8042/dicom-weborthanc:orthanc)http://localhost:8080/dcm4chee-arc/aets/DCM4CHEE/rshttps://healthcare.googleapis.com/v1/projects/{project}/locations/{location}/datasets/{dataset}/dicomStores/{store}/dicomWeb| Operation | Request Accept/Content-Type | Response Content-Type |
|---|---|---|
| QIDO-RS | Accept: application/dicom+json | application/dicom+json |
| WADO-RS (metadata) | Accept: application/dicom+json | application/dicom+json |
| WADO-RS (instance) | Accept: application/dicom | application/dicom or multipart/related |
type= parameter: multipart/related; type="application/dicom"; boundary=xxx. Missing the type parameter causes many servers to reject the request."00100010" not "PatientName". Use a lookup table or the DICOM standard browser.{"Alphabetic": "Family^Given"}, not plain strings.* as wildcard (not or ). Case sensitivity varies by server.Weekly Installs
0
Repository
GitHub Stars
6
First Seen
Jan 1, 1970
Security Audits
lark-cli 共享规则:飞书资源操作指南与权限配置详解
15,700 周安装
StudyDate20250101-20250131 |
| Date range |
ModalitiesInStudy | CT | Filter by modality |
StudyInstanceUID | 1.2.3... | Exact UID match |
AccessionNumber | ACC001 | RIS accession number |
StudyDescription | *CHEST* | Wildcard in description |
limit | 25 | Max results per page |
offset | 50 | Skip first N results |
includefield | all | Return all fields (or specify tags) |
| WADO-RS (rendered) | Accept: image/png or image/jpeg | image/png or image/jpeg |
| STOW-RS (DICOM) | Content-Type: multipart/related; type="application/dicom" | application/dicom+json |
| STOW-RS (JSON) | Content-Type: multipart/related; type="application/dicom+json" | application/dicom+json |
%?-: StudyDate=20250101-20250131, not StudyDate>=20250101&StudyDate<=20250131.limit/offset, others use Link headers. Check your server's documentation.