SPPAS 4.22

https://sppas.org/

Module sppas.src.videodata

Class sppasVideoReaderBuffer

Description

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
View Source
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.
View Source
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.

View Source
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.

View Source
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
View Source
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.

View Source
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 a MAXMEMORYSIZE Gb of RAM.

Parameters
  • value: (int) New size of the buffer.
Raises
  • ValueError: invalid given value
View Source
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))

get_buffer_overlap

Return the overlap value of the buffer.

View Source
def get_buffer_overlap(self):
    """Return the overlap value 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) Nb of image
Raises
  • ValueError: Invalid given value
View Source
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

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
View Source
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.

View Source
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
View Source
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
View Source
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()

check_buffer_index

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

Parameters
  • value: (int)
Raises

NegativeValueError IndexRangeException

View Source
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

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

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

View Source
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

View Source
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())

Protected functions

__load_frames

Browse a sequence of a video.

Returns
  • (list) List of either sppasImage or None
Parameters
  • start_frame
  • nb_frames
View Source
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

Overloads

__len__

Return the number of images in the current data buffer.

View Source
def __len__(self):
    """Return the number of images in the current data buffer."""
    return len(self.__images)

__iter__

Browse the current data buffer.

View Source
def __iter__(self):
    """Browse the current data buffer."""
    for img in self.__images:
        yield img

__getitem__

View Source
def __getitem__(self, item):
    return self.__images[item]

__str__

View Source
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

__repr__

View Source
def __repr__(self):
    return self.__class__.__name__

__format__

View Source
def __format__(self, fmt):
    return str(self).__format__(fmt)