1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156
|
import os
from typing import Union
import datetime
import moderngl
from moderngl_window.timers.clock import Timer
class BaseVideoCapture:
"""
``BaseVideoCapture`` is a base class to video capture
Args:
source (moderngl.Texture, moderngl.Framebuffer): the source of the capture
framerate (int, float) : the framerate of the video, by thefault is 60 fps
if the source is texture there are some requirements:
- dtype = 'f1';
- components >= 3.
"""
def __init__(
self,
source: Union[moderngl.Texture, moderngl.Framebuffer] = None,
framerate: Union[int, float] = 60,
):
self._source = source
self._framerate = framerate
self._recording = False
self._last_time: float = None
self._filename: str = None
self._width: int = None
self._height: int = None
self._timer = Timer()
self._components: int = None # for textures
if isinstance(self._source, moderngl.Texture):
self._components = self._source.components
def _dump_frame(self, frame):
"""
custom function called during self.save()
Args:
frame: frame data in bytes
"""
raise NotImplementedError("override this function")
def _start_func(self) -> bool:
"""
custom function called during self.start_capture()
must return a True if this function complete without errors
"""
raise NotImplementedError("override this function")
def _release_func(self):
"""
custom function called during self.realease()
"""
raise NotImplementedError("override this function")
def _get_wh(self):
"""
Return a tuple of the width and the height of the source
"""
return self._source.width, self._source.height
def _remove_file(self):
""" Remove the filename of the video is it exist """
if os.path.exists(self._filename):
os.remove(self._filename)
def start_capture(self, filename: str = None, framerate: Union[int, float] = 60):
"""
Start the capturing process
Args:
filename (str): name of the output file
framerate (int, float): framerate of the video
if filename is not specified it will be generated based
on the datetime.
"""
if self._recording:
print("Capturing is already started")
return
# ensure the texture has the correct dtype and components
if isinstance(self._source, moderngl.Texture):
if self._source.dtype != 'f1':
print("source type: moderngl.Texture must be type `f1` ")
return
if self._components < 3:
print("source type: moderngl.Texture must have at least 3 components")
return
if not filename:
now = datetime.datetime.now()
filename = f'video_{now:%Y%m%d_%H%M%S}.mp4'
self._filename = filename
self._framerate = framerate
self._width, self._height = self._get_wh()
# if something goes wrong with the start
# function, just stop and release the
# capturing process
if not self._start_func():
self.release()
print("Capturing failed")
return
self._timer.start()
self._last_time = self._timer.time
self._recording = True
def save(self):
"""
Save function to call at the end of render function
"""
if not self._recording:
return
dt = 1. / self._framerate
if self._timer.time - self._last_time > dt:
# start counting
self._last_time = self._timer.time
if isinstance(self._source, moderngl.Framebuffer):
# get data from framebuffer
frame = self._source.read(components=3)
self._dump_frame(frame)
else:
# get data from texture
frame = self._source.read()
self._dump_frame(frame)
def release(self):
"""
Stop the recording process
"""
if self._recording:
self._release_func()
self._timer.stop()
print(f"Video file succesfully saved as {self._filename}")
self._recording = None
|