Coverage for src/lektor_ng/videotools.py: 53%

1import decimal 

2import json 

3import os 

4import subprocess 

5from collections import namedtuple 

6from datetime import timedelta 

7 

8from lektor_ng.imagetools import Thumbnail, ThumbnailMode 

9from lektor_ng.reporter import reporter 

10from lektor_ng.utils import get_dependent_url, locate_executable, portable_popen 

11 

12THUMBNAIL_FORMATS = frozenset(["jpg", "jpeg", "png"]) 

13 

14 

15def _imround(x): 

16 """Round float pixel values like imagemagick does it.""" 

17 return decimal.Decimal(x).to_integral(decimal.ROUND_HALF_UP) 

18 

19 

20Rescaling = namedtuple("Rescaling", ["rescale", "crop"]) 

21 

22 

23class Dimensions(namedtuple("Dimensions", ["width", "height"])): 

24 __slots__ = () 

25 

26 def __new__(cls, width, height): 

27 width = int(width) 

28 height = int(height) 

29 

30 if width < 1 or height < 1: 

31 raise ValueError("Invalid dimensions") 

32 

33 return super().__new__(cls, width, height) 

34 

35 @property 

36 def aspect_ratio(self): 

37 return float(self.width) / float(self.height) 

38 

39 def _infer_dimensions(self, width, height): 

40 """Calculate dimensions based on aspect ratio if height, width or both 

41 are missing. 

42 """ 

43 if width is None and height is None: 

44 return self 

45 

46 if width is None: 

47 width = _imround(height * self.aspect_ratio) 

48 elif height is None: 

49 height = _imround(width / self.aspect_ratio) 

50 

51 return Dimensions(width, height) 

52 

53 def contains(self, other): 

54 """Return True if the given Dimensions can be completely enclosed by 

55 this Dimensions. 

56 """ 

57 return self.width >= other.width and self.height >= other.height 

58 

59 def fit_within(self, max_width=None, max_height=None, upscale=None): 

60 """Calculate resizing required to make these dimensions fit within the 

61 given dimensions. 

62 

63 Note that resizing only occurs if upscale is enabled. 

64 

65 >>> source = Dimensions(640, 480) 

66 >>> source.fit_within(max_width=320).rescale == Dimensions(320, 240) 

67 True 

68 

69 :param max_width: Maximum width for the new rescaled dimensions. 

70 :param max_height: Maximum height for the new rescaled dimensions. 

71 :param upscale: Allow making the dimensions larger (default False). 

72 :return: Rescaling operations 

73 :rtype: Rescaling 

74 """ 

75 if upscale is None: 

76 upscale = False 

77 

78 max_dim = self._infer_dimensions(max_width, max_height) 

79 

80 # Check if we should rescale at all 

81 if max_dim == self or (not upscale and max_dim.contains(self)): 

82 return Rescaling(self, self) 

83 

84 ar = self.aspect_ratio 

85 rescale_dim = Dimensions( 

86 width=_imround(min(max_dim.width, max_dim.height * ar)), 

87 height=_imround(min(max_dim.height, max_dim.width / ar)), 

88 ) 

89 

90 return Rescaling(rescale=rescale_dim, crop=rescale_dim) 

91 

92 def cover(self, min_width=None, min_height=None, upscale=None): 

93 """Calculate resizing required to make these dimensions cover the given 

94 dimensions. 

95 

96 Note that resizing only occurs if upscale is enabled. 

97 

98 >>> source = Dimensions(640, 480) 

99 >>> target = source.cover(240, 240) 

100 >>> target.rescale == Dimensions(320, 240) 

101 True 

102 >>> target.crop == Dimensions(240, 240) 

103 True 

104 

105 :param min_width: Minimum width for the new rescaled dimensions. 

106 :param min_height: Minimum height for the new rescaled dimensions. 

107 :param upscale: Allow making the dimensions larger (default True). 

108 :return: Rescaling operations 

109 :rtype: Rescaling 

110 """ 

111 if upscale is None: 

112 upscale = True 

113 

114 min_dim = self._infer_dimensions(min_width, min_height) 

115 

116 # Check if we should rescale at all 

117 if min_dim == self or (not upscale and min_dim.contains(self)): 

118 return Rescaling(self, self) 

119 

120 ar = self.aspect_ratio 

121 rescale_dim = Dimensions( 

122 width=_imround(max(min_dim.width, min_dim.height * ar)), 

123 height=_imround(max(min_dim.height, min_dim.width / ar)), 

124 ) 

125 

126 return Rescaling(rescale=rescale_dim, crop=min_dim) 

127 

128 def stretch(self, width=None, height=None, upscale=None): 

129 """Calculate resizing required to the given dimensions without 

130 considering aspect ratio. 

131 

132 Note that resizing only occurs if upscale is enabled. 

133 

134 >>> source = Dimensions(640, 480) 

135 >>> source.cover(240, 240).rescale == Dimensions(240, 240) 

136 True 

137 

138 :param min_width: Minimum width for the new rescaled dimensions. 

139 :param min_height: Minimum height for the new rescaled dimensions. 

140 :param upscale: Allow making the dimensions larger (default True). 

141 :return: Rescaling operations 

142 :rtype: Rescaling 

143 """ 

144 if upscale is None: 

145 upscale = True 

146 

147 dim = self._infer_dimensions(width, height) 

148 

149 # Check if we should rescale at all 

150 if dim == self or (not upscale and dim.contains(self)): 

151 return Rescaling(self, self) 

152 

153 return Rescaling(rescale=dim, crop=dim) 

154 

155 def resize(self, width=None, height=None, mode=ThumbnailMode.DEFAULT, upscale=None): 

156 if mode == ThumbnailMode.FIT: 

157 return self.fit_within(width, height, upscale) 

158 if mode == ThumbnailMode.CROP: 

159 return self.cover(width, height, upscale) 

160 if mode == ThumbnailMode.STRETCH: 

161 return self.stretch(width, height, upscale) 

162 

163 raise ValueError(f'Unexpected mode "{mode!r}"') 

164 

165 

166def get_timecode(td): 

167 """Convert a timedelta to an ffmpeg compatible string timecode. 

168 

169 A timecode has the format HH:MM:SS, with decimals if needed. 

170 """ 

171 seconds = td.total_seconds() 

172 

173 hours = int(seconds // 3600) 

174 seconds %= 3600 

175 

176 minutes = int(seconds // 60) 

177 seconds %= 60 

178 

179 str_seconds, str_decimals = str(float(seconds)).split(".") 

180 

181 timecode = f"{hours:02d}:{minutes:02d}:{str_seconds.zfill(2)}" 

182 if str_decimals != "0": 

183 timecode += f".{str_decimals}" 

184 

185 return timecode 

186 

187 

188def get_ffmpeg_quality(quality_percent): 

189 """Convert a value between 0-100 to an ffmpeg quality value (2-31). 

190 

191 Note that this is only applicable to the mjpeg encoder (which is used for 

192 jpeg images). mjpeg values works in reverse, i.e. lower is better. 

193 """ 

194 if not 0 <= quality_percent <= 100: 

195 raise ValueError("Video quality must be between 0 and 100") 

196 

197 low, high = 2, 31 

198 span = high - low 

199 factor = float(quality_percent) / 100.0 

200 return int(low + round(span * (1 - factor))) 

201 

202 

203def get_suffix(seek, width, height, mode, quality): 

204 """Make suffix for a thumbnail that is unique to the given parameters.""" 

205 timecode = get_timecode(seek).replace(":", "-").replace(".", "-") 

206 bits = [f"t{timecode}"] 

207 

208 if width is not None or height is not None: 

209 dimension = "x".join(str(x) for x in [width, height] if x is not None) 

210 bits.append(dimension) 

211 

212 if mode != ThumbnailMode.DEFAULT: 

213 bits.append(mode.value) 

214 

215 if quality is not None: 

216 bits.append(f"q{quality}") 

217 

218 return "_".join(bits) 

219 

220 

221def get_video_info(filename): 

222 """Read video information using ffprobe if available. 

223 

224 Returns a dict with: width, height and duration. 

225 """ 

226 ffprobe = locate_executable("ffprobe") 

227 if ffprobe is None: 

228 raise RuntimeError("Failed to locate ffprobe") 

229 

230 proc = portable_popen( 

231 [ 

232 ffprobe, 

233 "-v", 

234 "quiet", 

235 "-print_format", 

236 "json", 

237 "-show_format", 

238 "-show_streams", 

239 filename, 

240 ], 

241 stdout=subprocess.PIPE, 

242 ) 

243 stdout, _ = proc.communicate() 

244 

245 if proc.returncode != 0: 

246 raise RuntimeError(f"ffprobe exited with code {proc.returncode}") 

247 

248 ffprobe_data = json.loads(stdout.decode("utf8")) 

249 info = { 

250 "width": None, 

251 "height": None, 

252 "duration": None, 

253 } 

254 

255 # Try to extract total video duration 

256 try: 

257 info["duration"] = timedelta(seconds=float(ffprobe_data["format"]["duration"])) 

258 except (KeyError, TypeError, ValueError): 

259 pass 

260 

261 # Try to extract width and height from the first found video stream 

262 for stream in ffprobe_data["streams"]: 

263 if stream["codec_type"] != "video": 

264 continue 

265 

266 info["width"] = int(stream["width"]) 

267 info["height"] = int(stream["height"]) 

268 

269 # We currently don't bother with multiple video streams 

270 break 

271 

272 return info 

273 

274 

275def make_video_thumbnail( 

276 ctx, 

277 source_video, 

278 source_url_path, 

279 seek, 

280 *, 

281 width=None, 

282 height=None, 

283 mode=ThumbnailMode.DEFAULT, 

284 upscale=None, 

285 quality=None, 

286 format=None, 

287): 

288 if mode != ThumbnailMode.FIT and (width is None or height is None): 

289 msg = '"%s" mode requires both `width` and `height` to be defined.' 

290 raise ValueError(msg % mode.value) 

291 

292 if upscale is None: 

293 upscale = { 

294 ThumbnailMode.FIT: False, 

295 ThumbnailMode.CROP: True, 

296 ThumbnailMode.STRETCH: True, 

297 }[mode] 

298 

299 if format is None: 

300 format = "jpg" 

301 if format not in THUMBNAIL_FORMATS: 

302 raise ValueError(f'Invalid thumbnail format "{format}"') 

303 

304 if quality is not None and format != "jpg": 

305 raise ValueError("The quality parameter is only supported for jpeg images") 

306 

307 if seek < timedelta(0): 

308 raise ValueError("Seek must not be negative") 

309 

310 ffmpeg = locate_executable("ffmpeg") 

311 if ffmpeg is None: 

312 raise RuntimeError("Failed to locate ffmpeg") 

313 info = get_video_info(source_video) 

314 

315 source_dim = Dimensions(info["width"], info["height"]) 

316 resize_dim, crop_dim = source_dim.resize(width, height, mode, upscale) 

317 

318 # Construct a filename suffix unique to the given parameters 

319 suffix = get_suffix(seek, width, height, mode=mode, quality=quality) 

320 dst_url_path = get_dependent_url(source_url_path, suffix, ext=f".{format}") 

321 

322 if quality is None and format == "jpg": 

323 quality = 95 

324 

325 def build_thumbnail_artifact(artifact): 

326 artifact.ensure_dir() 

327 

328 vfilter = ( 

329 "thumbnail", 

330 f"scale={resize_dim.width}:{resize_dim.height}", 

331 f"crop={crop_dim.width}:{crop_dim.height}", 

332 ) 

333 

334 cmdline = [ 

335 ffmpeg, 

336 "-loglevel", 

337 "-8", 

338 "-ss", 

339 get_timecode(seek), # Input seeking since it's faster 

340 "-i", 

341 source_video, 

342 "-vf", 

343 ",".join(vfilter), 

344 "-frames:v", 

345 "1", 

346 "-qscale:v", 

347 str(get_ffmpeg_quality(quality)), 

348 artifact.dst_filename, 

349 ] 

350 

351 reporter.report_debug_info("ffmpeg cmd line", cmdline) 

352 proc = portable_popen(cmdline) 

353 if proc.wait() != 0: 

354 raise RuntimeError(f"ffmpeg exited with code {proc.returncode}") 

355 

356 if not os.path.exists(artifact.dst_filename): 

357 msg = ( 

358 f"Unable to create video thumbnail for {source_video!r}. " 

359 "Maybe the seek is outside of the video duration?" 

360 ) 

361 raise RuntimeError(msg) 

362 

363 ctx.sub_artifact(artifact_name=dst_url_path, sources=[source_video])(build_thumbnail_artifact) 

364 

365 return Thumbnail(dst_url_path, crop_dim.width, crop_dim.height)