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

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 

39import sys 

40from typing import NamedTuple, cast 

41 

42from . import ExifTags, Image 

43from ._util import DeferredError, is_path 

44 

45TYPE_CHECKING = False 

46if TYPE_CHECKING: 

47 from typing import IO, Any, Self 

48 

49 from ._typing import StrOrBytesPath 

50 

51logger = logging.getLogger(__name__) 

52 

53MAXBLOCK = 65536 

54""" 

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

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

57 

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

59 

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

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

62Plugins may also set a greater number. 

63 

64User code may set this to another number. 

65""" 

66 

67SAFEBLOCK = 1024 * 1024 

68 

69LOAD_TRUNCATED_IMAGES = False 

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

71 

72ERRORS = { 

73 -1: "image buffer overrun error", 

74 -2: "decoding error", 

75 -3: "unknown error", 

76 -8: "bad configuration", 

77 -9: "out of memory error", 

78} 

79""" 

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

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

82:meth:`.PyEncoder.encode_to_file`. 

83""" 

84 

85 

86# 

87# -------------------------------------------------------------------- 

88# Helpers 

89 

90 

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

92 try: 

93 msg = Image.core.getcodecstatus(error) 

94 except AttributeError: 

95 msg = ERRORS.get(error) 

96 if not msg: 

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

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

99 return OSError(msg) 

100 

101 

102def _get_fp_length(fp: IO[bytes]) -> int: 

103 if isinstance(fp, io.BytesIO): 

104 if sys.implementation.name == "pypy": 

105 pos = fp.tell() 

106 fp.seek(0, io.SEEK_END) 

107 length = fp.tell() 

108 fp.seek(pos) 

109 return length 

110 else: 

111 return fp.getbuffer().nbytes 

112 return os.fstat(fp.fileno()).st_size 

113 

114 

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

116 return t.offset 

117 

118 

119class _Tile(NamedTuple): 

120 codec_name: str 

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

122 offset: int = 0 

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

124 

125 

126# 

127# -------------------------------------------------------------------- 

128# ImageFile base class 

129 

130 

131class ImageFile(Image.Image, metaclass=abc.ABCMeta): 

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

133 

134 def __init__( 

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

136 ) -> None: 

137 super().__init__() 

138 

139 self._min_frame = 0 

140 

141 self.custom_mimetype: str | None = None 

142 

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

144 """ A list of tile descriptors """ 

145 

146 self.readonly = 1 # until we know better 

147 

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

149 self.decodermaxblock = MAXBLOCK 

150 

151 self.fp: IO[bytes] | None 

152 self._fp: IO[bytes] | DeferredError 

153 if is_path(fp): 

154 # filename 

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

156 self.filename = os.fspath(fp) 

157 self._exclusive_fp = True 

158 else: 

159 # stream 

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

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

162 # can be overridden 

163 self._exclusive_fp = False 

164 

165 try: 

166 try: 

167 self._open() 

168 

169 if isinstance(self, StubImageFile) and self._handler: 

170 self._handler.open(self) 

171 except ( 

172 IndexError, # end of data 

173 TypeError, # end of data (ord) 

174 KeyError, # unsupported mode 

175 EOFError, # got header but not the first frame 

176 struct.error, 

177 ) as v: 

178 raise SyntaxError(v) from v 

179 

180 if not self.mode or ( 

181 min(self.size) < 0 

182 if isinstance(self, StubImageFile) and self._handler is None 

183 else min(self.size) <= 0 

184 ): 

185 msg = "not identified by this driver" 

186 raise SyntaxError(msg) 

187 except BaseException: 

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

189 if self._exclusive_fp: 

190 self.fp.close() 

191 raise 

192 

193 @abc.abstractmethod 

194 def _open(self) -> None: 

195 pass 

196 

197 def _close_fp(self) -> None: 

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

199 if self._fp != self.fp: 

200 self._fp.close() 

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

202 if self.fp: 

203 self.fp.close() 

204 

205 # Context manager support 

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

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

208 self._close_fp() 

209 self.fp = None 

210 

211 def close(self) -> None: 

212 """ 

213 Closes the file pointer, if possible. 

214 

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

216 The image data will be unusable afterward. 

217 

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

219 have not had their file read and closed by the 

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

221 more information. 

222 """ 

223 try: 

224 self._close_fp() 

225 self.fp = None 

226 except Exception as msg: 

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

228 

229 super().close() 

230 

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

232 child_images = [] 

233 exif = self.getexif() 

234 ifds = [] 

235 if ExifTags.Base.SubIFDs in exif: 

236 subifd_offsets = exif[ExifTags.Base.SubIFDs] 

237 if subifd_offsets: 

238 if not isinstance(subifd_offsets, tuple): 

239 subifd_offsets = (subifd_offsets,) 

240 ifds = [ 

241 (exif._get_ifd_dict(subifd_offset), subifd_offset) 

242 for subifd_offset in subifd_offsets 

243 ] 

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

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

246 assert exif._info is not None 

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

248 

249 offset = None 

250 for ifd, ifd_offset in ifds: 

251 assert self.fp is not None 

252 current_offset = self.fp.tell() 

253 if offset is None: 

254 offset = current_offset 

255 

256 fp = self.fp 

257 if ifd is not None: 

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

259 if thumbnail_offset is not None: 

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

261 self.fp.seek(thumbnail_offset) 

262 

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

264 assert isinstance(length, int) 

265 data = self.fp.read(length) 

266 fp = io.BytesIO(data) 

267 

268 with Image.open(fp) as im: 

269 from . import TiffImagePlugin 

270 

271 if thumbnail_offset is None and isinstance( 

272 im, TiffImagePlugin.TiffImageFile 

273 ): 

274 im._frame_pos = [ifd_offset] 

275 im._seek(0) 

276 im.load() 

277 child_images.append(im) 

278 

279 if offset is not None: 

280 assert self.fp is not None 

281 self.fp.seek(offset) 

282 return child_images 

283 

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

285 if self.custom_mimetype: 

286 return self.custom_mimetype 

287 if self.format is not None: 

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

289 return None 

290 

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

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

293 

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

295 self.tile = [] 

296 if len(state) > 5: 

297 self.filename = state[5] 

298 super().__setstate__(state) 

299 

300 def verify(self) -> None: 

301 """Check file integrity""" 

302 

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

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

305 if self._exclusive_fp and self.fp: 

306 self.fp.close() 

307 self.fp = None 

308 

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

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

311 

312 if not self.tile: 

313 if self._im is None: 

314 msg = "cannot load this image" 

315 raise OSError(msg) 

316 return Image.Image.load(self) 

317 

318 if self._im is not None: 

319 # If this image has been closed, raise an error 

320 self.im 

321 

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

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

324 

325 assert self.fp is not None 

326 readonly = 0 

327 

328 # look for read/seek overrides 

329 if hasattr(self, "load_read"): 

330 read = self.load_read 

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

332 use_mmap = False 

333 else: 

334 read = self.fp.read 

335 

336 if hasattr(self, "load_seek"): 

337 seek = self.load_seek 

338 use_mmap = False 

339 else: 

340 seek = self.fp.seek 

341 

342 if use_mmap: 

343 # try memory mapping 

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

345 if isinstance(args, str): 

346 args = (args, 0, 1) 

347 if ( 

348 decoder_name == "raw" 

349 and extents in {None, (0, 0, *self.size)} 

350 and isinstance(args, tuple) 

351 and len(args) >= 3 

352 and args[0] == self.mode 

353 and args[0] in Image._MAPMODES 

354 ): 

355 if offset < 0: 

356 msg = "Tile offset cannot be negative" 

357 raise ValueError(msg) 

358 try: 

359 # use mmap, if possible 

360 import mmap 

361 

362 with open(self.filename) as fp: 

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

364 

365 if self.mode in {"L", "P"}: 

366 pixelsize = 1 

367 elif self.mode.startswith("I;16"): 

368 pixelsize = 2 

369 else: 

370 pixelsize = 4 

371 stride = max(args[1], self.size[0] * pixelsize) 

372 if offset + self.size[1] * stride > self.map.size(): 

373 msg = "buffer is not large enough" 

374 raise OSError(msg) 

375 self.im = Image.core.map_buffer( 

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

377 ) 

378 readonly = 1 

379 # After trashing self.im, 

380 # we might need to reload the palette data. 

381 if self.palette: 

382 self.palette.dirty = 1 

383 except (AttributeError, OSError, ImportError): 

384 self.map = None 

385 

386 self.load_prepare() 

387 err_code = -3 # initialize to unknown error 

388 if not self.map: 

389 # sort tiles in file order 

390 self.tile.sort(key=_tilesort) 

391 

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

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

394 

395 # Remove consecutive duplicates that only differ by their offset 

396 self.tile = [ 

397 list(tiles)[-1] 

398 for _, tiles in itertools.groupby( 

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

400 ) 

401 ] 

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

403 seek(offset) 

404 decoder = Image._getdecoder( 

405 self.mode, decoder_name, args, self.decoderconfig 

406 ) 

407 try: 

408 decoder.setimage(self.im, extents) 

409 if decoder.pulls_fd: 

410 decoder.setfd(self.fp) 

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

412 else: 

413 b = prefix 

414 while True: 

415 read_bytes = self.decodermaxblock 

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

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

418 if next_offset > offset: 

419 read_bytes = next_offset - offset 

420 try: 

421 s = read(read_bytes) 

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

423 # truncated png/gif 

424 if LOAD_TRUNCATED_IMAGES: 

425 break 

426 else: 

427 msg = "image file is truncated" 

428 raise OSError(msg) from e 

429 

430 if not s: # truncated jpeg 

431 if LOAD_TRUNCATED_IMAGES: 

432 break 

433 else: 

434 msg = ( 

435 "image file is truncated " 

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

437 ) 

438 raise OSError(msg) 

439 

440 b += s 

441 n, err_code = decoder.decode(b) 

442 if n < 0: 

443 break 

444 b = b[n:] 

445 finally: 

446 # Need to cleanup here to prevent leaks 

447 decoder.cleanup() 

448 

449 self.tile = [] 

450 self.readonly = readonly 

451 

452 self.load_end() 

453 

454 if self._exclusive_fp and self._close_exclusive_fp_after_loading: 

455 self.fp.close() 

456 self.fp = None 

457 

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

459 # still raised if decoder fails to return anything 

460 raise _get_oserror(err_code, encoder=False) 

461 

462 return Image.Image.load(self) 

463 

464 def load_prepare(self) -> None: 

465 # create image memory if necessary 

466 if self._im is None: 

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

468 

469 def load_end(self) -> None: 

470 # may be overridden 

471 pass 

472 

473 # may be defined for contained formats 

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

475 # pass 

476 

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

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

479 # pass 

480 

481 def _get_fp_length(self, fp: IO[bytes] | None = None) -> int: 

482 if fp is None: 

483 assert self.fp is not None 

484 fp = self.fp 

485 return _get_fp_length(fp) 

486 

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

488 if ( 

489 frame < self._min_frame 

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

491 # are not required to do so 

492 or ( 

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

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

495 ) 

496 ): 

497 msg = "attempt to seek outside sequence" 

498 raise EOFError(msg) 

499 

500 return self.tell() != frame 

501 

502 

503class StubHandler(abc.ABC): 

504 @abc.abstractmethod 

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

506 pass 

507 

508 @abc.abstractmethod 

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

510 pass 

511 

512 

513class StubImageFile(ImageFile): 

514 """ 

515 Base class for stub image loaders. 

516 

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

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

519 """ 

520 

521 _handler: StubHandler | None = None 

522 

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

524 if self._handler is None: 

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

526 raise OSError(msg) 

527 image = self._handler.load(self) 

528 

529 assert self.fp is not None 

530 if self._exclusive_fp and self._close_exclusive_fp_after_loading: 

531 self.fp.close() 

532 self.fp = None 

533 

534 assert image is not None 

535 # become the other object (!) 

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

537 self.__dict__ = image.__dict__ 

538 return image.load() 

539 

540 

541class Parser: 

542 """ 

543 Incremental image parser. This class implements the standard 

544 feed/close consumer interface. 

545 """ 

546 

547 incremental = None 

548 image: Image.Image | None = None 

549 data: bytes | None = None 

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

551 offset = 0 

552 finished = 0 

553 

554 def reset(self) -> None: 

555 """ 

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

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

558 instances cannot be reused. 

559 """ 

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

561 

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

563 """ 

564 (Consumer) Feed data to the parser. 

565 

566 :param data: A string buffer. 

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

568 """ 

569 # collect data 

570 

571 if self.finished: 

572 return 

573 

574 if self.data is None: 

575 self.data = data 

576 else: 

577 self.data += data 

578 

579 # parse what we have 

580 if self.decoder: 

581 if self.offset > 0: 

582 # skip header 

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

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

585 self.offset = self.offset - skip 

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

587 return 

588 

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

590 

591 if n < 0: 

592 # end of stream 

593 self.data = None 

594 self.finished = 1 

595 if e < 0: 

596 # decoding error 

597 self.image = None 

598 raise _get_oserror(e, encoder=False) 

599 else: 

600 # end of image 

601 return 

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

603 

604 elif self.image: 

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

606 # be incrementally parsed, due to load_seek, load_read, or pulls_fd 

607 # wait until we've gotten all available data 

608 pass 

609 

610 else: 

611 # attempt to open this file 

612 try: 

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

614 im = Image.open(fp) 

615 except OSError: 

616 pass # not enough data 

617 else: 

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

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

620 # initialize decoder 

621 decoder_name, extents, offset, args = im.tile[0] 

622 decoder = Image._getdecoder( 

623 im.mode, decoder_name, args, im.decoderconfig 

624 ) 

625 # a decoder that reads from the file itself 

626 # cannot be incrementally parsed. 

627 # wait until we've gotten all available data 

628 if not decoder.pulls_fd: 

629 im.load_prepare() 

630 im.tile = [] 

631 self.decoder = decoder 

632 self.decoder.setimage(im.im, extents) 

633 

634 # calculate decoder offset 

635 self.offset = offset 

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

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

638 self.offset = 0 

639 

640 self.image = im 

641 

642 def __enter__(self) -> Self: 

643 return self 

644 

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

646 self.close() 

647 

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

649 """ 

650 (Consumer) Close the stream. 

651 

652 :returns: An image object. 

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

654 because it cannot be identified or cannot be 

655 decoded. 

656 """ 

657 # finish decoding 

658 if self.decoder: 

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

660 self.feed(b"") 

661 self.data = self.decoder = None 

662 if not self.finished: 

663 msg = "image was incomplete" 

664 raise OSError(msg) 

665 if not self.image: 

666 msg = "cannot parse this image" 

667 raise OSError(msg) 

668 if self.data: 

669 # incremental parsing not possible; reopen the file 

670 # not that we have all data 

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

672 try: 

673 self.image = Image.open(fp) 

674 finally: 

675 self.image.load() 

676 return self.image 

677 

678 

679# -------------------------------------------------------------------- 

680 

681 

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

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

684 

685 :param im: Image object. 

686 :param fp: File object. 

687 :param tile: Tile list. 

688 :param bufsize: Optional buffer size 

689 """ 

690 

691 im.load() 

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

693 im.encoderconfig = () 

694 tile.sort(key=_tilesort) 

695 # FIXME: make MAXBLOCK a configuration parameter 

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

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

698 # a tricky case. 

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

700 try: 

701 fh = fp.fileno() 

702 fp.flush() 

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

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

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

706 if hasattr(fp, "flush"): 

707 fp.flush() 

708 

709 

710def _encode_tile( 

711 im: Image.Image, 

712 fp: IO[bytes], 

713 tile: list[_Tile], 

714 bufsize: int, 

715 fh: int | None, 

716 exc: BaseException | None = None, 

717) -> None: 

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

719 if offset > 0: 

720 fp.seek(offset) 

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

722 try: 

723 encoder.setimage(im.im, extents) 

724 if encoder.pushes_fd: 

725 encoder.setfd(fp) 

726 errcode = encoder.encode_to_pyfd()[1] 

727 else: 

728 if exc: 

729 # compress to Python file-compatible object 

730 while True: 

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

732 fp.write(data) 

733 if errcode: 

734 break 

735 else: 

736 # slight speedup: compress to real file object 

737 assert fh is not None 

738 errcode = encoder.encode_to_file(fh, bufsize) 

739 if errcode < 0: 

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

741 finally: 

742 encoder.cleanup() 

743 

744 

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

746 """ 

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

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

749 SAFEBLOCK, the file is read block by block. 

750 

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

752 :param size: Number of bytes to read. 

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

754 

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

756 

757 """ 

758 if size <= 0: 

759 return b"" 

760 if size <= SAFEBLOCK: 

761 data = fp.read(size) 

762 if len(data) < size: 

763 msg = "Truncated File Read" 

764 raise OSError(msg) 

765 return data 

766 blocks: list[bytes] = [] 

767 remaining_size = size 

768 while remaining_size > 0: 

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

770 if not block: 

771 break 

772 blocks.append(block) 

773 remaining_size -= len(block) 

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

775 msg = "Truncated File Read" 

776 raise OSError(msg) 

777 return b"".join(blocks) 

778 

779 

780class PyCodecState: 

781 def __init__(self) -> None: 

782 self.xsize = 0 

783 self.ysize = 0 

784 self.xoff = 0 

785 self.yoff = 0 

786 

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

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

789 

790 

791class PyCodec: 

792 fd: IO[bytes] | None 

793 

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

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

796 self.state = PyCodecState() 

797 self.fd = None 

798 self.mode = mode 

799 self.init(args) 

800 

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

802 """ 

803 Override to perform codec specific initialization 

804 

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

806 :returns: None 

807 """ 

808 self.args = args 

809 

810 def cleanup(self) -> None: 

811 """ 

812 Override to perform codec specific cleanup 

813 

814 :returns: None 

815 """ 

816 pass 

817 

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

819 """ 

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

821 

822 :param fd: A Python file-like object 

823 :returns: None 

824 """ 

825 self.fd = fd 

826 

827 def setimage( 

828 self, 

829 im: Image.core.ImagingCore, 

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

831 ) -> None: 

832 """ 

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

834 

835 :param im: A core image object 

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

837 for this tile 

838 :returns: None 

839 """ 

840 

841 # following c code 

842 self.im = im 

843 

844 if extents: 

845 x0, y0, x1, y1 = extents 

846 

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

848 msg = "Tile cannot extend outside image" 

849 raise ValueError(msg) 

850 

851 self.state.xoff = x0