Spaces:
Runtime error
Runtime error
| import torch | |
| import torch.nn.functional as F | |
| def stride_from_shape(shape): | |
| stride = [1] | |
| for x in reversed(shape[1:]): | |
| stride.append(stride[-1] * x) | |
| return list(reversed(stride)) | |
| def scatter_add_nd(input, indices, values): | |
| # input: [..., C], D dimension + C channel | |
| # indices: [N, D], long | |
| # values: [N, C] | |
| D = indices.shape[-1] | |
| C = input.shape[-1] | |
| size = input.shape[:-1] | |
| stride = stride_from_shape(size) | |
| assert len(size) == D | |
| input = input.view(-1, C) # [HW, C] | |
| flatten_indices = (indices * torch.tensor(stride, dtype=torch.long, device=indices.device)).sum(-1) # [N] | |
| input.scatter_add_(0, flatten_indices.unsqueeze(1).repeat(1, C), values) | |
| return input.view(*size, C) | |
| def scatter_add_nd_with_count(input, count, indices, values, weights=None): | |
| # input: [..., C], D dimension + C channel | |
| # count: [..., 1], D dimension | |
| # indices: [N, D], long | |
| # values: [N, C] | |
| D = indices.shape[-1] | |
| C = input.shape[-1] | |
| size = input.shape[:-1] | |
| stride = stride_from_shape(size) | |
| assert len(size) == D | |
| input = input.view(-1, C) # [HW, C] | |
| count = count.view(-1, 1) | |
| flatten_indices = (indices * torch.tensor(stride, dtype=torch.long, device=indices.device)).sum(-1) # [N] | |
| if weights is None: | |
| weights = torch.ones_like(values[..., :1]) | |
| input.scatter_add_(0, flatten_indices.unsqueeze(1).repeat(1, C), values) | |
| count.scatter_add_(0, flatten_indices.unsqueeze(1), weights) | |
| return input.view(*size, C), count.view(*size, 1) | |
| def nearest_grid_put_2d(H, W, coords, values, return_count=False): | |
| # coords: [N, 2], float in [-1, 1] | |
| # values: [N, C] | |
| C = values.shape[-1] | |
| indices = (coords * 0.5 + 0.5) * torch.tensor( | |
| [H - 1, W - 1], dtype=torch.float32, device=coords.device | |
| ) | |
| indices = indices.round().long() # [N, 2] | |
| result = torch.zeros(H, W, C, device=values.device, dtype=values.dtype) # [H, W, C] | |
| count = torch.zeros(H, W, 1, device=values.device, dtype=values.dtype) # [H, W, 1] | |
| weights = torch.ones_like(values[..., :1]) # [N, 1] | |
| result, count = scatter_add_nd_with_count(result, count, indices, values, weights) | |
| if return_count: | |
| return result, count | |
| mask = (count.squeeze(-1) > 0) | |
| result[mask] = result[mask] / count[mask].repeat(1, C) | |
| return result | |
| def linear_grid_put_2d(H, W, coords, values, return_count=False): | |
| # coords: [N, 2], float in [-1, 1] | |
| # values: [N, C] | |
| C = values.shape[-1] | |
| indices = (coords * 0.5 + 0.5) * torch.tensor( | |
| [H - 1, W - 1], dtype=torch.float32, device=coords.device | |
| ) | |
| indices_00 = indices.floor().long() # [N, 2] | |
| indices_00[:, 0].clamp_(0, H - 2) | |
| indices_00[:, 1].clamp_(0, W - 2) | |
| indices_01 = indices_00 + torch.tensor( | |
| [0, 1], dtype=torch.long, device=indices.device | |
| ) | |
| indices_10 = indices_00 + torch.tensor( | |
| [1, 0], dtype=torch.long, device=indices.device | |
| ) | |
| indices_11 = indices_00 + torch.tensor( | |
| [1, 1], dtype=torch.long, device=indices.device | |
| ) | |
| h = indices[..., 0] - indices_00[..., 0].float() | |
| w = indices[..., 1] - indices_00[..., 1].float() | |
| w_00 = (1 - h) * (1 - w) | |
| w_01 = (1 - h) * w | |
| w_10 = h * (1 - w) | |
| w_11 = h * w | |
| result = torch.zeros(H, W, C, device=values.device, dtype=values.dtype) # [H, W, C] | |
| count = torch.zeros(H, W, 1, device=values.device, dtype=values.dtype) # [H, W, 1] | |
| weights = torch.ones_like(values[..., :1]) # [N, 1] | |
| result, count = scatter_add_nd_with_count(result, count, indices_00, values * w_00.unsqueeze(1), weights* w_00.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_01, values * w_01.unsqueeze(1), weights* w_01.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_10, values * w_10.unsqueeze(1), weights* w_10.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_11, values * w_11.unsqueeze(1), weights* w_11.unsqueeze(1)) | |
| if return_count: | |
| return result, count | |
| mask = (count.squeeze(-1) > 0) | |
| result[mask] = result[mask] / count[mask].repeat(1, C) | |
| return result | |
| def mipmap_linear_grid_put_2d(H, W, coords, values, min_resolution=32, return_count=False): | |
| # coords: [N, 2], float in [-1, 1] | |
| # values: [N, C] | |
| C = values.shape[-1] | |
| result = torch.zeros(H, W, C, device=values.device, dtype=values.dtype) # [H, W, C] | |
| count = torch.zeros(H, W, 1, device=values.device, dtype=values.dtype) # [H, W, 1] | |
| cur_H, cur_W = H, W | |
| while min(cur_H, cur_W) > min_resolution: | |
| # try to fill the holes | |
| mask = (count.squeeze(-1) == 0) | |
| if not mask.any(): | |
| break | |
| cur_result, cur_count = linear_grid_put_2d(cur_H, cur_W, coords, values, return_count=True) | |
| result[mask] = result[mask] + F.interpolate(cur_result.permute(2,0,1).unsqueeze(0).contiguous(), (H, W), mode='bilinear', align_corners=False).squeeze(0).permute(1,2,0).contiguous()[mask] | |
| count[mask] = count[mask] + F.interpolate(cur_count.view(1, 1, cur_H, cur_W), (H, W), mode='bilinear', align_corners=False).view(H, W, 1)[mask] | |
| cur_H //= 2 | |
| cur_W //= 2 | |
| if return_count: | |
| return result, count | |
| mask = (count.squeeze(-1) > 0) | |
| result[mask] = result[mask] / count[mask].repeat(1, C) | |
| return result | |
| def nearest_grid_put_3d(H, W, D, coords, values, return_count=False): | |
| # coords: [N, 3], float in [-1, 1] | |
| # values: [N, C] | |
| C = values.shape[-1] | |
| indices = (coords * 0.5 + 0.5) * torch.tensor( | |
| [H - 1, W - 1, D - 1], dtype=torch.float32, device=coords.device | |
| ) | |
| indices = indices.round().long() # [N, 2] | |
| result = torch.zeros(H, W, D, C, device=values.device, dtype=values.dtype) # [H, W, C] | |
| count = torch.zeros(H, W, D, 1, device=values.device, dtype=values.dtype) # [H, W, 1] | |
| weights = torch.ones_like(values[..., :1]) # [N, 1] | |
| result, count = scatter_add_nd_with_count(result, count, indices, values, weights) | |
| if return_count: | |
| return result, count | |
| mask = (count.squeeze(-1) > 0) | |
| result[mask] = result[mask] / count[mask].repeat(1, C) | |
| return result | |
| def linear_grid_put_3d(H, W, D, coords, values, return_count=False): | |
| # coords: [N, 3], float in [-1, 1] | |
| # values: [N, C] | |
| C = values.shape[-1] | |
| indices = (coords * 0.5 + 0.5) * torch.tensor( | |
| [H - 1, W - 1, D - 1], dtype=torch.float32, device=coords.device | |
| ) | |
| indices_000 = indices.floor().long() # [N, 3] | |
| indices_000[:, 0].clamp_(0, H - 2) | |
| indices_000[:, 1].clamp_(0, W - 2) | |
| indices_000[:, 2].clamp_(0, D - 2) | |
| indices_001 = indices_000 + torch.tensor([0, 0, 1], dtype=torch.long, device=indices.device) | |
| indices_010 = indices_000 + torch.tensor([0, 1, 0], dtype=torch.long, device=indices.device) | |
| indices_011 = indices_000 + torch.tensor([0, 1, 1], dtype=torch.long, device=indices.device) | |
| indices_100 = indices_000 + torch.tensor([1, 0, 0], dtype=torch.long, device=indices.device) | |
| indices_101 = indices_000 + torch.tensor([1, 0, 1], dtype=torch.long, device=indices.device) | |
| indices_110 = indices_000 + torch.tensor([1, 1, 0], dtype=torch.long, device=indices.device) | |
| indices_111 = indices_000 + torch.tensor([1, 1, 1], dtype=torch.long, device=indices.device) | |
| h = indices[..., 0] - indices_000[..., 0].float() | |
| w = indices[..., 1] - indices_000[..., 1].float() | |
| d = indices[..., 2] - indices_000[..., 2].float() | |
| w_000 = (1 - h) * (1 - w) * (1 - d) | |
| w_001 = (1 - h) * w * (1 - d) | |
| w_010 = h * (1 - w) * (1 - d) | |
| w_011 = h * w * (1 - d) | |
| w_100 = (1 - h) * (1 - w) * d | |
| w_101 = (1 - h) * w * d | |
| w_110 = h * (1 - w) * d | |
| w_111 = h * w * d | |
| result = torch.zeros(H, W, D, C, device=values.device, dtype=values.dtype) # [H, W, D, C] | |
| count = torch.zeros(H, W, D, 1, device=values.device, dtype=values.dtype) # [H, W, D, 1] | |
| weights = torch.ones_like(values[..., :1]) # [N, 1] | |
| result, count = scatter_add_nd_with_count(result, count, indices_000, values * w_000.unsqueeze(1), weights * w_000.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_001, values * w_001.unsqueeze(1), weights * w_001.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_010, values * w_010.unsqueeze(1), weights * w_010.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_011, values * w_011.unsqueeze(1), weights * w_011.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_100, values * w_100.unsqueeze(1), weights * w_100.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_101, values * w_101.unsqueeze(1), weights * w_101.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_110, values * w_110.unsqueeze(1), weights * w_110.unsqueeze(1)) | |
| result, count = scatter_add_nd_with_count(result, count, indices_111, values * w_111.unsqueeze(1), weights * w_111.unsqueeze(1)) | |
| if return_count: | |
| return result, count | |
| mask = (count.squeeze(-1) > 0) | |
| result[mask] = result[mask] / count[mask].repeat(1, C) | |
| return result | |
| def mipmap_linear_grid_put_3d(H, W, D, coords, values, min_resolution=32, return_count=False): | |
| # coords: [N, 3], float in [-1, 1] | |
| # values: [N, C] | |
| C = values.shape[-1] | |
| result = torch.zeros(H, W, D, C, device=values.device, dtype=values.dtype) # [H, W, D, C] | |
| count = torch.zeros(H, W, D, 1, device=values.device, dtype=values.dtype) # [H, W, D, 1] | |
| cur_H, cur_W, cur_D = H, W, D | |
| while min(min(cur_H, cur_W), cur_D) > min_resolution: | |
| # try to fill the holes | |
| mask = (count.squeeze(-1) == 0) | |
| if not mask.any(): | |
| break | |
| cur_result, cur_count = linear_grid_put_3d(cur_H, cur_W, cur_D, coords, values, return_count=True) | |
| result[mask] = result[mask] + F.interpolate(cur_result.permute(3,0,1,2).unsqueeze(0).contiguous(), (H, W, D), mode='trilinear', align_corners=False).squeeze(0).permute(1,2,3,0).contiguous()[mask] | |
| count[mask] = count[mask] + F.interpolate(cur_count.view(1, 1, cur_H, cur_W, cur_D), (H, W, D), mode='trilinear', align_corners=False).view(H, W, D, 1)[mask] | |
| cur_H //= 2 | |
| cur_W //= 2 | |
| cur_D //= 2 | |
| if return_count: | |
| return result, count | |
| mask = (count.squeeze(-1) > 0) | |
| result[mask] = result[mask] / count[mask].repeat(1, C) | |
| return result | |
| def grid_put(shape, coords, values, mode='linear-mipmap', min_resolution=32, return_raw=False): | |
| # shape: [D], list/tuple | |
| # coords: [N, D], float in [-1, 1] | |
| # values: [N, C] | |
| D = len(shape) | |
| assert D in [2, 3], f'only support D == 2 or 3, but got D == {D}' | |
| if mode == 'nearest': | |
| if D == 2: | |
| return nearest_grid_put_2d(*shape, coords, values, return_raw) | |
| else: | |
| return nearest_grid_put_3d(*shape, coords, values, return_raw) | |
| elif mode == 'linear': | |
| if D == 2: | |
| return linear_grid_put_2d(*shape, coords, values, return_raw) | |
| else: | |
| return linear_grid_put_3d(*shape, coords, values, return_raw) | |
| elif mode == 'linear-mipmap': | |
| if D == 2: | |
| return mipmap_linear_grid_put_2d(*shape, coords, values, min_resolution, return_raw) | |
| else: | |
| return mipmap_linear_grid_put_3d(*shape, coords, values, min_resolution, return_raw) | |
| else: | |
| raise NotImplementedError(f"got mode {mode}") |