klammertext/sks/image/image_cache.py

"""
Does every target has a preferred format?
  HTML - JPEG (or SVG if available?)
  LaTeX - PDF or EPS?
  Plaintext - A URL? 
  EPUB - JPEG?
"""

import sys, os, glob, shutil, subprocess, re, codecs
import filecmp
import pdf_image
import kutil

import OpenImageIO as oiio

def find_ext(filenames, ext):
    for f in filenames:
        if f.endswith(ext):
            return f
    return None

def cache_update_required(source, cached):
    return cached is None \
        or not(os.path.exists(cached)) \
        or (os.path.getmtime(cached) < os.path.getmtime(source))


def image_dimensions(filename):
    resize = True
    if filename.endswith(".pdf"):
        width, height = pdf_image.dimensions(filename)
        resize = False
    else:
        inp = oiio.ImageInput.open(filename)
        if not inp:
            raise Exception(f'Cannot read image "{filename}": {oiio.geterror()}')
        spec = inp.spec()
        width, height = spec.width, spec.height
        inp.close()
    return width, height, resize


# Raster formats a photograph arrives in.  Vector sources (pdf, eps) are
# never re-encoded, nor is a gif or webp (they may be animated, and only
# html takes them).
photograph_formats = {"jpg", "jpeg", "png", "tif", "tiff"}

# Every extension an @image source may have: the targets' preferred formats
# plus what a conversion accepts (OpenImageIO for the rasters, pdftoppm for
# a pdf).  find_file() admits no other file -- the glob "basename.*" also
# returns a basename.kt, .aux or .log sitting beside a figure, and the old
# fallback took the first of those blindly, so a document whose PDF figure
# shared its stem with its source got "Cannot read image" (2026-09-25).
image_formats = {"jpg", "jpeg", "png", "gif", "webp", "tif", "tiff", "bmp", "pdf", "eps"}

def is_photograph(filename):
    """True if the file is a raster with no alpha channel: a photograph
    (or a grayscale one) that JPEG stores well.  A png with transparency,
    or a palette image, is not one -- JPEG would flatten or posterize it."""
    ext = os.path.splitext(filename)[1][1:].lower()
    if ext not in photograph_formats:
        return False
    inp = oiio.ImageInput.open(filename)
    if not inp:
        return False
    spec = inp.spec()
    inp.close()
    return spec.alpha_channel < 0 and spec.nchannels in (1, 3)


class Image_cache:
    max_width = 1600 # Should be a state variable
    cache_basedir = "_klammertext_imgsrc" # Should be a state variable
    verbose_label = "     [klammertext]" # Should parameterize verbosity
    def __init__(self, cache_dir, search_path, verbose=False, quality=100):
        #self.cache_dir = kutil.cache_directory(
        self.dir = kutil.cache_directory(cache_dir, "_image_cache")
        # JPEG quality (1-100) at which a photograph is written to the cache.
        # 100 means the source is copied or converted as it is (the
        # behaviour before 2026-09-09); below it, every photograph -- a
        # JPEG source included -- is re-encoded at that quality, which is
        # where a document's size mostly lives: a source JPEG is often
        # near-lossless (quality 95+), and a png photograph is stored
        # losslessly, at three to six times the size of a quality-85 JPEG
        # with no visible difference at print resolution.  The Image_quality
        # state variable (sks/image/image.k) sets it.
        try:
            self.quality = int(quality)
        except (TypeError, ValueError):
            raise Exception(f'Image_quality must be an integer from 1 to 100, not "{quality}"')
        if not 1 <= self.quality <= 100:
            raise Exception(f"Image_quality must be from 1 to 100, not {self.quality}")
        self.search = search_path.split(" ")
        self.verbose = verbose or os.environ.get("KLAMMERTEXT_IMAGE_CACHE_DISPLAY")
        self.required_format = { "html"  : ["jpg", "jpeg", "JPEG", "png", "gif", "webp"],
                                 "pdf"   : ["png", "jpg", "jpeg", "JPEG", "pdf", "eps"],
                                 "tex"   : ["png", "jpg", "jpeg", "JPEG", "pdf", "eps"],
                                 "latex" : ["png", "jpg", "jpeg", "JPEG", "pdf", "eps"] }
        if not os.path.exists(self.dir):
            os.makedirs(self.dir)
        self.search = self.search + [f"{os.path.dirname(cache_dir)}/imgsrc"]

    def construct_cache_filename(self, target, filename):
        basename, ext = os.path.splitext(filename)
        required = self.required_format[target]
        if ext in self.required_format[target]:
            cache_ext = ext
        else:
            cache_ext = required[0]
        return f"{self.dir}/{filename}.{cache_ext}"


    def photograph_cache_filename(self, basename):
        # The quality is part of the name, so changing it re-encodes and
        # two documents at different qualities do not fight over one file.
        return f"{self.dir}/{basename}.q{self.quality}.jpg"

    def reencode(self, target, source_filename):
        """Whether this source is written to the cache as a quality-N JPEG."""
        return self.quality < 100 \
            and target in self.required_format \
            and is_photograph(source_filename)

    def find_file(self, target, basename):
        result = None
        candidates = []
        matches = []
        for path in self.search:
            candidates += glob.glob(f"{path}/{basename}.*")
        candidates = [c for c in candidates
                      if os.path.splitext(c)[1][1:].lower() in image_formats]
        candidates = sorted(candidates, key=lambda p: 0 if p.endswith(".png") else 1)
        if False:
            print("Candidates:")
            for c in candidates:
                print(f" {c}")
        if candidates:
            matches = []
            for ext in self.required_format[target]:
                candidate = find_ext(candidates, ext)
                if candidate:
                    matches.append(candidate)
            if not matches:
                result = candidates[0]
        else:
            # An error, as any other missing input is: a document with an
            # image that is not there does not render.  (Until 2026-09-25
            # this substituted a red marker image and printed a message --
            # to a stdout the embedded interpreter captures, so the marker
            # appeared and nothing was said.)
            raise Exception(
                f'No image with basename "{basename}" found in the image search path: '
                + " | ".join(self.search))

        result = matches[0] if matches else candidates[0]
        """
        if len(matches) > 1:
            target = f"for .{target} format"
            print(f'{Image_cache.verbose_label}  Warning: Multiple matches for "{basename}" {target}:')
            sp = " "*len(Image_cache.verbose_label)
            for f in matches:
                print(f"{sp}    {f}")
        """
        return result


    def find_cached_file(self, target, basename):
        exts = self.required_format.get(target)
        if exts is None:
            raise Exception(
                f'No preferred image format defined for target "{target}"')
        result = None
        for ext in exts:
            cached_filename = f"{self.dir}/{basename}.{ext}"
            if os.path.exists(cached_filename):
                return cached_filename
        return None

        
    def describe_caching(self, source_filename, cached_filename, copied,
                         width, height, new_width, new_height):
        src = os.path.relpath(source_filename)
        if src[:2] == "..":
            src = source_filename
        dest = os.path.splitext(cached_filename)[1]
        if cached_filename.endswith(f".q{self.quality}.jpg"):
            dest = f"{dest} (quality {self.quality})"
        dim = "" if width == new_width else f" [{new_width}x{new_height}]"
        desc = "(copied)" if copied else f"-> {dest}{dim}"
        #print(f"{Image_cache.verbose_label}    To cache: {src} [{width}x{height}] {desc}")
        kutil.msg("To cache: {src} [{width}x{height}] {desc}")

    def update_cache(self, target, basename, source_filename):
        cached_filename = self.construct_cache_filename(target, basename)
        width, height, resize = image_dimensions(source_filename)
        new_width = width
        new_height = height

        source_format = os.path.splitext(source_filename)[-1][1:]
        #print("Source:", source_format, basename)
        if source_format in self.required_format[target]:
            cached_filename = f"{os.path.dirname(cached_filename)}/{os.path.basename(source_filename)}"

        copied = False
        if self.reencode(target, source_filename):
            cached_filename = self.photograph_cache_filename(basename)
            buf = oiio.ImageBuf(source_filename)
            if buf.has_error:
                raise Exception(f'Cannot read image "{source_filename}": {buf.geterror()}')
            if width > Image_cache.max_width:
                new_width = Image_cache.max_width
                new_height = int(round(new_width * float(height) / float(width)))
                roi = oiio.ROI(0, new_width, 0, new_height, 0, 1, 0, buf.nchannels)
                buf = oiio.ImageBufAlgo.resize(buf, roi=roi)
            buf.specmod().attribute("CompressionQuality", self.quality)
            if not buf.write(cached_filename):
                raise Exception(f'Cannot write "{cached_filename}": {buf.geterror()}')
        elif resize and width > Image_cache.max_width:
            new_width = Image_cache.max_width
            new_height = int(round(new_width * float(height) / float(width)))
            if source_filename.endswith(".pdf"):
                pdf_image.convert(source_filename, cached_filename)
                buf = oiio.ImageBuf(cached_filename)
            else:
                buf = oiio.ImageBuf(source_filename)
            roi = oiio.ROI(0, new_width, 0, new_height, 0, 1, 0, buf.nchannels)
            resized = oiio.ImageBufAlgo.resize(buf, roi=roi)
            resized.write(cached_filename)
        elif source_format in self.required_format[target]:
            #print("COPY", source_filename, cached_filename)
            if not os.path.exists(cached_filename) or \
               not filecmp.cmp(source_filename, cached_filename, shallow=False):
                shutil.copy2(source_filename, cached_filename)
            copied = True
        else:
            try:
                if source_filename.endswith(".pdf"):
                    pdf_image.convert(source_filename, cached_filename)
                else:
                    buf = oiio.ImageBuf(source_filename)
                    if not buf.has_error:
                        buf.write(cached_filename)
                    else:
                        raise Exception(buf.geterror())
            except Exception as err:
                # Raised, not printed and exited: the print would go to the
                # captured stdout and say nothing; the error report says it.
                raise Exception(f'Cannot convert "{source_filename}" to '
                                f'"{os.path.splitext(cached_filename)[-1][1:]}": {err}')

        if self.verbose:
            self.describe_caching(source_filename, cached_filename, copied,
                                  width, height, new_width, new_height)
        return cached_filename, new_width, new_height


    def get(self, target, basename):
        cached_filename = self.find_cached_file(target, basename)
        source_filename = self.find_file(target, basename)
        if self.reencode(target, source_filename):
            # A copy cached at another quality (or untouched, by an earlier
            # run at 100) does not serve: the wanted file is the quality-N
            # one, and cache_update_required() makes it if it is missing.
            cached_filename = self.photograph_cache_filename(basename)
        if cache_update_required(source_filename, cached_filename):
            cached_filename, width, height = self.update_cache(target, basename, source_filename)
        else:
            width, height, _ = image_dimensions(cached_filename)
        if self.verbose:
            base = os.path.basename(cached_filename)
            #print(f"{Image_cache.verbose_label}  From cache: {base} [{width}x{height}]")
            kutil.msg("From cache: {base} [{width}x{height}]")
        return cached_filename, width, height

    def clear(self):
        shutil.rmtree(self.dir)
        os.makedirs(self.dir)
            
if __name__ == '__main__':

    def getter(format, basename):
        C.get(format, basename)
        C.get(format, basename)

    def label(format=""):
        print("-"*80)
        print(format)

    home = os.environ.get("HOME")
    C = Image_cache(f"{home}",
        f"{home}/projects/klammertext/K/doc/handbook/imgsrc".split())
    a = "heringsdorf_stairwell"
    b = "usedom_clouds"

    C.clear()
    label("HTML")
    getter("html", a)
    getter("html", b)
    label("LATEX")
    getter("latex", a)
    getter("latex", b)
    label()