Team Ai
Datasetpublic

codekingpro/portable-devtools

sourceHugging Faceupdated 5mo agoView on Hugging Face
1likes15kdownloads
Jpeg2KImagePlugin.py461 linesDownload Raw Back to PIL
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    colr = None
180
181    while header.has_next_box():
182        tbox = header.next_box_type()
183
184        if tbox == b"ihdr":
185            height, width, nc, bpc = header.read_fields(">IIHB")
186            assert isinstance(height, int)
187            assert isinstance(width, int)
188            assert isinstance(bpc, int)
189            size = (width, height)
190            if nc == 1 and (bpc & 0x7F) > 8:
191                mode = "I;16"
192            elif nc == 1:
193                mode = "L"
194            elif nc == 2:
195                mode = "LA"
196            elif nc == 3:
197                mode = "RGB"
198            elif nc == 4:
199                mode = "RGBA"
200        elif tbox == b"colr":
201            meth, _, _, enumcs = header.read_fields(">BBBI")
202            if meth == 1:
203                if enumcs in (0, 15):
204                    colr = "1"
205                elif enumcs == 12:
206                    colr = "CMYK"
207                    if nc == 4:
208                        mode = "CMYK"
209                elif enumcs == 17:
210                    colr = "L"
211        elif tbox == b"pclr" and mode in ("L", "LA") and colr not in ("1", "L"):
212            ne, npc = header.read_fields(">HB")
213            assert isinstance(ne, int)
214            assert isinstance(npc, int)
215            max_bitdepth = 0
216            for bitdepth in header.read_fields(">" + ("B" * npc)):
217                assert isinstance(bitdepth, int)
218                if bitdepth > max_bitdepth:
219                    max_bitdepth = bitdepth
220            if max_bitdepth <= 8:
221                if npc == 4:
222                    palette_mode = "CMYK" if colr == "CMYK" else "RGBA"
223                else:
224                    palette_mode = "RGB"
225                palette = ImagePalette.ImagePalette(palette_mode)
226                for i in range(ne):
227                    color: list[int] = []
228                    for value in header.read_fields(">" + ("B" * npc)):
229                        assert isinstance(value, int)
230                        color.append(value)
231                    palette.getcolor(tuple(color))
232                mode = "P" if mode == "L" else "PA"
233        elif tbox == b"res ":
234            res = header.read_boxes()
235            while res.has_next_box():
236                tres = res.next_box_type()
237                if tres == b"resc":
238                    vrcn, vrcd, hrcn, hrcd, vrce, hrce = res.read_fields(">HHHHBB")
239                    assert isinstance(vrcn, int)
240                    assert isinstance(vrcd, int)
241                    assert isinstance(hrcn, int)
242                    assert isinstance(hrcd, int)
243                    assert isinstance(vrce, int)
244                    assert isinstance(hrce, int)
245                    hres = _res_to_dpi(hrcn, hrcd, hrce)
246                    vres = _res_to_dpi(vrcn, vrcd, vrce)
247                    if hres is not None and vres is not None:
248                        dpi = (hres, vres)
249                    break
250
251    if size is None or mode is None:
252        msg = "Malformed JP2 header"
253        raise SyntaxError(msg)
254
255    return size, mode, mimetype, dpi, palette
256
257
258##
259# Image plugin for JPEG2000 images.
260
261
262class Jpeg2KImageFile(ImageFile.ImageFile):
263    format = "JPEG2000"
264    format_description = "JPEG 2000 (ISO 15444)"
265
266    def _open(self) -> None:
267        assert self.fp is not None
268        sig = self.fp.read(4)
269        if sig == b"\xff\x4f\xff\x51":
270            self.codec = "j2k"
271            self._size, self._mode = _parse_codestream(self.fp)
272            self._parse_comment()
273        else:
274            sig = sig + self.fp.read(8)
275
276            if sig == b"\x00\x00\x00\x0cjP  \x0d\x0a\x87\x0a":
277                self.codec = "jp2"
278                header = _parse_jp2_header(self.fp)
279                self._size, self._mode, self.custom_mimetype, dpi, self.palette = header
280                if dpi is not None:
281                    self.info["dpi"] = dpi
282                if self.fp.read(12).endswith(b"jp2c\xff\x4f\xff\x51"):
283                    hdr = self.fp.read(2)
284                    length = _binary.i16be(hdr)
285                    self.fp.seek(length - 2, os.SEEK_CUR)
286                    self._parse_comment()
287            else:
288                msg = "not a JPEG 2000 file"
289                raise SyntaxError(msg)
290
291        self._reduce = 0
292        self.layers = 0
293
294        fd = -1
295        length = -1
296
297        try:
298            fd = self.fp.fileno()
299            length = os.fstat(fd).st_size
300        except Exception:
301            fd = -1
302            try:
303                pos = self.fp.tell()
304                self.fp.seek(0, io.SEEK_END)
305                length = self.fp.tell()
306                self.fp.seek(pos)
307            except Exception:
308                length = -1
309
310        self.tile = [
311            ImageFile._Tile(
312                "jpeg2k",
313                (0, 0) + self.size,
314                0,
315                (self.codec, self._reduce, self.layers, fd, length),
316            )
317        ]
318
319    def _parse_comment(self) -> None:
320        assert self.fp is not None
321        while True:
322            marker = self.fp.read(2)
323            if not marker:
324                break
325            typ = marker[1]
326            if typ in (0x90, 0xD9):
327                # Start of tile or end of codestream
328                break
329            hdr = self.fp.read(2)
330            length = _binary.i16be(hdr)
331            if typ == 0x64:
332                # Comment
333                self.info["comment"] = self.fp.read(length - 2)[2:]
334                break
335            else:
336                self.fp.seek(length - 2, os.SEEK_CUR)
337
338    @property  # type: ignore[override]
339    def reduce(
340        self,
341    ) -> (
342        Callable[[int | tuple[int, int], tuple[int, int, int, int] | None], Image.Image]
343        | int
344    ):
345        # https://github.com/python-pillow/Pillow/issues/4343 found that the
346        # new Image 'reduce' method was shadowed by this plugin's 'reduce'
347        # property. This attempts to allow for both scenarios
348        return self._reduce or super().reduce
349
350    @reduce.setter
351    def reduce(self, value: int) -> None:
352        self._reduce = value
353
354    def load(self) -> Image.core.PixelAccess | None:
355        if self.tile and self._reduce:
356            power = 1 << self._reduce
357            adjust = power >> 1
358            self._size = (
359                int((self.size[0] + adjust) / power),
360                int((self.size[1] + adjust) / power),
361            )
362
363            # Update the reduce and layers settings
364            t = self.tile[0]
365            assert isinstance(t[3], tuple)
366            t3 = (t[3][0], self._reduce, self.layers, t[3][3], t[3][4])
367            self.tile = [ImageFile._Tile(t[0], (0, 0) + self.size, t[2], t3)]
368
369        return ImageFile.ImageFile.load(self)
370
371
372def _accept(prefix: bytes) -> bool:
373    return prefix.startswith(
374        (b"\xff\x4f\xff\x51", b"\x00\x00\x00\x0cjP  \x0d\x0a\x87\x0a")
375    )
376
377
378# ------------------------------------------------------------
379# Save support
380
381
382def _save(im: Image.Image, fp: IO[bytes], filename: str | bytes) -> None:
383    # Get the keyword arguments
384    info = im.encoderinfo
385
386    if isinstance(filename, str):
387        filename = filename.encode()
388    if filename.endswith(b".j2k") or info.get("no_jp2", False):
389        kind = "j2k"
390    else:
391        kind = "jp2"
392
393    offset = info.get("offset", None)
394    tile_offset = info.get("tile_offset", None)
395    tile_size = info.get("tile_size", None)
396    quality_mode = info.get("quality_mode", "rates")
397    quality_layers = info.get("quality_layers", None)
398    if quality_layers is not None and not (
399        isinstance(quality_layers, (list, tuple))
400        and all(
401            isinstance(quality_layer, (int, float)) for quality_layer in quality_layers
402        )
403    ):
404        msg = "quality_layers must be a sequence of numbers"
405        raise ValueError(msg)
406
407    num_resolutions = info.get("num_resolutions", 0)
408    cblk_size = info.get("codeblock_size", None)
409    precinct_size = info.get("precinct_size", None)
410    irreversible = info.get("irreversible", False)
411    progression = info.get("progression", "LRCP")
412    cinema_mode = info.get("cinema_mode", "no")
413    mct = info.get("mct", 0)
414    signed = info.get("signed", False)
415    comment = info.get("comment")
416    if isinstance(comment, str):
417        comment = comment.encode()
418    plt = info.get("plt", False)
419
420    fd = -1
421    if hasattr(fp, "fileno"):
422        try:
423            fd = fp.fileno()
424        except Exception:
425            fd = -1
426
427    im.encoderconfig = (
428        offset,
429        tile_offset,
430        tile_size,
431        quality_mode,
432        quality_layers,
433        num_resolutions,
434        cblk_size,
435        precinct_size,
436        irreversible,
437        progression,
438        cinema_mode,
439        mct,
440        signed,
441        fd,
442        comment,
443        plt,
444    )
445
446    ImageFile._save(im, fp, [ImageFile._Tile("jpeg2k", (0, 0) + im.size, 0, kind)])
447
448
449# ------------------------------------------------------------
450# Registry stuff
451
452
453Image.register_open(Jpeg2KImageFile.format, Jpeg2KImageFile, _accept)
454Image.register_save(Jpeg2KImageFile.format, _save)
455
456Image.register_extensions(
457    Jpeg2KImageFile.format, [".jp2", ".j2k", ".jpc", ".jpf", ".jpx", ".j2c"]
458)
459
460Image.register_mime(Jpeg2KImageFile.format, "image/jp2")
461 
codekingpro/portable-devtools · Team Ai