Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / data / indexed / indexed_lazy_dataset.py: 0%

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""Lazy proxy for deferred indexed Dataset construction.""" 

16 

17from __future__ import annotations 

18 

19from collections.abc import Callable, Mapping 

20from typing import Any 

21 

22from hyper_parallel.data.dataset_logging import get_dataset_logger 

23 

24logger = get_dataset_logger(__name__) 

25 

26 

27class LazyDatasetProxy: 

28 """Construct and cache a Dataset when it is first accessed.""" 

29 

30 def __init__( 

31 self, 

32 dataset_factory: Callable[[], Any], 

33 *, 

34 unique_identifiers: Any, 

35 ) -> None: 

36 """Store lightweight metadata without constructing the Dataset. 

37 

38 Args: 

39 dataset_factory: Callable that constructs the real Dataset on first 

40 sample access. 

41 unique_identifiers: Stable Dataset cache identity assembled from 

42 construction inputs. 

43 """ 

44 self._dataset_factory = dataset_factory 

45 self._unique_identifiers = unique_identifiers 

46 self._dataset = None 

47 

48 def __len__(self) -> int: 

49 """Initialize the Dataset and return its length.""" 

50 return len(self._get_dataset()) 

51 

52 def __getitem__(self, index: int) -> Mapping[str, Any]: 

53 """Construct the Dataset if necessary and read one sample.""" 

54 dataset = self._get_dataset() 

55 sample = dataset[index] 

56 return sample 

57 

58 @property 

59 def unique_identifiers(self) -> Any: 

60 """Initialize the Dataset and return its cache identity.""" 

61 return self._get_dataset().unique_identifiers 

62 

63 def _get_dataset(self) -> Any: 

64 """Construct the Dataset once and reuse it for later accesses.""" 

65 if self._dataset is None: 

66 identifiers = self._unique_identifiers if isinstance(self._unique_identifiers, Mapping) else {} 

67 logger.debug( 

68 "Initializing lazy Dataset: class=%s, path=%s, split=%s, samples=%s", 

69 identifiers.get("class"), identifiers.get("dataset_path"), identifiers.get("index_split"), 

70 identifiers.get("num_samples"), 

71 ) 

72 self._dataset = self._dataset_factory() 

73 return self._dataset