1#
2# The Python Imaging Library
3# $Id$
4#
5# JPEG2000 file handling
6#
7# History:
8# 2014-03-12 ajh Created
9# 2021-06-30 rogermb Extract dpi information from the 'resc' header box
10#
11# Copyright (c) 2014 Coriolis Systems Limited
12# Copyright (c) 2014 Alastair Houghton
13#
14# See the README file for information on usage and redistribution.
15#
16from __future__ import annotations
17
18import io
19import os
20import struct
21from typing import cast
22
23from . import Image, ImageFile, ImagePalette, _binary
24
25TYPE_CHECKING = False
26if TYPE_CHECKING:
27 from collections.abc import Callable
28 from typing import IO
29
30
31class BoxReader:
32 """
33 A small helper class to read fields stored in JPEG2000 header boxes
34 and to easily step into and read sub-boxes.
35 """
36
37 def __init__(self, fp: IO[bytes], length: int = -1) -> None:
38 self.fp = fp
39 self.has_length = length >= 0
40 self.length = length
41 self.remaining_in_box = -1
42
43 def _can_read(self, num_bytes: int) -> bool:
44 if self.has_length and self.fp.tell() + num_bytes > self.length:
45 # Outside box: ensure we don't read past the known file length
46 return False
47 if self.remaining_in_box >= 0:
48 # Inside box contents: ensure read does not go past box boundaries
49 return num_bytes <= self.remaining_in_box
50 else:
51 return True # No length known, just read
52
53 def _read_bytes(self, num_bytes: int) -> bytes:
54 if not self._can_read(num_bytes):
55 msg = "Not enough data in header"
56 raise SyntaxError(msg)
57
58 data = self.fp.read(num_bytes)
59 if len(data) < num_bytes:
60 msg = f"Expected to read {num_bytes} bytes but only got {len(data)}."
61 raise OSError(msg)
62
63 if self.remaining_in_box > 0:
64 self.remaining_in_box -= num_bytes
65 return data
66
67 def read_fields(self, field_format: str) -> tuple[int | bytes, ...]:
68 size = struct.calcsize(field_format)
69 data = self._read_bytes(size)
70 return struct.unpack(field_format, data)
71
72 def read_boxes(self) -> BoxReader:
73 size = self.remaining_in_box
74 data = self._read_bytes(size)
75 return BoxReader(io.BytesIO(data), size)
76
77 def has_next_box(self) -> bool:
78 if self.has_length:
79 return self.fp.tell() + self.remaining_in_box < self.length
80 else:
81 return True
82
83 def next_box_type(self) -> bytes:
84 # Skip the rest of the box if it has not been read
85 if self.remaining_in_box > 0:
86 self.fp.seek(self.remaining_in_box, os.SEEK_CUR)
87 self.remaining_in_box = -1
88
89 # Read the length and type of the next box
90 lbox, tbox = cast(tuple[int, bytes], self.read_fields(">I4s"))
91 if lbox == 1:
92 lbox = cast(int, self.read_fields(">Q")[0])
93 hlen = 16
94 else:
95 hlen = 8
96
97 if lbox < hlen or not self._can_read(lbox - hlen):
98 msg = "Invalid header length"
99 raise SyntaxError(msg)
100
101 self.remaining_in_box = lbox - hlen
102 return tbox
103
104
105def _parse_codestream(fp: IO[bytes]) -> tuple[tuple[int, int], str]:
106 """Parse the JPEG 2000 codestream to extract the size and component
107 count from the SIZ marker segment, returning a PIL (size, mode) tuple."""
108
109 hdr = fp.read(2)
110 lsiz = _binary.i16be(hdr)
111 siz = hdr + fp.read(lsiz - 2)
112 lsiz, rsiz, xsiz, ysiz, xosiz, yosiz, _, _, _, _, csiz = struct.unpack_from(
113 ">HHIIIIIIIIH", siz
114 )
115
116 size = (xsiz - xosiz, ysiz - yosiz)
117 if csiz == 1:
118 ssiz = struct.unpack_from(">B", siz, 38)
119 if (ssiz[0] & 0x7F) + 1 > 8:
120 mode = "I;16"
121 else:
122 mode = "L"
123 elif csiz == 2:
124 mode = "LA"
125 elif csiz == 3:
126 mode = "RGB"
127 elif csiz == 4:
128 mode = "RGBA"
129 else:
130 msg = "unable to determine J2K image mode"
131 raise SyntaxError(msg)
132
133 return size, mode
134
135
136def _res_to_dpi(num: int, denom: int, exp: int) -> float | None:
137 """Convert JPEG2000's (numerator, denominator, exponent-base-10) resolution,
138 calculated as (num / denom) * 10^exp and stored in dots per meter,
139 to floating-point dots per inch."""
140 if denom == 0:
141 return None
142 return (254 * num * (10**exp)) / (10000 * denom)
143
144
145def _parse_jp2_header(
146 fp: IO[bytes],
147) -> tuple[
148 tuple[int, int],
149 str,
150 str | None,
151 tuple[float, float] | None,
152 ImagePalette.ImagePalette | None,
153]:
154 """Parse the JP2 header box to extract size, component count,
155 color space information, and optionally DPI information,
156 returning a (size, mode, mimetype, dpi) tuple."""
157
158 # Find the JP2 header box
159 reader = BoxReader(fp)
160 header = None
161 mimetype = None
162 while reader.has_next_box():
163 tbox = reader.next_box_type()
164
165 if tbox == b"jp2h":
166 header = reader.read_boxes()
167 break
168 elif tbox == b"ftyp":
169 if reader.read_fields(">4s")[0] == b"jpx ":
170 mimetype = "image/jpx"
171 assert header is not None
172
173 size = None
174 mode = None
175 bpc = None
176 nc = None
177 dpi = None # 2-tuple of DPI info, or None
178 palette = None
179
180 while header.has_next_box():
181 tbox = header.next_box_type()
182
183 if tbox == b"ihdr":
184 height, width, nc, bpc = header.read_fields(">IIHB")
185 assert isinstance(height, int)
186 assert isinstance(width, int)
187 assert isinstance(bpc, int)
188 size = (width, height)
189 if nc == 1 and (bpc & 0x7F) > 8:
190 mode = "I;16"
191 elif nc == 1:
192 mode = "L"
193 elif nc == 2:
194 mode = "LA"
195 elif nc == 3:
196 mode = "RGB"
197 elif nc == 4:
198 mode = "RGBA"
199 elif tbox == b"colr" and nc == 4:
200 meth, _, _, enumcs = header.read_fields(">BBBI")
201 if meth == 1 and enumcs == 12:
202 mode = "CMYK"
203 elif tbox == b"pclr" and mode in ("L", "LA"):
204 ne, npc = header.read_fields(">HB")
205 assert isinstance(ne, int)
206 assert isinstance(npc, int)
207 max_bitdepth = 0
208 for bitdepth in header.read_fields(">" + ("B" * npc)):
209 assert isinstance(bitdepth, int)
210 if bitdepth > max_bitdepth:
211 max_bitdepth = bitdepth
212 if max_bitdepth <= 8:
213 palette = ImagePalette.ImagePalette("RGBA" if npc == 4 else "RGB")
214 for i in range(ne):
215 color: list[int] = []
216 for value in header.read_fields(">" + ("B" * npc)):
217 assert isinstance(value, int)
218 color.append(value)
219 palette.getcolor(tuple(color))
220 mode = "P" if mode == "L" else "PA"
221 elif tbox == b"res ":
222 res = header.read_boxes()
223 while res.has_next_box():
224 tres = res.next_box_type()
225 if tres == b"resc":
226 vrcn, vrcd, hrcn, hrcd, vrce, hrce = res.read_fields(">HHHHBB")
227 assert isinstance(vrcn, int)
228 assert isinstance(vrcd, int)
229 assert isinstance(hrcn, int)
230 assert isinstance(hrcd, int)
231 assert isinstance(vrce, int)
232 assert isinstance(hrce, int)
233 hres = _res_to_dpi(hrcn, hrcd, hrce)
234 vres = _res_to_dpi(vrcn, vrcd, vrce)
235 if hres is not None and vres is not None:
236 dpi = (hres, vres)
237 break
238
239 if size is None or mode is None:
240 msg = "Malformed JP2 header"
241 raise SyntaxError(msg)
242
243 return size, mode, mimetype, dpi, palette
244
245
246##
247# Image plugin for JPEG2000 images.
248
249
250class Jpeg2KImageFile(ImageFile.ImageFile):
251 format = "JPEG2000"
252 format_description = "JPEG 2000 (ISO 15444)"
253
254 def _open(self) -> None:
255 sig = self.fp.read(4)
256 if sig == b"\xff\x4f\xff\x51":
257 self.codec = "j2k"
258 self._size, self._mode = _parse_codestream(self.fp)
259 self._parse_comment()
260 else:
261 sig = sig + self.fp.read(8)
262
263 if sig == b"\x00\x00\x00\x0cjP \x0d\x0a\x87\x0a":
264 self.codec = "jp2"
265 header = _parse_jp2_header(self.fp)
266 self._size, self._mode, self.custom_mimetype, dpi, self.palette = header
267 if dpi is not None:
268 self.info["dpi"] = dpi
269 if self.fp.read(12).endswith(b"jp2c\xff\x4f\xff\x51"):
270 hdr = self.fp.read(2)
271 length = _binary.i16be(hdr)
272 self.fp.seek(length - 2, os.SEEK_CUR)
273 self._parse_comment()
274 else:
275 msg = "not a JPEG 2000 file"
276 raise SyntaxError(msg)
277
278 self._reduce = 0
279 self.layers = 0
280
281 fd = -1
282 length = -1
283
284 try:
285 fd = self.fp.fileno()
286 length = os.fstat(fd).st_size
287 except Exception:
288 fd = -1
289 try:
290 pos = self.fp.tell()
291 self.fp.seek(0, io.SEEK_END)
292 length = self.fp.tell()
293 self.fp.seek(pos)
294 except Exception:
295 length = -1
296
297 self.tile = [
298 ImageFile._Tile(
299 "jpeg2k",
300 (0, 0) + self.size,
301 0,
302 (self.codec, self._reduce, self.layers, fd, length),
303 )
304 ]
305
306 def _parse_comment(self) -> None:
307 while True:
308 marker = self.fp.read(2)
309 if not marker:
310 break
311 typ = marker[1]
312 if typ in (0x90, 0xD9):
313 # Start of tile or end of codestream
314 break
315 hdr = self.fp.read(2)
316 length = _binary.i16be(hdr)
317 if typ == 0x64:
318 # Comment
319 self.info["comment"] = self.fp.read(length - 2)[2:]
320 break
321 else:
322 self.fp.seek(length - 2, os.SEEK_CUR)
323
324 @property # type: ignore[override]
325 def reduce(
326 self,
327 ) -> (
328 Callable[[int | tuple[int, int], tuple[int, int, int, int] | None], Image.Image]
329 | int
330 ):
331 # https://github.com/python-pillow/Pillow/issues/4343 found that the
332 # new Image 'reduce' method was shadowed by this plugin's 'reduce'
333 # property. This attempts to allow for both scenarios
334 return self._reduce or super().reduce
335
336 @reduce.setter
337 def reduce(self, value: int) -> None:
338 self._reduce = value
339
340 def load(self) -> Image.core.PixelAccess | None:
341 if self.tile and self._reduce:
342 power = 1 << self._reduce
343 adjust = power >> 1
344 self._size = (
345 int((self.size[0] + adjust) / power),
346 int((self.size[1] + adjust) / power),
347 )
348
349 # Update the reduce and layers settings
350 t = self.tile[0]
351 assert isinstance(t[3], tuple)
352 t3 = (t[3][0], self._reduce, self.layers, t[3][3], t[3][4])
353 self.tile = [ImageFile._Tile(t[0], (0, 0) + self.size, t[2], t3)]
354
355 return ImageFile.ImageFile.load(self)
356
357
358def _accept(prefix: bytes) -> bool:
359 return prefix.startswith(
360 (b"\xff\x4f\xff\x51", b"\x00\x00\x00\x0cjP \x0d\x0a\x87\x0a")
361 )
362
363
364# ------------------------------------------------------------
365# Save support
366
367
368def _save(im: Image.Image, fp: IO[bytes], filename: str | bytes) -> None:
369 # Get the keyword arguments
370 info = im.encoderinfo
371
372 if isinstance(filename, str):
373 filename = filename.encode()
374 if filename.endswith(b".j2k") or info.get("no_jp2", False):
375 kind = "j2k"
376 else:
377 kind = "jp2"
378
379 offset = info.get("offset", None)
380 tile_offset = info.get("tile_offset", None)
381 tile_size = info.get("tile_size", None)
382 quality_mode = info.get("quality_mode", "rates")
383 quality_layers = info.get("quality_layers", None)
384 if quality_layers is not None and not (
385 isinstance(quality_layers, (list, tuple))
386 and all(
387 isinstance(quality_layer, (int, float)) for quality_layer in quality_layers
388 )
389 ):
390 msg = "quality_layers must be a sequence of numbers"
391 raise ValueError(msg)
392
393 num_resolutions = info.get("num_resolutions", 0)
394 cblk_size = info.get("codeblock_size", None)
395 precinct_size = info.get("precinct_size", None)
396 irreversible = info.get("irreversible", False)
397 progression = info.get("progression", "LRCP")
398 cinema_mode = info.get("cinema_mode", "no")
399 mct = info.get("mct", 0)
400 signed = info.get("signed", False)
401 comment = info.get("comment")
402 if isinstance(comment, str):
403 comment = comment.encode()
404 plt = info.get("plt", False)
405
406 fd = -1
407 if hasattr(fp, "fileno"):
408 try:
409 fd = fp.fileno()
410 except Exception:
411 fd = -1
412
413 im.encoderconfig = (
414 offset,
415 tile_offset,
416 tile_size,
417 quality_mode,
418 quality_layers,
419 num_resolutions,
420 cblk_size,
421 precinct_size,
422 irreversible,
423 progression,
424 cinema_mode,
425 mct,
426 signed,
427 fd,
428 comment,
429 plt,
430 )
431
432 ImageFile._save(im, fp, [ImageFile._Tile("jpeg2k", (0, 0) + im.size, 0, kind)])
433
434
435# ------------------------------------------------------------
436# Registry stuff
437
438
439Image.register_open(Jpeg2KImageFile.format, Jpeg2KImageFile, _accept)
440Image.register_save(Jpeg2KImageFile.format, _save)
441
442Image.register_extensions(
443 Jpeg2KImageFile.format, [".jp2", ".j2k", ".jpc", ".jpf", ".jpx", ".j2c"]
444)
445
446Image.register_mime(Jpeg2KImageFile.format, "image/jp2")