forked from real-stanford/diffusion_policy
-
Notifications
You must be signed in to change notification settings - Fork 0
/
base_dataset.py
51 lines (39 loc) · 1.35 KB
/
base_dataset.py
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
from typing import Dict
import torch
import torch.nn
from diffusion_policy.model.common.normalizer import LinearNormalizer
class BaseLowdimDataset(torch.utils.data.Dataset):
def get_validation_dataset(self) -> 'BaseLowdimDataset':
# return an empty dataset by default
return BaseLowdimDataset()
def get_normalizer(self, **kwargs) -> LinearNormalizer:
raise NotImplementedError()
def get_all_actions(self) -> torch.Tensor:
raise NotImplementedError()
def __len__(self) -> int:
return 0
def __getitem__(self, idx: int) -> Dict[str, torch.Tensor]:
"""
output:
obs: T, Do
action: T, Da
"""
raise NotImplementedError()
class BaseImageDataset(torch.utils.data.Dataset):
def get_validation_dataset(self) -> 'BaseLowdimDataset':
# return an empty dataset by default
return BaseImageDataset()
def get_normalizer(self, **kwargs) -> LinearNormalizer:
raise NotImplementedError()
def get_all_actions(self) -> torch.Tensor:
raise NotImplementedError()
def __len__(self) -> int:
return 0
def __getitem__(self, idx: int) -> Dict[str, torch.Tensor]:
"""
output:
obs:
key: T, *
action: T, Da
"""
raise NotImplementedError()