Coverage for /usr/lib/python3/dist-packages/PIL/ImageFile.py: 18%

2# The Python Imaging Library.

3# $Id$

5# base class for image file handlers

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#

26#

27# See the README file for information on usage and redistribution.

28#

29from __future__ import annotations

31import io

32import itertools

33import struct

34import sys

35from typing import Any, NamedTuple

37from . import Image

38from ._deprecate import deprecate

39from ._util import is_path

41MAXBLOCK = 65536

43SAFEBLOCK = 1024 * 1024

45LOAD_TRUNCATED_IMAGES = False

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

48ERRORS = {

49 -1: "image buffer overrun error",

50 -2: "decoding error",

51 -3: "unknown error",

52 -8: "bad configuration",

53 -9: "out of memory error",

54}

55"""

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

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

58:meth:`.PyEncoder.encode_to_file`.

59"""

62#

63# --------------------------------------------------------------------

64# Helpers

67def _get_oserror(error, *, encoder):

68 try:

69 msg = Image.core.getcodecstatus(error)

70 except AttributeError:

71 msg = ERRORS.get(error)

72 if not msg:

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

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

75 return OSError(msg)

78def raise_oserror(error):

79 deprecate(

80 "raise_oserror",

81 12,

82 action="It is only useful for translating error codes returned by a codec's "

83 "decode() method, which ImageFile already does automatically.",

84 )

85 raise _get_oserror(error, encoder=False)

88def _tilesort(t):

89 # sort on offset

90 return t[2]

93class _Tile(NamedTuple):

94 encoder_name: str

95 extents: tuple[int, int, int, int]

96 offset: int

97 args: tuple[Any, ...] | str | None

100#

101# --------------------------------------------------------------------

102# ImageFile base class

103

104

105class ImageFile(Image.Image):

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

107

108 def __init__(self, fp=None, filename=None):

109 super().__init__()

110

111 self._min_frame = 0

112

113 self.custom_mimetype = None

114

115 self.tile = None

116 """ A list of tile descriptors, or ``None`` """

117

118 self.readonly = 1 # until we know better

119

120 self.decoderconfig = ()

121 self.decodermaxblock = MAXBLOCK

122

123 if is_path(fp):

124 # filename

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

126 self.filename = fp

127 self._exclusive_fp = True

128 else:

129 # stream

130 self.fp = fp

131 self.filename = filename

132 # can be overridden

133 self._exclusive_fp = None

134

135 try:

136 try:

137 self._open()

138 except (

139 IndexError, # end of data

140 TypeError, # end of data (ord)

141 KeyError, # unsupported mode

142 EOFError, # got header but not the first frame

143 struct.error,

144 ) as v:

145 raise SyntaxError(v) from v

146

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

148 msg = "not identified by this driver"

149 raise SyntaxError(msg)

150 except BaseException:

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

152 if self._exclusive_fp:

153 self.fp.close()

154 raise

155

156 def get_format_mimetype(self):

157 if self.custom_mimetype:

158 return self.custom_mimetype

159 if self.format is not None:

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

161

162 def __setstate__(self, state):

163 self.tile = []

164 super().__setstate__(state)

165

166 def verify(self):

167 """Check file integrity"""

168

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

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

171 if self._exclusive_fp:

172 self.fp.close()

173 self.fp = None

174

175 def load(self):

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

177

178 if self.tile is None:

179 msg = "cannot load this image"

180 raise OSError(msg)

181

182 pixel = Image.Image.load(self)

183 if not self.tile:

184 return pixel

185

186 self.map = None

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

188 # As of pypy 2.1.0, memory mapping was failing here.

189 use_mmap = use_mmap and not hasattr(sys, "pypy_version_info")

190

191 readonly = 0

192

193 # look for read/seek overrides

194 try:

195 read = self.load_read

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

197 use_mmap = False

198 except AttributeError:

199 read = self.fp.read

200

201 try:

202 seek = self.load_seek

203 use_mmap = False

204 except AttributeError:

205 seek = self.fp.seek

206

207 if use_mmap:

208 # try memory mapping

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

210 if isinstance(args, str):

211 args = (args, 0, 1)

212 if (

213 decoder_name == "raw"

214 and len(args) >= 3

215 and args[0] == self.mode

216 and args[0] in Image._MAPMODES

217 ):

218 try:

219 # use mmap, if possible

220 import mmap

221

222 with open(self.filename) as fp:

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

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

225 msg = "buffer is not large enough"

226 raise OSError(msg)

227 self.im = Image.core.map_buffer(

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

229 )

230 readonly = 1

231 # After trashing self.im,

232 # we might need to reload the palette data.

233 if self.palette:

234 self.palette.dirty = 1

235 except (AttributeError, OSError, ImportError):

236 self.map = None

237

238 self.load_prepare()

239 err_code = -3 # initialize to unknown error

240 if not self.map:

241 # sort tiles in file order

242 self.tile.sort(key=_tilesort)

243

244 try:

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

246 prefix = self.tile_prefix

247 except AttributeError:

248 prefix = b""

249

250 # Remove consecutive duplicates that only differ by their offset

251 self.tile = [

252 list(tiles)[-1]

253 for _, tiles in itertools.groupby(

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

255 )

256 ]

257 for decoder_name, extents, offset, args in self.tile:

258 seek(offset)

259 decoder = Image._getdecoder(

260 self.mode, decoder_name, args, self.decoderconfig

261 )

262 try:

263 decoder.setimage(self.im, extents)

264 if decoder.pulls_fd:

265 decoder.setfd(self.fp)

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

267 else:

268 b = prefix

269 while True:

270 try:

271 s = read(self.decodermaxblock)

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

273 # truncated png/gif

274 if LOAD_TRUNCATED_IMAGES:

275 break

276 else:

277 msg = "image file is truncated"

278 raise OSError(msg) from e

279

280 if not s: # truncated jpeg

281 if LOAD_TRUNCATED_IMAGES:

282 break

283 else:

284 msg = (

285 "image file is truncated "

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

287 )

288 raise OSError(msg)

289

290 b = b + s

291 n, err_code = decoder.decode(b)

292 if n < 0:

293 break

294 b = b[n:]

295 finally:

296 # Need to cleanup here to prevent leaks

297 decoder.cleanup()

298

299 self.tile = []

300 self.readonly = readonly

301

302 self.load_end()

303

304 if self._exclusive_fp and self._close_exclusive_fp_after_loading:

305 self.fp.close()

306 self.fp = None

307

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

309 # still raised if decoder fails to return anything

310 raise _get_oserror(err_code, encoder=False)

311

312 return Image.Image.load(self)

313

314 def load_prepare(self):

315 # create image memory if necessary

316 if not self.im or self.im.mode != self.mode or self.im.size != self.size:

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

318 # create palette (optional)

319 if self.mode == "P":

320 Image.Image.load(self)

321

322 def load_end(self):

323 # may be overridden

324 pass

325

326 # may be defined for contained formats

327 # def load_seek(self, pos):

328 # pass

329

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

331 # def load_read(self, bytes):

332 # pass

333

334 def _seek_check(self, frame):

335 if (

336 frame < self._min_frame

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

338 # are not required to do so

339 or (

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

341 and frame >= self.n_frames + self._min_frame

342 )

343 ):

344 msg = "attempt to seek outside sequence"

345 raise EOFError(msg)

346

347 return self.tell() != frame

348

349

350class StubImageFile(ImageFile):

351 """

352 Base class for stub image loaders.

353

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

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

356 """

357

358 def _open(self):

359 msg = "StubImageFile subclass must implement _open"

360 raise NotImplementedError(msg)

361

362 def load(self):

363 loader = self._load()

364 if loader is None:

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

366 raise OSError(msg)

367 image = loader.load(self)

368 assert image is not None

369 # become the other object (!)

370 self.__class__ = image.__class__

371 self.__dict__ = image.__dict__

372 return image.load()

373

374 def _load(self):

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

376 msg = "StubImageFile subclass must implement _load"

377 raise NotImplementedError(msg)

378

379

380class Parser:

381 """

382 Incremental image parser. This class implements the standard

383 feed/close consumer interface.

384 """

385

386 incremental = None

387 image = None

388 data = None

389 decoder = None

390 offset = 0

391 finished = 0

392

393 def reset(self):

394 """

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

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

397 instances cannot be reused.

398 """

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

400

401 def feed(self, data):

402 """

403 (Consumer) Feed data to the parser.

404

405 :param data: A string buffer.

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

407 """

408 # collect data

409

410 if self.finished:

411 return

412

413 if self.data is None:

414 self.data = data

415 else:

416 self.data = self.data + data

417

418 # parse what we have

419 if self.decoder:

420 if self.offset > 0:

421 # skip header

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

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

424 self.offset = self.offset - skip

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

426 return

427

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

429

430 if n < 0:

431 # end of stream

432 self.data = None

433 self.finished = 1

434 if e < 0:

435 # decoding error

436 self.image = None

437 raise _get_oserror(e, encoder=False)

438 else:

439 # end of image

440 return

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

442

443 elif self.image:

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

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

446 # available data

447 pass

448

449 else:

450 # attempt to open this file

451 try:

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

453 im = Image.open(fp)

454 except OSError:

455 pass # not enough data

456 else:

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

458 if flag or len(im.tile) != 1:

459 # custom load code, or multiple tiles

460 self.decode = None

461 else:

462 # initialize decoder

463 im.load_prepare()

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

465 im.tile = []

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

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

468

469 # calculate decoder offset

470 self.offset = o

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

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

473 self.offset = 0

474

475 self.image = im

476

477 def __enter__(self):

478 return self

479

480 def __exit__(self, *args):

481 self.close()

482

483 def close(self):

484 """

485 (Consumer) Close the stream.

486

487 :returns: An image object.

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

489 because it cannot be identified or cannot be

490 decoded.

491 """

492 # finish decoding

493 if self.decoder:

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

495 self.feed(b"")

496 self.data = self.decoder = None

497 if not self.finished:

498 msg = "image was incomplete"

499 raise OSError(msg)

500 if not self.image:

501 msg = "cannot parse this image"

502 raise OSError(msg)

503 if self.data:

504 # incremental parsing not possible; reopen the file

505 # not that we have all data

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

507 try:

508 self.image = Image.open(fp)

509 finally:

510 self.image.load()

511 return self.image

512

513

514# --------------------------------------------------------------------

515

516

517def _save(im, fp, tile, bufsize=0):

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

519

520 :param im: Image object.

521 :param fp: File object.

522 :param tile: Tile list.

523 :param bufsize: Optional buffer size

524 """

525

526 im.load()

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

528 im.encoderconfig = ()

529 tile.sort(key=_tilesort)

530 # FIXME: make MAXBLOCK a configuration parameter

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

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

533 # a tricky case.

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

535 try:

536 fh = fp.fileno()

537 fp.flush()

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

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

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

541 if hasattr(fp, "flush"):

542 fp.flush()

543

544

545def _encode_tile(im, fp, tile: list[_Tile], bufsize, fh, exc=None):

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

547 if offset > 0:

548 fp.seek(offset)

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

550 try:

551 encoder.setimage(im.im, extents)

552 if encoder.pushes_fd:

553 encoder.setfd(fp)

554 errcode = encoder.encode_to_pyfd()[1]

555 else:

556 if exc:

557 # compress to Python file-compatible object

558 while True:

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

560 fp.write(data)

561 if errcode:

562 break

563 else:

564 # slight speedup: compress to real file object

565 errcode = encoder.encode_to_file(fh, bufsize)

566 if errcode < 0:

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

568 finally:

569 encoder.cleanup()

570

571

572def _safe_read(fp, size):

573 """

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

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

576 SAFEBLOCK, the file is read block by block.

577

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

579 :param size: Number of bytes to read.

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

581

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

583

584 """

585 if size <= 0:

586 return b""

587 if size <= SAFEBLOCK:

588 data = fp.read(size)

589 if len(data) < size:

590 msg = "Truncated File Read"

591 raise OSError(msg)

592 return data

593 data = []

594 remaining_size = size

595 while remaining_size > 0:

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

597 if not block:

598 break

599 data.append(block)

600 remaining_size -= len(block)

601 if sum(len(d) for d in data) < size:

602 msg = "Truncated File Read"

603 raise OSError(msg)

604 return b"".join(data)

605

606

607class PyCodecState:

608 def __init__(self):

609 self.xsize = 0

610 self.ysize = 0

611 self.xoff = 0

612 self.yoff = 0

613

614 def extents(self):

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

616

617

618class PyCodec:

619 def __init__(self, mode, *args):

620 self.im = None

621 self.state = PyCodecState()

622 self.fd = None

623 self.mode = mode

624 self.init(args)

625

626 def init(self, args):

627 """

628 Override to perform codec specific initialization

629

630 :param args: Array of args items from the tile entry

631 :returns: None

632 """

633 self.args = args

634

635 def cleanup(self):

636 """

637 Override to perform codec specific cleanup

638

639 :returns: None

640 """

641 pass

642

643 def setfd(self, fd):

644 """

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

646

647 :param fd: A Python file-like object

648 :returns: None

649 """

650 self.fd = fd

651

652 def setimage(self, im, extents=None):

653 """

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

655

656 :param im: A core image object

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

658 for this tile

659 :returns: None

660 """

661

662 # following c code

663 self.im = im

664

665 if extents:

666 (x0, y0, x1, y1) = extents

667 else:

668 (x0, y0, x1, y1) = (0, 0, 0, 0)

669

670 if x0 == 0 and x1 == 0:

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

672 else:

673 self.state.xoff = x0

674 self.state.yoff = y0

675 self.state.xsize = x1 - x0

676 self.state.ysize = y1 - y0

677

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

679 msg = "Size cannot be negative"

680 raise ValueError(msg)

681

682 if (

683 self.state.xsize + self.state.xoff > self.im.size[0]

684 or self.state.ysize + self.state.yoff > self.im.size[1]

685 ):

686 msg = "Tile cannot extend outside image"

687 raise ValueError(msg)

688

689

690class PyDecoder(PyCodec):

691 """

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

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

694

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

696 """

697

698 _pulls_fd = False

699

700 @property

701 def pulls_fd(self):

702 return self._pulls_fd

703

704 def decode(self, buffer):

705 """

706 Override to perform the decoding process.

707

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

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

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

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

712 """

713 msg = "unavailable in base decoder"

714 raise NotImplementedError(msg)

715

716 def set_as_raw(self, data, rawmode=None):

717 """

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

719

720 :param data: Bytes to be set

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

722 If not specified, it will default to the mode of the image

723 :returns: None

724 """

725

726 if not rawmode:

727 rawmode = self.mode

728 d = Image._getdecoder(self.mode, "raw", rawmode)

729 d.setimage(self.im, self.state.extents())

730 s = d.decode(data)

731

732 if s[0] >= 0:

733 msg = "not enough image data"

734 raise ValueError(msg)

735 if s[1] != 0:

736 msg = "cannot decode image data"

737 raise ValueError(msg)

738

739

740class PyEncoder(PyCodec):

741 """

742 Python implementation of a format encoder. Override this class and

743 add the decoding logic in the :meth:`encode` method.

744

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

746 """

747

748 _pushes_fd = False

749

750 @property

751 def pushes_fd(self):

752 return self._pushes_fd

753

754 def encode(self, bufsize):

755 """

756 Override to perform the encoding process.

757

758 :param bufsize: Buffer size.

759 :returns: A tuple of ``(bytes encoded, errcode, bytes)``.

760 If finished with encoding return 1 for the error code.

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

762 """

763 msg = "unavailable in base encoder"

764 raise NotImplementedError(msg)

765

766 def encode_to_pyfd(self):

767 """

768 If ``pushes_fd`` is ``True``, then this method will be used,

769 and ``encode()`` will only be called once.

770

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

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

773 """

774 if not self.pushes_fd:

775 return 0, -8 # bad configuration

776 bytes_consumed, errcode, data = self.encode(0)

777 if data:

778 self.fd.write(data)

779 return bytes_consumed, errcode

780

781 def encode_to_file(self, fh, bufsize):

782 """

783 :param fh: File handle.

784 :param bufsize: Buffer size.

785

786 :returns: If finished successfully, return 0.

787 Otherwise, return an error code. Err codes are from

788 :data:`.ImageFile.ERRORS`.

789 """

790 errcode = 0

791 while errcode == 0:

792 status, errcode, buf = self.encode(bufsize)

793 if status > 0:

794 fh.write(buf[status:])

795 return errcode