#!/usr/bin/env python3
"""Defines the structure of an abstract video stream."""
from fractions import Fraction
import abc
import math
import numbers
import typing
import torch
from cutcutcodec.core.classes.filter import Filter
from cutcutcodec.core.classes.frame_video import FrameVideo
from cutcutcodec.core.classes.stream import Stream, StreamWrapper
[docs]
class StreamVideo(Stream):
"""Representation of any video stream.
Attributes
----------
is_space_continuous : boolean
True if the data is continuous in the spacial domain, False if it is discrete (readonly).
"""
def __init__(self, node):
super().__init__(node)
self._mask = None # for cache speedup
def _from_masked_items(self, flat_frame: torch.Tensor, mask: torch.Tensor) -> torch.Tensor:
"""Return a new frame with the initialised values at the position given by the mask.
A copy is made only if needed.
It is the complementary method of
``cutcutcodec.core.classes.stream_video.StreamVideo._to_masked_items``.
"""
if mask is self._mask:
return torch.reshape(flat_frame, (*mask.shape, 1))
frame = torch.empty(*mask.shape, dtype=flat_frame.dtype, device=flat_frame.device)
frame[mask] = flat_frame
frame = torch.unsqueeze(frame, 2)
return frame
def _snapshot(self, timestamp: Fraction, mask: torch.Tensor) -> torch.Tensor:
"""Abstract help for ``snapshot``.
Parameters
----------
timestamp : fractions.Fraction
The absolute time expressed in seconds
mask : torch.Tensor[boolean]
The 2d boolean matrix of the interested pixels.
Returns
-------
torch.Tensor[torch.float32]
A floting point matrix of pixel between 0 and 1.
"""
raise NotImplementedError
def _to_masked_items(self, frame: torch.Tensor, mask: torch.Tensor) -> torch.Tensor:
"""Return a contiguous flattened tensor equivalent to `frame[mask]`.
A copy is made only if needed.
It is the complementary method of
``cutcutcodec.core.classes.stream_video.StreamVideo._from_masked_items``.
"""
if mask is self._mask:
return torch.ravel(frame)
return torch.ravel(torch.masked_select(frame, mask).contiguous())
@property
@abc.abstractmethod
def is_space_continuous(self) -> bool:
"""Return True if the data is continuous in the spacial domain, False if it is discrete."""
raise NotImplementedError
[docs]
def snapshot(
self,
timestamp: numbers.Real,
shape: typing.Union[tuple[numbers.Integral, numbers.Integral], list[numbers.Integral]],
*, channels=None
) -> FrameVideo:
"""Extract the closest frame to the requested date.
Parameters
----------
timestamp : numbers.Real
The absolute time expressed in seconds, not relative to the beginning of the video.
For avoid the inacuracies of round, it is recomended to use fractional number.
shape : tuple[int, int] or list[int, int]
The pixel dimensions of the returned frame.
The convention adopted is the numpy convention (height, width).
channels : int, optional
Impose the numbers of channels, apply convertion if nescessary.
For the interpretation of the layers,
see ``cutcutcodec.core.classes.frame_video.FrameVideo``.
Returns
-------
frame : cutcutcodec.core.classes.frame_video.FrameVideo
Video frame with metadata.
Raises
------
cutcutcodec.core.exception.OutOfTimeRange
If we try to get a frame out of the definition range.
The valid range is [self.beginning, self.beginning+self.duration[.
"""
assert isinstance(timestamp, numbers.Real), timestamp.__class__.__name__
assert isinstance(shape, (tuple, list)), shape.__class__.__name__
assert len(shape) == 2, len(shape)
assert all(isinstance(s, numbers.Integral) and s >= 1 for s in shape), shape
shape = (int(shape[0]), int(shape[1]))
if math.isnan(timestamp): # default transparent video frame
frame = FrameVideo(0, torch.zeros((*shape, 2)))
else:
if self._mask is None or self._mask.shape != shape:
self._mask = torch.full(shape, True, dtype=bool)
frame = self._snapshot(Fraction(timestamp), self._mask)
frame = FrameVideo(timestamp, frame)
if channels is not None:
frame = frame.convert(channels)
return frame
@property
def type(self) -> str:
"""Implement ``cutcutcodec.core.classes.stream.Stream.type``."""
return "video"
[docs]
class StreamVideoWrapper(StreamWrapper, StreamVideo):
"""Allow to dynamically transfer the methods of an instanced video stream.
This can be very useful for implementing filters.
Attribute
---------
stream : cutcutcodec.core.classes.stream_video.StreamVideo
The video stream containing the properties to be transferred (readonly).
This stream is one of the input streams of the parent node.
"""
def __init__(self, node: Filter, index: numbers.Integral):
"""Initialise and create the class.
Parameters
----------
node : cutcutcodec.core.classes.filter.Filter
The parent node, transmitted to ``cutcutcodec.core.classes.stream.Stream``.
index : number.Integral
The index of the video stream among all the input streams of the ``node``.
0 for the first, 1 for the second ...
"""
assert isinstance(node, Filter), node.__class__.__name__
assert len(node.in_streams) > index, f"only {len(node.in_streams)} streams, no {index}"
assert isinstance(node.in_streams[index], StreamVideo), "the stream must be video type"
super().__init__(node, index)
def _snapshot(self, timestamp: Fraction, mask: torch.Tensor) -> torch.Tensor:
return self.stream._snapshot(timestamp, mask) # pylint: disable=W0212
@property
def is_space_continuous(self) -> bool:
"""Return True if the data is continuous in the spacial domain, False if it is discrete."""
return self.stream.is_space_continuous