(
self,
data: Sequence,
transform: Sequence[Callable] | Callable | None = None,
replace_rate: float = 0.1,
cache_num: int = sys.maxsize,
cache_rate: float = 1.0,
num_init_workers: int | None = 1,
num_replace_workers: int | None = 1,
progress: bool = True,
shuffle: bool = True,
seed: int = 0,
copy_cache: bool = True,
as_contiguous: bool = True,
runtime_cache=False,
)
| 1039 | """ |
| 1040 | |
| 1041 | def __init__( |
| 1042 | self, |
| 1043 | data: Sequence, |
| 1044 | transform: Sequence[Callable] | Callable | None = None, |
| 1045 | replace_rate: float = 0.1, |
| 1046 | cache_num: int = sys.maxsize, |
| 1047 | cache_rate: float = 1.0, |
| 1048 | num_init_workers: int | None = 1, |
| 1049 | num_replace_workers: int | None = 1, |
| 1050 | progress: bool = True, |
| 1051 | shuffle: bool = True, |
| 1052 | seed: int = 0, |
| 1053 | copy_cache: bool = True, |
| 1054 | as_contiguous: bool = True, |
| 1055 | runtime_cache=False, |
| 1056 | ) -> None: |
| 1057 | if shuffle: |
| 1058 | self.set_random_state(seed=seed) |
| 1059 | self.shuffle = shuffle |
| 1060 | |
| 1061 | self._start_pos: int = 0 |
| 1062 | self._update_lock: threading.Lock = threading.Lock() |
| 1063 | self._round: int = 1 |
| 1064 | self._replace_done: bool = False |
| 1065 | self._replace_mgr: threading.Thread | None = None |
| 1066 | if runtime_cache is not False: |
| 1067 | raise NotImplementedError("Options other than `runtime_cache=False` is not implemented yet.") |
| 1068 | |
| 1069 | super().__init__( |
| 1070 | data=data, |
| 1071 | transform=transform, |
| 1072 | cache_num=cache_num, |
| 1073 | cache_rate=cache_rate, |
| 1074 | num_workers=num_init_workers, |
| 1075 | progress=progress, |
| 1076 | copy_cache=copy_cache, |
| 1077 | as_contiguous=as_contiguous, |
| 1078 | runtime_cache=False, |
| 1079 | ) |
| 1080 | if self._cache is None: |
| 1081 | self._cache = self._fill_cache() |
| 1082 | if self.cache_num >= len(data): |
| 1083 | warnings.warn( |
| 1084 | "cache_num is greater or equal than dataset length, fall back to regular monai.data.CacheDataset." |
| 1085 | ) |
| 1086 | if replace_rate <= 0: |
| 1087 | raise ValueError("replace_rate must be greater than 0, otherwise, please use monai.data.CacheDataset.") |
| 1088 | |
| 1089 | self.num_replace_workers: int | None = num_replace_workers |
| 1090 | if self.num_replace_workers is not None: |
| 1091 | self.num_replace_workers = max(int(self.num_replace_workers), 1) |
| 1092 | |
| 1093 | self._total_num: int = len(data) |
| 1094 | self._replace_num: int = min(math.ceil(self.cache_num * replace_rate), len(data) - self.cache_num) |
| 1095 | self._replacements: list[Any] = [None for _ in range(self._replace_num)] |
| 1096 | self._replace_data_idx: list[int] = list(range(self._replace_num)) |
| 1097 | self._compute_data_idx() |
| 1098 |
nothing calls this directly
no test coverage detected