(self, images, scale=1.0, batch_size=4, num_iter=1, progress_bar=lambda x:x)
| 169 | |
| 170 | @torch.no_grad() |
| 171 | def interpolate(self, images, scale=1.0, batch_size=4, num_iter=1, progress_bar=lambda x:x): |
| 172 | # Preprocess |
| 173 | processed_images = self.process_images(images) |
| 174 | |
| 175 | for iter in range(num_iter): |
| 176 | # Input |
| 177 | input_tensor = torch.cat((processed_images[:-1], processed_images[1:]), dim=1) |
| 178 | |
| 179 | # Interpolate |
| 180 | output_tensor = [] |
| 181 | for batch_id in progress_bar(range(0, input_tensor.shape[0], batch_size)): |
| 182 | batch_id_ = min(batch_id + batch_size, input_tensor.shape[0]) |
| 183 | batch_input_tensor = input_tensor[batch_id: batch_id_] |
| 184 | batch_input_tensor = batch_input_tensor.to(device=self.device, dtype=self.torch_dtype) |
| 185 | flow, mask, merged = self.model(batch_input_tensor, [4/scale, 2/scale, 1/scale]) |
| 186 | output_tensor.append(merged[2].cpu()) |
| 187 | |
| 188 | # Output |
| 189 | output_tensor = torch.concat(output_tensor, dim=0).clip(0, 1) |
| 190 | processed_images = self.add_interpolated_images(processed_images, output_tensor) |
| 191 | processed_images = torch.stack(processed_images) |
| 192 | |
| 193 | # To images |
| 194 | output_images = self.decode_images(processed_images) |
| 195 | if output_images[0].size != images[0].size: |
| 196 | output_images = [image.resize(images[0].size) for image in output_images] |
| 197 | return output_images |
| 198 | |
| 199 | |
| 200 | class RIFESmoother(RIFEInterpolater): |
no test coverage detected