Update app.py

app.py

@@ -1,45 +1,336 @@
-import os
+import argparse
 import cv2
-import requests
+import glob
+import numpy as np
 import gradio as gr
-from PIL import Image
+from collections import OrderedDict
+import os
 import torch
-import shutil
+import requests
+from PIL import Image
+from models.network_swin2sr import Swin2SR as net
+from utils import util_calculate_psnr_ssim as util
 
-model_path = 'experiments/pretrained_models/Swin2SR_RealworldSR_X4_64_BSRGAN_PSNR.pth'
 
-if os.path.exists(model_path):
-    print(f'loading model from {model_path}')        
-else:
-    os.makedirs(os.path.dirname(model_path), exist_ok=True)
-    url = 'https://github.com/mv-lab/swin2sr/releases/download/v0.0.1/{}'.format(os.path.basename(model_path))
-    r = requests.get(url, allow_redirects=True)
-    print(f'downloading model {model_path}')
-    open(model_path, 'wb').write(r.content)
+def setup_model(args):
 
+    model = define_model(args)
+    model.eval()
+    model = model.to(device)
+    
+    return model
 
-os.makedirs("test", exist_ok=True)    
+def main(img):
 
-def inference(img):
+    # setup folder and path
     basewidth = 256
     wpercent = (basewidth/float(img.size[0]))
     hsize = int((float(img.size[1])*float(wpercent)))
     img = img.resize((basewidth,hsize), Image.ANTIALIAS)
     img.save("test/1.png", "PNG")
-    os.system('python main_test_swin2sr.py --task real_sr --model_path experiments/pretrained_models/Swin2SR_RealworldSR_X4_64_BSRGAN_PSNR.pth --folder_lq test --scale 4')
-    return 'results/swin2sr_real_sr_x4/1_Swin2SR.png'
+    
+    folder, save_dir, border, window_size = setup(args)
+    os.makedirs(save_dir, exist_ok=True)
+    test_results = OrderedDict()
+    test_results['psnr'] = []
+    test_results['ssim'] = []
+    test_results['psnr_y'] = []
+    test_results['ssim_y'] = []
+    test_results['psnrb'] = []
+    test_results['psnrb_y'] = []
+    psnr, ssim, psnr_y, ssim_y, psnrb, psnrb_y = 0, 0, 0, 0, 0, 0
+
+    for idx, path in enumerate(sorted(glob.glob(os.path.join(folder, '*')))):
+        # read image
+        imgname, img_lq, img_gt = get_image_pair(args, path)  # image to HWC-BGR, float32
+        img_lq = np.transpose(img_lq if img_lq.shape[2] == 1 else img_lq[:, :, [2, 1, 0]], (2, 0, 1))  # HCW-BGR to CHW-RGB
+        img_lq = torch.from_numpy(img_lq).float().unsqueeze(0).to(device)  # CHW-RGB to NCHW-RGB
+
+        # inference
+        with torch.no_grad():
+            # pad input image to be a multiple of window_size
+            _, _, h_old, w_old = img_lq.size()
+            h_pad = (h_old // window_size + 1) * window_size - h_old
+            w_pad = (w_old // window_size + 1) * window_size - w_old
+            img_lq = torch.cat([img_lq, torch.flip(img_lq, [2])], 2)[:, :, :h_old + h_pad, :]
+            img_lq = torch.cat([img_lq, torch.flip(img_lq, [3])], 3)[:, :, :, :w_old + w_pad]
+            output = test(img_lq, model, args, window_size)
+            
+            if args.task == 'compressed_sr':
+                output = output[0][..., :h_old * args.scale, :w_old * args.scale]
+            else:
+                output = output[..., :h_old * args.scale, :w_old * args.scale]
+
+        # save image
+        output = output.data.squeeze().float().cpu().clamp_(0, 1).numpy()
+        if output.ndim == 3:
+            output = np.transpose(output[[2, 1, 0], :, :], (1, 2, 0))  # CHW-RGB to HCW-BGR
+        output = (output * 255.0).round().astype(np.uint8)  # float32 to uint8
+        cv2.imwrite(f'{save_dir}/{imgname}_Swin2SR.png', output)
+
+            
+        # evaluate psnr/ssim/psnr_b
+        if img_gt is not None:
+            img_gt = (img_gt * 255.0).round().astype(np.uint8)  # float32 to uint8
+            img_gt = img_gt[:h_old * args.scale, :w_old * args.scale, ...]  # crop gt
+            img_gt = np.squeeze(img_gt)
+
+            psnr = util.calculate_psnr(output, img_gt, crop_border=border)
+            ssim = util.calculate_ssim(output, img_gt, crop_border=border)
+            test_results['psnr'].append(psnr)
+            test_results['ssim'].append(ssim)
+            if img_gt.ndim == 3:  # RGB image
+                psnr_y = util.calculate_psnr(output, img_gt, crop_border=border, test_y_channel=True)
+                ssim_y = util.calculate_ssim(output, img_gt, crop_border=border, test_y_channel=True)
+                test_results['psnr_y'].append(psnr_y)
+                test_results['ssim_y'].append(ssim_y)
+            if args.task in ['jpeg_car', 'color_jpeg_car']:
+                psnrb = util.calculate_psnrb(output, img_gt, crop_border=border, test_y_channel=False)
+                test_results['psnrb'].append(psnrb)
+                if args.task in ['color_jpeg_car']:
+                    psnrb_y = util.calculate_psnrb(output, img_gt, crop_border=border, test_y_channel=True)
+                    test_results['psnrb_y'].append(psnrb_y)
+            print('Testing {:d} {:20s} - PSNR: {:.2f} dB; SSIM: {:.4f}; PSNRB: {:.2f} dB;'
+                  'PSNR_Y: {:.2f} dB; SSIM_Y: {:.4f}; PSNRB_Y: {:.2f} dB.'.
+                  format(idx, imgname, psnr, ssim, psnrb, psnr_y, ssim_y, psnrb_y))
+        else:
+            print('Testing {:d} {:20s}'.format(idx, imgname))
+
+    # summarize psnr/ssim
+    if img_gt is not None:
+        ave_psnr = sum(test_results['psnr']) / len(test_results['psnr'])
+        ave_ssim = sum(test_results['ssim']) / len(test_results['ssim'])
+        print('\n{} \n-- Average PSNR/SSIM(RGB): {:.2f} dB; {:.4f}'.format(save_dir, ave_psnr, ave_ssim))
+        if img_gt.ndim == 3:
+            ave_psnr_y = sum(test_results['psnr_y']) / len(test_results['psnr_y'])
+            ave_ssim_y = sum(test_results['ssim_y']) / len(test_results['ssim_y'])
+            print('-- Average PSNR_Y/SSIM_Y: {:.2f} dB; {:.4f}'.format(ave_psnr_y, ave_ssim_y))
+        if args.task in ['jpeg_car', 'color_jpeg_car']:
+            ave_psnrb = sum(test_results['psnrb']) / len(test_results['psnrb'])
+            print('-- Average PSNRB: {:.2f} dB'.format(ave_psnrb))
+            if args.task in ['color_jpeg_car']:
+                ave_psnrb_y = sum(test_results['psnrb_y']) / len(test_results['psnrb_y'])
+                print('-- Average PSNRB_Y: {:.2f} dB'.format(ave_psnrb_y))
+                
+    return f"results/swin2sr_{args.task}_x{args.scale}/1_Swin2SR.png"
+
+
+def define_model(args):
+    # 001 classical image sr
+    if args.task == 'classical_sr':
+        model = net(upscale=args.scale, in_chans=3, img_size=args.training_patch_size, window_size=8,
+                    img_range=1., depths=[6, 6, 6, 6, 6, 6], embed_dim=180, num_heads=[6, 6, 6, 6, 6, 6],
+                    mlp_ratio=2, upsampler='pixelshuffle', resi_connection='1conv')
+        param_key_g = 'params'
+
+    # 002 lightweight image sr
+    # use 'pixelshuffledirect' to save parameters
+    elif args.task in ['lightweight_sr']:
+        model = net(upscale=args.scale, in_chans=3, img_size=64, window_size=8,
+                    img_range=1., depths=[6, 6, 6, 6], embed_dim=60, num_heads=[6, 6, 6, 6],
+                    mlp_ratio=2, upsampler='pixelshuffledirect', resi_connection='1conv')
+        param_key_g = 'params'
+        
+    elif args.task == 'compressed_sr':
+        model = net(upscale=args.scale, in_chans=3, img_size=args.training_patch_size, window_size=8,
+                    img_range=1., depths=[6, 6, 6, 6, 6, 6], embed_dim=180, num_heads=[6, 6, 6, 6, 6, 6],
+                    mlp_ratio=2, upsampler='pixelshuffle_aux', resi_connection='1conv')
+        param_key_g = 'params'                
+
+    # 003 real-world image sr
+    elif args.task == 'real_sr':
+        if not args.large_model:
+            # use 'nearest+conv' to avoid block artifacts
+            model = net(upscale=args.scale, in_chans=3, img_size=64, window_size=8,
+                        img_range=1., depths=[6, 6, 6, 6, 6, 6], embed_dim=180, num_heads=[6, 6, 6, 6, 6, 6],
+                        mlp_ratio=2, upsampler='nearest+conv', resi_connection='1conv')
+        else:
+            # larger model size; use '3conv' to save parameters and memory; use ema for GAN training
+            model = net(upscale=args.scale, in_chans=3, img_size=64, window_size=8,
+                        img_range=1., depths=[6, 6, 6, 6, 6, 6, 6, 6, 6], embed_dim=240,
+                        num_heads=[8, 8, 8, 8, 8, 8, 8, 8, 8],
+                        mlp_ratio=2, upsampler='nearest+conv', resi_connection='3conv')
+        param_key_g = 'params_ema'
+
+    # 006 grayscale JPEG compression artifact reduction
+    # use window_size=7 because JPEG encoding uses 8x8; use img_range=255 because it's sligtly better than 1
+    elif args.task == 'jpeg_car':
+        model = net(upscale=1, in_chans=1, img_size=126, window_size=7,
+                    img_range=255., depths=[6, 6, 6, 6, 6, 6], embed_dim=180, num_heads=[6, 6, 6, 6, 6, 6],
+                    mlp_ratio=2, upsampler='', resi_connection='1conv')
+        param_key_g = 'params'
+
+    # 006 color JPEG compression artifact reduction
+    # use window_size=7 because JPEG encoding uses 8x8; use img_range=255 because it's sligtly better than 1
+    elif args.task == 'color_jpeg_car':
+        model = net(upscale=1, in_chans=3, img_size=126, window_size=7,
+                    img_range=255., depths=[6, 6, 6, 6, 6, 6], embed_dim=180, num_heads=[6, 6, 6, 6, 6, 6],
+                    mlp_ratio=2, upsampler='', resi_connection='1conv')
+        param_key_g = 'params'
+
+    pretrained_model = torch.load(args.model_path)
+    model.load_state_dict(pretrained_model[param_key_g] if param_key_g in pretrained_model.keys() else pretrained_model, strict=True)
+
+    return model
+
+
+def setup(args):
+    # 001 classical image sr/ 002 lightweight image sr
+    if args.task in ['classical_sr', 'lightweight_sr', 'compressed_sr']:
+        save_dir = f'results/swin2sr_{args.task}_x{args.scale}'
+        if args.save_img_only:
+            folder = args.folder_lq
+        else:
+            folder = args.folder_gt
+        border = args.scale
+        window_size = 8
+
+    # 003 real-world image sr
+    elif args.task in ['real_sr']:
+        save_dir = f'results/swin2sr_{args.task}_x{args.scale}'
+        if args.large_model:
+            save_dir += '_large'
+        folder = args.folder_lq
+        border = 0
+        window_size = 8
+
+    # 006 JPEG compression artifact reduction
+    elif args.task in ['jpeg_car', 'color_jpeg_car']:
+        save_dir = f'results/swin2sr_{args.task}_jpeg{args.jpeg}'
+        folder = args.folder_gt
+        border = 0
+        window_size = 7
+
+    return folder, save_dir, border, window_size
+
+
+def get_image_pair(args, path):
+    (imgname, imgext) = os.path.splitext(os.path.basename(path))
+
+    # 001 classical image sr/ 002 lightweight image sr (load lq-gt image pairs)
+    if args.task in ['classical_sr', 'lightweight_sr']:
+        if args.save_img_only:
+            img_gt = None
+            img_lq = cv2.imread(path, cv2.IMREAD_COLOR).astype(np.float32) / 255.            
+        else:
+            img_gt = cv2.imread(path, cv2.IMREAD_COLOR).astype(np.float32) / 255.
+            img_lq = cv2.imread(f'{args.folder_lq}/{imgname}x{args.scale}{imgext}', cv2.IMREAD_COLOR).astype(
+                np.float32) / 255.            
         
-title = "Swin2SR"
-description = "Gradio demo for Swin2SR."
-article = "<p style='text-align: center'><a href='https://arxiv.org/abs/2209.11345' target='_blank'>Swin2SR: SwinV2 Transformer for Compressed Image Super-Resolution and Restoration</a> | <a href='https://github.com/mv-lab/swin2sr' target='_blank'>Github Repo</a></p>"
-
-examples=[['butterflyx4.png']]
-gr.Interface(
-    inference, 
-    gr.inputs.Image(type="pil", label="Input"),
-    "image",
-    title=title,
-    description=description,
-    article=article,
-    examples=examples,
-    ).launch(enable_queue=True)
+    elif args.task in ['compressed_sr']:
+        if args.save_img_only:
+            img_gt = None
+            img_lq = cv2.imread(path, cv2.IMREAD_COLOR).astype(np.float32) / 255.            
+        else:
+            img_gt = cv2.imread(path, cv2.IMREAD_COLOR).astype(np.float32) / 255.
+            img_lq = cv2.imread(f'{args.folder_lq}/{imgname}.jpg', cv2.IMREAD_COLOR).astype(
+                np.float32) / 255.        
+
+    # 003 real-world image sr (load lq image only)
+    elif args.task in ['real_sr', 'lightweight_sr_infer']:
+        img_gt = None
+        img_lq = cv2.imread(path, cv2.IMREAD_COLOR).astype(np.float32) / 255.
+
+    # 006 grayscale JPEG compression artifact reduction (load gt image and generate lq image on-the-fly)
+    elif args.task in ['jpeg_car']:
+        img_gt = cv2.imread(path, cv2.IMREAD_UNCHANGED)
+        if img_gt.ndim != 2:
+            img_gt = util.bgr2ycbcr(img_gt, y_only=True)
+        result, encimg = cv2.imencode('.jpg', img_gt, [int(cv2.IMWRITE_JPEG_QUALITY), args.jpeg])
+        img_lq = cv2.imdecode(encimg, 0)
+        img_gt = np.expand_dims(img_gt, axis=2).astype(np.float32) / 255.
+        img_lq = np.expand_dims(img_lq, axis=2).astype(np.float32) / 255.
+
+    # 006 JPEG compression artifact reduction (load gt image and generate lq image on-the-fly)
+    elif args.task in ['color_jpeg_car']:
+        img_gt = cv2.imread(path)
+        result, encimg = cv2.imencode('.jpg', img_gt, [int(cv2.IMWRITE_JPEG_QUALITY), args.jpeg])
+        img_lq = cv2.imdecode(encimg, 1)
+        img_gt = img_gt.astype(np.float32)/ 255.
+        img_lq = img_lq.astype(np.float32)/ 255.
+
+    return imgname, img_lq, img_gt
+
+
+def test(img_lq, model, args, window_size):
+    if args.tile is None:
+        # test the image as a whole
+        output = model(img_lq)
+    else:
+        # test the image tile by tile
+        b, c, h, w = img_lq.size()
+        tile = min(args.tile, h, w)
+        assert tile % window_size == 0, "tile size should be a multiple of window_size"
+        tile_overlap = args.tile_overlap
+        sf = args.scale
+
+        stride = tile - tile_overlap
+        h_idx_list = list(range(0, h-tile, stride)) + [h-tile]
+        w_idx_list = list(range(0, w-tile, stride)) + [w-tile]
+        E = torch.zeros(b, c, h*sf, w*sf).type_as(img_lq)
+        W = torch.zeros_like(E)
+
+        for h_idx in h_idx_list:
+            for w_idx in w_idx_list:
+                in_patch = img_lq[..., h_idx:h_idx+tile, w_idx:w_idx+tile]
+                out_patch = model(in_patch)
+                out_patch_mask = torch.ones_like(out_patch)
+
+                E[..., h_idx*sf:(h_idx+tile)*sf, w_idx*sf:(w_idx+tile)*sf].add_(out_patch)
+                W[..., h_idx*sf:(h_idx+tile)*sf, w_idx*sf:(w_idx+tile)*sf].add_(out_patch_mask)
+        output = E.div_(W)
+
+    return output
+
+if __name__ == '__main__':
+    
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--task', type=str, default='color_dn', help='classical_sr, lightweight_sr, real_sr, '
+                                                                     'gray_dn, color_dn, jpeg_car, color_jpeg_car')
+    parser.add_argument('--scale', type=int, default=1, help='scale factor: 1, 2, 3, 4, 8') # 1 for dn and jpeg car
+    parser.add_argument('--noise', type=int, default=15, help='noise level: 15, 25, 50')
+    parser.add_argument('--jpeg', type=int, default=40, help='scale factor: 10, 20, 30, 40')
+    parser.add_argument('--training_patch_size', type=int, default=128, help='patch size used in training Swin2SR. '
+                                       'Just used to differentiate two different settings in Table 2 of the paper. '
+                                       'Images are NOT tested patch by patch.')
+    parser.add_argument('--large_model', action='store_true', help='use large model, only provided for real image sr')
+    parser.add_argument('--model_path', type=str,
+                        default='model_zoo/swin2sr/Swin2SR_ClassicalSR_X2_64.pth')
+    parser.add_argument('--folder_lq', type=str, default=None, help='input low-quality test image folder')
+    parser.add_argument('--folder_gt', type=str, default=None, help='input ground-truth test image folder')
+    parser.add_argument('--tile', type=int, default=None, help='Tile size, None for no tile during testing (testing as a whole)')
+    parser.add_argument('--tile_overlap', type=int, default=32, help='Overlapping of different tiles')
+    parser.add_argument('--save_img_only', default=False, action='store_true', help='save image and do not evaluate')
+    args = parser.parse_args()
+
+    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+    # set up model
+    if os.path.exists(args.model_path):
+        print(f'loading model from {args.model_path}')        
+    else:
+        os.makedirs(os.path.dirname(args.model_path), exist_ok=True)
+        url = 'https://github.com/mv-lab/swin2sr/releases/download/v0.0.1/{}'.format(os.path.basename(args.model_path))
+        r = requests.get(url, allow_redirects=True)
+        print(f'downloading model {args.model_path}')
+        open(args.model_path, 'wb').write(r.content)
+    
+    model = setup_model(args)
+    
+    os.makedirs("test", exist_ok=True)
+    
+    #main(img)
+    
+    title = "Swin2SR"
+    description = "Gradio demo for Swin2SR."
+    article = "<p style='text-align: center'><a href='https://arxiv.org/abs/2209.11345' target='_blank'>Swin2SR: SwinV2 Transformer for Compressed Image Super-Resolution and Restoration</a> | <a href='https://github.com/mv-lab/swin2sr' target='_blank'>Github Repo</a></p>"
+
+    examples=[['butterflyx4.png']]
+    gr.Interface(
+        main, 
+        gr.inputs.Image(type="pil", label="Input"),
+        "image",
+        title=title,
+        description=description,
+        article=article,
+        examples=examples,
+        ).launch(enable_queue=True)