Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/PIL/ImageFile.py: 57%

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 

30 

31import abc 

32import io 

33import itertools 

34import logging 

35import os 

36import struct 

37from typing import IO, Any, NamedTuple, cast 

38 

39from . import ExifTags, Image 

40from ._util import DeferredError, is_path 

41 

42TYPE_CHECKING = False 

43if TYPE_CHECKING: 

44 from typing import Self 

45 

46 from ._typing import StrOrBytesPath 

47 

48logger = logging.getLogger(__name__) 

49 

50MAXBLOCK = 65536 

51""" 

52By default, Pillow processes image data in blocks. This helps to prevent excessive use 

53of resources. Codecs may disable this behaviour with ``_pulls_fd`` or ``_pushes_fd``. 

54 

55When reading an image, this is the number of bytes to read at once. 

56 

57When writing an image, this is the number of bytes to write at once. 

58If the image width times 4 is greater, then that will be used instead. 

59Plugins may also set a greater number. 

60 

61User code may set this to another number. 

62""" 

63 

64SAFEBLOCK = 1024 * 1024 

65 

66LOAD_TRUNCATED_IMAGES = False 

67"""Whether or not to load truncated image files. User code may change this.""" 

68 

69ERRORS = { 

70 -1: "image buffer overrun error", 

71 -2: "decoding error", 

72 -3: "unknown error", 

73 -8: "bad configuration", 

74 -9: "out of memory error", 

75} 

76""" 

77Dict of known error codes returned from :meth:`.PyDecoder.decode`, 

78:meth:`.PyEncoder.encode` :meth:`.PyEncoder.encode_to_pyfd` and 

79:meth:`.PyEncoder.encode_to_file`. 

80""" 

81 

82 

83# 

84# -------------------------------------------------------------------- 

85# Helpers 

86 

87 

88def _get_oserror(error: int, *, encoder: bool) -> OSError: 

89 try: 

90 msg = Image.core.getcodecstatus(error) 

91 except AttributeError: 

92 msg = ERRORS.get(error) 

93 if not msg: 

94 msg = f"{'encoder' if encoder else 'decoder'} error {error}" 

95 msg += f" when {'writing' if encoder else 'reading'} image file" 

96 return OSError(msg) 

97 

98 

99def _tilesort(t: _Tile) -> int: 

100 # sort on offset 

101 return t[2] 

102 

103 

104class _Tile(NamedTuple): 

105 codec_name: str 

106 extents: tuple[int, int, int, int] | None 

107 offset: int = 0 

108 args: tuple[Any, ...] | str | None = None 

109 

110 

111# 

112# -------------------------------------------------------------------- 

113# ImageFile base class 

114 

115 

116class ImageFile(Image.Image): 

117 """Base class for image file format handlers.""" 

118 

119 def __init__( 

120 self, fp: StrOrBytesPath | IO[bytes], filename: str | bytes | None = None 

121 ) -> None: 

122 super().__init__() 

123 

124 self._min_frame = 0 

125 

126 self.custom_mimetype: str | None = None 

127 

128 self.tile: list[_Tile] = [] 

129 """ A list of tile descriptors """ 

130 

131 self.readonly = 1 # until we know better 

132 

133 self.decoderconfig: tuple[Any, ...] = () 

134 self.decodermaxblock = MAXBLOCK 

135 

136 self.fp: IO[bytes] | None 

137 self._fp: IO[bytes] | DeferredError 

138 if is_path(fp): 

139 # filename 

140 self.fp = open(fp, "rb") 

141 self.filename = os.fspath(fp) 

142 self._exclusive_fp = True 

143 else: 

144 # stream 

145 self.fp = cast(IO[bytes], fp) 

146 self.filename = filename if filename is not None else "" 

147 # can be overridden 

148 self._exclusive_fp = False 

149 

150 try: 

151 try: 

152 self._open() 

153 

154 if isinstance(self, StubImageFile): 

155 if loader := self._load(): 

156 loader.open(self) 

157 except ( 

158 IndexError, # end of data 

159 TypeError, # end of data (ord) 

160 KeyError, # unsupported mode 

161 EOFError, # got header but not the first frame 

162 struct.error, 

163 ) as v: 

164 raise SyntaxError(v) from v 

165 

166 if not self.mode or self.size[0] <= 0 or self.size[1] <= 0: 

167 msg = "not identified by this driver" 

168 raise SyntaxError(msg) 

169 except BaseException: 

170 # close the file only if we have opened it this constructor 

171 if self._exclusive_fp: 

172 self.fp.close() 

173 raise 

174 

175 def _open(self) -> None: 

176 pass 

177 

178 def _close_fp(self) -> None: 

179 if getattr(self, "_fp", False) and not isinstance(self._fp, DeferredError): 

180 if self._fp != self.fp: 

181 self._fp.close() 

182 self._fp = DeferredError(ValueError("Operation on closed image")) 

183 if self.fp: 

184 self.fp.close() 

185 

186 # Context manager support 

187 def __exit__(self, *args: object) -> None: 

188 if getattr(self, "_exclusive_fp", False): 

189 self._close_fp() 

190 self.fp = None 

191 

192 def close(self) -> None: 

193 """ 

194 Closes the file pointer, if possible. 

195 

196 This operation will destroy the image core and release its memory. 

197 The image data will be unusable afterward. 

198 

199 This function is required to close images that have multiple frames or 

200 have not had their file read and closed by the 

201 :py:meth:`~PIL.Image.Image.load` method. See :ref:`file-handling` for 

202 more information. 

203 """ 

204 try: 

205 self._close_fp() 

206 self.fp = None 

207 except Exception as msg: 

208 logger.debug("Error closing: %s", msg) 

209 

210 super().close() 

211 

212 def get_child_images(self) -> list[ImageFile]: 

213 child_images = [] 

214 exif = self.getexif() 

215 ifds = [] 

216 if ExifTags.Base.SubIFDs in exif: 

217 subifd_offsets = exif[ExifTags.Base.SubIFDs] 

218 if subifd_offsets: 

219 if not isinstance(subifd_offsets, tuple): 

220 subifd_offsets = (subifd_offsets,) 

221 ifds = [ 

222 (exif._get_ifd_dict(subifd_offset), subifd_offset) 

223 for subifd_offset in subifd_offsets 

224 ] 

225 ifd1 = exif.get_ifd(ExifTags.IFD.IFD1) 

226 if ifd1 and ifd1.get(ExifTags.Base.JpegIFOffset): 

227 assert exif._info is not None 

228 ifds.append((ifd1, exif._info.next)) 

229 

230 offset = None 

231 for ifd, ifd_offset in ifds: 

232 assert self.fp is not None 

233 current_offset = self.fp.tell() 

234 if offset is None: 

235 offset = current_offset 

236 

237 fp = self.fp 

238 if ifd is not None: 

239 thumbnail_offset = ifd.get(ExifTags.Base.JpegIFOffset) 

240 if thumbnail_offset is not None: 

241 thumbnail_offset += getattr(self, "_exif_offset", 0) 

242 self.fp.seek(thumbnail_offset) 

243 

244 length = ifd.get(ExifTags.Base.JpegIFByteCount) 

245 assert isinstance(length, int) 

246 data = self.fp.read(length) 

247 fp = io.BytesIO(data) 

248 

249 with Image.open(fp) as im: 

250 from . import TiffImagePlugin 

251 

252 if thumbnail_offset is None and isinstance( 

253 im, TiffImagePlugin.TiffImageFile 

254 ): 

255 im._frame_pos = [ifd_offset] 

256 im._seek(0) 

257 im.load() 

258 child_images.append(im) 

259 

260 if offset is not None: 

261 assert self.fp is not None 

262 self.fp.seek(offset) 

263 return child_images 

264 

265 def get_format_mimetype(self) -> str | None: 

266 if self.custom_mimetype: 

267 return self.custom_mimetype 

268 if self.format is not None: 

269 return Image.MIME.get(self.format.upper()) 

270 return None 

271 

272 def __getstate__(self) -> list[Any]: 

273 return super().__getstate__() + [self.filename] 

274 

275 def __setstate__(self, state: list[Any]) -> None: 

276 self.tile = [] 

277 if len(state) > 5: 

278 self.filename = state[5] 

279 super().__setstate__(state) 

280 

281 def verify(self) -> None: 

282 """Check file integrity""" 

283 

284 # raise exception if something's wrong. must be called 

285 # directly after open, and closes file when finished. 

286 if self._exclusive_fp and self.fp: 

287 self.fp.close() 

288 self.fp = None 

289 

290 def load(self) -> Image.core.PixelAccess | None: 

291 """Load image data based on tile list""" 

292 

293 if not self.tile and self._im is None: 

294 msg = "cannot load this image" 

295 raise OSError(msg) 

296 

297 pixel = Image.Image.load(self) 

298 if not self.tile: 

299 return pixel 

300 

301 self.map: mmap.mmap | None = None 

302 use_mmap = self.filename and len(self.tile) == 1 

303 

304 assert self.fp is not None 

305 readonly = 0 

306 

307 # look for read/seek overrides 

308 if hasattr(self, "load_read"): 

309 read = self.load_read 

310 # don't use mmap if there are custom read/seek functions 

311 use_mmap = False 

312 else: 

313 read = self.fp.read 

314 

315 if hasattr(self, "load_seek"): 

316 seek = self.load_seek 

317 use_mmap = False 

318 else: 

319 seek = self.fp.seek 

320 

321 if use_mmap: 

322 # try memory mapping 

323 decoder_name, extents, offset, args = self.tile[0] 

324 if isinstance(args, str): 

325 args = (args, 0, 1) 

326 if ( 

327 decoder_name == "raw" 

328 and isinstance(args, tuple) 

329 and len(args) >= 3 

330 and args[0] == self.mode 

331 and args[0] in Image._MAPMODES 

332 ): 

333 if offset < 0: 

334 msg = "Tile offset cannot be negative" 

335 raise ValueError(msg) 

336 try: 

337 # use mmap, if possible 

338 import mmap 

339 

340 with open(self.filename) as fp: 

341 self.map = mmap.mmap(fp.fileno(), 0, access=mmap.ACCESS_READ) 

342 if offset + self.size[1] * args[1] > self.map.size(): 

343 msg = "buffer is not large enough" 

344 raise OSError(msg) 

345 self.im = Image.core.map_buffer( 

346 self.map, self.size, decoder_name, offset, args 

347 ) 

348 readonly = 1 

349 # After trashing self.im, 

350 # we might need to reload the palette data. 

351 if self.palette: 

352 self.palette.dirty = 1 

353 except (AttributeError, OSError, ImportError): 

354 self.map = None 

355 

356 self.load_prepare() 

357 err_code = -3 # initialize to unknown error 

358 if not self.map: 

359 # sort tiles in file order 

360 self.tile.sort(key=_tilesort) 

361 

362 # FIXME: This is a hack to handle TIFF's JpegTables tag. 

363 prefix = getattr(self, "tile_prefix", b"") 

364 

365 # Remove consecutive duplicates that only differ by their offset 

366 self.tile = [ 

367 list(tiles)[-1] 

368 for _, tiles in itertools.groupby( 

369 self.tile, lambda tile: (tile[0], tile[1], tile[3]) 

370 ) 

371 ] 

372 for i, (decoder_name, extents, offset, args) in enumerate(self.tile): 

373 seek(offset) 

374 decoder = Image._getdecoder( 

375 self.mode, decoder_name, args, self.decoderconfig 

376 ) 

377 try: 

378 decoder.setimage(self.im, extents) 

379 if decoder.pulls_fd: 

380 decoder.setfd(self.fp) 

381 err_code = decoder.decode(b"")[1] 

382 else: 

383 b = prefix 

384 while True: 

385 read_bytes = self.decodermaxblock 

386 if i + 1 < len(self.tile): 

387 next_offset = self.tile[i + 1].offset 

388 if next_offset > offset: 

389 read_bytes = next_offset - offset 

390 try: 

391 s = read(read_bytes) 

392 except (IndexError, struct.error) as e: 

393 # truncated png/gif 

394 if LOAD_TRUNCATED_IMAGES: 

395 break 

396 else: 

397 msg = "image file is truncated" 

398 raise OSError(msg) from e 

399 

400 if not s: # truncated jpeg 

401 if LOAD_TRUNCATED_IMAGES: 

402 break 

403 else: 

404 msg = ( 

405 "image file is truncated " 

406 f"({len(b)} bytes not processed)" 

407 ) 

408 raise OSError(msg) 

409 

410 b = b + s 

411 n, err_code = decoder.decode(b) 

412 if n < 0: 

413 break 

414 b = b[n:] 

415 finally: 

416 # Need to cleanup here to prevent leaks 

417 decoder.cleanup() 

418 

419 self.tile = [] 

420 self.readonly = readonly 

421 

422 self.load_end() 

423 

424 if self._exclusive_fp and self._close_exclusive_fp_after_loading: 

425 self.fp.close() 

426 self.fp = None 

427 

428 if not self.map and not LOAD_TRUNCATED_IMAGES and err_code < 0: 

429 # still raised if decoder fails to return anything 

430 raise _get_oserror(err_code, encoder=False) 

431 

432 return Image.Image.load(self) 

433 

434 def load_prepare(self) -> None: 

435 # create image memory if necessary 

436 if self._im is None: 

437 self.im = Image.core.new(self.mode, self.size) 

438 # create palette (optional) 

439 if self.mode == "P": 

440 Image.Image.load(self) 

441 

442 def load_end(self) -> None: 

443 # may be overridden 

444 pass 

445 

446 # may be defined for contained formats 

447 # def load_seek(self, pos: int) -> None: 

448 # pass 

449 

450 # may be defined for blocked formats (e.g. PNG) 

451 # def load_read(self, read_bytes: int) -> bytes: 

452 # pass 

453 

454 def _seek_check(self, frame: int) -> bool: 

455 if ( 

456 frame < self._min_frame 

457 # Only check upper limit on frames if additional seek operations 

458 # are not required to do so 

459 or ( 

460 not (hasattr(self, "_n_frames") and self._n_frames is None) 

461 and frame >= getattr(self, "n_frames") + self._min_frame 

462 ) 

463 ): 

464 msg = "attempt to seek outside sequence" 

465 raise EOFError(msg) 

466 

467 return self.tell() != frame 

468 

469 

470class StubHandler(abc.ABC): 

471 def open(self, im: StubImageFile) -> None: 

472 pass 

473 

474 @abc.abstractmethod 

475 def load(self, im: StubImageFile) -> Image.Image: 

476 pass 

477 

478 

479class StubImageFile(ImageFile, metaclass=abc.ABCMeta): 

480 """ 

481 Base class for stub image loaders. 

482 

483 A stub loader is an image loader that can identify files of a 

484 certain format, but relies on external code to load the file. 

485 """ 

486 

487 @abc.abstractmethod 

488 def _open(self) -> None: 

489 pass 

490 

491 def load(self) -> Image.core.PixelAccess | None: 

492 loader = self._load() 

493 if loader is None: 

494 msg = f"cannot find loader for this {self.format} file" 

495 raise OSError(msg) 

496 image = loader.load(self) 

497 assert image is not None 

498 # become the other object (!) 

499 self.__class__ = image.__class__ # type: ignore[assignment] 

500 self.__dict__ = image.__dict__ 

501 return image.load() 

502 

503 @abc.abstractmethod 

504 def _load(self) -> StubHandler | None: 

505 """(Hook) Find actual image loader.""" 

506 pass 

507 

508 

509class Parser: 

510 """ 

511 Incremental image parser. This class implements the standard 

512 feed/close consumer interface. 

513 """ 

514 

515 incremental = None 

516 image: Image.Image | None = None 

517 data: bytes | None = None 

518 decoder: Image.core.ImagingDecoder | PyDecoder | None = None 

519 offset = 0 

520 finished = 0 

521 

522 def reset(self) -> None: 

523 """ 

524 (Consumer) Reset the parser. Note that you can only call this 

525 method immediately after you've created a parser; parser 

526 instances cannot be reused. 

527 """ 

528 assert self.data is None, "cannot reuse parsers" 

529 

530 def feed(self, data: bytes) -> None: 

531 """ 

532 (Consumer) Feed data to the parser. 

533 

534 :param data: A string buffer. 

535 :exception OSError: If the parser failed to parse the image file. 

536 """ 

537 # collect data 

538 

539 if self.finished: 

540 return 

541 

542 if self.data is None: 

543 self.data = data 

544 else: 

545 self.data = self.data + data 

546 

547 # parse what we have 

548 if self.decoder: 

549 if self.offset > 0: 

550 # skip header 

551 skip = min(len(self.data), self.offset) 

552 self.data = self.data[skip:] 

553 self.offset = self.offset - skip 

554 if self.offset > 0 or not self.data: 

555 return 

556 

557 n, e = self.decoder.decode(self.data) 

558 

559 if n < 0: 

560 # end of stream 

561 self.data = None 

562 self.finished = 1 

563 if e < 0: 

564 # decoding error 

565 self.image = None 

566 raise _get_oserror(e, encoder=False) 

567 else: 

568 # end of image 

569 return 

570 self.data = self.data[n:] 

571 

572 elif self.image: 

573 # if we end up here with no decoder, this file cannot 

574 # be incrementally parsed. wait until we've gotten all 

575 # available data 

576 pass 

577 

578 else: 

579 # attempt to open this file 

580 try: 

581 with io.BytesIO(self.data) as fp: 

582 im = Image.open(fp) 

583 except OSError: 

584 pass # not enough data 

585 else: 

586 flag = hasattr(im, "load_seek") or hasattr(im, "load_read") 

587 if not flag and len(im.tile) == 1: 

588 # initialize decoder 

589 im.load_prepare() 

590 d, e, o, a = im.tile[0] 

591 im.tile = [] 

592 self.decoder = Image._getdecoder(im.mode, d, a, im.decoderconfig) 

593 self.decoder.setimage(im.im, e) 

594 

595 # calculate decoder offset 

596 self.offset = o 

597 if self.offset <= len(self.data): 

598 self.data = self.data[self.offset :] 

599 self.offset = 0 

600 

601 self.image = im 

602 

603 def __enter__(self) -> Self: 

604 return self 

605 

606 def __exit__(self, *args: object) -> None: 

607 self.close() 

608 

609 def close(self) -> Image.Image: 

610 """ 

611 (Consumer) Close the stream. 

612 

613 :returns: An image object. 

614 :exception OSError: If the parser failed to parse the image file either 

615 because it cannot be identified or cannot be 

616 decoded. 

617 """ 

618 # finish decoding 

619 if self.decoder: 

620 # get rid of what's left in the buffers 

621 self.feed(b"") 

622 self.data = self.decoder = None 

623 if not self.finished: 

624 msg = "image was incomplete" 

625 raise OSError(msg) 

626 if not self.image: 

627 msg = "cannot parse this image" 

628 raise OSError(msg) 

629 if self.data: 

630 # incremental parsing not possible; reopen the file 

631 # not that we have all data 

632 with io.BytesIO(self.data) as fp: 

633 try: 

634 self.image = Image.open(fp) 

635 finally: 

636 self.image.load() 

637 return self.image 

638 

639 

640# -------------------------------------------------------------------- 

641 

642 

643def _save(im: Image.Image, fp: IO[bytes], tile: list[_Tile], bufsize: int = 0) -> None: 

644 """Helper to save image based on tile list 

645 

646 :param im: Image object. 

647 :param fp: File object. 

648 :param tile: Tile list. 

649 :param bufsize: Optional buffer size 

650 """ 

651 

652 im.load() 

653 if not hasattr(im, "encoderconfig"): 

654 im.encoderconfig = () 

655 tile.sort(key=_tilesort) 

656 # FIXME: make MAXBLOCK a configuration parameter 

657 # It would be great if we could have the encoder specify what it needs 

658 # But, it would need at least the image size in most cases. RawEncode is 

659 # a tricky case. 

660 bufsize = max(MAXBLOCK, bufsize, im.size[0] * 4) # see RawEncode.c 

661 try: 

662 fh = fp.fileno() 

663 fp.flush() 

664 _encode_tile(im, fp, tile, bufsize, fh) 

665 except (AttributeError, io.UnsupportedOperation) as exc: 

666 _encode_tile(im, fp, tile, bufsize, None, exc) 

667 if hasattr(fp, "flush"): 

668 fp.flush() 

669 

670 

671def _encode_tile( 

672 im: Image.Image, 

673 fp: IO[bytes], 

674 tile: list[_Tile], 

675 bufsize: int, 

676 fh: int | None, 

677 exc: BaseException | None = None, 

678) -> None: 

679 for encoder_name, extents, offset, args in tile: 

680 if offset > 0: 

681 fp.seek(offset) 

682 encoder = Image._getencoder(im.mode, encoder_name, args, im.encoderconfig) 

683 try: 

684 encoder.setimage(im.im, extents) 

685 if encoder.pushes_fd: 

686 encoder.setfd(fp) 

687 errcode = encoder.encode_to_pyfd()[1] 

688 else: 

689 if exc: 

690 # compress to Python file-compatible object 

691 while True: 

692 errcode, data = encoder.encode(bufsize)[1:] 

693 fp.write(data) 

694 if errcode: 

695 break 

696 else: 

697 # slight speedup: compress to real file object 

698 assert fh is not None 

699 errcode = encoder.encode_to_file(fh, bufsize) 

700 if errcode < 0: 

701 raise _get_oserror(errcode, encoder=True) from exc 

702 finally: 

703 encoder.cleanup() 

704 

705 

706def _safe_read(fp: IO[bytes], size: int) -> bytes: 

707 """ 

708 Reads large blocks in a safe way. Unlike fp.read(n), this function 

709 doesn't trust the user. If the requested size is larger than 

710 SAFEBLOCK, the file is read block by block. 

711 

712 :param fp: File handle. Must implement a <b>read</b> method. 

713 :param size: Number of bytes to read. 

714 :returns: A string containing <i>size</i> bytes of data. 

715 

716 Raises an OSError if the file is truncated and the read cannot be completed 

717 

718 """ 

719 if size <= 0: 

720 return b"" 

721 if size <= SAFEBLOCK: 

722 data = fp.read(size) 

723 if len(data) < size: 

724 msg = "Truncated File Read" 

725 raise OSError(msg) 

726 return data 

727 blocks: list[bytes] = [] 

728 remaining_size = size 

729 while remaining_size > 0: 

730 block = fp.read(min(remaining_size, SAFEBLOCK)) 

731 if not block: 

732 break 

733 blocks.append(block) 

734 remaining_size -= len(block) 

735 if sum(len(block) for block in blocks) < size: 

736 msg = "Truncated File Read" 

737 raise OSError(msg) 

738 return b"".join(blocks) 

739 

740 

741class PyCodecState: 

742 def __init__(self) -> None: 

743 self.xsize = 0 

744 self.ysize = 0 

745 self.xoff = 0 

746 self.yoff = 0 

747 

748 def extents(self) -> tuple[int, int, int, int]: 

749 return self.xoff, self.yoff, self.xoff + self.xsize, self.yoff + self.ysize 

750 

751 

752class PyCodec: 

753 fd: IO[bytes] | None 

754 

755 def __init__(self, mode: str, *args: Any) -> None: 

756 self.im: Image.core.ImagingCore | None = None 

757 self.state = PyCodecState() 

758 self.fd = None 

759 self.mode = mode 

760 self.init(args) 

761 

762 def init(self, args: tuple[Any, ...]) -> None: 

763 """ 

764 Override to perform codec specific initialization 

765 

766 :param args: Tuple of arg items from the tile entry 

767 :returns: None 

768 """ 

769 self.args = args 

770 

771 def cleanup(self) -> None: 

772 """ 

773 Override to perform codec specific cleanup 

774 

775 :returns: None 

776 """ 

777 pass 

778 

779 def setfd(self, fd: IO[bytes]) -> None: 

780 """ 

781 Called from ImageFile to set the Python file-like object 

782 

783 :param fd: A Python file-like object 

784 :returns: None 

785 """ 

786 self.fd = fd 

787 

788 def setimage( 

789 self, 

790 im: Image.core.ImagingCore, 

791 extents: tuple[int, int, int, int] | None = None, 

792 ) -> None: 

793 """ 

794 Called from ImageFile to set the core output image for the codec 

795 

796 :param im: A core image object 

797 :param extents: a 4 tuple of (x0, y0, x1, y1) defining the rectangle 

798 for this tile 

799 :returns: None 

800 """ 

801 

802 # following c code 

803 self.im = im 

804 

805 if extents: 

806 x0, y0, x1, y1 = extents 

807 

808 if x0 < 0 or y0 < 0 or x1 > self.im.size[0] or y1 > self.im.size[1]: 

809 msg = "Tile cannot extend outside image" 

810 raise ValueError(msg) 

811 

812 self.state.xoff = x0 

813 self.state.yoff = y0 

814 self.state.xsize = x1 - x0 

815 self.state.ysize = y1 - y0 

816 else: 

817 self.state.xsize, self.state.ysize = self.im.size 

818 

819 if self.state.xsize <= 0 or self.state.ysize <= 0: 

820 msg = "Size must be positive" 

821 raise ValueError(msg) 

822 

823 

824class PyDecoder(PyCodec): 

825 """ 

826 Python implementation of a format decoder. Override this class and 

827 add the decoding logic in the :meth:`decode` method. 

828 

829 See :ref:`Writing Your Own File Codec in Python<file-codecs-py>` 

830 """ 

831 

832 _pulls_fd = False 

833 

834 @property 

835 def pulls_fd(self) -> bool: 

836 return self._pulls_fd 

837 

838 def decode(self, buffer: Image.DecoderInput) -> tuple[int, int]: 

839 """ 

840 Override to perform the decoding process. 

841 

842 :param buffer: A bytes object with the data to be decoded. 

843 :returns: A tuple of ``(bytes consumed, errcode)``. 

844 If finished with decoding return -1 for the bytes consumed. 

845 Err codes are from :data:`.ImageFile.ERRORS`. 

846 """ 

847 msg = "unavailable in base decoder" 

848 raise NotImplementedError(msg) 

849 

850 def set_as_raw( 

851 self, 

852 data: bytes | bytearray, 

853 rawmode: str | None = None, 

854 extra: tuple[Any, ...] = (), 

855 ) -> None: 

856 """ 

857 Convenience method to set the internal image from a stream of raw data 

858 

859 :param data: Bytes to be set 

860 :param rawmode: The rawmode to be used for the decoder.