Coverage for src/bob/pad/face/transformer/VideoToFrames.py: 86%

1import logging 

2 

3from functools import partial 

4 

5from sklearn.base import BaseEstimator, TransformerMixin 

6 

7from bob.pipelines.sample import DelayedSample, Sample 

8from bob.pipelines.wrappers import _frmt 

9 

10logger = logging.getLogger(__name__) 

11 

12 

13def _get(sth): 

14 return sth 

15 

16 

17class VideoToFrames(TransformerMixin, BaseEstimator): 

18 """Expands video samples to frame-based samples only when transform is called.""" 

19 

20 def __init__(self, delayed_output=True): 

21 self.delayed_output = delayed_output 

22 

23 def transform(self, video_samples): 

24 logger.debug(f"{_frmt(self)}.transform") 

25 outputs = [] 

26 for vid_sample in video_samples: 

27 annotations = getattr(vid_sample, "annotations", None) 

28 # Define groups with `sample.key`` since we need a unique ID for 

29 # each video. The `groups` attribute is used to do cross-validation 

30 if not hasattr(vid_sample, "key"): 

31 raise ValueError( 

32 "Video sample must have a unique `key` " 

33 "attribute to be used with {}".format( 

34 self.__class__.__name__ 

35 ) 

36 ) 

37 groups = vid_sample.key 

38 # video is an instance of VideoAsArray or VideoLikeContainer 

39 video = vid_sample.data 

40 for frame, frame_id in zip(video, video.indices): 

41 if frame is None: 

42 continue 

43 # Do we have frame annotations? 

44 frame_annotations = None 

45 if annotations is not None: 

46 # Global annotation are present -> query them 

47 frame_annotations = annotations.get(str(frame_id)) 

48 # Update key, otherwise get the one from parent and each frames 

49 # get the same one, breaking checkpoint mechanic for steps 

50 # later down the pipelines 

51 key = "{}_{}".format(vid_sample.key, frame_id) 

52 if self.delayed_output: 

53 # create a load method so that we can create DelayedSamples 

54 # because the input samples could be DelayedSamples with 

55 # delayed attributes as well and we don't want to load 

56 # those delayed attributes. 

57 sample = DelayedSample( 

58 partial(_get, frame), 

59 frame_id=frame_id, 

60 video_key=groups, 

61 parent=vid_sample, 

62 # Override parent's attributes 

63 annotations=frame_annotations, 

64 key=key, 

65 ) 

66 else: 

67 sample = Sample( 

68 frame, 

69 frame_id=frame_id, 

70 video_key=groups, 

71 parent=vid_sample, 

72 # Override parent's attributes 

73 annotations=frame_annotations, 

74 key=key, 

75 ) 

76 outputs.append(sample) 

77 return outputs 

78 

79 def fit(self, X, y=None): 

80 return self 

81 

82 def _more_tags(self): 

83 return { 

84 "requires_fit": False, 

85 "bob_checkpoint_features": False, 

86 }