File size: 6,123 Bytes
0b70b07
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
import os
import argparse
import cv2
import numpy as np

from PIL import Image
from pathlib import Path
from glob import glob
from tqdm import tqdm

import torch
import torchvision.transforms as tf

from segmentation_models_pytorch import Linknet
import myutils

# ROOT_DIR = str(Path(__file__).resolve().parents[0])
ROOT_DIR = './'
# time_str = timestr = time.strftime("%Y-%m-%d %H-%M-%S")
# DEFAULT_OUT = os.path.join(ROOT_DIR, 'output', 'test_waterseg', time_str)
DEFAULT_OUT = os.path.join(ROOT_DIR, 'output', 'segs')
# DEFAULT_PALETTE = os.path.join(ROOT_DIR, "assets", "mask_palette.png")
# sys.path.append(ROOT_DIR)
# print("Added", ROOT_DIR, "to PATH.")


def norm_imagenet(img_pil, dims):
    """
    Normalizes and resizes input image
    :param img_pil: PIL Image
    :param dims: Model's expected input dimensions
    :return: Normalized Image as a Tensor
    """

    # Mean and stddev of ImageNet dataset
    mean = torch.tensor([0.485, 0.456, 0.406])
    std = torch.tensor([0.229, 0.224, 0.225])

    # Resize, convert to tensor, normalize
    transform_norm = tf.Compose([
        tf.Resize([dims[0], dims[1]]),
        tf.ToTensor(),
        tf.Normalize(mean, std)
    ])

    img_norm = transform_norm(img_pil)
    return img_norm


def predict_one(path, model, mask_outdir, overlay_outdir, device):
    """
    Predicts a single image from path
    :param path: Path to image
    :param model: Loaded Torch Model
    :param mask_outdir: Filepath to mask out directory
    :param overlay_outdir: Filepath to overlay out directory
    :return: None
    """
    img_pil = myutils.load_image_in_PIL(path)

    # Prediction is an PIL Image of 0s and 1s
    prediction = predict_pil(model, img_pil, model_dims=(416, 416), device=device)

    basename = str(Path(os.path.basename(path)).stem)
    mask_savepth = os.path.join(mask_outdir, basename + '.png')
    # mask_save = prediction.convert('RGB')
    prediction.save(mask_savepth)

    over_savepth = os.path.join(overlay_outdir, basename + '.png')
    img_np = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)
    overlay_np = myutils.add_overlay(img_np, np.array(prediction))
    cv2.imwrite(over_savepth, overlay_np)
    # overlay_np = np.array(img_pil) * 1 + np.array(prediction.convert('RGB')) * 0.8
    # overlay_np = overlay_np.clip(0, 255)
    # Image.fromarray(overlay_np.astype(np.uint8)).save(over_savepth)


def predict_pil(model, img_pil, model_dims, device):
    """
    Predicts a single PIL Image
    :param model: Loaded PyTorch model
    :param img_pil: PIL image
    :param model_dims: Model input dimensions
    :return: Segmentation prediction as PIL Image
    """

    img_np = np.array(img_pil)
    img_tensor_norm = norm_imagenet(img_pil, model_dims)

    # Pipeline to resize the prediction to the original image dimensions
    pred_resize = tf.Compose([tf.Resize([img_np.shape[0], img_np.shape[1]])])

    # Add extra dimension at front as model expects input 1*3*dimX*dimY (batch size of 1)
    input_data = img_tensor_norm.unsqueeze(0)

    try:
        # print("Converted input image to cuda.")
        prediction = model.predict(input_data.to(device))
    except:
        print("Did not convert input image to cuda.")
        prediction = model.predict(input_data)

    prediction = pred_resize(prediction)
    prediction = myutils.postprocessing_pred(prediction.squeeze().cpu().round().numpy().astype(np.uint8))
    prediction = Image.fromarray(prediction).convert('P')
    prediction.putpalette(myutils.color_palette)
    return prediction


def test_waterseg(model_path, test_path, test_name, out_path, device):
    """
    Tests either a single or an entire folder of images
    :param args: Command line args
    :return: None
    """
    model = torch.load(model_path)
    print('############################################')
    print('############################################')
    print('############################################')
    test_path = test_path
    out_path = os.path.join(out_path, test_name)

    mask_out = os.path.join(out_path, 'mask')
    overlay_out = os.path.join(out_path, 'overlay')
    if not os.path.exists(mask_out):
        os.makedirs(mask_out)
    if not os.path.exists(overlay_out):
        os.makedirs(overlay_out)

    if os.path.isfile(test_path):
        predict_one(test_path, model, mask_out, overlay_out, device)
    elif os.path.isdir(test_path):
        paths = glob(os.path.join(test_path, '*.jpg')) + glob(os.path.join(test_path, '*.png'))
        for path in tqdm(paths):
            predict_one(path, model, mask_out, overlay_out, device)
    else:
        print("Error: Unknown path type:", test_path)


if __name__ == '__main__':
    # Hyper parameters
    parser = argparse.ArgumentParser(description='V-FloodNet: Water Image Segmentation')
    # Required: Path to the .pth file.
    parser.add_argument('--model-path',
                        default='./records/link_efficientb4_model.pth',
                        type=str,
                        metavar='PATH',
                        help='Path to the model')
    # Required: Path to either the single file or directory of files containing .jpg or .png images
    parser.add_argument('--test-path',
                        type=str,
                        metavar='PATH',
                        required=True,
                        help='Can point to folder or an individual jpg/png image')
    parser.add_argument('--test-name',
                        type=str,
                        required=True,
                        help='Test name')
    parser.add_argument('--out-path',
                        default=DEFAULT_OUT,
                        type=str,
                        metavar='PATH',
                        help='(OPTIONAL) Path to output folder, defaults to project root/output')
    args = parser.parse_args()

    # Device
    device = torch.device('cpu')
    if torch.cuda.is_available():
        device = torch.device('cuda')

    test_waterseg(args.model_path, args.test_path, args.test_name, args.out_path, device)

    print(myutils.gct(), 'Test image segmentation done.')