Spaces:
Running
on
Zero
Running
on
Zero
File size: 4,279 Bytes
917a889 |
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 |
from typing import *
import torch
import torch.nn as nn
from .. import SparseTensor
__all__ = [
'SparseDownsample',
'SparseUpsample',
]
class SparseDownsample(nn.Module):
"""
Downsample a sparse tensor by a factor of `factor`.
Implemented as average pooling.
"""
def __init__(self, factor: int, mode: Literal['mean', 'max'] = 'mean'):
super(SparseDownsample, self).__init__()
self.factor = factor
self.mode = mode
assert self.mode in ['mean', 'max'], f'Invalid mode: {self.mode}'
def forward(self, x: SparseTensor) -> SparseTensor:
cache = x.get_spatial_cache(f'downsample_{self.factor}')
if cache is None:
DIM = x.coords.shape[-1] - 1
coord = list(x.coords.unbind(dim=-1))
for i in range(DIM):
coord[i+1] = coord[i+1] // self.factor
MAX = [(s + self.factor - 1) // self.factor for s in x.spatial_shape]
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_coords = torch.stack(
[code // OFFSET[0]] +
[(code // OFFSET[i+1]) % MAX[i] for i in range(DIM)],
dim=-1
)
else:
new_coords, idx = cache
new_feats = torch.scatter_reduce(
torch.zeros(new_coords.shape[0], x.feats.shape[1], device=x.feats.device, dtype=x.feats.dtype),
dim=0,
index=idx.unsqueeze(1).expand(-1, x.feats.shape[1]),
src=x.feats,
reduce=self.mode,
include_self=False,
)
out = SparseTensor(new_feats, new_coords, x._shape)
out._scale = tuple([s * self.factor for s in x._scale])
out._spatial_cache = x._spatial_cache
if cache is None:
x.register_spatial_cache(f'downsample_{self.factor}', (new_coords, idx))
out.register_spatial_cache(f'upsample_{self.factor}', (x.coords, idx))
out.register_spatial_cache(f'shape', torch.Size(MAX))
if self.training:
subidx = x.coords[:, 1:] % self.factor
subidx = sum([subidx[..., i] * self.factor ** i for i in range(DIM)])
subdivision = torch.zeros((new_coords.shape[0], self.factor ** DIM), device=x.device, dtype=torch.bool)
subdivision[idx, subidx] = True
out.register_spatial_cache(f'subdivision', subdivision)
return out
class SparseUpsample(nn.Module):
"""
Upsample a sparse tensor by a factor of `factor`.
Implemented as nearest neighbor interpolation.
"""
def __init__(
self, factor: int
):
super(SparseUpsample, self).__init__()
self.factor = factor
def forward(self, x: SparseTensor, subdivision: Optional[SparseTensor] = None) -> SparseTensor:
DIM = x.coords.shape[-1] - 1
cache = x.get_spatial_cache(f'upsample_{self.factor}')
if cache is None:
if subdivision is None:
raise ValueError('Cache not found. Provide subdivision tensor or pair SparseUpsample with SparseDownsample.')
else:
sub = subdivision.feats
N_leaf = sub.sum(dim=-1)
subidx = sub.nonzero()[:, -1]
new_coords = x.coords.clone().detach()
new_coords[:, 1:] *= self.factor
new_coords = torch.repeat_interleave(new_coords, N_leaf, dim=0, output_size=subidx.shape[0])
for i in range(DIM):
new_coords[:, i+1] += subidx // self.factor ** i % self.factor
idx = torch.repeat_interleave(torch.arange(x.coords.shape[0], device=x.device), N_leaf, dim=0, output_size=subidx.shape[0])
else:
new_coords, idx = cache
new_feats = x.feats[idx]
out = SparseTensor(new_feats, new_coords, x._shape)
out._scale = tuple([s / self.factor for s in x._scale])
if cache is not None: # only keep cache when subdiv following it
out._spatial_cache = x._spatial_cache
return out
|