mmcm/data/clip/clip.py

from __future__ import annotations
from copy import deepcopy

from typing import Iterable, List, Tuple, Dict, Hashable, Any, Union

import numpy as np

from ...utils.util import convert_class_attr_to_dict


from ..general.items import Items, Item
from .clipid import MatchedClipIds


import logging

logger = logging.getLogger(__name__)  # pylint: disable=invalid-name


__all__ = ["Clip", "ClipSeq"]


class Clip(Item):
    """媒体片段, 指转场点与转场点之间的部分"""

    def __init__(
        self,
        time_start: float,
        duration: float,
        clipid: int = None,
        media_type: str = None,
        mediaid: str = None,
        timepoint_type: str = None,
        text: str = None,
        stage: str = None,
        path: str = None,
        duration_num: int = None,
        similar_clipseq: MatchedClipIds = None,
        dynamic: float = None,
        **kwargs,
    ):
        """
        Args:
            time_start (float): 开始时间,秒为单位,对应该媒体文件的, 和media_map.json上的序号一一对应
            duration (_type_): 片段持续时间
            clipid (int, or [int]): 由media_map提供的片段序号, 和media_map.json上的序号一一对应
            media_type (str, optional): music, video,text, Defaults to None.
            mediaid (int): 多媒体id, 当clipid是列表时,表示该片段是个融合片段
            timepoint_type(int, ): 开始点的转场类型. Defaults to None.
            text(str, optional): 该片段的文本描述,音乐可以是歌词,视频可以是台词,甚至可以是弹幕. Defaults to None.
            stage(str, optional): 该片段在整个媒体文件中的结构位置,如音乐的intro、chrous、vesa,视频的片头、片尾、开始、高潮、转场等. Defaults to None.
            path (str, optional): 该媒体文件的路径,用于后续媒体读取、处理. Defaults to None.
            duration_num (_type_, optional): 片段持续帧数, Defaults to None.
            similar_clipseq ([Clip]], optional): 与该片段相似的片段，具体结构待定义. Defaults to None.
        """
        self.media_type = media_type
        self.mediaid = mediaid
        self.time_start = time_start
        self.duration = duration
        self.clipid = clipid
        self.path = path
        self.timepoint_type = timepoint_type
        self.text = text
        self.stage = stage
        self.duration_num = duration_num
        self.similar_clipseq = similar_clipseq
        self.dynamic = dynamic
        self.__dict__.update(**kwargs)

    def preprocess(self):
        pass

    def spread_parameters(self):
        pass

    @property
    def time_end(
        self,
    ) -> float:
        return self.time_start + self.duration

    def get_emb(self, key: str, idx: int) -> np.float:
        return self.emb.get_value(key, idx)


class ClipSeq(Items):
    """媒体片段序列"""

    def __init__(self, items: List[Clip] = None):
        super().__init__(items)
        self.clipseq = self.data

    def preprocess(self):
        pass

    def set_clip_value(self, k: Hashable, v: Any) -> None:
        """给序列中的每一个clip 赋值"""
        for i in range(len(self.clipseq)):
            self.clipseq[i].__setattr__(k, v)

    def __len__(
        self,
    ) -> int:
        return len(self.clipseq)

    @property
    def duration(
        self,
    ) -> float:
        """Clip.duration的和

        Returns:
            float: 序列总时长
        """
        if len(self.clipseq) == 0:
            return 0
        else:
            return sum([c.duration for c in self.clipseq])

    def __getitem__(self, i: Union[int, Iterable]) -> Union[Clip, ClipSeq]:
        """支持索引和切片操作，如果输入是整数则返回Clip，如果是切片，则返回ClipSeq

        Args:
            i (int or slice): 索引

        Raises:
            ValueError: 需要按照给的输入类型索引

        Returns:
            Clip or ClipSeq:
        """
        if "int" in str(type(i)):
            i = int(i)
        if isinstance(i, int):
            clip = self.clipseq[i]
            return clip
        elif isinstance(i, Iterable):
            clipseq = [self.__getitem__(x) for x in i]
            clipseq = ClipSeq(clipseq)
            return clipseq
        elif isinstance(i, slice):
            if i.step is None:
                step = 1
            else:
                step = i.step
            clipseq = [self.__getitem__(x) for x in range(i.start, i.stop, step)]
            clipseq = ClipSeq(clipseq)
            return clipseq
        else:
            raise ValueError(
                "unsupported input, should be int or slice, but given {}, type={}".format(
                    i, type(i)
                )
            )

    @property
    def mvp_clip(self):
        """读取实际的片段数据为moviepy格式

        Raises:
            NotImplementedError: _description_
        """
        raise NotImplementedError

    @property
    def duration_seq_emb(
        self,
    ) -> np.array:
        emb = np.array([c.duration for c in self.clipseq])
        return emb

    @property
    def timestamp_seq_emb(self) -> np.array:
        emb = np.array([c.time_start for c in self.clipseq])
        return emb

    @property
    def rela_timestamp_seq_emb(self) -> np.array:
        duration_seq = [c.duration for c in self.clipseq]
        emb = np.cumsum(duration_seq) / self.duration
        return emb

    def get_emb(self, key: str, idx: int) -> np.float:
        clip_start_idx = self.clipseq[0].clipid
        clip_end_idx = self.clipseq[-1].clipid
        # TODO: 待修改为更通用的形式
        if idx is None:
            idx = range(clip_start_idx, clip_end_idx + 1)
        elif isinstance(idx, int):
            idx += clip_start_idx
        elif isinstance(idx, Iterable):
            idx = [x + clip_start_idx for x in idx]
        else:
            raise ValueError(
                f"idx only support None, int, Iterable, but given {idx},type is {type(idx)}"
            )
        return self.emb.get_value(key, idx=idx)