Spaces:
Runtime error
Runtime error
| import torch | |
| import torch.nn.functional as F | |
| import unfoldNd | |
| def extract_patches(x, patch_size, stride, loop=False): | |
| """Extract patches from a motion sequence""" | |
| b, c, _t = x.shape | |
| # manually padding to loop | |
| if loop: | |
| half = patch_size // 2 | |
| front, tail = x[:,:,:half], x[:,:,-half:] | |
| x = torch.concat([tail, x, front], dim=-1) | |
| x_patches = unfoldNd.unfoldNd(x, kernel_size=patch_size, stride=stride).transpose(1, 2).reshape(b, -1, c, patch_size) | |
| return x_patches.view(b, -1, c * patch_size) | |
| def combine_patches(x_shape, ys, patch_size, stride, loop=False): | |
| """Combine motion patches""" | |
| # manually handle to loop | |
| out_shape = [*x_shape] | |
| if loop: | |
| padding = patch_size // 2 | |
| out_shape[-1] = out_shape[-1] + padding * 2 | |
| combined = unfoldNd.foldNd(ys.permute(0, 2, 1), output_size=tuple(out_shape[-1:]), kernel_size=patch_size, stride=stride) | |
| # normal fold matrix | |
| input_ones = torch.ones(tuple(out_shape), dtype=ys.dtype, device=ys.device) | |
| divisor = unfoldNd.unfoldNd(input_ones, kernel_size=patch_size, stride=stride) | |
| divisor = unfoldNd.foldNd(divisor, output_size=tuple(out_shape[-1:]), kernel_size=patch_size, stride=stride) | |
| combined = (combined / divisor).squeeze(dim=0).unsqueeze(0) | |
| if loop: | |
| half = patch_size // 2 | |
| front, tail = combined[:,:,:half], combined[:,:,-half:] | |
| combined[:, :, half:2 * half] = (combined[:, :, half:2 * half] + tail) / 2 | |
| combined[:, :, - 2 * half:-half] = (front + combined[:, :, - 2 * half:-half]) / 2 | |
| combined = combined[:, :, half:-half] | |
| return combined | |
| def efficient_cdist(X, Y): | |
| """ | |
| Pytorch efficient way of computing distances between all vectors in X and Y, i.e (X[:, None] - Y[None, :])**2 | |
| Get the nearest neighbor index from Y for each X | |
| :param X: (n1, d) tensor | |
| :param Y: (n2, d) tensor | |
| Returns a n2 n1 of indices | |
| """ | |
| dist = (X * X).sum(1)[:, None] + (Y * Y).sum(1)[None, :] - 2.0 * torch.mm(X, torch.transpose(Y, 0, 1)) | |
| d = X.shape[1] | |
| dist /= d # normalize by size of vector to make dists independent of the size of d ( use same alpha for all patche-sizes) | |
| return dist # DO NOT use torch.sqrt | |
| def get_col_mins_efficient(dist_fn, X, Y, b=1024): | |
| """ | |
| Computes the l2 distance to the closest x or each y. | |
| :param X: (n1, d) tensor | |
| :param Y: (n2, d) tensor | |
| Returns n1 long array of L2 distances | |
| """ | |
| n_batches = len(Y) // b | |
| mins = torch.zeros(Y.shape[0], dtype=X.dtype, device=X.device) | |
| for i in range(n_batches): | |
| mins[i * b:(i + 1) * b] = dist_fn(X, Y[i * b:(i + 1) * b]).min(0)[0] | |
| if len(Y) % b != 0: | |
| mins[n_batches * b:] = dist_fn(X, Y[n_batches * b:]).min(0)[0] | |
| return mins | |
| def get_NNs_Dists(dist_fn, X, Y, alpha=None, b=1024): | |
| """ | |
| Get the nearest neighbor index from Y for each X. | |
| Avoids holding a (n1 * n2) amtrix in order to reducing memory footprint to (b * max(n1,n2)). | |
| :param X: (n1, d) tensor | |
| :param Y: (n2, d) tensor | |
| Returns a n2 n1 of indices amd distances | |
| """ | |
| if alpha is not None: | |
| normalizing_row = get_col_mins_efficient(dist_fn, X, Y, b=b) | |
| normalizing_row = alpha + normalizing_row[None, :] | |
| else: | |
| normalizing_row = 1 | |
| NNs = torch.zeros(X.shape[0], dtype=torch.long, device=X.device) | |
| Dists = torch.zeros(X.shape[0], dtype=torch.float, device=X.device) | |
| n_batches = len(X) // b | |
| for i in range(n_batches): | |
| dists = dist_fn(X[i * b:(i + 1) * b], Y) / normalizing_row | |
| NNs[i * b:(i + 1) * b] = dists.min(1)[1] | |
| Dists[i * b:(i + 1) * b] = dists.min(1)[0] | |
| if len(X) % b != 0: | |
| dists = dist_fn(X[n_batches * b:], Y) / normalizing_row | |
| NNs[n_batches * b:] = dists.min(1)[1] | |
| Dists[n_batches * b: ] = dists.min(1)[0] | |
| return NNs, Dists | |