Mercurial > hg > orthanc-book
changeset 1262:a737d4cca0d3
python storage sample
| author | Alain Mazy <am@orthanc.team> |
|---|---|
| date | Tue, 10 Feb 2026 15:18:53 +0100 |
| parents | 6775156f8493 |
| children | 57976f197f2a |
| files | Sphinx/source/plugins/python.rst Sphinx/source/plugins/python/storage-area-3.py |
| diffstat | 2 files changed, 135 insertions(+), 2 deletions(-) [+] |
line wrap: on
line diff
--- 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 @@ <https://en.wikipedia.org/wiki/Object_storage>`__ technology available in Python (such as `Ceph <https://en.wikipedia.org/wiki/Ceph_(software)>`__ 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 <object-storage>`, the :ref:`PostgreSQL <postgresql>` and the :ref:`MySQL <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
--- /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
