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