imaging_api.services.projects ============================= .. py:module:: imaging_api.services.projects Attributes ---------- .. autoapisummary:: imaging_api.services.projects.XNAT_URL Functions --------- .. autoapisummary:: imaging_api.services.projects.get_project_from_central_hub_project_id imaging_api.services.projects.get_project imaging_api.services.projects.get_all_projects imaging_api.services.projects.create_payload_for_project_creation imaging_api.services.projects.create_project imaging_api.services.projects.to_create_project imaging_api.services.projects.set_project_prearchive_settings imaging_api.services.projects.get_command_info imaging_api.services.projects.create_project_event_subscription imaging_api.services.projects.add_central_hub_users_to_project imaging_api.services.projects.delete_queued_import_requests imaging_api.services.projects.delete_project imaging_api.services.projects.get_subjects imaging_api.services.projects.get_experiments imaging_api.services.projects.get_experiment imaging_api.services.projects.get_subject_id_from_experiment_response Module Contents --------------- .. py:data:: XNAT_URL :value: 'http://xnat-web:8080' .. py:function:: get_project_from_central_hub_project_id(central_hub_project_id: str, headers: dict[str, str]) -> imaging_api.routers.schemas.Project Gets the XNAT project from a central hub project ID (corresponds to XNAT project secondary ID) :param central_hub_project_id: Central hub project ID :type central_hub_project_id: str :param headers: XNAT authentication headers :type headers: dict[str, str] :returns: XNAT project object :rtype: Project :raises imaging_api.utils.exceptions.NotFoundError: If no XNAT project has a matching ``secondary_ID``. :raises Exception: If the upstream ``get_all_projects`` call fails. .. py:function:: get_project(project_id: str, headers: dict[str, str]) -> imaging_api.routers.schemas.Project Fetches a specific XNAT project by selecting from all projects. :param project_id: Unique identifier for the project :type project_id: str :param headers: XNAT authentication headers :type headers: dict[str, str] :returns: XNAT project object :rtype: Project :raises imaging_api.utils.exceptions.NotFoundError: If no XNAT project matches ``project_id``. :raises Exception: If the upstream ``get_all_projects`` call fails. .. py:function:: get_all_projects(headers: dict[str, str]) -> list[imaging_api.routers.schemas.Project] Fetches all XNAT projects using the correct REST API endpoint. :param headers: XNAT authentication headers :type headers: dict[str, str] :returns: List of XNAT project objects :rtype: list[Project] :raises Exception: If the HTTP request to XNAT fails, or if XNAT returns a non-200 response. .. py:function:: create_payload_for_project_creation(xnat_projects_uri: str, project_id: str, project_secondary_id: str, project_name: str, project_description: str = '') -> str Creates the payload for creating a new project in XNAT. Builds the XML using ``xml.etree.ElementTree`` so that XML control characters (``<``, ``>``, ``&``, ``"``, ``'``) in any field are escaped as entity references rather than interpolated raw — this defeats XML injection that could otherwise mutate the projectData document sent to XNAT. :param xnat_projects_uri: XNAT projects URI. :type xnat_projects_uri: str :param project_id: Unique identifier for the project. :type project_id: str :param project_secondary_id: Secondary ID for the project. :type project_secondary_id: str :param project_name: Name of the project. :type project_name: str :param project_description: Description of the project. :type project_description: str, optional :returns: XML payload for creating the project. :rtype: str .. py:function:: create_project(project_id: str, project_secondary_id: str, project_name: str, project_description: str, headers: dict[str, str]) -> imaging_api.routers.schemas.Project Core function to create a new project in XNAT. Note that ID, secondary_ID and name are required. Uses the XNAT REST API endpoint: ``POST - /data/projects`` Note this endpoint only accepts XML payload, not JSON. See also https://wiki.xnat.org/xnat-api/project-api#ProjectAPI-Createoneormoreprojects :param project_id: Unique identifier for the project. :type project_id: str :param project_secondary_id: Secondary ID for the project. :type project_secondary_id: str :param project_name: Name of the project. :type project_name: str :param project_description: Description of the project. :type project_description: str :param headers: XNAT authentication headers. :type headers: dict[str, str] :returns: XNAT project object. :rtype: Project :raises AlreadyExistsError: If a project with the same ID already exists in XNAT. :raises Exception: If there is an error during the creation of the project. .. py:function:: to_create_project(imaging_project: imaging_api.routers.schemas.CentralHubProject) -> imaging_api.routers.schemas.CreateProject Maps Central Hub project information to XNAT project input to make a request to create a project. :param imaging_project: Central Hub project object. :type imaging_project: imaging_api.routers.schemas.CentralHubProject :returns: XNAT create project request object. :rtype: CreateProject .. py:function:: set_project_prearchive_settings(project_id: str, headers: dict[str, str]) -> None Set Project Prearchive Settings. See also https://wiki.xnat.org/xnat-api/prearchive-api#PrearchiveAPI-SetProjectPrearchiveSettings :param project_id: Unique identifier for the project :type project_id: str :param headers: XNAT authentication headers :type headers: dict[str, str] :returns: None :raises Exception: If there is an error during the process of setting the project prearchive settings. .. py:function:: get_command_info(container: str, headers: dict[str, str]) -> tuple[int, str] Fetches the XNAT command ID and wrapper name for a given container image. :param container: Container image name, e.g. "xnat/dcm2niix:latest". :type container: str :param headers: XNAT authentication headers. :type headers: dict[str, str] :returns: A tuple of (command_id, wrapper_name). :rtype: tuple[int, str] :raises Exception: If the command cannot be fetched from XNAT. .. py:function:: create_project_event_subscription(project_id: str, container: str, active: bool, headers: dict[str, str]) -> None Creates a project-scoped event subscription in XNAT that auto-triggers a command on scan upload. The subscription listens for ScanEvent:CREATED events within the specified project and triggers the given container command when a scan with DICOM resources is created. The active flag controls whether the subscription is enabled or deactivated on creation. :param project_id: XNAT project ID to scope the subscription to. :type project_id: str :param container: Container image name, e.g. "xnat/dcm2niix:latest". :type container: str :param headers: XNAT authentication headers. :type headers: dict[str, str] :param active: If True, the subscription is active immediately. If False, it is created but deactivated (can be toggled later via the XNAT API). :type active: bool :raises Exception: If the subscription creation fails. .. py:function:: add_central_hub_users_to_project(central_hub_project: imaging_api.routers.schemas.CentralHubProject, project_id: str, headers: dict[str, str]) -> tuple[list[imaging_api.routers.schemas.CreatedUser], list[imaging_api.routers.schemas.User]] Adds list of central hub users to an imaging project on XNAT. Note users that are disabled will not be created or added to the XNAT project. TODO reassess this decision. :param central_hub_project: Central Hub project object :type central_hub_project: imaging_api.routers.schemas.CentralHubProject :param project_id: Unique identifier for the project :type project_id: str :param headers: XNAT authentication headers :type headers: dict[str, str] :returns: List of created users and added users. :rtype: tuple[list[imaging_api.routers.schemas.CreatedUser], list[imaging_api.routers.schemas.User]] .. py:function:: delete_queued_import_requests(project_id: str, headers: dict[str, str]) -> bool :async: Deletes queued import requests from PACS for a specific project. Does not raise an exception if it fails to delete the queued imports. :param project_id: Unique identifier for the project :type project_id: str :param headers: XNAT authentication headers :type headers: dict[str, str] :returns: True if deletion was successful, False otherwise :rtype: bool .. py:function:: delete_project(project_id: str, headers: dict[str, str]) -> imaging_api.routers.schemas.Project :async: Deletes an existing project in XNAT. :param project_id: Unique identifier for the project :type project_id: str :param headers: XNAT authentication headers :type headers: dict[str, str] :returns: XNAT project object :rtype: Project :raises imaging_api.utils.exceptions.NotFoundError: If the project does not exist on XNAT. :raises Exception: If XNAT returns a non-200 response for the delete call. .. py:function:: get_subjects(project_id: str, headers: dict[str, str]) -> list[imaging_api.routers.schemas.Subject] Retrieves a list of subjects in a specific project in XNAT. :param project_id: Unique identifier for the project. :type project_id: str :param headers: XNAT authentication headers. :type headers: dict[str, str] :returns: List of XNAT subject objects. :rtype: list[Subject] :raises Exception: If there is an error while fetching the subjects from XNAT. .. py:function:: get_experiments(project_id: str, headers: dict[str, str]) -> list[imaging_api.routers.schemas.Experiment] Fetches all XNAT experiments from a project. :param project_id: Unique identifier for the project. :type project_id: str :param headers: XNAT authentication headers. :type headers: dict[str, str] :returns: List of XNAT experiment objects. :rtype: list[Experiment] :raises Exception: If there is an error while fetching the experiments from XNAT. .. py:function:: get_experiment(project_id: str, experiment_id_or_label: str, headers: dict[str, str]) -> dict Fetches a specific XNAT experiment from a project. Note the XNAT Experiment API supports getting an experiment by either its label or ID: ``GET - /data/projects/{project-id}/experiments/{experiment-label | experiment-id}`` :param project_id: Unique identifier for the project. :type project_id: str :param experiment_id_or_label: Unique identifier or label for the experiment. :type experiment_id_or_label: str :param headers: XNAT authentication headers. :type headers: dict[str, str] :returns: XNAT experiment dictionary response :rtype: dict :raises imaging_api.utils.exceptions.NotFoundError: If the experiment with the given ID or label is not found in the :raises project.: :raises Exception: If there is an error during the fetch process. .. py:function:: get_subject_id_from_experiment_response(experiment_response: dict[str, Any]) -> str Extracts the XNAT subject ID from the XNAT experiment response JSON. :param experiment_response: XNAT experiment response JSON. :type experiment_response: dict[str, Any] :returns: Subject ID :rtype: str :raises Exception: If there is an error during the parsing of the experiment response JSON.