Spaces:
Running
on
Zero
Running
on
Zero
File size: 4,567 Bytes
db6a3b7 |
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 |
from typing import *
import torch
import torch.nn as nn
from . import SparseTensor
__all__ = [
'SparseDownsample',
'SparseUpsample',
'SparseSubdivide'
]
class SparseDownsample(nn.Module):
"""
Downsample a sparse tensor by a factor of `factor`.
Implemented as average pooling.
"""
def __init__(self, factor: Union[int, Tuple[int, ...], List[int]]):
super(SparseDownsample, self).__init__()
self.factor = tuple(factor) if isinstance(factor, (list, tuple)) else factor
def forward(self, input: SparseTensor) -> SparseTensor:
DIM = input.coords.shape[-1] - 1
factor = self.factor if isinstance(self.factor, tuple) else (self.factor,) * DIM
assert DIM == len(factor), 'Input coordinates must have the same dimension as the downsample factor.'
coord = list(input.coords.unbind(dim=-1))
for i, f in enumerate(factor):
coord[i+1] = coord[i+1] // f
MAX = [coord[i+1].max().item() + 1 for i in range(DIM)]
OFFSET = torch.cumprod(torch.tensor(MAX[::-1]), 0).tolist()[::-1] + [1]
code = sum([c * o for c, o in zip(coord, OFFSET)])
code, idx = code.unique(return_inverse=True)
new_feats = torch.scatter_reduce(
torch.zeros(code.shape[0], input.feats.shape[1], device=input.feats.device, dtype=input.feats.dtype),
dim=0,
index=idx.unsqueeze(1).expand(-1, input.feats.shape[1]),
src=input.feats,
reduce='mean'
)
new_coords = torch.stack(
[code // OFFSET[0]] +
[(code // OFFSET[i+1]) % MAX[i] for i in range(DIM)],
dim=-1
)
out = SparseTensor(new_feats, new_coords, input.shape,)
out._scale = tuple([s // f for s, f in zip(input._scale, factor)])
out._spatial_cache = input._spatial_cache
out.register_spatial_cache(f'upsample_{factor}_coords', input.coords)
out.register_spatial_cache(f'upsample_{factor}_layout', input.layout)
out.register_spatial_cache(f'upsample_{factor}_idx', idx)
return out
class SparseUpsample(nn.Module):
"""
Upsample a sparse tensor by a factor of `factor`.
Implemented as nearest neighbor interpolation.
"""
def __init__(self, factor: Union[int, Tuple[int, int, int], List[int]]):
super(SparseUpsample, self).__init__()
self.factor = tuple(factor) if isinstance(factor, (list, tuple)) else factor
def forward(self, input: SparseTensor) -> SparseTensor:
DIM = input.coords.shape[-1] - 1
factor = self.factor if isinstance(self.factor, tuple) else (self.factor,) * DIM
assert DIM == len(factor), 'Input coordinates must have the same dimension as the upsample factor.'
new_coords = input.get_spatial_cache(f'upsample_{factor}_coords')
new_layout = input.get_spatial_cache(f'upsample_{factor}_layout')
idx = input.get_spatial_cache(f'upsample_{factor}_idx')
if any([x is None for x in [new_coords, new_layout, idx]]):
raise ValueError('Upsample cache not found. SparseUpsample must be paired with SparseDownsample.')
new_feats = input.feats[idx]
out = SparseTensor(new_feats, new_coords, input.shape, new_layout)
out._scale = tuple([s * f for s, f in zip(input._scale, factor)])
out._spatial_cache = input._spatial_cache
return out
class SparseSubdivide(nn.Module):
"""
Upsample a sparse tensor by a factor of `factor`.
Implemented as nearest neighbor interpolation.
"""
def __init__(self):
super(SparseSubdivide, self).__init__()
def forward(self, input: SparseTensor) -> SparseTensor:
DIM = input.coords.shape[-1] - 1
# upsample scale=2^DIM
n_cube = torch.ones([2] * DIM, device=input.device, dtype=torch.int)
n_coords = torch.nonzero(n_cube)
n_coords = torch.cat([torch.zeros_like(n_coords[:, :1]), n_coords], dim=-1)
factor = n_coords.shape[0]
assert factor == 2 ** DIM
# print(n_coords.shape)
new_coords = input.coords.clone()
new_coords[:, 1:] *= 2
new_coords = new_coords.unsqueeze(1) + n_coords.unsqueeze(0).to(new_coords.dtype)
new_feats = input.feats.unsqueeze(1).expand(input.feats.shape[0], factor, *input.feats.shape[1:])
out = SparseTensor(new_feats.flatten(0, 1), new_coords.flatten(0, 1), input.shape)
out._scale = input._scale * 2
out._spatial_cache = input._spatial_cache
return out
|