Coverage for src/bob/bio/video/annotator.py: 90%

81 statements  

« prev     ^ index     » next       coverage.py v7.6.0, created at 2024-07-13 00:44 +0200

1import collections 

2import logging 

3 

4import bob.bio.base 

5import bob.bio.face 

6 

7from . import utils 

8 

9logger = logging.getLogger(__name__) 

10 

11 

12def normalize_annotations(annotations, validator, max_age=-1): 

13 """Normalizes the annotations of one video sequence. It fills the 

14 annotations for frames from previous ones if the annotation for the current 

15 frame is not valid. 

16 

17 Parameters 

18 ---------- 

19 annotations : OrderedDict 

20 A dict of dict where the keys to the first dict are frame indices as 

21 strings (starting from 0). The inside dicts contain annotations for that 

22 frame. The dictionary needs to be an ordered dict in order for this to 

23 work. 

24 validator : ``callable`` 

25 Takes a dict (annotations) and returns True if the annotations are valid. 

26 This can be a check based on minimal face size for example: see 

27 :any:`bob.bio.face.annotator.min_face_size_validator`. 

28 max_age : :obj:`int`, optional 

29 An integer indicating for a how many frames a detected face is valid if 

30 no detection occurs after such frame. A value of -1 == forever 

31 

32 Yields 

33 ------ 

34 str 

35 The index of frame. 

36 dict 

37 The corrected annotations of the frame. 

38 """ 

39 # the annotations for the current frame 

40 current = None 

41 age = 0 

42 

43 for k, annot in annotations.items(): 

44 if validator(annot): 

45 current = annot 

46 age = 0 

47 elif max_age < 0 or age < max_age: 

48 age += 1 

49 else: # no detections and age is larger than maximum allowed 

50 current = None 

51 

52 yield k, current 

53 

54 

55class Base(bob.bio.base.annotator.Annotator): 

56 """The base class for video annotators.""" 

57 

58 @staticmethod 

59 def frame_ids_and_frames(frames): 

60 """Takes the frames and yields frame_ids and frames. 

61 

62 Parameters 

63 ---------- 

64 frames : :any:`bob.bio.video.VideoLikeContainer` or :any:`bob.bio.video.VideoAsArray` or :any:`numpy.array` 

65 The frames of the video file. 

66 

67 Yields 

68 ------ 

69 frame_id : str 

70 A string that represents the frame id. 

71 frame : :any:`numpy.array` 

72 The frame of the video file as an array. 

73 """ 

74 if isinstance(frames, (utils.VideoAsArray, utils.VideoLikeContainer)): 

75 for fid, fr in zip(frames.indices, frames): 

76 yield fid, fr 

77 else: 

78 for fid, fr in enumerate(frames): 

79 yield str(fid), fr 

80 

81 def annotate(self, frames, **kwargs): 

82 """Annotates videos. 

83 

84 Parameters 

85 ---------- 

86 frames : :any:`bob.bio.video.VideoLikeContainer` or :any:`bob.bio.video.VideoAsArray` or :any:`numpy.array` 

87 The frames of the video file. 

88 **kwargs 

89 Extra arguments that annotators may need. 

90 

91 Returns 

92 ------- 

93 OrderedDict 

94 A dictionary where its key is the frame id as a string and its value 

95 is a dictionary that are the annotations for that frame. 

96 

97 

98 .. note:: 

99 

100 You can use the :any:`Base.frame_ids_and_frames` functions to normalize 

101 the input in your implementation. 

102 """ 

103 raise NotImplementedError() 

104 

105 def transform(self, samples): 

106 """Takes a batch of data and annotates them. 

107 

108 Each ``kwargs`` value is a list of parameters, with each element of those 

109 lists corresponding to each element of ``samples`` (for example: 

110 with ``[s1, s2, ...]`` as ``samples``, ``kwargs['annotations']`` 

111 should contain ``[{<s1_annotations>}, {<s2_annotations>}, ...]``). 

112 """ 

113 return [self.annotate(sample) for sample in samples] 

114 

115 

116class FailSafeVideo(Base): 

117 """A fail-safe video annotator. 

118 It tries several annotators in order and tries the next one if the previous 

119 one fails. However, the difference between this annotator and 

120 :any:`bob.bio.base.annotator.FailSafe` is that this one tries to use 

121 annotations from older frames (if valid) before trying the next annotator. 

122 

123 .. warning:: 

124 

125 You must be careful in using this annotator since different annotators 

126 could have different results. For example the bounding box of one 

127 annotator be totally different from another annotator. 

128 

129 Parameters 

130 ---------- 

131 annotators : list 

132 A list of annotators to try. 

133 max_age : int 

134 The maximum number of frames that an annotation is valid for next frames. 

135 This value should be positive. If you want to set max_age to infinite, 

136 then you can use the :any:`bob.bio.video.annotator.Wrapper` instead. 

137 validator : ``callable`` 

138 A function that takes the annotations of a frame and validates it. 

139 

140 

141 Please see :any:`Base` for more accepted parameters. 

142 """ 

143 

144 def __init__( 

145 self, 

146 annotators, 

147 max_age=15, 

148 validator=None, 

149 **kwargs, 

150 ): 

151 super().__init__(**kwargs) 

152 

153 if max_age <= 0: 

154 raise ValueError( 

155 f"max_age: `{max_age}' cannot be less than 1, If you want to set max_age to infinite," 

156 "then you can use the :any:`bob.bio.video.annotator.Wrapper` with `normalize` set to True." 

157 ) 

158 self.max_age = max_age 

159 

160 if validator is None: 

161 validator = bob.bio.face.annotator.min_face_size_validator 

162 self.validator = validator 

163 

164 self.annotators = [] 

165 for annotator in annotators: 

166 if isinstance(annotator, str): 

167 annotator = bob.bio.base.load_resource(annotator, "annotator") 

168 self.annotators.append(annotator) 

169 

170 def annotate(self, frames): 

171 """See :any:`Base.annotate`""" 

172 video_annotations = collections.OrderedDict() 

173 current = None 

174 age = 0 

175 for i, frame in self.frame_ids_and_frames(frames): 

176 for annotator in self.annotators: 

177 annot = annotator.transform([frame])[0] 

178 if annot and self.validator(annot): 

179 current = annot 

180 age = 0 

181 break 

182 elif age < self.max_age: 

183 age += 1 

184 break 

185 else: # no detections and age is larger than maximum allowed 

186 current = None 

187 

188 if current is not annot: 

189 logger.debug("Annotator `%s' failed.", annotator) 

190 

191 video_annotations[i] = current 

192 return video_annotations 

193 

194 

195class Wrapper(Base): 

196 """Annotates video files using the provided image annotator. 

197 See the documentation of :any:`Base` too. 

198 

199 Parameters 

200 ---------- 

201 annotator : :any:`bob.bio.base.annotator.Annotator` or str 

202 The image annotator to be used. The annotator could also be the name of a 

203 bob.bio.annotator resource which will be loaded. 

204 max_age : int 

205 see :any:`normalize_annotations`. 

206 normalize : bool 

207 If True, it will normalize annotations using :any:`normalize_annotations` 

208 validator : object 

209 See :any:`normalize_annotations` and 

210 :any:`bob.bio.face.annotator.min_face_size_validator` for one example. 

211 

212 

213 Please see :any:`Base` for more accepted parameters. 

214 

215 .. warning:: 

216 

217 You should only set ``normalize`` to True only if you are annotating 

218 **all** frames of the video file. 

219 

220 """ 

221 

222 def __init__( 

223 self, 

224 annotator, 

225 normalize=False, 

226 validator=None, 

227 max_age=-1, 

228 **kwargs, 

229 ): 

230 super().__init__(**kwargs) 

231 

232 # load annotator configuration 

233 if isinstance(annotator, str): 

234 annotator = bob.bio.base.load_resource(annotator, "annotator") 

235 self.annotator = annotator 

236 

237 if validator is None: 

238 validator = bob.bio.face.annotator.min_face_size_validator 

239 self.validator = validator 

240 

241 self.normalize = normalize 

242 self.max_age = max_age 

243 

244 def annotate(self, frames): 

245 """See :any:`Base.annotate`""" 

246 annotations = collections.OrderedDict() 

247 for i, frame in self.frame_ids_and_frames(frames): 

248 logger.debug("Annotating frame %s", i) 

249 annotations[i] = self.annotator.transform([frame])[0] 

250 if self.normalize: 

251 annotations = collections.OrderedDict( 

252 normalize_annotations(annotations, self.validator, self.max_age) 

253 ) 

254 return annotations