From 00f84322ab85db7333d064337b9411ef242d0dcc Mon Sep 17 00:00:00 2001 From: Stephan Vedder Date: Mon, 26 Oct 2026 09:48:52 +0100 Subject: [PATCH] feat(mpps): add service for modality performed procedure step Validated against dcm4chee, which is the only open MPPS implementation i know about --- CHANGELOG.md | 4 + Cargo.lock | 1 + Cargo.toml | 1 + README.md | 19 +++ docs/topics/configuration.md | 7 + src/api/mod.rs | 4 +- src/api/mpps/mod.rs | 5 + src/api/mpps/routes.rs | 120 +++++++++++++++ src/api/mpps/service.rs | 230 +++++++++++++++++++++++++++++ src/backend/dimse/mod.rs | 3 + src/backend/dimse/mpps/mod.rs | 235 ++++++++++++++++++++++++++++++ src/backend/dimse/mpps/scu.rs | 72 +++++++++ src/backend/dimse/mpps/service.rs | 178 ++++++++++++++++++++++ src/backend/mod.rs | 8 + src/config/mod.rs | 14 ++ src/utils/dicom_json.rs | 45 ++++++ src/utils/mod.rs | 1 + tests/common/mod.rs | 2 +- tests/fixtures/mpps_scp.py | 79 ++++++++++ tests/mpps.rs | 167 +++++++++++++++++++++ 20 files changed, 1193 insertions(+), 2 deletions(-) create mode 100644 src/api/mpps/mod.rs create mode 100644 src/api/mpps/routes.rs create mode 100644 src/api/mpps/service.rs create mode 100644 src/backend/dimse/mpps/mod.rs create mode 100644 src/backend/dimse/mpps/scu.rs create mode 100644 src/backend/dimse/mpps/service.rs create mode 100644 src/utils/dicom_json.rs create mode 100644 tests/fixtures/mpps_scp.py create mode 100644 tests/mpps.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 766dabc..398c814 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- MPPS-RS (Modality Performed Procedure Step) support for the DIMSE backend: the Create, Update and + Retrieve transactions are translated into N-CREATE, N-SET and N-GET requests. The timeout can be + configured with the new `mpps-rs.timeout` option. + ### Changed ### Fixed diff --git a/Cargo.lock b/Cargo.lock index ade1275..97e066f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1723,6 +1723,7 @@ dependencies = [ "mimalloc", "mime", "multer", + "reqwest 0.13.3", "sentry", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 9215765..75603a2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,6 +65,7 @@ aws-credential-types = { version = "1.2.13", optional = true } testcontainers = "0.27.3" dicom-test-files = "0.4.0" dicom-web = "0.6.0" +reqwest = { version = "0.13.2", default-features = false, features = ["json"] } [lints.rust] unsafe_code = "forbid" diff --git a/README.md b/README.md index 22adc2e..8d3e8d8 100644 --- a/README.md +++ b/README.md @@ -177,6 +177,25 @@ https://www.dicomstandard.org/news-dir/current/docs/sups/sup246.pdf | limit | Return only {n} results | ✅ | | offset | Skip {n} results | ✅ | +### Report Modality Performed Procedure Steps (MPPS-RS) + +https://dicom.nema.org/medical/dicom/current/output/chtml/part18/chapter_15.html + +The DIMSE backend translates the transactions into N-CREATE, N-SET and N-GET requests of the +Modality Performed Procedure Step SOP Classes. + +#### Transactions + +| Transaction | Method | URI Template | Support Status | +| ----------- | ------ | -------------------------------------------------------------- | :------------: | +| Create | POST | `/modality-performed-procedure-steps/{mppsUID}` | ✅ | +| Update | POST | `/modality-performed-procedure-steps/{mppsUID}?update` | ✅ | +| Retrieve | GET | `/modality-performed-procedure-steps/{mppsUID}{?includefield}` | ✅¹ | + +¹ Requires a PACS that supports the Modality Performed Procedure Step Retrieve SOP Class. + +Only the `application/dicom+json` media type is supported. + ### Manage worklist items (UPS-RS) https://www.dicomstandard.org/using/dicomweb/workflow-ups-rs diff --git a/docs/topics/configuration.md b/docs/topics/configuration.md index 547443f..ef1301c 100644 --- a/docs/topics/configuration.md +++ b/docs/topics/configuration.md @@ -197,6 +197,8 @@ aets: timeout: 3000 stow-rs: timeout: 3000 + mpps-rs: + timeout: 3000 ``` Each AET (regardless of the backend) has additional settings specific to the DICOMweb endpoints. @@ -236,6 +238,11 @@ Each AET (regardless of the backend) has additional settings specific to the DIC This is the timeout for a single operation (e.g. receiving a DIMSE-C response primitive). If you want to set a timeout for the total execution time, use the server.http.request-timeout option instead. + + How many milliseconds to wait until a MPPS-RS request should time out. + This is the timeout for a single operation (e.g. receiving a N-CREATE-RSP, N-SET-RSP or N-GET-RSP). + Defaults to 30000. + ## S3 Backend Config diff --git a/src/api/mod.rs b/src/api/mod.rs index 657f1db..ae3f4a0 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -12,6 +12,7 @@ use serde::{Deserialize, Deserializer}; mod aets; mod home; +pub mod mpps; pub mod mwl; pub mod qido; pub mod stow; @@ -27,7 +28,8 @@ pub fn routes(base_path: &str) -> Router { .merge(qido::routes()) .merge(wado::routes()) .merge(stow::routes()) - .merge(mwl::routes()), + .merge(mwl::routes()) + .merge(mpps::routes()), ); // axum no longer supports nesting at the root diff --git a/src/api/mpps/mod.rs b/src/api/mpps/mod.rs new file mode 100644 index 0000000..f87ca45 --- /dev/null +++ b/src/api/mpps/mod.rs @@ -0,0 +1,5 @@ +mod routes; +mod service; + +pub use routes::routes; +pub use service::*; diff --git a/src/api/mpps/routes.rs b/src/api/mpps/routes.rs new file mode 100644 index 0000000..a7c0ba9 --- /dev/null +++ b/src/api/mpps/routes.rs @@ -0,0 +1,120 @@ +use crate::api::mpps::{ + MppsCreateRequest, MppsError, MppsRetrieveQueryParameters, MppsRetrieveRequest, + MppsUpdateRequest, +}; +use crate::backend::ServiceProvider; +use crate::utils::dicom_json::DicomJsonBody; +use crate::AppState; +use axum::body::Body; +use axum::extract::{Path, RawQuery}; +use axum::http::header::CONTENT_TYPE; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use axum::routing::post; +use axum::Router; +use axum_extra::extract::Query; +use dicom_json::DicomJson; +use tracing::instrument; + +const APPLICATION_DICOM_JSON: &str = "application/dicom+json"; + +/// HTTP Router for the Modality Performed Procedure Step Service. +/// +pub fn routes() -> Router { + Router::new().route( + "/modality-performed-procedure-steps/{mpps_uid}", + post(create_or_update).get(retrieve), + ) +} + +/// The Create and Update Transaction share the same method and resource, but the Update +/// Transaction requires the `update` query parameter, which has no value. +/// This cannot be deserialized into an `Option`, as an empty value deserializes to `None`. +fn is_update(query: Option<&str>) -> bool { + query.is_some_and(|query| { + query + .split('&') + .any(|parameter| parameter.split('=').next() == Some("update")) + }) +} + +fn disabled() -> Response { + ( + StatusCode::SERVICE_UNAVAILABLE, + "MPPS-RS endpoint is disabled", + ) + .into_response() +} + +/// Create Transaction and Update Transaction. +/// +/// +#[instrument(skip_all)] +async fn create_or_update( + provider: ServiceProvider, + Path((_aet, mpps_uid)): Path<(String, String)>, + RawQuery(query): RawQuery, + DicomJsonBody(data_set): DicomJsonBody, +) -> Response { + let Some(mpps) = provider.mpps else { + return disabled(); + }; + + let result = if is_update(query.as_deref()) { + match MppsUpdateRequest::new(mpps_uid, data_set) { + Ok(request) => mpps.update(request).await.map(|()| StatusCode::OK), + Err(err) => Err(err), + } + } else { + match MppsCreateRequest::new(mpps_uid, data_set) { + Ok(request) => mpps.create(request).await.map(|()| StatusCode::CREATED), + Err(err) => Err(err), + } + }; + + result.map_or_else(MppsError::into_response, IntoResponse::into_response) +} + +/// Retrieve Transaction. +/// +#[instrument(skip_all)] +async fn retrieve( + provider: ServiceProvider, + Path((_aet, mpps_uid)): Path<(String, String)>, + Query(parameters): Query, +) -> Response { + let Some(mpps) = provider.mpps else { + return disabled(); + }; + + let request = match MppsRetrieveRequest::new(mpps_uid, parameters) { + Ok(request) => request, + Err(err) => return err.into_response(), + }; + + match mpps.retrieve(request).await { + Ok(data_set) => Response::builder() + .status(StatusCode::OK) + .header(CONTENT_TYPE, APPLICATION_DICOM_JSON) + .body(Body::from( + serde_json::to_string(&DicomJson::from(data_set)).unwrap(), + )) + .unwrap(), + Err(err) => err.into_response(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn detects_update_parameter() { + assert!(is_update(Some("update"))); + assert!(is_update(Some("update="))); + assert!(is_update(Some("foo=bar&update"))); + assert!(!is_update(None)); + assert!(!is_update(Some(""))); + assert!(!is_update(Some("updated"))); + } +} diff --git a/src/api/mpps/service.rs b/src/api/mpps/service.rs new file mode 100644 index 0000000..e161d1f --- /dev/null +++ b/src/api/mpps/service.rs @@ -0,0 +1,230 @@ +use crate::api::{deserialize_includefield, IncludeField}; +use crate::types::UI; +use async_trait::async_trait; +use axum::body::Body; +use axum::http::{Response, StatusCode}; +use axum::response::IntoResponse; +use dicom::dictionary_std::tags; +use dicom::object::mem::InMemElement; +use dicom::object::InMemDicomObject; +use serde::Deserialize; +use thiserror::Error; + +/// The only value of Performed Procedure Step Status (0040,0252) that is allowed when an MPPS +/// is created. +/// +const STATUS_IN_PROGRESS: &str = "IN PROGRESS"; + +/// Provides the functionality of the Modality Performed Procedure Step service. +/// +#[async_trait] +pub trait MppsService: Send + Sync { + /// Create Transaction. + /// + async fn create(&self, request: MppsCreateRequest) -> Result<(), MppsError>; + + /// Update Transaction. + /// + async fn update(&self, request: MppsUpdateRequest) -> Result<(), MppsError>; + + /// Retrieve Transaction. + /// + async fn retrieve(&self, request: MppsRetrieveRequest) -> Result; +} + +pub struct MppsCreateRequest { + pub mpps_uid: UI, + pub data_set: InMemDicomObject, +} + +impl MppsCreateRequest { + pub fn new(mpps_uid: UI, data_set: InMemDicomObject) -> Result { + let data_set = validate_data_set(&mpps_uid, data_set)?; + + let status = data_set + .get(tags::PERFORMED_PROCEDURE_STEP_STATUS) + .map(InMemElement::to_str) + .and_then(Result::ok); + if status.as_deref() != Some(STATUS_IN_PROGRESS) { + return Err(MppsError::InvalidRequest(format!( + "Performed Procedure Step Status (0040,0252) must be {STATUS_IN_PROGRESS}" + ))); + } + + Ok(Self { mpps_uid, data_set }) + } +} + +pub struct MppsUpdateRequest { + pub mpps_uid: UI, + pub data_set: InMemDicomObject, +} + +impl MppsUpdateRequest { + pub fn new(mpps_uid: UI, data_set: InMemDicomObject) -> Result { + let data_set = validate_data_set(&mpps_uid, data_set)?; + Ok(Self { mpps_uid, data_set }) + } +} + +pub struct MppsRetrieveRequest { + pub mpps_uid: UI, + pub include_field: IncludeField, +} + +impl MppsRetrieveRequest { + pub fn new(mpps_uid: UI, parameters: MppsRetrieveQueryParameters) -> Result { + validate_uid(&mpps_uid)?; + Ok(Self { + mpps_uid, + include_field: parameters.include_field, + }) + } +} + +/// Query parameters for the Retrieve Transaction. +/// +#[derive(Debug, Default, Deserialize, PartialEq, Eq)] +#[serde(default)] +pub struct MppsRetrieveQueryParameters { + #[serde(rename = "includefield")] + #[serde(deserialize_with = "deserialize_includefield")] + pub include_field: IncludeField, +} + +/// Validates the MPPS UID of the resource path and removes the SOP Instance UID from the data +/// set, as it is conveyed in the command set of the N-CREATE-RQ and N-SET-RQ instead. +fn validate_data_set( + mpps_uid: &str, + mut data_set: InMemDicomObject, +) -> Result { + validate_uid(mpps_uid)?; + + if let Ok(element) = data_set.take_element(tags::SOP_INSTANCE_UID) { + let sop_instance_uid = element.to_str().unwrap_or_default(); + if sop_instance_uid != mpps_uid { + return Err(MppsError::InvalidRequest(format!( + "SOP Instance UID (0008,0018) {sop_instance_uid} does not match the MPPS UID {mpps_uid}" + ))); + } + } + + Ok(data_set) +} + +/// Checks if `uid` is a valid UID, see PS3.5 Section 9.1. +/// +fn validate_uid(uid: &str) -> Result<(), MppsError> { + let is_valid = !uid.is_empty() + && uid.len() <= 64 + && uid.split('.').all(|component| { + !component.is_empty() + && component.bytes().all(|byte| byte.is_ascii_digit()) + && (component == "0" || !component.starts_with('0')) + }); + + if is_valid { + Ok(()) + } else { + Err(MppsError::InvalidRequest(format!("Invalid MPPS UID {uid}"))) + } +} + +#[derive(Debug, Error)] +pub enum MppsError { + #[error("{0}")] + InvalidRequest(String), + #[error("A Modality Performed Procedure Step with UID {0} already exists")] + AlreadyExists(UI), + #[error("Modality Performed Procedure Step {0} not found")] + NotFound(UI), + #[error("Modality Performed Procedure Step {0} cannot be updated: {1}")] + Conflict(UI, String), + #[error(transparent)] + Backend(#[from] anyhow::Error), +} + +impl IntoResponse for MppsError { + fn into_response(self) -> Response { + let status = match &self { + Self::InvalidRequest(_) => StatusCode::BAD_REQUEST, + Self::AlreadyExists(_) | Self::Conflict(..) => StatusCode::CONFLICT, + Self::NotFound(_) => StatusCode::NOT_FOUND, + Self::Backend(_) => StatusCode::SERVICE_UNAVAILABLE, + }; + + Response::builder() + .status(status) + .body(Body::from(format!("{self:#}"))) + .unwrap() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::extract::Query; + use axum::http::Uri; + use dicom::core::{DataElement, VR}; + use dicom::dicom_value; + + const MPPS_UID: &str = "1.2.826.0.1.3680043.8.498.1"; + + fn data_set(status: &str) -> InMemDicomObject { + InMemDicomObject::from_element_iter([DataElement::new( + tags::PERFORMED_PROCEDURE_STEP_STATUS, + VR::CS, + dicom_value!(Str, status), + )]) + } + + #[test] + fn validates_uids() { + assert!(validate_uid(MPPS_UID).is_ok()); + assert!(validate_uid("2.25.0").is_ok()); + for uid in ["", "1..2", "1.02", "1.2.", "1.2a", &"1".repeat(65)] { + assert!(validate_uid(uid).is_err(), "{uid} should be invalid"); + } + } + + #[test] + fn create_requires_in_progress_status() { + assert!(MppsCreateRequest::new(MPPS_UID.into(), data_set("IN PROGRESS")).is_ok()); + assert!(MppsCreateRequest::new(MPPS_UID.into(), data_set("COMPLETED")).is_err()); + assert!(MppsCreateRequest::new(MPPS_UID.into(), InMemDicomObject::new_empty()).is_err()); + } + + #[test] + fn sop_instance_uid_must_match_mpps_uid() { + let mut matching = data_set("IN PROGRESS"); + matching.put(DataElement::new( + tags::SOP_INSTANCE_UID, + VR::UI, + dicom_value!(Str, MPPS_UID), + )); + let request = MppsCreateRequest::new(MPPS_UID.into(), matching).unwrap(); + assert!(request.data_set.get(tags::SOP_INSTANCE_UID).is_none()); + + let mut mismatching = data_set("COMPLETED"); + mismatching.put(DataElement::new( + tags::SOP_INSTANCE_UID, + VR::UI, + dicom_value!(Str, "1.2.3"), + )); + assert!(MppsUpdateRequest::new(MPPS_UID.into(), mismatching).is_err()); + } + + #[test] + fn parse_retrieve_query_params() { + let uri = + Uri::from_static("http://test?includefield=PatientID,PerformedProcedureStepStatus"); + let Query(params) = Query::::try_from_uri(&uri).unwrap(); + assert_eq!( + params.include_field, + IncludeField::List(vec![ + tags::PATIENT_ID, + tags::PERFORMED_PROCEDURE_STEP_STATUS + ]) + ); + } +} diff --git a/src/backend/dimse/mod.rs b/src/backend/dimse/mod.rs index 283479f..cc4f701 100644 --- a/src/backend/dimse/mod.rs +++ b/src/backend/dimse/mod.rs @@ -4,6 +4,8 @@ //! It depends on a store service class provider that must run in the background. //! - STOR-RS is implemented as a store service class user (C-STORE service). //! - MWL-RS is implemented as a find service class user (C-FIND service). +//! - MPPS-RS is implemented as a modality performed procedure step service class user +//! (N-CREATE, N-SET and N-GET services). //! mod cfind; @@ -11,6 +13,7 @@ pub mod cmove; mod cstore; pub mod association; +pub mod mpps; pub mod mwl; pub mod qido; pub mod stow; diff --git a/src/backend/dimse/mpps/mod.rs b/src/backend/dimse/mpps/mod.rs new file mode 100644 index 0000000..ec52c94 --- /dev/null +++ b/src/backend/dimse/mpps/mod.rs @@ -0,0 +1,235 @@ +//! Modality Performed Procedure Step SOP Classes. +//! + +use crate::backend::dimse::{DicomMessage, ReadError, DATA_SET_EXISTS, DATA_SET_MISSING}; +use crate::types::{UI, US}; +use dicom::core::{DataElement, PrimitiveValue, Tag, VR}; +use dicom::dicom_value; +use dicom::dictionary_std::{tags, uids}; +use dicom::object::mem::InMemElement; +use dicom::object::InMemDicomObject; + +pub mod scu; +pub mod service; + +pub use service::DimseMppsService; + +// Magic numbers defined by the DICOM specification. +// https://dicom.nema.org/medical/dicom/current/output/chtml/part07/chapter_E.html +pub const COMMAND_FIELD_N_GET_REQUEST: US = 0x0110; +pub const COMMAND_FIELD_N_SET_REQUEST: US = 0x0120; +pub const COMMAND_FIELD_N_CREATE_REQUEST: US = 0x0140; + +/// N-CREATE-RQ for the Modality Performed Procedure Step SOP Class. +/// +pub struct NCreateRequest { + pub message_id: US, + pub sop_instance_uid: UI, + pub data_set: InMemDicomObject, +} + +impl From for DicomMessage { + #[rustfmt::skip] + fn from(request: NCreateRequest) -> Self { + let command = InMemDicomObject::command_from_element_iter([ + DataElement::new(tags::AFFECTED_SOP_CLASS_UID, VR::UI, dicom_value!(Str, uids::MODALITY_PERFORMED_PROCEDURE_STEP)), + DataElement::new(tags::COMMAND_FIELD, VR::US, dicom_value!(U16, [COMMAND_FIELD_N_CREATE_REQUEST])), + DataElement::new(tags::MESSAGE_ID, VR::US, dicom_value!(U16, [request.message_id])), + DataElement::new(tags::COMMAND_DATA_SET_TYPE, VR::US, dicom_value!(U16, [DATA_SET_EXISTS])), + DataElement::new(tags::AFFECTED_SOP_INSTANCE_UID, VR::UI, dicom_value!(Str, request.sop_instance_uid)), + ]); + + Self { + command, + data: Some(request.data_set), + presentation_context_id: None, + } + } +} + +/// N-SET-RQ for the Modality Performed Procedure Step SOP Class. +/// +pub struct NSetRequest { + pub message_id: US, + pub sop_instance_uid: UI, + pub data_set: InMemDicomObject, +} + +impl From for DicomMessage { + #[rustfmt::skip] + fn from(request: NSetRequest) -> Self { + let command = InMemDicomObject::command_from_element_iter([ + DataElement::new(tags::REQUESTED_SOP_CLASS_UID, VR::UI, dicom_value!(Str, uids::MODALITY_PERFORMED_PROCEDURE_STEP)), + DataElement::new(tags::COMMAND_FIELD, VR::US, dicom_value!(U16, [COMMAND_FIELD_N_SET_REQUEST])), + DataElement::new(tags::MESSAGE_ID, VR::US, dicom_value!(U16, [request.message_id])), + DataElement::new(tags::COMMAND_DATA_SET_TYPE, VR::US, dicom_value!(U16, [DATA_SET_EXISTS])), + DataElement::new(tags::REQUESTED_SOP_INSTANCE_UID, VR::UI, dicom_value!(Str, request.sop_instance_uid)), + ]); + + Self { + command, + data: Some(request.data_set), + presentation_context_id: None, + } + } +} + +/// N-GET-RQ for the Modality Performed Procedure Step Retrieve SOP Class. +/// +pub struct NGetRequest { + pub message_id: US, + pub sop_instance_uid: UI, + /// The attributes to retrieve. All attributes are retrieved if empty. + pub attributes: Vec, +} + +impl From for DicomMessage { + #[rustfmt::skip] + fn from(request: NGetRequest) -> Self { + let mut command = InMemDicomObject::command_from_element_iter([ + DataElement::new(tags::REQUESTED_SOP_CLASS_UID, VR::UI, dicom_value!(Str, uids::MODALITY_PERFORMED_PROCEDURE_STEP_RETRIEVE)), + DataElement::new(tags::COMMAND_FIELD, VR::US, dicom_value!(U16, [COMMAND_FIELD_N_GET_REQUEST])), + DataElement::new(tags::MESSAGE_ID, VR::US, dicom_value!(U16, [request.message_id])), + DataElement::new(tags::COMMAND_DATA_SET_TYPE, VR::US, dicom_value!(U16, [DATA_SET_MISSING])), + DataElement::new(tags::REQUESTED_SOP_INSTANCE_UID, VR::UI, dicom_value!(Str, request.sop_instance_uid)), + ]); + + if !request.attributes.is_empty() { + command.put(DataElement::new( + tags::ATTRIBUTE_IDENTIFIER_LIST, + VR::AT, + PrimitiveValue::Tags(request.attributes.into_iter().collect()), + )); + } + + Self { + command, + data: None, + presentation_context_id: None, + } + } +} + +/// N-CREATE-RSP, N-SET-RSP or N-GET-RSP. +#[derive(Debug)] +pub struct NResponse { + pub status: US, + pub error_comment: Option, + pub error_id: Option, + pub data: Option, +} + +impl TryFrom for NResponse { + type Error = ReadError; + + fn try_from(message: DicomMessage) -> Result { + let status = message + .command + .get(tags::STATUS) + .map(InMemElement::to_int::) + .and_then(Result::ok) + .ok_or(ReadError::MissingAttribute(tags::STATUS))?; + + let error_comment = message + .command + .get(tags::ERROR_COMMENT) + .map(InMemElement::to_str) + .and_then(Result::ok) + .map(|comment| comment.trim().to_owned()); + + let error_id = message + .command + .get(tags::ERROR_ID) + .map(InMemElement::to_int::) + .and_then(Result::ok); + + Ok(Self { + status, + error_comment, + error_id, + data: message.data, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn command_str(message: &DicomMessage, tag: Tag) -> String { + message + .command + .get(tag) + .unwrap() + .to_str() + .unwrap() + .into_owned() + } + + #[test] + fn n_create_uses_affected_sop_instance_uid() { + let message = DicomMessage::from(NCreateRequest { + message_id: 1, + sop_instance_uid: UI::from("1.2.3"), + data_set: InMemDicomObject::new_empty(), + }); + assert_eq!( + command_str(&message, tags::AFFECTED_SOP_INSTANCE_UID), + "1.2.3" + ); + assert_eq!( + command_str(&message, tags::AFFECTED_SOP_CLASS_UID), + uids::MODALITY_PERFORMED_PROCEDURE_STEP + ); + assert!(message.data.is_some()); + } + + #[test] + fn n_set_uses_requested_sop_instance_uid() { + let message = DicomMessage::from(NSetRequest { + message_id: 1, + sop_instance_uid: UI::from("1.2.3"), + data_set: InMemDicomObject::new_empty(), + }); + assert_eq!( + command_str(&message, tags::REQUESTED_SOP_INSTANCE_UID), + "1.2.3" + ); + assert_eq!( + command_str(&message, tags::REQUESTED_SOP_CLASS_UID), + uids::MODALITY_PERFORMED_PROCEDURE_STEP + ); + } + + #[test] + fn n_get_only_lists_requested_attributes() { + let all = DicomMessage::from(NGetRequest { + message_id: 1, + sop_instance_uid: UI::from("1.2.3"), + attributes: Vec::new(), + }); + assert!(all.command.get(tags::ATTRIBUTE_IDENTIFIER_LIST).is_none()); + assert_eq!( + command_str(&all, tags::REQUESTED_SOP_CLASS_UID), + uids::MODALITY_PERFORMED_PROCEDURE_STEP_RETRIEVE + ); + + let some = DicomMessage::from(NGetRequest { + message_id: 1, + sop_instance_uid: UI::from("1.2.3"), + attributes: vec![tags::PATIENT_ID, tags::PERFORMED_PROCEDURE_STEP_STATUS], + }); + let attributes = some + .command + .get(tags::ATTRIBUTE_IDENTIFIER_LIST) + .unwrap() + .value() + .primitive() + .unwrap() + .to_owned(); + assert_eq!( + attributes, + PrimitiveValue::Tags([tags::PATIENT_ID, tags::PERFORMED_PROCEDURE_STEP_STATUS].into()) + ); + } +} diff --git a/src/backend/dimse/mpps/scu.rs b/src/backend/dimse/mpps/scu.rs new file mode 100644 index 0000000..ec42bff --- /dev/null +++ b/src/backend/dimse/mpps/scu.rs @@ -0,0 +1,72 @@ +use crate::backend::dimse::association; +use crate::backend::dimse::mpps::NResponse; +use crate::backend::dimse::{ + DicomMessage, DicomMessageReader, DicomMessageWriter, ReadError, WriteError, +}; +use crate::types::UI; +use association::pool::{AssociationPool, PoolError, PresentationParameter}; +use association::AssociationError; +use dicom::dictionary_std::uids; +use std::time::Duration; +use thiserror::Error; +use tracing::trace; + +pub struct MppsServiceClassUser { + pool: AssociationPool, + timeout: Duration, +} + +impl MppsServiceClassUser { + pub const fn new(pool: AssociationPool, timeout: Duration) -> Self { + Self { pool, timeout } + } + + /// Sends a single N-CREATE-RQ, N-SET-RQ or N-GET-RQ over an association that negotiated + /// the given SOP Class and returns its response. + #[allow(clippy::significant_drop_tightening)] + pub async fn invoke( + &self, + sop_class_uid: &str, + request: impl Into + Send, + ) -> Result { + let mut association = self + .pool + .get(PresentationParameter { + abstract_syntax_uid: UI::from(sop_class_uid), + transfer_syntax_uids: vec![UI::from(uids::IMPLICIT_VR_LITTLE_ENDIAN)], + }) + .await?; + + let written = association.write_message(request, None, self.timeout).await; + association.discard_on_err(written)?; + trace!("Sent request"); + + let read = association.read_message(self.timeout).await; + let message = association.discard_on_err(read)?; + trace!("Received response"); + + Ok(NResponse::try_from(message)?) + } +} + +#[derive(Debug, Error)] +pub enum MppsScuError { + #[error(transparent)] + Read(Box), + #[error(transparent)] + Write(Box), + #[error(transparent)] + Association(#[from] PoolError), +} + +impl From for MppsScuError { + fn from(err: ReadError) -> Self { + Self::Read(Box::new(err)) + } +} + +impl From for MppsScuError { + fn from(err: WriteError) -> Self { + Self::Write(Box::new(err)) + } +} diff --git a/src/backend/dimse/mpps/service.rs b/src/backend/dimse/mpps/service.rs new file mode 100644 index 0000000..09a020c --- /dev/null +++ b/src/backend/dimse/mpps/service.rs @@ -0,0 +1,178 @@ +use crate::api::mpps::{ + MppsCreateRequest, MppsError, MppsRetrieveRequest, MppsService, MppsUpdateRequest, +}; +use crate::api::IncludeField; +use crate::backend::dimse::association::pool::AssociationPool; +use crate::backend::dimse::mpps::scu::MppsServiceClassUser; +use crate::backend::dimse::mpps::{NCreateRequest, NGetRequest, NResponse, NSetRequest}; +use crate::backend::dimse::{next_message_id, StatusType}; +use async_trait::async_trait; +use dicom::dictionary_std::uids; +use dicom::object::InMemDicomObject; +use std::time::Duration; +use tracing::warn; + +/// Error ID (0000,0903) of the Processing Failure that is returned for an N-SET-RQ after the +/// Performed Procedure Step reached its final state. +/// +const ERROR_ID_MAY_NO_LONGER_BE_UPDATED: u16 = 0xA710; + +pub struct DimseMppsService { + scu: MppsServiceClassUser, +} + +impl DimseMppsService { + pub const fn new(pool: AssociationPool, timeout: Duration) -> Self { + let scu = MppsServiceClassUser::new(pool, timeout); + Self { scu } + } +} + +#[async_trait] +impl MppsService for DimseMppsService { + async fn create(&self, request: MppsCreateRequest) -> Result<(), MppsError> { + let response = self + .scu + .invoke( + uids::MODALITY_PERFORMED_PROCEDURE_STEP, + NCreateRequest { + message_id: next_message_id(), + sop_instance_uid: request.mpps_uid.clone(), + data_set: request.data_set, + }, + ) + .await + .map_err(|err| anyhow::Error::new(err).context("N-CREATE failed"))?; + + check_status(&response, &request.mpps_uid, "N-CREATE") + } + + async fn update(&self, request: MppsUpdateRequest) -> Result<(), MppsError> { + let response = self + .scu + .invoke( + uids::MODALITY_PERFORMED_PROCEDURE_STEP, + NSetRequest { + message_id: next_message_id(), + sop_instance_uid: request.mpps_uid.clone(), + data_set: request.data_set, + }, + ) + .await + .map_err(|err| anyhow::Error::new(err).context("N-SET failed"))?; + + check_status(&response, &request.mpps_uid, "N-SET") + } + + async fn retrieve(&self, request: MppsRetrieveRequest) -> Result { + let attributes = match request.include_field { + IncludeField::All => Vec::new(), + IncludeField::List(tags) => tags, + }; + + let response = self + .scu + .invoke( + uids::MODALITY_PERFORMED_PROCEDURE_STEP_RETRIEVE, + NGetRequest { + message_id: next_message_id(), + sop_instance_uid: request.mpps_uid.clone(), + attributes, + }, + ) + .await + .map_err(|err| anyhow::Error::new(err).context("N-GET failed"))?; + + check_status(&response, &request.mpps_uid, "N-GET")?; + response + .data + .ok_or_else(|| anyhow::Error::msg("N-GET-RSP did not contain a data set").into()) + } +} + +/// Maps the status of an N-CREATE-RSP, N-SET-RSP or N-GET-RSP to the status codes of the +/// MPPS transactions. +/// +/// +fn check_status(response: &NResponse, mpps_uid: &str, operation: &str) -> Result<(), MppsError> { + let detail = response.error_comment.as_ref().map_or_else( + || format!("{operation} failed with status 0x{:04X}", response.status), + |comment| { + format!( + "{operation} failed with status 0x{:04X}: {comment}", + response.status + ) + }, + ); + + match StatusType::try_from(response.status) { + Ok(StatusType::Success) => return Ok(()), + Ok(StatusType::Warning) => { + warn!( + "{operation} succeeded with warning status 0x{:04X}", + response.status + ); + return Ok(()); + } + _ => {} + } + + match (response.status, response.error_id) { + // Duplicate SOP Instance + (0x0111, _) => Err(MppsError::AlreadyExists(mpps_uid.to_owned())), + // No Such SOP Instance + (0x0112, _) => Err(MppsError::NotFound(mpps_uid.to_owned())), + // Processing Failure: Performed Procedure Step Object may no longer be updated + (0x0110, Some(ERROR_ID_MAY_NO_LONGER_BE_UPDATED)) => { + Err(MppsError::Conflict(mpps_uid.to_owned(), detail)) + } + // No Such Attribute, Invalid Attribute Value, Invalid Object Instance, + // Missing Attribute, Missing Attribute Value + (0x0105 | 0x0106 | 0x0117 | 0x0120 | 0x0121, _) => Err(MppsError::InvalidRequest(detail)), + _ => Err(MppsError::Backend(anyhow::Error::msg(detail))), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn response(status: u16) -> NResponse { + NResponse { + status, + error_comment: Some(String::from("comment")), + error_id: None, + data: None, + } + } + + #[test] + fn maps_dimse_status() { + assert!(check_status(&response(0x0000), "1.2", "N-SET").is_ok()); + assert!(check_status(&response(0x0116), "1.2", "N-SET").is_ok()); + assert!(matches!( + check_status(&response(0x0111), "1.2", "N-CREATE"), + Err(MppsError::AlreadyExists(_)) + )); + assert!(matches!( + check_status(&response(0x0112), "1.2", "N-SET"), + Err(MppsError::NotFound(_)) + )); + let final_state = NResponse { + error_id: Some(ERROR_ID_MAY_NO_LONGER_BE_UPDATED), + ..response(0x0110) + }; + assert!(matches!( + check_status(&final_state, "1.2", "N-SET"), + Err(MppsError::Conflict(..)) + )); + assert!(matches!( + check_status(&response(0x0120), "1.2", "N-CREATE"), + Err(MppsError::InvalidRequest(detail)) if detail.ends_with("comment") + )); + assert!(matches!( + check_status(&response(0x0110), "1.2", "N-SET"), + Err(MppsError::Backend(_)) + )); + } +} diff --git a/src/backend/mod.rs b/src/backend/mod.rs index 973e37c..94d05c2 100644 --- a/src/backend/mod.rs +++ b/src/backend/mod.rs @@ -1,3 +1,4 @@ +use crate::api::mpps::MppsService; use crate::api::mwl::MwlService; use crate::api::qido::QidoService; use crate::api::stow::StowService; @@ -20,6 +21,7 @@ pub struct ServiceProvider { pub wado: Option>, pub stow: Option>, pub mwl: Option>, + pub mpps: Option>, } impl FromRequestParts for ServiceProvider @@ -51,6 +53,7 @@ where // TODO: Use a singleton to avoid re-creating on every request. let provider = match ae_config.backend { BackendConfig::Dimse { .. } => { + use crate::backend::dimse::mpps::DimseMppsService; use crate::backend::dimse::mwl::DimseMwlService; use crate::backend::dimse::qido::DimseQidoService; use crate::backend::dimse::stow::DimseStowService; @@ -77,6 +80,10 @@ where pool.to_owned(), Duration::from_millis(ae_config.mwl.timeout), ))), + mpps: Some(Box::new(DimseMppsService::new( + pool.to_owned(), + Duration::from_millis(ae_config.mpps.timeout), + ))), } } #[cfg(feature = "s3")] @@ -88,6 +95,7 @@ where wado: Some(Box::new(S3WadoService::new(&config))), stow: None, mwl: None, + mpps: None, } } }; diff --git a/src/config/mod.rs b/src/config/mod.rs index d7047ef..907a2b3 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -31,6 +31,8 @@ pub struct ApplicationEntityConfig { pub stow: StowConfig, #[serde(default, rename = "mwl-rs")] pub mwl: MwlConfig, + #[serde(default, rename = "mpps-rs")] + pub mpps: MppsConfig, } #[derive(Debug, Clone, Deserialize)] @@ -195,6 +197,18 @@ impl Default for MwlConfig { } } +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub struct MppsConfig { + pub timeout: u64, +} + +impl Default for MppsConfig { + fn default() -> Self { + Self { timeout: 30_000 } + } +} + impl AppConfig { /// Loads the application configuration from the following sources: /// 1. Defaults (defined in `defaults.toml`) diff --git a/src/utils/dicom_json.rs b/src/utils/dicom_json.rs new file mode 100644 index 0000000..6315949 --- /dev/null +++ b/src/utils/dicom_json.rs @@ -0,0 +1,45 @@ +use axum::body::Bytes; +use axum::extract::{FromRequest, Request}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use dicom::object::InMemDicomObject; + +/// Extracts an [`InMemDicomObject`] from a request body encoded as DICOM JSON. +/// +pub struct DicomJsonBody(pub InMemDicomObject); + +pub enum DicomJsonBodyRejection { + InvalidBody(axum::extract::rejection::BytesRejection), + InvalidJson(serde_json::Error), +} + +impl IntoResponse for DicomJsonBodyRejection { + fn into_response(self) -> Response { + match self { + Self::InvalidBody(err) => (StatusCode::BAD_REQUEST, err.to_string()).into_response(), + Self::InvalidJson(err) => ( + StatusCode::BAD_REQUEST, + format!("Failed to parse DICOM JSON payload: {err}"), + ) + .into_response(), + } + } +} + +impl FromRequest for DicomJsonBody +where + S: Send + Sync, +{ + type Rejection = DicomJsonBodyRejection; + + async fn from_request(request: Request, state: &S) -> Result { + let bytes = Bytes::from_request(request, state) + .await + .map_err(DicomJsonBodyRejection::InvalidBody)?; + + let object: InMemDicomObject = + dicom_json::from_slice(&bytes).map_err(DicomJsonBodyRejection::InvalidJson)?; + + Ok(Self(object)) + } +} diff --git a/src/utils/mod.rs b/src/utils/mod.rs index 30ba0c2..fc2bc47 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -1 +1,2 @@ +pub mod dicom_json; pub mod multipart; diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 7e74ab9..611be84 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -29,7 +29,7 @@ pub struct ServerProcess { child: Child, stdout: Lines>, workdir: PathBuf, - http_port: u16, + pub http_port: u16, } impl ServerProcess { diff --git a/tests/fixtures/mpps_scp.py b/tests/fixtures/mpps_scp.py new file mode 100644 index 0000000..dde1a21 --- /dev/null +++ b/tests/fixtures/mpps_scp.py @@ -0,0 +1,79 @@ +"""Minimal MPPS SCP (PS3.4 Annex F) for the MPPS-RS integration tests. + +Keeps all Performed Procedure Steps in memory and supports N-CREATE, N-SET and N-GET. +""" +from pydicom.dataset import Dataset +from pynetdicom import AE, evt +from pynetdicom.sop_class import ( + ModalityPerformedProcedureStep, + ModalityPerformedProcedureStepRetrieve, +) + +steps = {} + + +def handle_create(event): + uid = event.request.AffectedSOPInstanceUID + if uid in steps: + return 0x0111, None # Duplicate SOP Instance + + data_set = event.attribute_list + if data_set.get("PerformedProcedureStepStatus") != "IN PROGRESS": + return 0x0106, None # Invalid Attribute Value + + data_set.SOPClassUID = ModalityPerformedProcedureStep + data_set.SOPInstanceUID = uid + steps[uid] = data_set + return 0x0000, data_set + + +def handle_set(event): + uid = event.request.RequestedSOPInstanceUID + if uid not in steps: + return 0x0112, None # No Such SOP Instance + + data_set = steps[uid] + if data_set.PerformedProcedureStepStatus != "IN PROGRESS": + # PS3.4 F.7.2.2.4 + status = Dataset() + status.Status = 0x0110 + status.ErrorComment = "Performed Procedure Step Object may no longer be updated" + status.ErrorID = 0xA710 + return status, None + + data_set.update(event.modification_list) + return 0x0000, data_set + + +def handle_get(event): + uid = event.request.RequestedSOPInstanceUID + if uid not in steps: + return 0x0112, None # No Such SOP Instance + + data_set = steps[uid] + attributes = event.request.AttributeIdentifierList + if not attributes: + return 0x0000, data_set + + if not isinstance(attributes, list): + attributes = [attributes] + subset = Dataset() + for tag in attributes: + if tag in data_set: + subset[tag] = data_set[tag] + return 0x0000, subset + + +ae = AE(ae_title="MPPS-SCP") +ae.add_supported_context(ModalityPerformedProcedureStep) +ae.add_supported_context(ModalityPerformedProcedureStepRetrieve) +print("MPPS SCP started", flush=True) +ae.start_server( + ("0.0.0.0", 11112), + block=True, + evt_handlers=[ + (evt.EVT_N_CREATE, handle_create), + (evt.EVT_N_SET, handle_set), + (evt.EVT_N_GET, handle_get), + ], +) diff --git a/tests/mpps.rs b/tests/mpps.rs new file mode 100644 index 0000000..f146421 --- /dev/null +++ b/tests/mpps.rs @@ -0,0 +1,167 @@ +mod common; + +use anyhow::Context; +use common::*; +use reqwest::{Client, StatusCode}; +use serde_json::{json, Value}; +use testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers::runners::AsyncRunner; +use testcontainers::{ContainerAsync, GenericImage, ImageExt}; + +/// Starts a minimal MPPS SCP, see `tests/fixtures/mpps_scp.py`. +async fn spawn_mpps_scp() -> anyhow::Result> { + GenericImage::new("python", "3.13-alpine") + .with_exposed_port(11112.tcp()) + .with_wait_for(WaitFor::message_on_stdout("MPPS SCP started")) + .with_copy_to( + "/mpps_scp.py", + include_bytes!("fixtures/mpps_scp.py").to_vec(), + ) + .with_cmd([ + "sh", + "-c", + "pip install -q --disable-pip-version-check --root-user-action=ignore pynetdicom \ + && python -u /mpps_scp.py", + ]) + .start() + .await + .context("failed to start MPPS SCP container") +} + +fn new_uid() -> String { + format!("2.25.{}", uuid::Uuid::new_v4().as_u128()) +} + +/// N-CREATE attributes, see PS3.4 Table F.7.2-1. +fn create_payload() -> Value { + json!({ + "00400270": { "vr": "SQ", "Value": [{ + "0020000D": { "vr": "UI", "Value": [new_uid()] }, + "00080050": { "vr": "SH", "Value": ["ACC-1"] }, + }]}, + "00100010": { "vr": "PN", "Value": [{ "Alphabetic": "MPPS^TEST" }] }, + "00100020": { "vr": "LO", "Value": ["MPPS-PAT-1"] }, + "00400253": { "vr": "SH", "Value": ["PPS-1"] }, + "00400241": { "vr": "AE", "Value": ["MODALITY1"] }, + "00400244": { "vr": "DA", "Value": ["20261026"] }, + "00400245": { "vr": "TM", "Value": ["093000"] }, + "00400252": { "vr": "CS", "Value": ["IN PROGRESS"] }, + "00080060": { "vr": "CS", "Value": ["CT"] }, + }) +} + +fn status_payload(status: &str) -> Value { + json!({ "00400252": { "vr": "CS", "Value": [status] } }) +} + +async fn post(http: &Client, url: &str, payload: &Value) -> anyhow::Result { + let response = http + .post(url) + .header("Content-Type", "application/dicom+json") + .header("Accept", "application/dicom+json") + .body(payload.to_string()) + .send() + .await?; + Ok(response.status()) +} + +async fn retrieve(http: &Client, url: &str) -> anyhow::Result<(StatusCode, Value)> { + let response = http.get(url).send().await?; + let status = response.status(); + let body = response.text().await?; + Ok((status, serde_json::from_str(&body).unwrap_or(Value::Null))) +} + +#[tokio::test] +async fn can_create_update_and_retrieve_mpps() -> anyhow::Result<()> { + let scp = spawn_mpps_scp().await?; + let scp_port = scp + .get_host_port_ipv4(11112.tcp()) + .await + .context("failed to get mapped MPPS SCP port")?; + + let server = spawn_dicomrst(&format!( + " + server: + http: + port: 0 + dimse: + - aet: DICOM-RST + interface: 0.0.0.0 + port: 0 + aets: + - aet: MPPS-SCP + host: 127.0.0.1 + port: {scp_port} + backend: DIMSE + " + )) + .await?; + let http = Client::new(); + let base_url = format!( + "http://localhost:{}/aets/MPPS-SCP/modality-performed-procedure-steps", + server.http_port + ); + let url = format!("{base_url}/{}", new_uid()); + + // Create Transaction + assert_eq!( + post(&http, &url, &create_payload()).await?, + StatusCode::CREATED + ); + assert_eq!( + post(&http, &url, &create_payload()).await?, + StatusCode::CONFLICT + ); + let mut completed = create_payload(); + completed["00400252"] = status_payload("COMPLETED")["00400252"].clone(); + assert_eq!( + post(&http, &format!("{base_url}/{}", new_uid()), &completed).await?, + StatusCode::BAD_REQUEST, + "An MPPS must be created IN PROGRESS" + ); + assert_eq!( + post(&http, &format!("{base_url}/1.02"), &create_payload()).await?, + StatusCode::BAD_REQUEST, + "The MPPS UID must be a valid UID" + ); + + // Retrieve Transaction + let (status, mpps) = retrieve(&http, &url).await?; + assert_eq!(status, StatusCode::OK); + assert_eq!(mpps["00100020"]["Value"][0], "MPPS-PAT-1"); + let (status, mpps) = retrieve(&http, &format!("{url}?includefield=00400252")).await?; + assert_eq!(status, StatusCode::OK); + assert_eq!( + mpps, + status_payload("IN PROGRESS"), + "Only the included field should be returned" + ); + let (status, _) = retrieve(&http, &format!("{base_url}/{}", new_uid())).await?; + assert_eq!(status, StatusCode::NOT_FOUND); + + // Update Transaction + let update_url = format!("{url}?update"); + assert_eq!( + post(&http, &update_url, &status_payload("COMPLETED")).await?, + StatusCode::OK + ); + let (_, mpps) = retrieve(&http, &url).await?; + assert_eq!(mpps["00400252"]["Value"][0], "COMPLETED"); + assert_eq!( + post(&http, &update_url, &status_payload("DISCONTINUED")).await?, + StatusCode::CONFLICT, + "A completed MPPS may no longer be updated" + ); + assert_eq!( + post( + &http, + &format!("{base_url}/{}?update", new_uid()), + &status_payload("COMPLETED") + ) + .await?, + StatusCode::NOT_FOUND + ); + + Ok(()) +}