WebPImagePlugin.py 11 KB

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