automatic/modules/postprocessing.py

import os
import tempfile

from PIL import Image

from modules import shared, images, devices, scripts_manager, scripts_postprocessing, infotext
from modules.logger import log
from modules.shared import opts
from modules.paths import resolve_output_path


def run_postprocessing(extras_mode, image, image_folder: list[tempfile.NamedTemporaryFile], input_dir, output_dir, show_extras_results, *args, save_output: bool = True):
    devices.torch_gc()
    shared.state.begin('Extras')
    image_data = []
    image_names = []
    image_fullnames = []
    image_ext = []
    outputs = []
    params = {}
    info = ''
    if extras_mode == 1:
        for img in image_folder:
            if isinstance(img, Image.Image):
                image = img
                fn = ''
                ext = None
            else:
                try:
                    image = Image.open(os.path.abspath(img.name))
                except Exception as e:
                    log.error(f'Failed to open image: file="{img.name}" {e}')
                    continue
                fn, ext = os.path.splitext(img.orig_name)
                image_fullnames.append(img.name)
            image_data.append(image)
            image_names.append(fn)
            image_ext.append(ext)
        log.debug(f'Process: mode=batch inputs={len(image_folder)} images={len(image_data)}')
    elif extras_mode == 2:
        assert input_dir, 'input directory not selected'
        image_list = os.listdir(input_dir)
        for filename in image_list:
            fn = os.path.join(input_dir, filename)
            try:
                image = Image.open(fn)
            except Exception as e:
                log.error(f'Failed to open image: file="{fn}" {e}')
                continue
            image_fullnames.append(fn)
            image_data.append(image)
            image_names.append(fn)
            image_ext.append(None)
        log.debug(f'Process: mode=folder inputs={input_dir} files={len(image_list)} images={len(image_data)}')
    else:
        image_data.append(image)
        image_names.append(None)
        image_ext.append(None)
    if extras_mode == 2 and output_dir != '':
        outpath = output_dir
    else:
        outpath = resolve_output_path(opts.outdir_samples, opts.outdir_extras_samples)
    processed_images = []
    for image, name, ext in zip(image_data, image_names, image_ext, strict=False): # pylint: disable=redefined-argument-from-local
        log.debug(f'Process: image={image} {args}')
        info = ''
        if shared.state.interrupted:
            log.debug('Postprocess interrupted')
            break
        if image is None:
            continue
        shared.state.textinfo = name
        pp = scripts_postprocessing.PostprocessedImage(image.convert("RGB"))
        scripts_manager.scripts_postproc.run(pp, args)
        geninfo, items = images.read_info_from_image(image)
        params = infotext.parse(geninfo)
        for k, v in items.items():
            pp.image.info[k] = v
        if 'parameters' in items:
            info = items['parameters'] + ', '
        if (params.get('size-1', 0) != pp.image.width) or (params.get('size-2', 0) != pp.image.height):
            params['size-1'] = pp.image.width
            params['size-2'] = pp.image.height
            info += f"Size: {pp.image.width}x{pp.image.height}, "
        info = info + ", ".join([k if k == v else f'{k}: {infotext.quote(v)}' for k, v in pp.info.items() if v is not None])
        pp.image.info["postprocessing"] = info
        processed_images.append(pp.image)
        if save_output:
            if opts.use_original_name_batch and name is not None:
                forced_filename = os.path.splitext(os.path.basename(name))[0]
                images.save_image(pp.image, path=outpath, extension=ext or opts.samples_format, info=info, grid=False, pnginfo_section_name="extras", existing_info=pp.image.info, forced_filename=forced_filename)
            else:
                images.save_image(pp.image, path=outpath, extension=ext or opts.samples_format, info=info, grid=False, pnginfo_section_name="extras", existing_info=pp.image.info)
        if extras_mode != 2 or show_extras_results:
            outputs.append(pp.image)
        image.close()
    scripts_manager.scripts_postproc.postprocess(processed_images, args)

    devices.torch_gc()
    return outputs, info, params


def run_extras(extras_mode, resize_mode, image, image_folder, input_dir, output_dir, show_extras_results, upscaling_resize, upscaling_resize_w, upscaling_resize_h, upscaling_crop, extras_upscaler_1, extras_upscaler_2, extras_upscaler_2_visibility, save_output: bool = True, script_args: dict | None = None):
    """old handler for API"""

    merged = {
        "Upscale": {
            "upscale_mode": resize_mode,
            "upscale_by": upscaling_resize,
            "upscale_to_width": upscaling_resize_w,
            "upscale_to_height": upscaling_resize_h,
            "upscale_crop": upscaling_crop,
            "upscaler_1_name": extras_upscaler_1,
            "upscaler_2_name": extras_upscaler_2,
            "upscaler_2_visibility": extras_upscaler_2_visibility,
        },
    }
    if script_args:
        for name, kvs in script_args.items():
            merged.setdefault(name, {}).update(kvs or {})
    args = scripts_manager.scripts_postproc.create_args_for_run(merged)

    return run_postprocessing(extras_mode, image, image_folder, input_dir, output_dir, show_extras_results, *args, save_output=save_output)