308 lines
10 KiB
Python
Executable File
308 lines
10 KiB
Python
Executable File
# Copyright 2008-2020 pydicom authors. See LICENSE file for details.
|
|
"""Hold DicomFile class, which does basic I/O for a dicom file."""
|
|
|
|
from io import BytesIO
|
|
import os
|
|
from struct import Struct
|
|
from types import TracebackType
|
|
from typing import TYPE_CHECKING, cast, Any, TypeVar, Protocol
|
|
|
|
if TYPE_CHECKING: # pragma: no cover
|
|
from collections.abc import Callable
|
|
|
|
|
|
ExitException = tuple[
|
|
type[BaseException] | None, BaseException | None, TracebackType | None
|
|
]
|
|
Self = TypeVar("Self", bound="DicomIO")
|
|
|
|
|
|
class ReadableBuffer(Protocol):
|
|
def read(self, size: int = ..., /) -> bytes: ... # pragma: no cover
|
|
|
|
def seek(self, offset: int, whence: int = ..., /) -> int: ... # pragma: no cover
|
|
|
|
def tell(self) -> int: ... # pragma: no cover
|
|
|
|
|
|
class WriteableBuffer(Protocol):
|
|
def seek(self, offset: int, whence: int = ..., /) -> int: ... # pragma: no cover
|
|
|
|
def tell(self) -> int: ... # pragma: no cover
|
|
|
|
def write(
|
|
self, b: bytes | bytearray | memoryview, /
|
|
) -> int: ... # pragma: no cover
|
|
|
|
|
|
class DicomIO:
|
|
"""Wrapper for managing buffer-like objects used when reading or writing
|
|
DICOM datasets.
|
|
"""
|
|
|
|
def __init__(self, buffer: ReadableBuffer | WriteableBuffer) -> None:
|
|
"""Create a new ``DicomIO`` instance.
|
|
|
|
Parameters
|
|
----------
|
|
buffer : buffer-like object
|
|
A buffer-like object that implements:
|
|
|
|
* ``seek()`` and ``tell()`` methods with the same signatures as
|
|
:meth:`io.IOBase.seek` and :meth:`io.IOBase.tell`
|
|
* a ``read()`` method with the same signature as
|
|
:meth:`io.RawIOBase.read` if it supports reading data from
|
|
itself, and/or
|
|
* a ``write()`` method with the same signature as
|
|
:meth:`io.RawIOBase.write` if it supports writing data to itself
|
|
|
|
If `buffer` supports reading it can be used with
|
|
:func:`~pydicom.filereader.dcmread` as the source to decode a DICOM
|
|
dataset from, and if it supports writing it can be used with
|
|
:func:`~pydicom.filewriter.dcmwrite` as the destination for the
|
|
encoded DICOM dataset.
|
|
"""
|
|
# Data packers/unpackers
|
|
self._us_unpacker: Callable[[bytes], tuple[Any, ...]]
|
|
self._us_packer: Callable[[int], bytes]
|
|
self._ul_unpacker: Callable[[bytes], tuple[Any, ...]]
|
|
self._ul_packer: Callable[[int], bytes]
|
|
self._tag_unpacker: Callable[[bytes], tuple[Any, ...]]
|
|
self._tag_packer: Callable[[int, int], bytes]
|
|
|
|
# Store the encoding method
|
|
self._implicit_vr: bool
|
|
self._little_endian: bool
|
|
|
|
# The buffer-like object being wrapped
|
|
self._buffer = buffer
|
|
|
|
# The filename associated with the buffer-like
|
|
self._name: str | None = getattr(self._buffer, "name", None)
|
|
|
|
# It's more efficient to replace the existing class methods
|
|
# instead of wrapping them
|
|
if hasattr(buffer, "read"):
|
|
self.read = buffer.read
|
|
|
|
if hasattr(buffer, "write"):
|
|
self.write = buffer.write
|
|
|
|
if hasattr(buffer, "close"):
|
|
self.close = buffer.close
|
|
|
|
# seek() and tell() are always required
|
|
self.seek = buffer.seek
|
|
self.tell = buffer.tell
|
|
|
|
def close(self, *args: Any, **kwargs: Any) -> Any:
|
|
"""Close the buffer (if possible)"""
|
|
pass
|
|
|
|
def __enter__(self: Self) -> Self:
|
|
return self
|
|
|
|
def __exit__(self, *exc_info: ExitException) -> None:
|
|
self.close()
|
|
|
|
@property
|
|
def is_little_endian(self) -> bool:
|
|
"""Get/set the endianness for encoding/decoding, ``True`` for little
|
|
endian and ``False`` for big endian.
|
|
"""
|
|
if not hasattr(self, "_little_endian"):
|
|
raise AttributeError(
|
|
f"{type(self).__name__}.is_little_endian' has not been set"
|
|
)
|
|
|
|
return self._little_endian
|
|
|
|
@is_little_endian.setter
|
|
def is_little_endian(self, value: bool) -> None:
|
|
if not isinstance(value, bool):
|
|
raise TypeError(f"'{type(self).__name__}.is_little_endian' must be bool")
|
|
|
|
self._little_endian = value
|
|
|
|
endianness = "><"[value]
|
|
self._us_packer = Struct(f"{endianness}H").pack
|
|
self._us_unpacker = Struct(f"{endianness}H").unpack
|
|
self._ul_packer = Struct(f"{endianness}L").pack
|
|
self._ul_unpacker = Struct(f"{endianness}L").unpack
|
|
self._tag_packer = Struct(f"{endianness}2H").pack
|
|
self._tag_unpacker = Struct(f"{endianness}2H").unpack
|
|
|
|
@property
|
|
def is_implicit_VR(self) -> bool:
|
|
"""Get/set the VR mode for encoding/decoding. ``True`` for implicit VR
|
|
and ``False`` for explicit VR.
|
|
"""
|
|
if not hasattr(self, "_implicit_vr"):
|
|
raise AttributeError(
|
|
f"{type(self).__name__}.is_implicit_VR' has not been set"
|
|
)
|
|
|
|
return self._implicit_vr
|
|
|
|
@is_implicit_VR.setter
|
|
def is_implicit_VR(self, value: bool) -> None:
|
|
if not isinstance(value, bool):
|
|
raise TypeError(f"'{type(self).__name__}.is_implicit_VR' must be bool")
|
|
|
|
self._implicit_vr = value
|
|
|
|
@property
|
|
def name(self) -> str | None:
|
|
"""Return the value of the :attr:`~pydicom.filebase.DicomIO.parent`'s
|
|
``name`` attribute, or ``None`` if no such attribute.
|
|
"""
|
|
return self._name
|
|
|
|
@name.setter
|
|
def name(self, name: str) -> None:
|
|
self._name = name
|
|
|
|
@property
|
|
def parent(self) -> ReadableBuffer | WriteableBuffer:
|
|
"""Return the buffer object being wrapped."""
|
|
return self._buffer
|
|
|
|
def read(self, size: int = -1, /) -> bytes:
|
|
"""Read up to `size` bytes from the buffer and return them. If `size`
|
|
is unspecified, all bytes until EOF are returned.
|
|
|
|
Fewer than `size` bytes may be returned if the operating system call
|
|
returns fewer than `size` bytes.
|
|
"""
|
|
raise TypeError(
|
|
f"'{type(self).__name__}' cannot be used with "
|
|
f"'{type(self._buffer).__name__}': object has no read() method"
|
|
)
|
|
|
|
def read_exact(self, length: int, nr_retries: int = 3) -> bytes:
|
|
"""Return `length` bytes read from the buffer.
|
|
|
|
Parameters
|
|
----------
|
|
length : int
|
|
The number of bytes to be read. If ``None`` (default) then read all
|
|
the bytes available.
|
|
nr_retries : int, optional
|
|
The number of tries to read data when the number of bytes read
|
|
from the buffer is less than `length`. Default ``3``.
|
|
|
|
Returns
|
|
-------
|
|
bytes
|
|
The read data.
|
|
|
|
Raises
|
|
------
|
|
EOFError
|
|
If unable to read `length` bytes.
|
|
"""
|
|
bytes_read = self.read(length)
|
|
if len(bytes_read) == length:
|
|
return bytes_read
|
|
|
|
# Use a bytearray because concatenating bytes is expensive
|
|
bytes_read = bytearray(bytes_read)
|
|
attempts = 0
|
|
while (num_bytes := len(bytes_read)) < length and attempts < nr_retries:
|
|
bytes_read += self.read(length - num_bytes)
|
|
attempts += 1
|
|
|
|
if num_bytes == length:
|
|
return bytes(bytes_read)
|
|
|
|
raise EOFError(
|
|
f"Unexpected end of file. Read {num_bytes} bytes of {length} "
|
|
f"expected starting at position 0x{self.tell() - num_bytes:x}"
|
|
)
|
|
|
|
def read_tag(self) -> tuple[int, int]:
|
|
"""Return a DICOM tag value read from the buffer."""
|
|
return cast(
|
|
tuple[int, int],
|
|
self._tag_unpacker(self.read_exact(4)),
|
|
)
|
|
|
|
def read_UL(self) -> int:
|
|
"""Return a UL value read from the buffer."""
|
|
return cast(int, self._ul_unpacker(self.read(4))[0])
|
|
|
|
def read_US(self) -> int:
|
|
"""Return a US value read from the buffer."""
|
|
return cast(int, self._us_unpacker(self.read(2))[0])
|
|
|
|
def seek(self, offset: int, whence: int = os.SEEK_SET, /) -> int:
|
|
"""Change the buffer position to the given byte `offset`, relative to
|
|
the position indicated by `whence` and return the new absolute position.
|
|
"""
|
|
raise NotImplementedError() # pragma: no cover
|
|
|
|
def tell(self) -> int:
|
|
"""Return the current stream position of the buffer"""
|
|
raise NotImplementedError() # pragma: no cover
|
|
|
|
def write(self, b: bytes | bytearray | memoryview, /) -> int:
|
|
"""Write the bytes-like object `b` to the buffer and return the number
|
|
of bytes written.
|
|
"""
|
|
raise TypeError(
|
|
f"'{type(self).__name__}' cannot be used with "
|
|
f"'{type(self._buffer).__name__}': object has no write() method"
|
|
)
|
|
|
|
def write_tag(self, tag: int) -> None:
|
|
"""Write a DICOM tag to the buffer."""
|
|
self.write(self._tag_packer(tag >> 16, tag & 0xFFFF))
|
|
|
|
def write_UL(self, val: int) -> None:
|
|
"""Write a UL value to the buffer."""
|
|
self.write(self._ul_packer(val))
|
|
|
|
def write_US(self, val: int) -> None:
|
|
"""Write a US value to the buffer."""
|
|
self.write(self._us_packer(val))
|
|
|
|
|
|
class DicomFileLike(DicomIO):
|
|
"""Wrapper for file-likes to simplify encoding/decoding DICOM datasets.
|
|
|
|
See Also
|
|
--------
|
|
:class:`~pydicom.filebase.DicomIO`
|
|
:class:`~pydicom.filebase.DicomBytesIO`
|
|
"""
|
|
|
|
pass
|
|
|
|
|
|
def DicomFile(*args: Any, **kwargs: Any) -> DicomFileLike:
|
|
"""Return an opened :class:`~pydicom.filebase.DicomFileLike` from a file-like."""
|
|
return DicomFileLike(open(*args, **kwargs))
|
|
|
|
|
|
class DicomBytesIO(DicomIO):
|
|
"""Wrapper for :class:`io.BytesIO` to simplify encoding/decoding DICOM datasets.
|
|
|
|
See Also
|
|
--------
|
|
:class:`~pydicom.filebase.DicomIO`
|
|
:class:`~pydicom.filebase.DicomFileLike`
|
|
"""
|
|
|
|
def __init__(self, initial_bytes: bytes | bytearray | memoryview = b"") -> None:
|
|
"""Create a new DicomBytesIO instance.
|
|
|
|
Parameters
|
|
----------
|
|
buffer : bytes | bytearray | memoryview, optional
|
|
The buffer to write to or read from, default is an empty buffer.
|
|
"""
|
|
buffer = BytesIO(initial_bytes)
|
|
super().__init__(buffer)
|
|
|
|
self.getvalue = buffer.getvalue
|