# HG changeset patch # User Alain Mazy # Date 1770733133 -3600 # Node ID a737d4cca0d377462639107018444cbd353642e4 # Parent 6775156f849399c67f4032b65a12cbeee440a3e6 python storage sample diff -r 6775156f8493 -r a737d4cca0d3 Sphinx/source/plugins/python.rst --- a/Sphinx/source/plugins/python.rst Tue Feb 10 14:56:37 2026 +0100 +++ b/Sphinx/source/plugins/python.rst Tue Feb 10 15:18:53 2026 +0100 @@ -870,12 +870,21 @@ `__ technology available in Python (such as `Ceph `__ or AWS S3-like -tools). The performance will not be as good as a C/C++ native plugin +tools). The performance will likely not be as good as a C/C++ native plugin (cf. the :ref:`cloud storage `, the :ref:`PostgreSQL ` and the :ref:`MySQL ` plugins), but it can be used for prototyping or for basic setups. -Here is a full, self-explaining sample: +Note that the signatures of these callbacks has changed between +version 3.3 and 7.1 (not released yet) + +Here is a full self-explaining sample for plugin from 7.1 + +.. literalinclude:: python/storage-area-3.py + :language: python + +Here is a full, self-explaining sample for plugin 3.3 and later +(still compatible with 7.1 but with reduced features): .. literalinclude:: python/storage-area.py :language: python diff -r 6775156f8493 -r a737d4cca0d3 Sphinx/source/plugins/python/storage-area-3.py --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/Sphinx/source/plugins/python/storage-area-3.py Tue Feb 10 15:18:53 2026 +0100 @@ -0,0 +1,124 @@ +# Compared to the default Orthanc storage, this storage plugin +# uses a flat folder structure but prefixes all file paths with +# the PatientID. This is a sample only and should not be used +# in production. + +import orthanc +from typing import Tuple, Optional +import json +import os +import re + +orthanc_full_configuration = json.loads(orthanc.GetConfiguration()) + +root_storage_directory = orthanc_full_configuration.get('StorageDirectory', orthanc_full_configuration.get('IndexDirectory', os.curdir)) + +orthanc.LogWarning(f"Python Storage: root directory: {root_storage_directory}") + +def normalize_filename(s: str, replacement_char: str = "_") -> str: + # Replace invalid characters with the replacement character + s = re.sub(r'[\\/*?:"<>|]', replacement_char, s) + # Replace spaces with the replacement character + s = re.sub(r'\s+', replacement_char, s) + # Remove leading/trailing replacement characters + s = s.strip(replacement_char) + # Ensure the string is not empty + if not s: + s = "undefined" + return s + +def get_relative_path(patient_id: str, uuid: str, content_type: orthanc.ContentType) -> str: + return f"{normalize_filename(patient_id)} - {uuid}{get_extension(content_type)}" + +def get_absolute_path(relative_path: str) -> str: + return os.path.join(root_storage_directory, relative_path) + +def get_extension(content_type: orthanc.ContentType): + if content_type == orthanc.ContentType.DICOM: + return ".dcm" + elif content_type == orthanc.ContentType.DICOM_AS_JSON: + return ".json" + elif content_type == orthanc.ContentType.DICOM_UNTIL_PIXEL_DATA: + return ".dcm.head" + else: + return ".unk" + +def StorageCreate(uuid: str, + content_type: orthanc.ContentType, + compression_type: orthanc.CompressionType, + content: bytes, + dicom_instance: orthanc.DicomInstance) -> Tuple[orthanc.ErrorCode, Optional[bytes]]: + # orthanc.LogInfo("---- StorageCreate") + + instance_tags = json.loads(dicom_instance.GetInstanceJson()) + patient_id = instance_tags["0010,0010"]["Value"] + try: + relative_path = get_relative_path(patient_id, uuid, content_type) + path = get_absolute_path(relative_path) + with open(path, "wb") as f: + f.write(content) + + # orthanc.LogInfo(f"Successfully created file {path} for {uuid}, wrote {len(content)} bytes") + str_custom_data = relative_path + + return orthanc.ErrorCode.SUCCESS, str_custom_data.encode('utf-8') + except IOError as e: + orthanc.LogError(f"Failed to create storage for {uuid}: {str(e)}") + return orthanc.ErrorCode.PLUGIN, None + except Exception as e: + orthanc.LogError(f"Unexpected error creating storage for {uuid}: {str(e)}") + return orthanc.ErrorCode.PLUGIN, None + +def StorageReadRange(uuid: str, + content_type: orthanc.ContentType, + range_start: int, + size: int, + custom_data: bytes) -> Tuple[orthanc.ErrorCode, Optional[bytes]]: + # orthanc.LogInfo("---- StorageReadRange") + str_custom_data = custom_data.decode('utf-8') + # orthanc.LogInfo(f"---- CustomData: {str_custom_data}") + try: + relative_path = str_custom_data + path = get_absolute_path(relative_path) + + with open(path, "rb") as f: + # Move to the start position + f.seek(range_start) + # Read the specified number of bytes + data = f.read(size) + + # orthanc.LogInfo(f"Read {len(data)} bytes from position {range_start}") + return orthanc.ErrorCode.SUCCESS, data + except FileNotFoundError: + orthanc.LogError(f"File {path} not found for {uuid}") + return orthanc.ErrorCode.UNKNOWN_RESOURCE, None + except Exception as e: + orthanc.LogError(f"Error reading file {path} for {uuid}: {str(e)}") + return orthanc.ErrorCode.PLUGIN, None + +def StorageRemove(uuid: str, + content_type: orthanc.ContentType, + custom_data: bytes) -> orthanc.ErrorCode: + # orthanc.LogInfo("---- StorageRemove") + + str_custom_data = custom_data.decode('utf-8') + # orthanc.LogInfo(f"---- CustomData: {str_custom_data}") + try: + relative_path = str_custom_data + path = get_absolute_path(relative_path) + if os.path.exists(path): + os.remove(path) + # orthanc.LogInfo(f"Successfully removed file {path} for {uuid}") + else: + orthanc.LogWarning(f"Storage file not found for removal: {path}") + + return orthanc.ErrorCode.SUCCESS + except IOError as e: + orthanc.LogError(f"Failed to remove {path} for {uuid}: {str(e)}") + return orthanc.ErrorCode.PLUGIN + except Exception as e: + orthanc.LogError(f"Unexpected error removing {path} for {uuid}: {str(e)}") + return orthanc.ErrorCode.PLUGIN + + +orthanc.RegisterStorageArea3(StorageCreate, StorageReadRange, StorageRemove) \ No newline at end of file