https://sppas.org/

SPPAS 4.16

Module sppas.src.videodata

Class sppasVideoReaderBuffer

Description

Class to 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.

Example

Initialize a VideoBuffer with a size of 100 images and overlap of 10:

Example

>>> v = sppasVideoReaderBuffer(video, 100, 10)

Bufferize the next sequence of images of the video:

Example

>>> v.next()

Release the flow taken by the reading of the video:

Example

>>> v.close()

Constructor

Create a new sppasVideoReaderBuffer instance.

Parameters

  • video: (mp4, etc...) The video filename to browse
  • size: (int) Number of images of the buffer or -1 for auto
  • overlap: (overlap) The number of images to keep from the previous buffer
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)

Public functions

open

Override. Create an opencv video capture from the given video.

Parameters

  • video: (name of video file, image sequence, url or video stream, GStreamer pipeline, IP camera) The video to browse.
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)
close

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

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_with_start_idx

Reset AND set the index of the 1st frame of the current buffer.

Returns

  • (tuple) start index of the frames in the buffer
Parameters
  • idx
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)
get_buffer_size

Return the defined size of the buffer.

def get_buffer_size(self):
    """Return the defined size of the buffer."""
    return self.__nb_img
set_buffer_size

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

MAXMEMORYSIZE Gb of RAM.

Parameters

  • value: (int) The new size of the buffer.

Raises

ValueError

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
        MAX_MEMORY_SIZE Gb of RAM.

        :param value: (int) The new size of the buffer.
        :raise: ValueError

        """
    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))
get_buffer_overlap

Return the overlap of the buffer.

def get_buffer_overlap(self):
    """Return the overlap of the buffer."""
    return self.__overlap
set_buffer_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.

Parameters

  • value: (int)

Raises

ValueError

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)
        :raise: ValueError

        """
    overlap = int(value)
    if overlap >= self.__nb_img or overlap < 0:
        raise ValueError('invalid buffer overlap')
    self.__overlap = value
seek_buffer

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.

Parameters

  • value: (int) Frame position
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)
tell_buffer

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
get_buffer_range

Return the indexes of the frames of the current buffer.

Returns

  • (tuple) start index, end index of the frames in the 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
next

Fill in the buffer with the next sequence of images of the video.

Returns

  • False if we reached the end 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
    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
    self.seek(start_frame)
    result = self.__load_frames(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()
check_buffer_index

Raise an exception if the given image index is not valid.

Parameters

  • value: (int)

Raises

NegativeValueError, IndexRangeException

def check_buffer_index(self, value):
    """Raise an exception if the given image index is not valid.

        :param value: (int)
        :raise: NegativeValueError, 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

Append an image into the buffer and pop the first if full queue.

Parameters

  • image: (sppasImage) A new image to append to the list
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

Pop an image of the buffer.

Parameters

  • img_idx: (int) Index of the image in the buffer

Raises

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_at

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!

Parameters

  • img: (sppasImage) Set the image in the buffer
  • img_idx: (int) Index of the image in the buffer

Raises

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
        :raise: 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())

Protected functions

__load_frames

Browse a sequence of a video.

Returns

  • (list) List of sppasImage instances or None if reading a frame failed
Parameters
  • nb_frames
def __load_frames(self, nb_frames):
    """Browse a sequence of a video.

        :returns: (list) List of sppasImage instances or None if reading
        a frame failed

        """
    w, h = (0, 0)
    from_pos = self.tell()
    to_pos = self.get_nframes()
    if from_pos + nb_frames > to_pos:
        nb_frames = to_pos - from_pos
    images = list()
    for i in range(nb_frames):
        image_array = self.read_frame()
        if image_array is not None and w == 0:
            w, h = image_array.size()
        if image_array is None and w != 0:
            image_array = sppasImage(0).blank_image(w, h)
        images.append(image_array)
    return images

Overloads

__len__

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)
__iter__

Browse the current data buffer.

def __iter__(self):
    """Browse the current data buffer."""
    for img in self.__images:
        yield img
__getitem__
def __getitem__(self, item):
    return self.__images[item]
__str__
def __str__(self):
    liste = list()
    iterator = self.__iter__()
    for i in range(len(self.__images)):
        liste.append(str(next(iterator)) + '\n')
    return liste
__repr__
def __repr__(self):
    return self.__class__.__name__
__format__
def __format__(self, fmt):
    return str(self).__format__(fmt)