|
|
from typing import * |
|
|
import torch |
|
|
import torch.nn as nn |
|
|
from .. import SparseTensor |
|
|
|
|
|
|
|
|
class SparseSpatial2Channel(nn.Module): |
|
|
""" |
|
|
Downsample a sparse tensor by a factor of `factor`. |
|
|
Implemented as rearranging its features from spatial to channel. |
|
|
""" |
|
|
def __init__(self, factor: int = 2): |
|
|
super(SparseSpatial2Channel, self).__init__() |
|
|
self.factor = factor |
|
|
|
|
|
def forward(self, x: SparseTensor) -> SparseTensor: |
|
|
DIM = x.coords.shape[-1] - 1 |
|
|
cache = x.get_spatial_cache(f'spatial2channel_{self.factor}') |
|
|
if cache is None: |
|
|
coord = list(x.coords.unbind(dim=-1)) |
|
|
for i in range(DIM): |
|
|
coord[i+1] = coord[i+1] // self.factor |
|
|
subidx = x.coords[:, 1:] % self.factor |
|
|
subidx = sum([subidx[..., i] * self.factor ** i for i in range(DIM)]) |
|
|
|
|
|
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, subidx = cache |
|
|
|
|
|
new_feats = torch.zeros(new_coords.shape[0] * self.factor ** DIM, x.feats.shape[1], device=x.feats.device, dtype=x.feats.dtype) |
|
|
new_feats[idx * self.factor ** DIM + subidx] = x.feats |
|
|
|
|
|
out = SparseTensor(new_feats.reshape(new_coords.shape[0], -1), new_coords, None if x._shape is None else torch.Size([x._shape[0], x._shape[1] * self.factor ** DIM])) |
|
|
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'spatial2channel_{self.factor}', (new_coords, idx, subidx)) |
|
|
out.register_spatial_cache(f'channel2spatial_{self.factor}', (x.coords, idx, subidx)) |
|
|
out.register_spatial_cache(f'shape', torch.Size(MAX)) |
|
|
if self.training: |
|
|
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 SparseChannel2Spatial(nn.Module): |
|
|
""" |
|
|
Upsample a sparse tensor by a factor of `factor`. |
|
|
Implemented as rearranging its features from channel to spatial. |
|
|
""" |
|
|
def __init__(self, factor: int = 2): |
|
|
super(SparseChannel2Spatial, 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'channel2spatial_{self.factor}') |
|
|
if cache is None: |
|
|
if subdivision is None: |
|
|
raise ValueError('Cache not found. Provide subdivision tensor or pair SparseChannel2Spatial with SparseSpatial2Channel.') |
|
|
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, subidx = cache |
|
|
|
|
|
x_feats = x.feats.reshape(x.feats.shape[0] * self.factor ** DIM, -1) |
|
|
new_feats = x_feats[idx * self.factor ** DIM + subidx] |
|
|
out = SparseTensor(new_feats, new_coords, None if x._shape is None else torch.Size([x._shape[0], x._shape[1] // self.factor ** DIM])) |
|
|
out._scale = tuple([s / self.factor for s in x._scale]) |
|
|
if cache is not None: |
|
|
out._spatial_cache = x._spatial_cache |
|
|
return out |
|
|
|
