mmcm/data/clip.py

from copy import deepcopy
from typing import Iterable
import logging

import numpy as np

from ..utils.util import convert_class_attr_to_dict

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


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

    def __init__(
        self,
        time_start,
        duration,
        clipid=None,
        media_type=None,
        mediaid=None,
        timepoint_type=None,
        text=None,
        stage=None,
        path=None,
        duration_num=None,
        group_time_start=0,
        group_clipid=None,
        original_clipid=None,
        emb=None,
        multi_factor=None,
        similar_clipseq=None,
        rythm: 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 (_type_, optional): 该媒体文件的路径,用于后续媒体读取、处理. Defaults to None.
            duration_num (_type_, optional): 片段持续帧数, Defaults to None.
            group_time_start (int, optional): 当多歌曲、多视频剪辑时,group_time_start 表示该片段所对应的子媒体前所有子媒体的片段时长总和。
                默认0, 表示只有1个媒体文件. Defaults to 0.
            group_clipid (int, optional):  # MediaInfo.sub_meta_info 中的实际序号.
            original_clipid (None or [int], optional): 有些片段由其他片段合并,该字段用于片段来源,id是 media_map.json 中的实际序号. Defaults to None.
            emb (np.array, optional): 片段 综合emb,. Defaults to None.
            multi_factor (MultiFactorFeature), 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.group_time_start = group_time_start
        self.group_clipid = group_clipid
        self.duration_num = duration_num
        self.original_clipid = original_clipid if original_clipid is not None else []
        self.emb = emb
        self.multi_factor = multi_factor
        self.similar_clipseq = similar_clipseq
        self.rythm = rythm
        # TODO: 目前谱面中会有一些不必要的中间结果，比较占内存，现在代码里删掉，待后续数据协议确定
        kwargs = {k: v for k, v in kwargs.items()}
        self.__dict__.update(kwargs)
        self.preprocess()

    def preprocess(self):
        pass

    def spread_parameters(self):
        pass

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

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

        Raises:
            NotImplementedError: _description_
        """
        raise NotImplementedError


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

    ClipClass = Clip

    def __init__(self, clips) -> None:
        """_summary_

        Args:
            clips ([Clip]]): 媒体片段序列
        """
        if not isinstance(clips, list):
            clips = [clips]
        if len(clips) == 0:
            self.clips = []
        elif isinstance(clips[0], dict):
            self.clips = [self.ClipClass(**d) for d in clips]
        else:
            self.clips = clips

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

    def __len__(
        self,
    ):
        return len(self.clips)

    def merge(self, other, group_time_start_delta=None, groupid_delta=None):
        """融合其他ClipSeq。media_info 融合时需要记录 clip 所在的 groupid 和 group_time_start，delta用于表示变化

        Args:
            other (ClipSeq): 待融合的ClipSeq
            group_time_start_delta (float, optional): . Defaults to None.
            groupid_delta (int, optional): _description_. Defaults to None.
        """
        if group_time_start_delta is not None or groupid_delta is not None:
            for i, clip in enumerate(other):
                if group_time_start_delta is not None:
                    clip.group_time_start += group_time_start_delta
                if groupid_delta is not None:
                    clip.groupid += groupid_delta
        self.clips.extend(other.clips)
        for i in range(len(self.clips)):
            self.clips[i].group_clipid = i

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

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

    def __getitem__(self, i) -> Clip:
        """支持索引和切片操作，如果输入是整数则返回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.clips[i]
            return clip
        elif isinstance(i, Iterable):
            clips = [self.__getitem__(x) for x in i]
            clipseq = ClipSeq(clips)
            return clipseq
        elif isinstance(i, slice):
            if i.step is None:
                step = 1
            else:
                step = i.step
            clips = [self.__getitem__(x) for x in range(i.start, i.stop, step)]
            clipseq = ClipSeq(clips)
            return clipseq
        else:
            raise ValueError(
                "unsupported input, should be int or slice, but given {}, type={}".format(
                    i, type(i)
                )
            )

    def insert(self, idx, obj):
        self.clips.insert(idx, obj)

    def append(self, obj):
        self.clips.append(obj)

    def extend(self, objs):
        self.clips.extend(objs)

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

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

    @property
    def rela_timestamp_seq_emb(self):
        emb = self.timestamp_seq_emb / self.duration
        return emb

    def get_factor_seq_emb(self, factor, dim):
        emb = []
        for c in self.clips:
            if factor not in c.multi_factor or c.multi_factor[factor] is None:
                v = np.full(dim, np.inf)
            else:
                v = c.multi_factor[factor]
            emb.append(v)
        emb = np.stack(emb, axis=0)
        return emb

    def semantic_seq_emb(self, dim):
        return self.get_factor_seq_emb(factor="semantics", dim=dim)

    def emotion_seq_emb(self, dim):
        return self.get_factor_seq_emb(factor="emotion", dim=dim)

    def theme_seq_emb(self, dim):
        return self.get_factor_seq_emb(factor="theme", dim=dim)

    def to_dct(
        self,
        target_keys=None,
        ignored_keys=None,
    ):
        if ignored_keys is None:
            ignored_keys = ["kwargs", "audio_path", "lyric_path", "start", "end"]
        clips = [
            clip.to_dct(target_keys=target_keys, ignored_keys=ignored_keys)
            for clip in self.clips
        ]
        return clips

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

        Raises:
            NotImplementedError: _description_
        """
        raise NotImplementedError


class ClipIds(object):
    def __init__(
        self,
        clipids: list or int,
    ) -> None:
        """ClipSeq 中的 Clip序号，主要用于多个 Clip 融合后的 Clip, 使用场景如
        1. 一个 MusicClip 可以匹配到多个 VideoClip，VideoClip 的索引便可以使用 ClipIds 定义。

        Args:
            clipids (list or int): ClipSeq 中的序号
        """
        self.clipids = clipids if isinstance(clipids, list) else [clipids]


class ClipIdsSeq(object):
    def __init__(self, clipids_seq: list) -> None:
        """多个 ClipIds，使用场景可以是
        1. 将MediaClipSeq 进行重组，拆分重组成更粗粒度的ClipSeq；

        Args:
            clipids_seq (list): 组合后的 ClipIds 列表
        """
        self.clipids_seq = (
            clipids_seq if isinstance(clipids_seq, ClipIds) else [clipids_seq]
        )


# TODO: metric后续可能是字典
class MatchedClipIds(object):
    def __init__(
        self, id1: ClipIds, id2: ClipIds, metric: float = None, **kwargs
    ) -> None:
        """两种模态数据的片段匹配对，使用场景 可以是
        1. 音乐片段和视频片段 之间的匹配关系，

        Args:
            id1 (ClipIds): 第一种模态的片段
            id2 (ClipIds): 第二种模态的片段
            metric (float): 匹配度量距离
        """
        self.id1 = id1 if isinstance(id1, ClipIds) else ClipIds(id1)
        self.id2 = id2 if isinstance(id2, ClipIds) else ClipIds(id2)
        self.metric = metric
        self.__dict__.update(**kwargs)


class MatchedClipIdsSeq(object):
    def __init__(self, seq: list, metric: float = None, **kwargs) -> None:
        """两种模态数据的序列匹配对，使用场景可以是
        1. 音乐片段序列和视频片段序列 之间的匹配，每一个元素都是MatchedClipIds:

        Args:
            seq (list): 两种模态数据的序列匹配对列表
            metric (float): 匹配度量距离
        """
        self.seq = seq
        self.metric = metric
        self.__dict__.update(**kwargs)