Manage a video with a buffer (a queue) of images.
This class allows to use a buffer of images on a video to manage it
sequentially, and to have a better control on it.
Initialize a VideoBuffer with a size of 100 images and overlap of 10:
>>> v = sppasVideoReaderBuffer(video, 100, 10)
Bufferize the next sequence of images of the video:
>>> v.next()
Release the flow taken by the reading of the video:
>>> v.close()
Create a new sppasVideoReaderBuffer instance.
def __init__(self, video=None, size=-1, overlap=DEFAULT_BUFFER_OVERLAP):
"""Create a new sppasVideoReaderBuffer instance.
:param video: (mp4, etc...) The video filename to browse
:param size: (int) Number of images of the buffer or -1 for auto
:param overlap: (overlap) The number of images to keep
from the previous buffer
"""
super(sppasVideoReaderBuffer, self).__init__()
self.__nb_img = 0
self.__overlap = 0
self.set_buffer_size(size)
self.set_buffer_overlap(overlap)
self.__images = list()
self.__buffer_idx = (-1, -1)
if video is not None:
self.open(video)
if size == -1:
self.set_buffer_size(size)
Override. Create an opencv video capture from the given video.
def open(self, video):
"""Override. Create an opencv video capture from the given video.
:param video: (name of video file, image sequence, url or video
stream, GStreamer pipeline, IP camera) The video to browse.
"""
self.reset()
sppasVideoReader.open(self, video)
Override. Release the flow taken by the reading of the video.
def close(self):
"""Override. Release the flow taken by the reading of the video."""
self.reset()
sppasVideoReader.close(self)
Reset the buffer but does not change anything to the video.
def reset(self):
"""Reset the buffer but does not change anything to the video."""
self.__images = list()
self.__buffer_idx = (-1, -1)
Reset AND set the index of the 1st frame of the current buffer.
def reset_with_start_idx(self, idx):
"""Reset AND set the index of the 1st frame of the current buffer.
:return: (tuple) start index of the frames in the buffer
"""
self.__images = list()
idx = int(idx)
self.__buffer_idx = (idx, idx)
Return the defined size of the buffer.
def get_buffer_size(self):
"""Return the defined size of the buffer."""
return self.__nb_img
Set the size of the buffer.
The new value is applied to the next buffer, it won't affect the currently in-use data.
A value of -1 will fix automatically the buffer to use a MAXMEMORYSIZE Gb of RAM.
def set_buffer_size(self, value=-1):
"""Set the size of the buffer.
The new value is applied to the next buffer, it won't affect the currently in-use data.
A value of -1 will fix automatically the buffer to use a MAX_MEMORY_SIZE Gb of RAM.
:param value: (int) New size of the buffer.
:raises: ValueError: invalid given value
"""
value = int(value)
if value == -1:
if self.is_opened() is False:
w, h = (1920, 1080)
else:
w, h = (self.get_width(), self.get_height())
nbytes = w * h * 3
value = sppasVideoReaderBuffer.MAX_MEMORY_SIZE // nbytes
if self.is_opened() is True and value > self.get_nframes():
value = self.get_nframes()
if value <= 0:
raise NegativeValueError(value)
if self.is_opened() is True and value > self.get_nframes():
value = self.get_nframes()
if self.__overlap >= value:
raise ValueError("The already defined overlap value {:d} can't be greater than the buffer size.")
self.__nb_img = value
logging.info('The video buffer is set to {:d} images'.format(self.__nb_img))
Return the overlap value of the buffer.
def get_buffer_overlap(self):
"""Return the overlap value of the buffer."""
return self.__overlap
Set the number of images to keep from the previous buffer.
The new value is applied to the next buffer, it won't affect the
currently in-use data.
def set_buffer_overlap(self, value):
"""Set the number of images to keep from the previous buffer.
The new value is applied to the next buffer, it won't affect the
currently in-use data.
:param value: (int) Nb of image
:raises: ValueError: Invalid given value
"""
overlap = int(value)
if overlap >= self.__nb_img or overlap < 0:
raise ValueError('Invalid buffer overlap')
self.__overlap = value
Set the position of the frame for the next buffer to be read.
It won't change the current position in the video until "next" is
invoked. It invalidates the current buffer.
def seek_buffer(self, value):
"""Set the position of the frame for the next buffer to be read.
It won't change the current position in the video until "next" is
invoked. It invalidates the current buffer.
:param value: (int) Frame position
"""
value = self.check_frame(value)
self.reset()
self.__buffer_idx = (-1, value - 1)
Return the frame position for the next buffer to be read.
Possibly, it can't match the current position in the stream, if
video.read() was invoked for example.
def tell_buffer(self):
"""Return the frame position for the next buffer to be read.
Possibly, it can't match the current position in the stream, if
video.read() was invoked for example.
"""
return self.__buffer_idx[1] + 1
Return the indexes of the frames of the current buffer.
def get_buffer_range(self):
"""Return the indexes of the frames of the current buffer.
:return: (tuple) start index, end index of the frames in the buffer
"""
if -1 in self.__buffer_idx:
return (-1, -1)
return self.__buffer_idx
Fill in the buffer with the next sequence of images of the video.
def next(self):
"""Fill in the buffer with the next sequence of images of the video.
:return: False if we reached the end of the video
"""
if self.is_opened() is False:
return False
self.__images = list()
nb_frames = self.__nb_img - self.__overlap
if self.__buffer_idx[1] == -1:
nb_frames = self.__nb_img
start_frame = self.__buffer_idx[1] + 1
if start_frame >= self.get_nframes():
return False
result = self.__load_frames(start_frame, nb_frames)
next_frame = start_frame + len(result)
delta = self.__nb_img - self.__overlap
self.__images = self.__images[delta:]
self.__buffer_idx = (start_frame - len(self.__images), next_frame - 1)
self.__images.extend(result)
result.clear()
return next_frame < self.get_nframes()
Raise an exception if the given image index is not valid.
NegativeValueError
IndexRangeException
def check_buffer_index(self, value):
"""Raise an exception if the given image index is not valid.
:param value: (int)
:raises: NegativeValueError
:raises: IndexRangeException
"""
value = int(value)
if value < 0:
raise NegativeValueError(value)
begin, end = self.get_buffer_range()
if value < self.get_buffer_size():
return value
raise IndexRangeException(value, 0, self.get_buffer_size())
Append an image into the buffer and pop the first if full queue.
def append(self, image):
"""Append an image into the buffer and pop the first if full queue.
:param image: (sppasImage) A new image to append to the list
"""
if isinstance(image, sppasImage) is False:
raise sppasTypeError(type(image), 'sppasImage')
self.__images.append(image)
if len(self.__images) > self.__nb_img:
self.__images.pop(0)
Pop an image of the buffer.
IndexRangeException
def pop(self, img_idx):
"""Pop an image of the buffer.
:param img_idx: (int) Index of the image in the buffer
:raise: IndexRangeException
"""
img_idx = int(img_idx)
if 0 <= img_idx < self.get_buffer_size():
self.__images.pop(img_idx)
else:
raise IndexRangeException(img_idx, 0, self.get_buffer_size())
Set an image of the buffer.
No verification is performed on the image. It should be the
same format (size, nb channels, etc) than the other ones.
Use this method with caution!
IndexRangeException
def set_at(self, img, img_idx):
"""Set an image of the buffer.
No verification is performed on the image. It should be the
same format (size, nb channels, etc) than the other ones.
Use this method with caution!
:param img: (sppasImage) Set the image in the buffer
:param img_idx: (int) Index of the image in the buffer
:raises: IndexRangeException
"""
img_idx = int(img_idx)
if 0 <= img_idx < self.get_buffer_size():
self.__images[img_idx] = img
else:
raise IndexRangeException(img_idx, 0, self.get_buffer_size())
Browse a sequence of a video.
def __load_frames(self, start_frame, nb_frames):
"""Browse a sequence of a video.
:return: (list) List of either sppasImage or None
"""
self.seek(start_frame)
if start_frame + nb_frames > self.get_nframes():
logging.info('Not enough remaining {:d} frames. Will read {:d} instead.'.format(nb_frames, self.get_nframes() - start_frame))
nb_frames = self.get_nframes() - start_frame
images = [None] * nb_frames
for i in range(nb_frames):
images[i] = self.read_frame()
if images[i] is None:
logging.warning("A problem occurred when reading image at index {:d}. The image is set to 'None' instead.".format(i))
return images
Return the number of images in the current data buffer.
def __len__(self):
"""Return the number of images in the current data buffer."""
return len(self.__images)
Browse the current data buffer.
def __iter__(self):
"""Browse the current data buffer."""
for img in self.__images:
yield img
def __getitem__(self, item):
return self.__images[item]
def __str__(self):
liste = [''] * len(self.__images)
iterator = self.__iter__()
for i in range(len(self.__images)):
liste[i] = str(next(iterator)) + '\n'
return liste
def __repr__(self):
return self.__class__.__name__
def __format__(self, fmt):
return str(self).__format__(fmt)