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

477 statements  

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, metaclass=abc.ABCMeta): 

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 @abc.abstractmethod 

181 def _open(self) -> None: 

182 pass 

183 

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() 

191 

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 

197 

198 def close(self) -> None: 

199 """ 

200 Closes the file pointer, if possible. 

201 

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

203 The image data will be unusable afterward. 

204 

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) 

215 

216 super().close() 

217 

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

235 

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 

242 

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) 

249 

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

251 assert isinstance(length, int) 

252 data = self.fp.read(length) 

253 fp = io.BytesIO(data) 

254 

255 with Image.open(fp) as im: 

256 from . import TiffImagePlugin 

257 

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) 

265 

266 if offset is not None: 

267 assert self.fp is not None 

268 self.fp.seek(offset) 

269 return child_images 

270 

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 

277 

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

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

280 

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) 

286 

287 def verify(self) -> None: 

288 """Check file integrity""" 

289 

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 

295 

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

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

298 

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

300 msg = "cannot load this image" 

301 raise OSError(msg) 

302 

303 pixel = Image.Image.load(self) 

304 if not self.tile: 

305 return pixel 

306 

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

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

309 

310 assert self.fp is not None 

311 readonly = 0 

312 

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 

320 

321 if hasattr(self, "load_seek"): 

322 seek = self.load_seek 

323 use_mmap = False 

324 else: 

325 seek = self.fp.seek 

326 

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 

345 

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 

361 

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) 

367 

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

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

370 

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 

405 

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) 

415 

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() 

424 

425 self.tile = [] 

426 self.readonly = readonly 

427 

428 self.load_end() 

429 

430 if self._exclusive_fp and self._close_exclusive_fp_after_loading: 

431 self.fp.close() 

432 self.fp = None 

433 

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) 

437 

438 return Image.Image.load(self) 

439 

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) 

447 

448 def load_end(self) -> None: 

449 # may be overridden 

450 pass 

451 

452 # may be defined for contained formats 

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

454 # pass 

455 

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

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

458 # pass 

459 

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) 

472 

473 return self.tell() != frame 

474 

475 

476class StubHandler(abc.ABC): 

477 @abc.abstractmethod 

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

479 pass 

480 

481 @abc.abstractmethod 

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

483 pass 

484 

485 

486class StubImageFile(ImageFile): 

487 """ 

488 Base class for stub image loaders. 

489 

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

493 

494 _handler: StubHandler | None = None 

495 

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() 

506 

507 

508class Parser: 

509 """ 

510 Incremental image parser. This class implements the standard 

511 feed/close consumer interface. 

512 """ 

513 

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 

520 

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" 

528 

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

530 """ 

531 (Consumer) Feed data to the parser. 

532 

533 :param data: A string buffer. 

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

535 """ 

536 # collect data 

537 

538 if self.finished: 

539 return 

540 

541 if self.data is None: 

542 self.data = data 

543 else: 

544 self.data = self.data + data 

545 

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 

555 

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

557 

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:] 

570 

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 

576 

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) 

593 

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 

599 

600 self.image = im 

601 

602 def __enter__(self) -> Self: 

603 return self 

604 

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

606 self.close() 

607 

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

609 """ 

610 (Consumer) Close the stream. 

611 

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 

637 

638 

639# -------------------------------------------------------------------- 

640 

641 

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

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

644 

645 :param im: Image object. 

646 :param fp: File object. 

647 :param tile: Tile list. 

648 :param bufsize: Optional buffer size 

649 """ 

650 

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() 

668 

669 

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() 

703 

704 

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. 

710 

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. 

714 

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

716 

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) 

738 

739 

740class PyCodecState: 

741 def __init__(self) -> None: 

742 self.xsize = 0 

743 self.ysize = 0 

744 self.xoff = 0 

745 self.yoff = 0 

746 

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

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

749 

750 

751class PyCodec: 

752 fd: IO[bytes] | None 

753 

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) 

760 

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

762 """ 

763 Override to perform codec specific initialization 

764 

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

766 :returns: None 

767 """ 

768 self.args = args 

769 

770 def cleanup(self) -> None: 

771 """ 

772 Override to perform codec specific cleanup 

773 

774 :returns: None 

775 """ 

776 pass 

777 

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

779 """ 

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

781 

782 :param fd: A Python file-like object 

783 :returns: None 

784 """ 

785 self.fd = fd 

786 

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 

794 

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

800 

801 # following c code 

802 self.im = im 

803 

804 if extents: 

805 x0, y0, x1, y1 = extents 

806 

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) 

810 

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 

817 

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

819 msg = "Size must be positive" 

820 raise ValueError(msg) 

821 

822 

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. 

827 

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

829 """ 

830 

831 _pulls_fd = False 

832 

833 @property 

834 def pulls_fd(self) -> bool: 

835 return self._pulls_fd 

836 

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

838 """ 

839 Override to perform the decoding process. 

840 

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) 

848 

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 

857 

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

864 

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) 

871 

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) 

878 

879 

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. 

884 

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

886 """ 

887 

888 _pushes_fd = False 

889 

890 @property 

891 def pushes_fd(self) -> bool: 

892 return self._pushes_fd 

893 

894 def encode(self, bufsize: int) -> tuple[int, int, bytes]: 

895 """ 

896 Override to perform the encoding process. 

897 

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) 

905 

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. 

910 

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 

921 

922 def encode_to_file(self, fh: int, bufsize: int) -> int: 

923 """ 

924 :param fh: File handle. 

925 :param bufsize: Buffer size. 

926 

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