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

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 

31__lazy_modules__ = {"PIL._util", "io", "itertools", "struct"} 

32 

33import abc 

34import io 

35import itertools 

36import logging 

37import os 

38import struct 

39from typing import NamedTuple, cast 

40 

41from . import ExifTags, Image 

42from ._util import DeferredError, is_path 

43 

44TYPE_CHECKING = False 

45if TYPE_CHECKING: 

46 from typing import IO, Any, Self 

47 

48 from ._typing import StrOrBytesPath 

49 

50logger = logging.getLogger(__name__) 

51 

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``. 

56 

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

58 

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. 

62 

63User code may set this to another number. 

64""" 

65 

66SAFEBLOCK = 1024 * 1024 

67 

68LOAD_TRUNCATED_IMAGES = False 

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

70 

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""" 

83 

84 

85# 

86# -------------------------------------------------------------------- 

87# Helpers 

88 

89 

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) 

99 

100 

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

102 # sort on offset 

103 return t[2] 

104 

105 

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 

111 

112 

113# 

114# -------------------------------------------------------------------- 

115# ImageFile base class 

116 

117 

118class ImageFile(Image.Image): 

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

120 

121 def __init__( 

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

123 ) -> None: 

124 super().__init__() 

125 

126 self._min_frame = 0 

127 

128 self.custom_mimetype: str | None = None 

129 

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

131 """ A list of tile descriptors """ 

132 

133 self.readonly = 1 # until we know better 

134 

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

136 self.decodermaxblock = MAXBLOCK 

137 

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 

151 

152 try: 

153 try: 

154 self._open() 

155 

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 

166 

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 

179 

180 def _open(self) -> None: 

181 pass 

182 

183 def _close_fp(self) -> None: 

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

185 if self._fp != self.fp: 

186 self._fp.close() 

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

188 if self.fp: 

189 self.fp.close() 

190 

191 # Context manager support 

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

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

194 self._close_fp() 

195 self.fp = None 

196 

197 def close(self) -> None: 

198 """ 

199 Closes the file pointer, if possible. 

200 

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

202 The image data will be unusable afterward. 

203 

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

205 have not had their file read and closed by the 

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

207 more information. 

208 """ 

209 try: 

210 self._close_fp() 

211 self.fp = None 

212 except Exception as msg: 

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

214 

215 super().close() 

216 

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

218 child_images = [] 

219 exif = self.getexif() 

220 ifds = [] 

221 if ExifTags.Base.SubIFDs in exif: 

222 subifd_offsets = exif[ExifTags.Base.SubIFDs] 

223 if subifd_offsets: 

224 if not isinstance(subifd_offsets, tuple): 

225 subifd_offsets = (subifd_offsets,) 

226 ifds = [ 

227 (exif._get_ifd_dict(subifd_offset), subifd_offset) 

228 for subifd_offset in subifd_offsets 

229 ] 

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

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

232 assert exif._info is not None 

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

234 

235 offset = None 

236 for ifd, ifd_offset in ifds: 

237 assert self.fp is not None 

238 current_offset = self.fp.tell() 

239 if offset is None: 

240 offset = current_offset 

241 

242 fp = self.fp 

243 if ifd is not None: 

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

245 if thumbnail_offset is not None: 

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

247 self.fp.seek(thumbnail_offset) 

248 

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

250 assert isinstance(length, int) 

251 data = self.fp.read(length) 

252 fp = io.BytesIO(data) 

253 

254 with Image.open(fp) as im: 

255 from . import TiffImagePlugin 

256 

257 if thumbnail_offset is None and isinstance( 

258 im, TiffImagePlugin.TiffImageFile 

259 ): 

260 im._frame_pos = [ifd_offset] 

261 im._seek(0) 

262 im.load() 

263 child_images.append(im) 

264 

265 if offset is not None: 

266 assert self.fp is not None 

267 self.fp.seek(offset) 

268 return child_images 

269 

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

271 if self.custom_mimetype: 

272 return self.custom_mimetype 

273 if self.format is not None: 

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

275 return None 

276 

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

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

279 

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

281 self.tile = [] 

282 if len(state) > 5: 

283 self.filename = state[5] 

284 super().__setstate__(state) 

285 

286 def verify(self) -> None: 

287 """Check file integrity""" 

288 

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

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

291 if self._exclusive_fp and self.fp: 

292 self.fp.close() 

293 self.fp = None 

294 

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

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

297 

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

299 msg = "cannot load this image" 

300 raise OSError(msg) 

301 

302 pixel = Image.Image.load(self) 

303 if not self.tile: 

304 return pixel 

305 

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

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

308 

309 assert self.fp is not None 

310 readonly = 0 

311 

312 # look for read/seek overrides 

313 if hasattr(self, "load_read"): 

314 read = self.load_read 

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

316 use_mmap = False 

317 else: 

318 read = self.fp.read 

319 

320 if hasattr(self, "load_seek"): 

321 seek = self.load_seek 

322 use_mmap = False 

323 else: 

324 seek = self.fp.seek 

325 

326 if use_mmap: 

327 # try memory mapping 

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

329 if isinstance(args, str): 

330 args = (args, 0, 1) 

331 if ( 

332 decoder_name == "raw" 

333 and isinstance(args, tuple) 

334 and len(args) >= 3 

335 and args[0] == self.mode 

336 and args[0] in Image._MAPMODES 

337 ): 

338 if offset < 0: 

339 msg = "Tile offset cannot be negative" 

340 raise ValueError(msg) 

341 try: 

342 # use mmap, if possible 

343 import mmap 

344 

345 with open(self.filename) as fp: 

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

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

348 msg = "buffer is not large enough" 

349 raise OSError(msg) 

350 self.im = Image.core.map_buffer( 

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

352 ) 

353 readonly = 1 

354 # After trashing self.im, 

355 # we might need to reload the palette data. 

356 if self.palette: 

357 self.palette.dirty = 1 

358 except (AttributeError, OSError, ImportError): 

359 self.map = None 

360 

361 self.load_prepare() 

362 err_code = -3 # initialize to unknown error 

363 if not self.map: 

364 # sort tiles in file order 

365 self.tile.sort(key=_tilesort) 

366 

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

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

369 

370 # Remove consecutive duplicates that only differ by their offset 

371 self.tile = [ 

372 list(tiles)[-1] 

373 for _, tiles in itertools.groupby( 

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

375 ) 

376 ] 

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

378 seek(offset) 

379 decoder = Image._getdecoder( 

380 self.mode, decoder_name, args, self.decoderconfig 

381 ) 

382 try: 

383 decoder.setimage(self.im, extents) 

384 if decoder.pulls_fd: 

385 decoder.setfd(self.fp) 

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

387 else: 

388 b = prefix 

389 while True: 

390 read_bytes = self.decodermaxblock 

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

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

393 if next_offset > offset: 

394 read_bytes = next_offset - offset 

395 try: 

396 s = read(read_bytes) 

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

398 # truncated png/gif 

399 if LOAD_TRUNCATED_IMAGES: 

400 break 

401 else: 

402 msg = "image file is truncated" 

403 raise OSError(msg) from e 

404 

405 if not s: # truncated jpeg 

406 if LOAD_TRUNCATED_IMAGES: 

407 break 

408 else: 

409 msg = ( 

410 "image file is truncated " 

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

412 ) 

413 raise OSError(msg) 

414 

415 b = b + s 

416 n, err_code = decoder.decode(b) 

417 if n < 0: 

418 break 

419 b = b[n:] 

420 finally: 

421 # Need to cleanup here to prevent leaks 

422 decoder.cleanup() 

423 

424 self.tile = [] 

425 self.readonly = readonly 

426 

427 self.load_end() 

428 

429 if self._exclusive_fp and self._close_exclusive_fp_after_loading: 

430 self.fp.close() 

431 self.fp = None 

432 

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

434 # still raised if decoder fails to return anything 

435 raise _get_oserror(err_code, encoder=False) 

436 

437 return Image.Image.load(self) 

438 

439 def load_prepare(self) -> None: 

440 # create image memory if necessary 

441 if self._im is None: 

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

443 # create palette (optional) 

444 if self.mode == "P": 

445 Image.Image.load(self) 

446 

447 def load_end(self) -> None: 

448 # may be overridden 

449 pass 

450 

451 # may be defined for contained formats 

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

453 # pass 

454 

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

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

457 # pass 

458 

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

460 if ( 

461 frame < self._min_frame 

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

463 # are not required to do so 

464 or ( 

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

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

467 ) 

468 ): 

469 msg = "attempt to seek outside sequence" 

470 raise EOFError(msg) 

471 

472 return self.tell() != frame 

473 

474 

475class StubHandler(abc.ABC): 

476 @abc.abstractmethod 

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

478 pass 

479 

480 @abc.abstractmethod 

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

482 pass 

483 

484 

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

486 """ 

487 Base class for stub image loaders. 

488 

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

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

491 """ 

492 

493 _handler: StubHandler | None = None 

494 

495 @abc.abstractmethod 

496 def _open(self) -> None: 

497 pass 

498 

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

500 if self._handler is None: 

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

502 raise OSError(msg) 

503 image = self._handler.load(self) 

504 assert image is not None 

505 # become the other object (!) 

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

507 self.__dict__ = image.__dict__ 

508 return image.load() 

509 

510 

511class Parser: 

512 """ 

513 Incremental image parser. This class implements the standard 

514 feed/close consumer interface. 

515 """ 

516 

517 incremental = None 

518 image: Image.Image | None = None 

519 data: bytes | None = None 

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

521 offset = 0 

522 finished = 0 

523 

524 def reset(self) -> None: 

525 """ 

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

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

528 instances cannot be reused. 

529 """ 

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

531 

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

533 """ 

534 (Consumer) Feed data to the parser. 

535 

536 :param data: A string buffer. 

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

538 """ 

539 # collect data 

540 

541 if self.finished: 

542 return 

543 

544 if self.data is None: 

545 self.data = data 

546 else: 

547 self.data = self.data + data 

548 

549 # parse what we have 

550 if self.decoder: 

551 if self.offset > 0: 

552 # skip header 

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

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

555 self.offset = self.offset - skip 

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

557 return 

558 

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

560 

561 if n < 0: 

562 # end of stream 

563 self.data = None 

564 self.finished = 1 

565 if e < 0: 

566 # decoding error 

567 self.image = None 

568 raise _get_oserror(e, encoder=False) 

569 else: 

570 # end of image 

571 return 

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

573 

574 elif self.image: 

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

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

577 # available data 

578 pass 

579 

580 else: 

581 # attempt to open this file 

582 try: 

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

584 im = Image.open(fp) 

585 except OSError: 

586 pass # not enough data 

587 else: 

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

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

590 # initialize decoder 

591 im.load_prepare() 

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

593 im.tile = [] 

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

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

596 

597 # calculate decoder offset 

598 self.offset = o 

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

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

601 self.offset = 0 

602 

603 self.image = im 

604 

605 def __enter__(self) -> Self: 

606 return self 

607 

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

609 self.close() 

610 

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

612 """ 

613 (Consumer) Close the stream. 

614 

615 :returns: An image object. 

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

617 because it cannot be identified or cannot be 

618 decoded. 

619 """ 

620 # finish decoding 

621 if self.decoder: 

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

623 self.feed(b"") 

624 self.data = self.decoder = None 

625 if not self.finished: 

626 msg = "image was incomplete" 

627 raise OSError(msg) 

628 if not self.image: 

629 msg = "cannot parse this image" 

630 raise OSError(msg) 

631 if self.data: 

632 # incremental parsing not possible; reopen the file 

633 # not that we have all data 

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

635 try: 

636 self.image = Image.open(fp) 

637 finally: 

638 self.image.load() 

639 return self.image 

640 

641 

642# -------------------------------------------------------------------- 

643 

644 

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

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

647 

648 :param im: Image object. 

649 :param fp: File object. 

650 :param tile: Tile list. 

651 :param bufsize: Optional buffer size 

652 """ 

653 

654 im.load() 

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

656 im.encoderconfig = () 

657 tile.sort(key=_tilesort) 

658 # FIXME: make MAXBLOCK a configuration parameter 

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

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

661 # a tricky case. 

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

663 try: 

664 fh = fp.fileno() 

665 fp.flush() 

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

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

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

669 if hasattr(fp, "flush"): 

670 fp.flush() 

671 

672 

673def _encode_tile( 

674 im: Image.Image, 

675 fp: IO[bytes], 

676 tile: list[_Tile], 

677 bufsize: int, 

678 fh: int | None, 

679 exc: BaseException | None = None, 

680) -> None: 

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

682 if offset > 0: 

683 fp.seek(offset) 

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

685 try: 

686 encoder.setimage(im.im, extents) 

687 if encoder.pushes_fd: 

688 encoder.setfd(fp) 

689 errcode = encoder.encode_to_pyfd()[1] 

690 else: 

691 if exc: 

692 # compress to Python file-compatible object 

693 while True: 

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

695 fp.write(data) 

696 if errcode: 

697 break 

698 else: 

699 # slight speedup: compress to real file object 

700 assert fh is not None 

701 errcode = encoder.encode_to_file(fh, bufsize) 

702 if errcode < 0: 

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

704 finally: 

705 encoder.cleanup() 

706 

707 

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

709 """ 

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

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

712 SAFEBLOCK, the file is read block by block. 

713 

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

715 :param size: Number of bytes to read. 

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

717 

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

719 

720 """ 

721 if size <= 0: 

722 return b"" 

723 if size <= SAFEBLOCK: 

724 data = fp.read(size) 

725 if len(data) < size: 

726 msg = "Truncated File Read" 

727 raise OSError(msg) 

728 return data 

729 blocks: list[bytes] = [] 

730 remaining_size = size 

731 while remaining_size > 0: 

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

733 if not block: 

734 break 

735 blocks.append(block) 

736 remaining_size -= len(block) 

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

738 msg = "Truncated File Read" 

739 raise OSError(msg) 

740 return b"".join(blocks) 

741 

742 

743class PyCodecState: 

744 def __init__(self) -> None: 

745 self.xsize = 0 

746 self.ysize = 0 

747 self.xoff = 0 

748 self.yoff = 0 

749 

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

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

752 

753 

754class PyCodec: 

755 fd: IO[bytes] | None 

756 

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

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

759 self.state = PyCodecState() 

760 self.fd = None 

761 self.mode = mode 

762 self.init(args) 

763 

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

765 """ 

766 Override to perform codec specific initialization 

767 

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

769 :returns: None 

770 """ 

771 self.args = args 

772 

773 def cleanup(self) -> None: 

774 """ 

775 Override to perform codec specific cleanup 

776 

777 :returns: None 

778 """ 

779 pass 

780 

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

782 """ 

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

784 

785 :param fd: A Python file-like object 

786 :returns: None 

787 """ 

788 self.fd = fd 

789 

790 def setimage( 

791 self, 

792 im: Image.core.ImagingCore, 

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

794 ) -> None: 

795 """ 

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

797 

798 :param im: A core image object 

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

800 for this tile 

801 :returns: None 

802 """ 

803 

804 # following c code 

805 self.im = im 

806 

807 if extents: 

808 x0, y0, x1, y1 = extents 

809 

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

811 msg = "Tile cannot extend outside image" 

812 raise ValueError(msg) 

813 

814 self.state.xoff = x0 

815 self.state.yoff = y0 

816 self.state.xsize = x1 - x0 

817 self.state.ysize = y1 - y0 

818 else: 

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

820 

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

822 msg = "Size must be positive" 

823 raise ValueError(msg) 

824 

825 

826class PyDecoder(PyCodec): 

827 """ 

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

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

830 

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

832 """ 

833 

834 _pulls_fd = False 

835 

836 @property 

837 def pulls_fd(self) -> bool: 

838 return self._pulls_fd 

839 

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

841 """ 

842 Override to perform the decoding process. 

843 

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

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

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

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

848 """ 

849 msg = "unavailable in base decoder" 

850 raise NotImplementedError(msg) 

851 

852 def set_as_raw( 

853 self, 

854 data: bytes | bytearray, 

855 rawmode: str | None = None, 

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

857 ) -> None: 

858 """