Module dempy.acquisitions.video_sample

Expand source code
from typing import List, Dict, Any, Union, ByteString

from dempy._base import Entity
from dempy._protofiles import VideoMessage


class VideoSample(Entity):
    """VideoSample class"""
    def __init__(self, type: str, id: str, tags: List[str], metadata: Dict[str, str], timestamp: int, acquisition_id: str, device_id: str,
                 sensor_id: str, media_type: str, video_source: str):
        super().__init__(type, id, tags, metadata)
        self.timestamp = timestamp
        self.acquisition_id = acquisition_id
        self.device_id = device_id
        self.sensor_id = sensor_id
        self.media_type = media_type
        self.video_source = video_source

    @staticmethod
    def to_protobuf(obj: "VideoSample") -> VideoMessage:
        """Encode a video sample to a Protobuf message

        Arguments:
            obj {VideoSample} -- video sample to be encoded

        Returns:
            VideoMessage -- encoded video sample
        """
        video_message = VideoMessage()
        video_message.entity.CopyFrom(Entity.to_protobuf(obj))

        video_message.timestamp = obj.timestamp
        video_message.acquisition_id = obj.acquisition_id

        if obj.device_id is not None:
            video_message.device_id = obj.device_id
        if obj.sensor_id is not None:
            video_message.sensor_id = obj.sensor_id

        video_message.media_type = obj.media_type
        video_message.video_source = obj.video_source

        return video_message

    @staticmethod
    def from_protobuf(obj: Union[ByteString, VideoMessage]) -> "VideoSample":
        """Decode a Protobuf message to {VideoSample}

        Arguments:
            obj {Union[ByteString, VideoMessage]} -- message to be decoded

        Returns:
            VideoSample -- decoded video sample
        """
        video_message = obj if isinstance(obj, VideoMessage) else VideoMessage().ParseFromString(obj)

        return VideoSample(
            type=video_message.entity.type,
            id=video_message.entity.id,
            tags=video_message.entity.tags,
            metadata=video_message.entity.metadata,
            timestamp=video_message.timestamp,
            acquisition_id=video_message.acquisition_id,
            device_id=video_message.device_id if video_message.HasField("device_id") else None,
            sensor_id=video_message.sensor_id if video_message.HasField("sensor_id") else None,
            media_type=video_message.media_type,
            video_source=video_message.video_source
        )

    @staticmethod
    def from_json(obj: Dict[str, str]) -> Any:
        """Parse a JSON dictionary to {VideoSample}

        Arguments:
            obj {Dict[str, str]} -- JSON object

        Returns:
            Any -- parsed object and sub-objects
        """
        if "type" in obj and obj["type"] == "VideoSample":
            return VideoSample(
                type=obj["type"],
                id=obj["id"],
                tags=obj["tags"],
                metadata=obj["metadata"],
                timestamp=obj["timestamp"],
                acquisition_id=obj["acquisitionId"],
                device_id=obj["deviceId"],
                sensor_id=obj["sensorId"],
                media_type=obj["mediaType"],
                video_source=obj["videoSource"],
            )

        return obj


__all__ = [
    "VideoSample"
]

Classes

class VideoSample (type: str, id: str, tags: List[str], metadata: Dict[str, str], timestamp: int, acquisition_id: str, device_id: str, sensor_id: str, media_type: str, video_source: str)

VideoSample class

Expand source code
class VideoSample(Entity):
    """VideoSample class"""
    def __init__(self, type: str, id: str, tags: List[str], metadata: Dict[str, str], timestamp: int, acquisition_id: str, device_id: str,
                 sensor_id: str, media_type: str, video_source: str):
        super().__init__(type, id, tags, metadata)
        self.timestamp = timestamp
        self.acquisition_id = acquisition_id
        self.device_id = device_id
        self.sensor_id = sensor_id
        self.media_type = media_type
        self.video_source = video_source

    @staticmethod
    def to_protobuf(obj: "VideoSample") -> VideoMessage:
        """Encode a video sample to a Protobuf message

        Arguments:
            obj {VideoSample} -- video sample to be encoded

        Returns:
            VideoMessage -- encoded video sample
        """
        video_message = VideoMessage()
        video_message.entity.CopyFrom(Entity.to_protobuf(obj))

        video_message.timestamp = obj.timestamp
        video_message.acquisition_id = obj.acquisition_id

        if obj.device_id is not None:
            video_message.device_id = obj.device_id
        if obj.sensor_id is not None:
            video_message.sensor_id = obj.sensor_id

        video_message.media_type = obj.media_type
        video_message.video_source = obj.video_source

        return video_message

    @staticmethod
    def from_protobuf(obj: Union[ByteString, VideoMessage]) -> "VideoSample":
        """Decode a Protobuf message to {VideoSample}

        Arguments:
            obj {Union[ByteString, VideoMessage]} -- message to be decoded

        Returns:
            VideoSample -- decoded video sample
        """
        video_message = obj if isinstance(obj, VideoMessage) else VideoMessage().ParseFromString(obj)

        return VideoSample(
            type=video_message.entity.type,
            id=video_message.entity.id,
            tags=video_message.entity.tags,
            metadata=video_message.entity.metadata,
            timestamp=video_message.timestamp,
            acquisition_id=video_message.acquisition_id,
            device_id=video_message.device_id if video_message.HasField("device_id") else None,
            sensor_id=video_message.sensor_id if video_message.HasField("sensor_id") else None,
            media_type=video_message.media_type,
            video_source=video_message.video_source
        )

    @staticmethod
    def from_json(obj: Dict[str, str]) -> Any:
        """Parse a JSON dictionary to {VideoSample}

        Arguments:
            obj {Dict[str, str]} -- JSON object

        Returns:
            Any -- parsed object and sub-objects
        """
        if "type" in obj and obj["type"] == "VideoSample":
            return VideoSample(
                type=obj["type"],
                id=obj["id"],
                tags=obj["tags"],
                metadata=obj["metadata"],
                timestamp=obj["timestamp"],
                acquisition_id=obj["acquisitionId"],
                device_id=obj["deviceId"],
                sensor_id=obj["sensorId"],
                media_type=obj["mediaType"],
                video_source=obj["videoSource"],
            )

        return obj

Ancestors

  • dempy._base.Entity

Static methods

def from_json(obj: Dict[str, str]) -> Any

Parse a JSON dictionary to {VideoSample}

Arguments

obj {Dict[str, str]} – JSON object

Returns

Any -- parsed object and sub-objects
 
Expand source code
@staticmethod
def from_json(obj: Dict[str, str]) -> Any:
    """Parse a JSON dictionary to {VideoSample}

    Arguments:
        obj {Dict[str, str]} -- JSON object

    Returns:
        Any -- parsed object and sub-objects
    """
    if "type" in obj and obj["type"] == "VideoSample":
        return VideoSample(
            type=obj["type"],
            id=obj["id"],
            tags=obj["tags"],
            metadata=obj["metadata"],
            timestamp=obj["timestamp"],
            acquisition_id=obj["acquisitionId"],
            device_id=obj["deviceId"],
            sensor_id=obj["sensorId"],
            media_type=obj["mediaType"],
            video_source=obj["videoSource"],
        )

    return obj
def from_protobuf(obj: Union[ByteString, dempy_pb2.VideoSample]) -> VideoSample

Decode a Protobuf message to {VideoSample}

Arguments

obj {Union[ByteString, VideoMessage]} – message to be decoded

Returns

VideoSample -- decoded video sample
 
Expand source code
@staticmethod
def from_protobuf(obj: Union[ByteString, VideoMessage]) -> "VideoSample":
    """Decode a Protobuf message to {VideoSample}

    Arguments:
        obj {Union[ByteString, VideoMessage]} -- message to be decoded

    Returns:
        VideoSample -- decoded video sample
    """
    video_message = obj if isinstance(obj, VideoMessage) else VideoMessage().ParseFromString(obj)

    return VideoSample(
        type=video_message.entity.type,
        id=video_message.entity.id,
        tags=video_message.entity.tags,
        metadata=video_message.entity.metadata,
        timestamp=video_message.timestamp,
        acquisition_id=video_message.acquisition_id,
        device_id=video_message.device_id if video_message.HasField("device_id") else None,
        sensor_id=video_message.sensor_id if video_message.HasField("sensor_id") else None,
        media_type=video_message.media_type,
        video_source=video_message.video_source
    )
def to_protobuf(obj: VideoSample) -> dempy_pb2.VideoSample

Encode a video sample to a Protobuf message

Arguments

obj {VideoSample} – video sample to be encoded

Returns

VideoMessage -- encoded video sample
 
Expand source code
@staticmethod
def to_protobuf(obj: "VideoSample") -> VideoMessage:
    """Encode a video sample to a Protobuf message

    Arguments:
        obj {VideoSample} -- video sample to be encoded

    Returns:
        VideoMessage -- encoded video sample
    """
    video_message = VideoMessage()
    video_message.entity.CopyFrom(Entity.to_protobuf(obj))

    video_message.timestamp = obj.timestamp
    video_message.acquisition_id = obj.acquisition_id

    if obj.device_id is not None:
        video_message.device_id = obj.device_id
    if obj.sensor_id is not None:
        video_message.sensor_id = obj.sensor_id

    video_message.media_type = obj.media_type
    video_message.video_source = obj.video_source

    return video_message