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")