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

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

169 statements  

1from __future__ import annotations 

2 

3import os 

4from io import BytesIO 

5from typing import IO 

6 

7from . import ExifTags, Image, ImageFile 

8 

9try: 

10 from . import _avif 

11 

12 SUPPORTED = True 

13except ImportError: 

14 SUPPORTED = False 

15 

16# Decoder options as module globals, until there is a way to pass parameters 

17# to Image.open (see https://github.com/python-pillow/Pillow/issues/569) 

18DECODE_CODEC_CHOICE = "auto" 

19DEFAULT_MAX_THREADS = 0 

20 

21 

22def get_codec_version(codec_name: str) -> str | None: 

23 versions = _avif.codec_versions() 

24 for version in versions.split(", "): 

25 if version.split(" [")[0] == codec_name: 

26 return version.split(":")[-1].split(" ")[0] 

27 return None 

28 

29 

30def _accept(prefix: bytes) -> bool | str: 

31 if prefix[4:8] != b"ftyp": 

32 return False 

33 major_brand = prefix[8:12] 

34 if major_brand in ( 

35 # coding brands 

36 b"avif", 

37 b"avis", 

38 # We accept files with AVIF container brands; we can't yet know if 

39 # the ftyp box has the correct compatible brands, but if it doesn't 

40 # then the plugin will raise a SyntaxError which Pillow will catch 

41 # before moving on to the next plugin that accepts the file. 

42 # 

43 # Also, because this file might not actually be an AVIF file, we 

44 # don't raise an error if AVIF support isn't properly compiled. 

45 b"mif1", 

46 b"msf1", 

47 ): 

48 if not SUPPORTED: 

49 return ( 

50 "image file could not be identified because AVIF support not installed" 

51 ) 

52 return True 

53 return False 

54 

55 

56def _get_default_max_threads() -> int: 

57 if DEFAULT_MAX_THREADS: 

58 return DEFAULT_MAX_THREADS 

59 if hasattr(os, "sched_getaffinity"): 

60 return len(os.sched_getaffinity(0)) 

61 else: 

62 return os.cpu_count() or 1 

63 

64 

65class AvifImageFile(ImageFile.ImageFile): 

66 format = "AVIF" 

67 format_description = "AVIF image" 

68 __frame = -1 

69 

70 def _open(self) -> None: 

71 if not SUPPORTED: 

72 msg = "image file could not be opened because AVIF support not installed" 

73 raise SyntaxError(msg) 

74 

75 if DECODE_CODEC_CHOICE != "auto" and not _avif.decoder_codec_available( 

76 DECODE_CODEC_CHOICE 

77 ): 

78 msg = "Invalid opening codec" 

79 raise ValueError(msg) 

80 self._decoder = _avif.AvifDecoder( 

81 self.fp.read(), 

82 DECODE_CODEC_CHOICE, 

83 _get_default_max_threads(), 

84 ) 

85 

86 # Get info from decoder 

87 self._size, self.n_frames, self._mode, icc, exif, exif_orientation, xmp = ( 

88 self._decoder.get_info() 

89 ) 

90 self.is_animated = self.n_frames > 1 

91 

92 if icc: 

93 self.info["icc_profile"] = icc 

94 if xmp: 

95 self.info["xmp"] = xmp 

96 

97 if exif_orientation != 1 or exif: 

98 exif_data = Image.Exif() 

99 if exif: 

100 exif_data.load(exif) 

101 original_orientation = exif_data.get(ExifTags.Base.Orientation, 1) 

102 else: 

103 original_orientation = 1 

104 if exif_orientation != original_orientation: 

105 exif_data[ExifTags.Base.Orientation] = exif_orientation 

106 exif = exif_data.tobytes() 

107 if exif: 

108 self.info["exif"] = exif 

109 self.seek(0) 

110 

111 def seek(self, frame: int) -> None: 

112 if not self._seek_check(frame): 

113 return 

114 

115 # Set tile 

116 self.__frame = frame 

117 self.tile = [ImageFile._Tile("raw", (0, 0) + self.size, 0, self.mode)] 

118 

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

120 if self.tile: 

121 # We need to load the image data for this frame 

122 data, timescale, pts_in_timescales, duration_in_timescales = ( 

123 self._decoder.get_frame(self.__frame) 

124 ) 

125 self.info["timestamp"] = round(1000 * (pts_in_timescales / timescale)) 

126 self.info["duration"] = round(1000 * (duration_in_timescales / timescale)) 

127 

128 if self.fp and self._exclusive_fp: 

129 self.fp.close() 

130 self.fp = BytesIO(data) 

131 

132 return super().load() 

133 

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

135 pass 

136 

137 def tell(self) -> int: 

138 return self.__frame 

139 

140 

141def _save_all(im: Image.Image, fp: IO[bytes], filename: str | bytes) -> None: 

142 _save(im, fp, filename, save_all=True) 

143 

144 

145def _save( 

146 im: Image.Image, fp: IO[bytes], filename: str | bytes, save_all: bool = False 

147) -> None: 

148 info = im.encoderinfo.copy() 

149 if save_all: 

150 append_images = list(info.get("append_images", [])) 

151 else: 

152 append_images = [] 

153 

154 total = 0 

155 for ims in [im] + append_images: 

156 total += getattr(ims, "n_frames", 1) 

157 

158 quality = info.get("quality", 75) 

159 if not isinstance(quality, int) or quality < 0 or quality > 100: 

160 msg = "Invalid quality setting" 

161 raise ValueError(msg) 

162 

163 duration = info.get("duration", 0) 

164 subsampling = info.get("subsampling", "4:2:0") 

165 speed = info.get("speed", 6) 

166 max_threads = info.get("max_threads", _get_default_max_threads()) 

167 codec = info.get("codec", "auto") 

168 if codec != "auto" and not _avif.encoder_codec_available(codec): 

169 msg = "Invalid saving codec" 

170 raise ValueError(msg) 

171 range_ = info.get("range", "full") 

172 tile_rows_log2 = info.get("tile_rows", 0) 

173 tile_cols_log2 = info.get("tile_cols", 0) 

174 alpha_premultiplied = bool(info.get("alpha_premultiplied", False)) 

175 autotiling = bool(info.get("autotiling", tile_rows_log2 == tile_cols_log2 == 0)) 

176 

177 icc_profile = info.get("icc_profile", im.info.get("icc_profile")) 

178 exif_orientation = 1 

179 if exif := info.get("exif"): 

180 if isinstance(exif, Image.Exif): 

181 exif_data = exif 

182 else: 

183 exif_data = Image.Exif() 

184 exif_data.load(exif) 

185 if ExifTags.Base.Orientation in exif_data: 

186 exif_orientation = exif_data.pop(ExifTags.Base.Orientation) 

187 exif = exif_data.tobytes() if exif_data else b"" 

188 elif isinstance(exif, Image.Exif): 

189 exif = exif_data.tobytes() 

190 

191 xmp = info.get("xmp") 

192 

193 if isinstance(xmp, str): 

194 xmp = xmp.encode("utf-8") 

195 

196 advanced = info.get("advanced") 

197 if advanced is not None: 

198 if isinstance(advanced, dict): 

199 advanced = advanced.items() 

200 try: 

201 advanced = tuple(advanced) 

202 except TypeError: 

203 invalid = True 

204 else: 

205 invalid = any(not isinstance(v, tuple) or len(v) != 2 for v in advanced) 

206 if invalid: 

207 msg = ( 

208 "advanced codec options must be a dict of key-value string " 

209 "pairs or a series of key-value two-tuples" 

210 ) 

211 raise ValueError(msg) 

212 

213 # Setup the AVIF encoder 

214 enc = _avif.AvifEncoder( 

215 im.size, 

216 subsampling, 

217 quality, 

218 speed, 

219 max_threads, 

220 codec, 

221 range_, 

222 tile_rows_log2, 

223 tile_cols_log2, 

224 alpha_premultiplied, 

225 autotiling, 

226 icc_profile or b"", 

227 exif or b"", 

228 exif_orientation, 

229 xmp or b"", 

230 advanced, 

231 ) 

232 

233 # Add each frame 

234 frame_idx = 0 

235 frame_duration = 0 

236 cur_idx = im.tell() 

237 is_single_frame = total == 1 

238 try: 

239 for ims in [im] + append_images: 

240 # Get number of frames in this image 

241 nfr = getattr(ims, "n_frames", 1) 

242 

243 for idx in range(nfr): 

244 ims.seek(idx) 

245 

246 # Make sure image mode is supported 

247 frame = ims 

248 rawmode = ims.mode 

249 if ims.mode not in {"RGB", "RGBA"}: 

250 rawmode = "RGBA" if ims.has_transparency_data else "RGB" 

251 frame = ims.convert(rawmode) 

252 

253 # Update frame duration 

254 if isinstance(duration, (list, tuple)): 

255 frame_duration = duration[frame_idx] 

256 else: 

257 frame_duration = duration 

258 

259 # Append the frame to the animation encoder 

260 enc.add( 

261 frame.tobytes("raw", rawmode), 

262 frame_duration, 

263 frame.size, 

264 rawmode, 

265 is_single_frame, 

266 ) 

267 

268 # Update frame index 

269 frame_idx += 1 

270 

271 if not save_all: 

272 break 

273 

274 finally: 

275 im.seek(cur_idx) 

276 

277 # Get the final output from the encoder 

278 data = enc.finish() 

279 if data is None: 

280 msg = "cannot write file as AVIF (encoder returned None)" 

281 raise OSError(msg) 

282 

283 fp.write(data) 

284 

285 

286Image.register_open(AvifImageFile.format, AvifImageFile, _accept) 

287if SUPPORTED: 

288 Image.register_save(AvifImageFile.format, _save) 

289 Image.register_save_all(AvifImageFile.format, _save_all) 

290 Image.register_extensions(AvifImageFile.format, [".avif", ".avifs"]) 

291 Image.register_mime(AvifImageFile.format, "image/avif")