WebPImagePlugin.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363
  1. from __future__ import annotations
  2. from io import BytesIO
  3. from typing import IO, Any
  4. from . import Image, ImageFile
  5. try:
  6. from . import _webp
  7. SUPPORTED = True
  8. except ImportError:
  9. SUPPORTED = False
  10. _VALID_WEBP_MODES = {"RGBX": True, "RGBA": True, "RGB": True}
  11. _VALID_WEBP_LEGACY_MODES = {"RGB": True, "RGBA": True}
  12. _VP8_MODES_BY_IDENTIFIER = {
  13. b"VP8 ": "RGB",
  14. b"VP8X": "RGBA",
  15. b"VP8L": "RGBA", # lossless
  16. }
  17. def _accept(prefix: bytes) -> bool | str:
  18. is_riff_file_format = prefix[:4] == b"RIFF"
  19. is_webp_file = prefix[8:12] == b"WEBP"
  20. is_valid_vp8_mode = prefix[12:16] in _VP8_MODES_BY_IDENTIFIER
  21. if is_riff_file_format and is_webp_file and is_valid_vp8_mode:
  22. if not SUPPORTED:
  23. return (
  24. "image file could not be identified because WEBP support not installed"
  25. )
  26. return True
  27. return False
  28. class WebPImageFile(ImageFile.ImageFile):
  29. format = "WEBP"
  30. format_description = "WebP image"
  31. __loaded = 0
  32. __logical_frame = 0
  33. def _open(self) -> None:
  34. if not _webp.HAVE_WEBPANIM:
  35. # Legacy mode
  36. data, width, height, self._mode, icc_profile, exif = _webp.WebPDecode(
  37. self.fp.read()
  38. )
  39. if icc_profile:
  40. self.info["icc_profile"] = icc_profile
  41. if exif:
  42. self.info["exif"] = exif
  43. self._size = width, height
  44. self.fp = BytesIO(data)
  45. self.tile = [("raw", (0, 0) + self.size, 0, self.mode)]
  46. self.n_frames = 1
  47. self.is_animated = False
  48. return
  49. # Use the newer AnimDecoder API to parse the (possibly) animated file,
  50. # and access muxed chunks like ICC/EXIF/XMP.
  51. self._decoder = _webp.WebPAnimDecoder(self.fp.read())
  52. # Get info from decoder
  53. width, height, loop_count, bgcolor, frame_count, mode = self._decoder.get_info()
  54. self._size = width, height
  55. self.info["loop"] = loop_count
  56. bg_a, bg_r, bg_g, bg_b = (
  57. (bgcolor >> 24) & 0xFF,
  58. (bgcolor >> 16) & 0xFF,
  59. (bgcolor >> 8) & 0xFF,
  60. bgcolor & 0xFF,
  61. )
  62. self.info["background"] = (bg_r, bg_g, bg_b, bg_a)
  63. self.n_frames = frame_count
  64. self.is_animated = self.n_frames > 1
  65. self._mode = "RGB" if mode == "RGBX" else mode
  66. self.rawmode = mode
  67. self.tile = []
  68. # Attempt to read ICC / EXIF / XMP chunks from file
  69. icc_profile = self._decoder.get_chunk("ICCP")
  70. exif = self._decoder.get_chunk("EXIF")
  71. xmp = self._decoder.get_chunk("XMP ")
  72. if icc_profile:
  73. self.info["icc_profile"] = icc_profile
  74. if exif:
  75. self.info["exif"] = exif
  76. if xmp:
  77. self.info["xmp"] = xmp
  78. # Initialize seek state
  79. self._reset(reset=False)
  80. def _getexif(self) -> dict[str, Any] | None:
  81. if "exif" not in self.info:
  82. return None
  83. return self.getexif()._get_merged_dict()
  84. def seek(self, frame: int) -> None:
  85. if not self._seek_check(frame):
  86. return
  87. # Set logical frame to requested position
  88. self.__logical_frame = frame
  89. def _reset(self, reset: bool = True) -> None:
  90. if reset:
  91. self._decoder.reset()
  92. self.__physical_frame = 0
  93. self.__loaded = -1
  94. self.__timestamp = 0
  95. def _get_next(self):
  96. # Get next frame
  97. ret = self._decoder.get_next()
  98. self.__physical_frame += 1
  99. # Check if an error occurred
  100. if ret is None:
  101. self._reset() # Reset just to be safe
  102. self.seek(0)
  103. msg = "failed to decode next frame in WebP file"
  104. raise EOFError(msg)
  105. # Compute duration
  106. data, timestamp = ret
  107. duration = timestamp - self.__timestamp
  108. self.__timestamp = timestamp
  109. # libwebp gives frame end, adjust to start of frame
  110. timestamp -= duration
  111. return data, timestamp, duration
  112. def _seek(self, frame: int) -> None:
  113. if self.__physical_frame == frame:
  114. return # Nothing to do
  115. if frame < self.__physical_frame:
  116. self._reset() # Rewind to beginning
  117. while self.__physical_frame < frame:
  118. self._get_next() # Advance to the requested frame
  119. def load(self):
  120. if _webp.HAVE_WEBPANIM:
  121. if self.__loaded != self.__logical_frame:
  122. self._seek(self.__logical_frame)
  123. # We need to load the image data for this frame
  124. data, timestamp, duration = self._get_next()
  125. self.info["timestamp"] = timestamp
  126. self.info["duration"] = duration
  127. self.__loaded = self.__logical_frame
  128. # Set tile
  129. if self.fp and self._exclusive_fp:
  130. self.fp.close()
  131. self.fp = BytesIO(data)
  132. self.tile = [("raw", (0, 0) + self.size, 0, self.rawmode)]
  133. return super().load()
  134. def load_seek(self, pos: int) -> None:
  135. pass
  136. def tell(self) -> int:
  137. if not _webp.HAVE_WEBPANIM:
  138. return super().tell()
  139. return self.__logical_frame
  140. def _save_all(im: Image.Image, fp: IO[bytes], filename: str | bytes) -> None:
  141. encoderinfo = im.encoderinfo.copy()
  142. append_images = list(encoderinfo.get("append_images", []))
  143. # If total frame count is 1, then save using the legacy API, which
  144. # will preserve non-alpha modes
  145. total = 0
  146. for ims in [im] + append_images:
  147. total += getattr(ims, "n_frames", 1)
  148. if total == 1:
  149. _save(im, fp, filename)
  150. return
  151. background: int | tuple[int, ...] = (0, 0, 0, 0)
  152. if "background" in encoderinfo:
  153. background = encoderinfo["background"]
  154. elif "background" in im.info:
  155. background = im.info["background"]
  156. if isinstance(background, int):
  157. # GifImagePlugin stores a global color table index in
  158. # info["background"]. So it must be converted to an RGBA value
  159. palette = im.getpalette()
  160. if palette:
  161. r, g, b = palette[background * 3 : (background + 1) * 3]
  162. background = (r, g, b, 255)
  163. else:
  164. background = (background, background, background, 255)
  165. duration = im.encoderinfo.get("duration", im.info.get("duration", 0))
  166. loop = im.encoderinfo.get("loop", 0)
  167. minimize_size = im.encoderinfo.get("minimize_size", False)
  168. kmin = im.encoderinfo.get("kmin", None)
  169. kmax = im.encoderinfo.get("kmax", None)
  170. allow_mixed = im.encoderinfo.get("allow_mixed", False)
  171. verbose = False
  172. lossless = im.encoderinfo.get("lossless", False)
  173. quality = im.encoderinfo.get("quality", 80)
  174. alpha_quality = im.encoderinfo.get("alpha_quality", 100)
  175. method = im.encoderinfo.get("method", 0)
  176. icc_profile = im.encoderinfo.get("icc_profile") or ""
  177. exif = im.encoderinfo.get("exif", "")
  178. if isinstance(exif, Image.Exif):
  179. exif = exif.tobytes()
  180. xmp = im.encoderinfo.get("xmp", "")
  181. if allow_mixed:
  182. lossless = False
  183. # Sensible keyframe defaults are from gif2webp.c script
  184. if kmin is None:
  185. kmin = 9 if lossless else 3
  186. if kmax is None:
  187. kmax = 17 if lossless else 5
  188. # Validate background color
  189. if (
  190. not isinstance(background, (list, tuple))
  191. or len(background) != 4
  192. or not all(0 <= v < 256 for v in background)
  193. ):
  194. msg = f"Background color is not an RGBA tuple clamped to (0-255): {background}"
  195. raise OSError(msg)
  196. # Convert to packed uint
  197. bg_r, bg_g, bg_b, bg_a = background
  198. background = (bg_a << 24) | (bg_r << 16) | (bg_g << 8) | (bg_b << 0)
  199. # Setup the WebP animation encoder
  200. enc = _webp.WebPAnimEncoder(
  201. im.size[0],
  202. im.size[1],
  203. background,
  204. loop,
  205. minimize_size,
  206. kmin,
  207. kmax,
  208. allow_mixed,
  209. verbose,
  210. )
  211. # Add each frame
  212. frame_idx = 0
  213. timestamp = 0
  214. cur_idx = im.tell()
  215. try:
  216. for ims in [im] + append_images:
  217. # Get # of frames in this image
  218. nfr = getattr(ims, "n_frames", 1)
  219. for idx in range(nfr):
  220. ims.seek(idx)
  221. ims.load()
  222. # Make sure image mode is supported
  223. frame = ims
  224. rawmode = ims.mode
  225. if ims.mode not in _VALID_WEBP_MODES:
  226. alpha = (
  227. "A" in ims.mode
  228. or "a" in ims.mode
  229. or (ims.mode == "P" and "A" in ims.im.getpalettemode())
  230. )
  231. rawmode = "RGBA" if alpha else "RGB"
  232. frame = ims.convert(rawmode)
  233. if rawmode == "RGB":
  234. # For faster conversion, use RGBX
  235. rawmode = "RGBX"
  236. # Append the frame to the animation encoder
  237. enc.add(
  238. frame.tobytes("raw", rawmode),
  239. round(timestamp),
  240. frame.size[0],
  241. frame.size[1],
  242. rawmode,
  243. lossless,
  244. quality,
  245. alpha_quality,
  246. method,
  247. )
  248. # Update timestamp and frame index
  249. if isinstance(duration, (list, tuple)):
  250. timestamp += duration[frame_idx]
  251. else:
  252. timestamp += duration
  253. frame_idx += 1
  254. finally:
  255. im.seek(cur_idx)
  256. # Force encoder to flush frames
  257. enc.add(None, round(timestamp), 0, 0, "", lossless, quality, alpha_quality, 0)
  258. # Get the final output from the encoder
  259. data = enc.assemble(icc_profile, exif, xmp)
  260. if data is None:
  261. msg = "cannot write file as WebP (encoder returned None)"
  262. raise OSError(msg)
  263. fp.write(data)
  264. def _save(im: Image.Image, fp: IO[bytes], filename: str | bytes) -> None:
  265. lossless = im.encoderinfo.get("lossless", False)
  266. quality = im.encoderinfo.get("quality", 80)
  267. alpha_quality = im.encoderinfo.get("alpha_quality", 100)
  268. icc_profile = im.encoderinfo.get("icc_profile") or ""
  269. exif = im.encoderinfo.get("exif", b"")
  270. if isinstance(exif, Image.Exif):
  271. exif = exif.tobytes()
  272. if exif.startswith(b"Exif\x00\x00"):
  273. exif = exif[6:]
  274. xmp = im.encoderinfo.get("xmp", "")
  275. method = im.encoderinfo.get("method", 4)
  276. exact = 1 if im.encoderinfo.get("exact") else 0
  277. if im.mode not in _VALID_WEBP_LEGACY_MODES:
  278. im = im.convert("RGBA" if im.has_transparency_data else "RGB")
  279. data = _webp.WebPEncode(
  280. im.tobytes(),
  281. im.size[0],
  282. im.size[1],
  283. lossless,
  284. float(quality),
  285. float(alpha_quality),
  286. im.mode,
  287. icc_profile,
  288. method,
  289. exact,
  290. exif,
  291. xmp,
  292. )
  293. if data is None:
  294. msg = "cannot write file as WebP (encoder returned None)"
  295. raise OSError(msg)
  296. fp.write(data)
  297. Image.register_open(WebPImageFile.format, WebPImageFile, _accept)
  298. if SUPPORTED:
  299. Image.register_save(WebPImageFile.format, _save)
  300. if _webp.HAVE_WEBPANIM:
  301. Image.register_save_all(WebPImageFile.format, _save_all)
  302. Image.register_extension(WebPImageFile.format, ".webp")
  303. Image.register_mime(WebPImageFile.format, "image/webp")