Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/PIL/ImageFile.py: 19%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1#
2# The Python Imaging Library.
3# $Id$
4#
5# base class for image file handlers
6#
7# history:
8# 1995-09-09 fl Created
9# 1996-03-11 fl Fixed load mechanism.
10# 1996-04-15 fl Added pcx/xbm decoders.
11# 1996-04-30 fl Added encoders.
12# 1996-12-14 fl Added load helpers
13# 1997-01-11 fl Use encode_to_file where possible
14# 1997-08-27 fl Flush output in _save
15# 1998-03-05 fl Use memory mapping for some modes
16# 1999-02-04 fl Use memory mapping also for "I;16" and "I;16B"
17# 1999-05-31 fl Added image parser
18# 2000-10-12 fl Set readonly flag on memory-mapped images
19# 2002-03-20 fl Use better messages for common decoder errors
20# 2003-04-21 fl Fall back on mmap/map_buffer if map is not available
21# 2003-10-30 fl Added StubImageFile class
22# 2004-02-25 fl Made incremental parser more robust
23#
24# Copyright (c) 1997-2004 by Secret Labs AB
25# Copyright (c) 1995-2004 by Fredrik Lundh
26#
27# See the README file for information on usage and redistribution.
28#
29from __future__ import annotations
31__lazy_modules__ = {"PIL._util", "io", "itertools", "struct"}
33import abc
34import io
35import itertools
36import logging
37import os
38import struct
39from typing import NamedTuple, cast
41from . import ExifTags, Image
42from ._util import DeferredError, is_path
44TYPE_CHECKING = False
45if TYPE_CHECKING:
46 from typing import IO, Any, Self
48 from ._typing import StrOrBytesPath
50logger = logging.getLogger(__name__)
52MAXBLOCK = 65536
53"""
54By default, Pillow processes image data in blocks. This helps to prevent excessive use
55of resources. Codecs may disable this behaviour with ``_pulls_fd`` or ``_pushes_fd``.
57When reading an image, this is the number of bytes to read at once.
59When writing an image, this is the number of bytes to write at once.
60If the image width times 4 is greater, then that will be used instead.
61Plugins may also set a greater number.
63User code may set this to another number.
64"""
66SAFEBLOCK = 1024 * 1024
68LOAD_TRUNCATED_IMAGES = False
69"""Whether or not to load truncated image files. User code may change this."""
71ERRORS = {
72 -1: "image buffer overrun error",
73 -2: "decoding error",
74 -3: "unknown error",
75 -8: "bad configuration",
76 -9: "out of memory error",
77}
78"""
79Dict of known error codes returned from :meth:`.PyDecoder.decode`,
80:meth:`.PyEncoder.encode` :meth:`.PyEncoder.encode_to_pyfd` and
81:meth:`.PyEncoder.encode_to_file`.
82"""
85#
86# --------------------------------------------------------------------
87# Helpers
90def _get_oserror(error: int, *, encoder: bool) -> OSError:
91 try:
92 msg = Image.core.getcodecstatus(error)
93 except AttributeError:
94 msg = ERRORS.get(error)
95 if not msg:
96 msg = f"{'encoder' if encoder else 'decoder'} error {error}"
97 msg += f" when {'writing' if encoder else 'reading'} image file"
98 return OSError(msg)
101def _tilesort(t: _Tile) -> int:
102 # sort on offset
103 return t[2]
106class _Tile(NamedTuple):
107 codec_name: str
108 extents: tuple[int, int, int, int] | None
109 offset: int = 0
110 args: tuple[Any, ...] | str | None = None
113#
114# --------------------------------------------------------------------
115# ImageFile base class
118class ImageFile(Image.Image, metaclass=abc.ABCMeta):
119 """Base class for image file format handlers."""
121 def __init__(
122 self, fp: StrOrBytesPath | IO[bytes], filename: str | bytes | None = None
123 ) -> None:
124 super().__init__()
126 self._min_frame = 0
128 self.custom_mimetype: str | None = None
130 self.tile: list[_Tile] = []
131 """ A list of tile descriptors """
133 self.readonly = 1 # until we know better
135 self.decoderconfig: tuple[Any, ...] = ()
136 self.decodermaxblock = MAXBLOCK
138 self.fp: IO[bytes] | None
139 self._fp: IO[bytes] | DeferredError
140 if is_path(fp):
141 # filename
142 self.fp = open(fp, "rb")
143 self.filename = os.fspath(fp)
144 self._exclusive_fp = True
145 else:
146 # stream
147 self.fp = cast("IO[bytes]", fp)
148 self.filename = filename if filename is not None else ""
149 # can be overridden
150 self._exclusive_fp = False
152 try:
153 try:
154 self._open()
156 if isinstance(self, StubImageFile) and self._handler:
157 self._handler.open(self)
158 except (
159 IndexError, # end of data
160 TypeError, # end of data (ord)
161 KeyError, # unsupported mode
162 EOFError, # got header but not the first frame
163 struct.error,
164 ) as v:
165 raise SyntaxError(v) from v
167 if not self.mode or (
168 min(self.size) < 0
169 if isinstance(self, StubImageFile) and self._handler is None
170 else min(self.size) <= 0
171 ):
172 msg = "not identified by this driver"
173 raise SyntaxError(msg)
174 except BaseException:
175 # close the file only if we have opened it this constructor
176 if self._exclusive_fp:
177 self.fp.close()
178 raise
180 @abc.abstractmethod
181 def _open(self) -> None:
182 pass
184 def _close_fp(self) -> None:
185 if getattr(self, "_fp", False) and not isinstance(self._fp, DeferredError):
186 if self._fp != self.fp:
187 self._fp.close()
188 self._fp = DeferredError(ValueError("Operation on closed image"))
189 if self.fp:
190 self.fp.close()
192 # Context manager support
193 def __exit__(self, *args: object) -> None:
194 if getattr(self, "_exclusive_fp", False):
195 self._close_fp()
196 self.fp = None
198 def close(self) -> None:
199 """
200 Closes the file pointer, if possible.
202 This operation will destroy the image core and release its memory.
203 The image data will be unusable afterward.
205 This function is required to close images that have multiple frames or
206 have not had their file read and closed by the
207 :py:meth:`~PIL.Image.Image.load` method. See :ref:`file-handling` for
208 more information.
209 """
210 try:
211 self._close_fp()
212 self.fp = None
213 except Exception as msg:
214 logger.debug("Error closing: %s", msg)
216 super().close()
218 def get_child_images(self) -> list[ImageFile]:
219 child_images = []
220 exif = self.getexif()
221 ifds = []
222 if ExifTags.Base.SubIFDs in exif:
223 subifd_offsets = exif[ExifTags.Base.SubIFDs]
224 if subifd_offsets:
225 if not isinstance(subifd_offsets, tuple):
226 subifd_offsets = (subifd_offsets,)
227 ifds = [
228 (exif._get_ifd_dict(subifd_offset), subifd_offset)
229 for subifd_offset in subifd_offsets
230 ]
231 ifd1 = exif.get_ifd(ExifTags.IFD.IFD1)
232 if ifd1 and ifd1.get(ExifTags.Base.JpegIFOffset):
233 assert exif._info is not None
234 ifds.append((ifd1, exif._info.next))
236 offset = None
237 for ifd, ifd_offset in ifds:
238 assert self.fp is not None
239 current_offset = self.fp.tell()
240 if offset is None:
241 offset = current_offset
243 fp = self.fp
244 if ifd is not None:
245 thumbnail_offset = ifd.get(ExifTags.Base.JpegIFOffset)
246 if thumbnail_offset is not None:
247 thumbnail_offset += getattr(self, "_exif_offset", 0)
248 self.fp.seek(thumbnail_offset)
250 length = ifd.get(ExifTags.Base.JpegIFByteCount)
251 assert isinstance(length, int)
252 data = self.fp.read(length)
253 fp = io.BytesIO(data)
255 with Image.open(fp) as im:
256 from . import TiffImagePlugin
258 if thumbnail_offset is None and isinstance(
259 im, TiffImagePlugin.TiffImageFile
260 ):
261 im._frame_pos = [ifd_offset]
262 im._seek(0)
263 im.load()
264 child_images.append(im)
266 if offset is not None:
267 assert self.fp is not None
268 self.fp.seek(offset)
269 return child_images
271 def get_format_mimetype(self) -> str | None:
272 if self.custom_mimetype:
273 return self.custom_mimetype
274 if self.format is not None:
275 return Image.MIME.get(self.format.upper())
276 return None
278 def __getstate__(self) -> list[Any]:
279 return super().__getstate__() + [self.filename]
281 def __setstate__(self, state: list[Any]) -> None:
282 self.tile = []
283 if len(state) > 5:
284 self.filename = state[5]
285 super().__setstate__(state)
287 def verify(self) -> None:
288 """Check file integrity"""
290 # raise exception if something's wrong. must be called
291 # directly after open, and closes file when finished.
292 if self._exclusive_fp and self.fp:
293 self.fp.close()
294 self.fp = None
296 def load(self) -> Image.core.PixelAccess | None:
297 """Load image data based on tile list"""
299 if not self.tile and self._im is None:
300 msg = "cannot load this image"
301 raise OSError(msg)
303 pixel = Image.Image.load(self)
304 if not self.tile:
305 return pixel
307 self.map: mmap.mmap | None = None
308 use_mmap = self.filename and len(self.tile) == 1
310 assert self.fp is not None
311 readonly = 0
313 # look for read/seek overrides
314 if hasattr(self, "load_read"):
315 read = self.load_read
316 # don't use mmap if there are custom read/seek functions
317 use_mmap = False
318 else:
319 read = self.fp.read
321 if hasattr(self, "load_seek"):
322 seek = self.load_seek
323 use_mmap = False
324 else:
325 seek = self.fp.seek
327 if use_mmap:
328 # try memory mapping
329 decoder_name, extents, offset, args = self.tile[0]
330 if isinstance(args, str):
331 args = (args, 0, 1)
332 if (
333 decoder_name == "raw"
334 and isinstance(args, tuple)
335 and len(args) >= 3
336 and args[0] == self.mode
337 and args[0] in Image._MAPMODES
338 ):
339 if offset < 0:
340 msg = "Tile offset cannot be negative"
341 raise ValueError(msg)
342 try:
343 # use mmap, if possible
344 import mmap
346 with open(self.filename) as fp:
347 self.map = mmap.mmap(fp.fileno(), 0, access=mmap.ACCESS_READ)
348 if offset + self.size[1] * args[1] > self.map.size():
349 msg = "buffer is not large enough"
350 raise OSError(msg)
351 self.im = Image.core.map_buffer(
352 self.map, self.size, decoder_name, offset, args
353 )
354 readonly = 1
355 # After trashing self.im,
356 # we might need to reload the palette data.
357 if self.palette:
358 self.palette.dirty = 1
359 except (AttributeError, OSError, ImportError):
360 self.map = None
362 self.load_prepare()
363 err_code = -3 # initialize to unknown error
364 if not self.map:
365 # sort tiles in file order
366 self.tile.sort(key=_tilesort)
368 # FIXME: This is a hack to handle TIFF's JpegTables tag.
369 prefix = getattr(self, "tile_prefix", b"")
371 # Remove consecutive duplicates that only differ by their offset
372 self.tile = [
373 list(tiles)[-1]
374 for _, tiles in itertools.groupby(
375 self.tile, lambda tile: (tile[0], tile[1], tile[3])
376 )
377 ]
378 for i, (decoder_name, extents, offset, args) in enumerate(self.tile):
379 seek(offset)
380 decoder = Image._getdecoder(
381 self.mode, decoder_name, args, self.decoderconfig
382 )
383 try:
384 decoder.setimage(self.im, extents)
385 if decoder.pulls_fd:
386 decoder.setfd(self.fp)
387 err_code = decoder.decode(b"")[1]
388 else:
389 b = prefix
390 while True:
391 read_bytes = self.decodermaxblock
392 if i + 1 < len(self.tile):
393 next_offset = self.tile[i + 1].offset
394 if next_offset > offset:
395 read_bytes = next_offset - offset
396 try:
397 s = read(read_bytes)
398 except (IndexError, struct.error) as e:
399 # truncated png/gif
400 if LOAD_TRUNCATED_IMAGES:
401 break
402 else:
403 msg = "image file is truncated"
404 raise OSError(msg) from e
406 if not s: # truncated jpeg
407 if LOAD_TRUNCATED_IMAGES:
408 break
409 else:
410 msg = (
411 "image file is truncated "
412 f"({len(b)} bytes not processed)"
413 )
414 raise OSError(msg)
416 b = b + s
417 n, err_code = decoder.decode(b)
418 if n < 0:
419 break
420 b = b[n:]
421 finally:
422 # Need to cleanup here to prevent leaks
423 decoder.cleanup()
425 self.tile = []
426 self.readonly = readonly
428 self.load_end()
430 if self._exclusive_fp and self._close_exclusive_fp_after_loading:
431 self.fp.close()
432 self.fp = None
434 if not self.map and not LOAD_TRUNCATED_IMAGES and err_code < 0:
435 # still raised if decoder fails to return anything
436 raise _get_oserror(err_code, encoder=False)
438 return Image.Image.load(self)
440 def load_prepare(self) -> None:
441 # create image memory if necessary
442 if self._im is None:
443 self.im = Image.core.new(self.mode, self.size)
444 # create palette (optional)
445 if self.mode == "P":
446 Image.Image.load(self)
448 def load_end(self) -> None:
449 # may be overridden
450 pass
452 # may be defined for contained formats
453 # def load_seek(self, pos: int) -> None:
454 # pass
456 # may be defined for blocked formats (e.g. PNG)
457 # def load_read(self, read_bytes: int) -> bytes:
458 # pass
460 def _seek_check(self, frame: int) -> bool:
461 if (
462 frame < self._min_frame
463 # Only check upper limit on frames if additional seek operations
464 # are not required to do so
465 or (
466 not (hasattr(self, "_n_frames") and self._n_frames is None)
467 and frame >= getattr(self, "n_frames") + self._min_frame
468 )
469 ):
470 msg = "attempt to seek outside sequence"
471 raise EOFError(msg)
473 return self.tell() != frame
476class StubHandler(abc.ABC):
477 @abc.abstractmethod
478 def open(self, im: StubImageFile) -> None:
479 pass
481 @abc.abstractmethod
482 def load(self, im: StubImageFile) -> Image.Image:
483 pass
486class StubImageFile(ImageFile):
487 """
488 Base class for stub image loaders.
490 A stub loader is an image loader that can identify files of a
491 certain format, but relies on external code to load the file.
492 """
494 _handler: StubHandler | None = None
496 def load(self) -> Image.core.PixelAccess | None:
497 if self._handler is None:
498 msg = f"cannot find loader for this {self.format} file"
499 raise OSError(msg)
500 image = self._handler.load(self)
501 assert image is not None
502 # become the other object (!)
503 self.__class__ = image.__class__ # type: ignore[assignment]
504 self.__dict__ = image.__dict__
505 return image.load()
508class Parser:
509 """
510 Incremental image parser. This class implements the standard
511 feed/close consumer interface.
512 """
514 incremental = None
515 image: Image.Image | None = None
516 data: bytes | None = None
517 decoder: Image.core.ImagingDecoder | PyDecoder | None = None
518 offset = 0
519 finished = 0
521 def reset(self) -> None:
522 """
523 (Consumer) Reset the parser. Note that you can only call this
524 method immediately after you've created a parser; parser
525 instances cannot be reused.
526 """
527 assert self.data is None, "cannot reuse parsers"
529 def feed(self, data: bytes) -> None:
530 """
531 (Consumer) Feed data to the parser.
533 :param data: A string buffer.
534 :exception OSError: If the parser failed to parse the image file.
535 """
536 # collect data
538 if self.finished:
539 return
541 if self.data is None:
542 self.data = data
543 else:
544 self.data = self.data + data
546 # parse what we have
547 if self.decoder:
548 if self.offset > 0:
549 # skip header
550 skip = min(len(self.data), self.offset)
551 self.data = self.data[skip:]
552 self.offset = self.offset - skip
553 if self.offset > 0 or not self.data:
554 return
556 n, e = self.decoder.decode(self.data)
558 if n < 0:
559 # end of stream
560 self.data = None
561 self.finished = 1
562 if e < 0:
563 # decoding error
564 self.image = None
565 raise _get_oserror(e, encoder=False)
566 else:
567 # end of image
568 return
569 self.data = self.data[n:]
571 elif self.image:
572 # if we end up here with no decoder, this file cannot
573 # be incrementally parsed. wait until we've gotten all
574 # available data
575 pass
577 else:
578 # attempt to open this file
579 try:
580 with io.BytesIO(self.data) as fp:
581 im = Image.open(fp)
582 except OSError:
583 pass # not enough data
584 else:
585 flag = hasattr(im, "load_seek") or hasattr(im, "load_read")
586 if not flag and len(im.tile) == 1:
587 # initialize decoder
588 im.load_prepare()
589 d, e, o, a = im.tile[0]
590 im.tile = []
591 self.decoder = Image._getdecoder(im.mode, d, a, im.decoderconfig)
592 self.decoder.setimage(im.im, e)
594 # calculate decoder offset
595 self.offset = o
596 if self.offset <= len(self.data):
597 self.data = self.data[self.offset :]
598 self.offset = 0
600 self.image = im
602 def __enter__(self) -> Self:
603 return self
605 def __exit__(self, *args: object) -> None:
606 self.close()
608 def close(self) -> Image.Image:
609 """
610 (Consumer) Close the stream.
612 :returns: An image object.
613 :exception OSError: If the parser failed to parse the image file either
614 because it cannot be identified or cannot be
615 decoded.
616 """
617 # finish decoding
618 if self.decoder:
619 # get rid of what's left in the buffers
620 self.feed(b"")
621 self.data = self.decoder = None
622 if not self.finished:
623 msg = "image was incomplete"
624 raise OSError(msg)
625 if not self.image:
626 msg = "cannot parse this image"
627 raise OSError(msg)
628 if self.data:
629 # incremental parsing not possible; reopen the file
630 # not that we have all data
631 with io.BytesIO(self.data) as fp:
632 try:
633 self.image = Image.open(fp)
634 finally:
635 self.image.load()
636 return self.image
639# --------------------------------------------------------------------
642def _save(im: Image.Image, fp: IO[bytes], tile: list[_Tile], bufsize: int = 0) -> None:
643 """Helper to save image based on tile list
645 :param im: Image object.
646 :param fp: File object.
647 :param tile: Tile list.
648 :param bufsize: Optional buffer size
649 """
651 im.load()
652 if not hasattr(im, "encoderconfig"):
653 im.encoderconfig = ()
654 tile.sort(key=_tilesort)
655 # FIXME: make MAXBLOCK a configuration parameter
656 # It would be great if we could have the encoder specify what it needs
657 # But, it would need at least the image size in most cases. RawEncode is
658 # a tricky case.
659 bufsize = max(MAXBLOCK, bufsize, im.size[0] * 4) # see RawEncode.c
660 try:
661 fh = fp.fileno()
662 fp.flush()
663 _encode_tile(im, fp, tile, bufsize, fh)
664 except (AttributeError, io.UnsupportedOperation) as exc:
665 _encode_tile(im, fp, tile, bufsize, None, exc)
666 if hasattr(fp, "flush"):
667 fp.flush()
670def _encode_tile(
671 im: Image.Image,
672 fp: IO[bytes],
673 tile: list[_Tile],
674 bufsize: int,
675 fh: int | None,
676 exc: BaseException | None = None,
677) -> None:
678 for encoder_name, extents, offset, args in tile:
679 if offset > 0:
680 fp.seek(offset)
681 encoder = Image._getencoder(im.mode, encoder_name, args, im.encoderconfig)
682 try:
683 encoder.setimage(im.im, extents)
684 if encoder.pushes_fd:
685 encoder.setfd(fp)
686 errcode = encoder.encode_to_pyfd()[1]
687 else:
688 if exc:
689 # compress to Python file-compatible object
690 while True:
691 errcode, data = encoder.encode(bufsize)[1:]
692 fp.write(data)
693 if errcode:
694 break
695 else:
696 # slight speedup: compress to real file object
697 assert fh is not None
698 errcode = encoder.encode_to_file(fh, bufsize)
699 if errcode < 0:
700 raise _get_oserror(errcode, encoder=True) from exc
701 finally:
702 encoder.cleanup()
705def _safe_read(fp: IO[bytes], size: int) -> bytes:
706 """
707 Reads large blocks in a safe way. Unlike fp.read(n), this function
708 doesn't trust the user. If the requested size is larger than
709 SAFEBLOCK, the file is read block by block.
711 :param fp: File handle. Must implement a <b>read</b> method.
712 :param size: Number of bytes to read.
713 :returns: A string containing <i>size</i> bytes of data.
715 Raises an OSError if the file is truncated and the read cannot be completed
717 """
718 if size <= 0:
719 return b""
720 if size <= SAFEBLOCK:
721 data = fp.read(size)
722 if len(data) < size:
723 msg = "Truncated File Read"
724 raise OSError(msg)
725 return data
726 blocks: list[bytes] = []
727 remaining_size = size
728 while remaining_size > 0:
729 block = fp.read(min(remaining_size, SAFEBLOCK))
730 if not block:
731 break
732 blocks.append(block)
733 remaining_size -= len(block)
734 if sum(len(block) for block in blocks) < size:
735 msg = "Truncated File Read"
736 raise OSError(msg)
737 return b"".join(blocks)
740class PyCodecState:
741 def __init__(self) -> None:
742 self.xsize = 0
743 self.ysize = 0
744 self.xoff = 0
745 self.yoff = 0
747 def extents(self) -> tuple[int, int, int, int]:
748 return self.xoff, self.yoff, self.xoff + self.xsize, self.yoff + self.ysize
751class PyCodec:
752 fd: IO[bytes] | None
754 def __init__(self, mode: str, *args: Any) -> None:
755 self.im: Image.core.ImagingCore | None = None
756 self.state = PyCodecState()
757 self.fd = None
758 self.mode = mode
759 self.init(args)
761 def init(self, args: tuple[Any, ...]) -> None:
762 """
763 Override to perform codec specific initialization
765 :param args: Tuple of arg items from the tile entry
766 :returns: None
767 """
768 self.args = args
770 def cleanup(self) -> None:
771 """
772 Override to perform codec specific cleanup
774 :returns: None
775 """
776 pass
778 def setfd(self, fd: IO[bytes]) -> None:
779 """
780 Called from ImageFile to set the Python file-like object
782 :param fd: A Python file-like object
783 :returns: None
784 """
785 self.fd = fd
787 def setimage(
788 self,
789 im: Image.core.ImagingCore,
790 extents: tuple[int, int, int, int] | None = None,
791 ) -> None:
792 """
793 Called from ImageFile to set the core output image for the codec
795 :param im: A core image object
796 :param extents: a 4 tuple of (x0, y0, x1, y1) defining the rectangle
797 for this tile
798 :returns: None
799 """
801 # following c code
802 self.im = im
804 if extents:
805 x0, y0, x1, y1 = extents
807 if x0 < 0 or y0 < 0 or x1 > self.im.size[0] or y1 > self.im.size[1]:
808 msg = "Tile cannot extend outside image"
809 raise ValueError(msg)
811 self.state.xoff = x0
812 self.state.yoff = y0
813 self.state.xsize = x1 - x0
814 self.state.ysize = y1 - y0
815 else:
816 self.state.xsize, self.state.ysize = self.im.size
818 if self.state.xsize <= 0 or self.state.ysize <= 0:
819 msg = "Size must be positive"
820 raise ValueError(msg)
823class PyDecoder(PyCodec):
824 """
825 Python implementation of a format decoder. Override this class and
826 add the decoding logic in the :meth:`decode` method.
828 See :ref:`Writing Your Own File Codec in Python<file-codecs-py>`
829 """
831 _pulls_fd = False
833 @property
834 def pulls_fd(self) -> bool:
835 return self._pulls_fd
837 def decode(self, buffer: Image.DecoderInput) -> tuple[int, int]:
838 """
839 Override to perform the decoding process.
841 :param buffer: A bytes object with the data to be decoded.
842 :returns: A tuple of ``(bytes consumed, errcode)``.
843 If finished with decoding return -1 for the bytes consumed.
844 Err codes are from :data:`.ImageFile.ERRORS`.
845 """
846 msg = "unavailable in base decoder"
847 raise NotImplementedError(msg)
849 def set_as_raw(
850 self,
851 data: bytes | bytearray,
852 rawmode: str | None = None,
853 extra: tuple[Any, ...] = (),
854 ) -> None:
855 """
856 Convenience method to set the internal image from a stream of raw data
858 :param data: Bytes to be set
859 :param rawmode: The rawmode to be used for the decoder.
860 If not specified, it will default to the mode of the image
861 :param extra: Extra arguments for the decoder.
862 :returns: None
863 """
865 if not rawmode:
866 rawmode = self.mode
867 d = Image._getdecoder(self.mode, "raw", rawmode, extra)
868 assert self.im is not None
869 d.setimage(self.im, self.state.extents())
870 s = d.decode(data)
872 if s[0] >= 0:
873 msg = "not enough image data"
874 raise ValueError(msg)
875 if s[1] != 0:
876 msg = "cannot decode image data"
877 raise ValueError(msg)
880class PyEncoder(PyCodec):
881 """
882 Python implementation of a format encoder. Override this class and
883 add the decoding logic in the :meth:`encode` method.
885 See :ref:`Writing Your Own File Codec in Python<file-codecs-py>`
886 """
888 _pushes_fd = False
890 @property
891 def pushes_fd(self) -> bool:
892 return self._pushes_fd
894 def encode(self, bufsize: int) -> tuple[int, int, bytes]:
895 """
896 Override to perform the encoding process.
898 :param bufsize: Buffer size.
899 :returns: A tuple of ``(bytes encoded, errcode, bytes)``.
900 If finished with encoding return 1 for the error code.
901 Err codes are from :data:`.ImageFile.ERRORS`.
902 """
903 msg = "unavailable in base encoder"
904 raise NotImplementedError(msg)
906 def encode_to_pyfd(self) -> tuple[int, int]:
907 """
908 If ``pushes_fd`` is ``True``, then this method will be used,
909 and ``encode()`` will only be called once.
911 :returns: A tuple of ``(bytes consumed, errcode)``.
912 Err codes are from :data:`.ImageFile.ERRORS`.
913 """
914 if not self.pushes_fd:
915 return 0, -8 # bad configuration
916 bytes_consumed, errcode, data = self.encode(0)
917 if data:
918 assert self.fd is not None
919 self.fd.write(data)
920 return bytes_consumed, errcode
922 def encode_to_file(self, fh: int, bufsize: int) -> int:
923 """
924 :param fh: File handle.
925 :param bufsize: Buffer size.
927 :returns: If finished successfully, return 0.
928 Otherwise, return an error code. Err codes are from
929 :data:`.ImageFile.ERRORS`.
930 """
931 errcode = 0
932 while errcode == 0:
933 status, errcode, buf = self.encode(bufsize)
934 if status > 0:
935 os.write(fh, buf[status:])
936 return errcode