class EventList(MugenList):
    """
    A list of Events which occur in some time sequence
    """

    end: Optional[float]

    def __init__(
        self,
        events: Optional[List[Union[Event, TIME_FORMAT]]] = None,
        *,
        end: TIME_FORMAT = None,
    ):
        """
        Parameters
        ----------
        events
            events which occur in the time sequence

        end
            duration of the time sequence
        """
        if events is not None:
            for index, event in enumerate(events):
                if not isinstance(event, Event):
                    # Convert event location to Event
                    events[index] = Event(event)

        self.end = end

        super().__init__(events)

    def __eq__(self, other):
        return super().__eq__(other) and self.end == other.end

    def __add__(self, rhs):
        return type(self)((super().__add__(rhs)), end=rhs.end)

    def __getitem__(self, item):
        result = list.__getitem__(self, item)
        if isinstance(item, slice):
            return type(self)(result, end=self.end)
        else:
            return result

    def __repr__(self):
        event_reprs = [event.index_repr(index) for index, event in enumerate(self)]
        pretty_repr = super().pretty_repr(event_reprs)
        return f"<{pretty_repr}, end: {self.end}>"

    def list_repr(self, indexes: range, selected: bool):
        """
        Repr for use in lists
        """
        return (
            f"<{self.__class__.__name__} {indexes.start}-{indexes.stop} ({len(self)}), "
            f"type: {self.type}, selected: {selected}>"
        )

    @property
    def type(self) -> Union[str, None]:
        if len(self) == 0:
            return None
        elif len(set([event.__class__.__name__ for event in self])) == 1:
            return self[0].__class__.__name__
        else:
            return "mixed"

    @property
    def locations(self) -> List[float]:
        return [event.location for event in self]

    @property
    def intervals(self) -> List[float]:
        return location.intervals_from_locations(self.locations)

    @property
    def segment_locations(self):
        """
        Returns
        -------
        locations of segments between events
        """
        return [0] + [event.location for event in self]

    @property
    @requires_end
    def segment_durations(self):
        """
        Returns
        -------
        durations of segments between events
        """
        return location.intervals_from_locations(self.locations + [self.end])

    @property
    def durations(self) -> List[float]:
        return [event.duration for event in self]

    @property
    def types(self) -> List[str]:
        return [event.__class__.__name__ for event in self]

    def add_events(self, events: List[Union[Event, TIME_FORMAT]]):
        for event in events:
            if not isinstance(event, Event):
                event = Event(event)
            bisect.insort(self, event)

    def offset(self, offset: float):
        """
        Offsets all events by the given amount
        """
        for event in self:
            event.location += offset

    @convert_float_to_fraction("speed")
    def speed_multiply(
        self, speed: Union[float, Fraction], offset: Optional[int] = None
    ):
        """
        Speeds up or slows down events by grouping them together or splitting them up.
        For slowdowns, event type group boundaries and isolated events are preserved.

        Parameters
        ----------
        speed
            Factor to speedup or slowdown by.
            Must be of the form x (speedup) or 1/x (slowdown), where x is a natural number.
            Otherwise, 0 to remove all events.

        offset
            Offsets the grouping of events for slowdowns.
            Takes a max offset of x - 1 for a slowdown of 1/x, where x is a natural number
        """
        if speed == 0:
            self.clear()
        elif speed > 1:
            self._split(speed.numerator)
        elif speed < 1:
            self._merge(speed.denominator, offset)

    def _split(self, pieces_per_split: int):
        """
        Splits events up to form shorter intervals

        Parameters
        ----------
        pieces_per_split
            Number of pieces to split each event into
        """
        splintered_events = EventList()

        # Group events by type
        type_groups = self.group_by_type()

        for group in type_groups:
            if len(group) == 1:
                # Always keep isolated events
                splintered_events.append(group[0])
            else:
                for index, event in enumerate(group):
                    splintered_events.append(event)

                    if index == len(group) - 1:
                        # Skip last event
                        continue

                    next_event = group[index + 1]
                    interval = next_event.location - event.location
                    interval_piece = interval / pieces_per_split

                    location = event.location
                    for _ in range(pieces_per_split - 1):
                        location += interval_piece
                        event_splinter = copy.deepcopy(event)
                        event_splinter.location = location
                        splintered_events.append(event_splinter)

        self[:] = splintered_events

    def _merge(self, pieces_per_merge: int, offset: Optional[int] = None):
        """
        Merges adjacent events of identical type to form longer intervals

        Parameters
        ----------
        pieces_per_merge
            Number of adjacent events to merge at a time

        offset
            Offset for the merging of events
        """
        if offset is None:
            offset = 0

        combined_events = EventList()

        # Group events by type
        type_groups = self.group_by_type()
        for group in type_groups:
            if len(group) == 1:
                # Always keep isolated events
                combined_events.append(group[0])
            else:
                for index, event in enumerate(group):
                    if (index - offset) % pieces_per_merge == 0:
                        combined_events.append(event)

        self[:] = combined_events

    def group_by_type(self, select_types: List[str] = None) -> "EventGroupList":
        """
        Groups events by type

        Attributes
        ----------
        select_types
            A list of types for which to select groups in the resulting EventGroupList.
            If no types are specified, all resulting groups will be selected.

        Returns
        -------
        An EventGroupList partitioned by type
        """
        if select_types is None:
            select_types = []

        groups = [
            EventList(list(group), end=self.end)
            for index, group in groupby(self, key=attrgetter("__class__"))
        ]
        if not select_types:
            selected_groups = groups
        else:
            selected_groups = [group for group in groups if group.type in select_types]

        return EventGroupList(groups, selected=selected_groups)

    def group_by_slices(self, slices: (int, int)) -> "EventGroupList":
        """
        Groups events by slices.
        Does not support negative indexing.

        Slices explicitly passed in will become selected groups in the resulting EventGroupList.

        Returns
        -------
        An EventGroupList partitioned by slice
        """
        slices = [slice(sl[0], sl[1]) for sl in slices]

        # Fill in rest of slices
        all_slices = general.fill_slices(slices, len(self))
        target_indexes = [index for index, sl in enumerate(all_slices) if sl in slices]

        # Group events by slices
        groups = [self[sl] for sl in all_slices]
        selected_groups = [
            group for index, group in enumerate(groups) if index in target_indexes
        ]
        groups = EventGroupList(groups, selected=selected_groups)

        return groups class Persistable:
    """
    Mixin for persisting objects
    """

    @use_temporary_file_fallback("output_path", PICKLE_EXTENSION)
    def save(self, output_path: Optional[str] = None):
        with open(output_path, "wb") as output_file:
            dill.dump(self, output_file)

        return output_path

    @classmethod
    def load(cls, input_path: str):
        with open(input_path, "rb") as input_file:
            persistable = dill.load(input_file)

        return persistable def intervals_from_locations(locations: List[float]) -> List[float]:
def locations_from_intervals(intervals: List[float]) -> List[float]:
def start_end_locations_from_locations(
    locations: List[float],
) -> Tuple[List[float], List[float]]:
def start_end_locations_from_intervals(
    intervals: List[float],
) -> Tuple[List[float], List[float]]: TEMP_PATH_BASE = tempfile.TemporaryDirectory().name
def touch(filename):
def ensure_directory_exists(*directories):
def recreate_directory(*directories):
def list_directory(path, include_hidden=False):
def list_directory_files(directory: str, include_hidden=False) -> List[str]:
def run_command(command) -> CompletedProcess:
def generate_temp_file_path(extension: str) -> str:
def use_temporary_file_fallback(path_var: str, extension: str):
    def _use_temporary_file_path(path_variable): def use_temporary_file_fallback(path_var: str, extension: str):
    """
    Decorator to set path_var to a temporary file path if it is None. Does not create the file.

    Parameters
    ----------
    path_var
        A variable expecting a file path

    extension
        extension for the temporary file
    """

    def _use_temporary_file_path(path_variable):
        return path_variable or generate_temp_file_path(extension)

    return preprocess_args(_use_temporary_file_path, [path_var]) def ensure_json_serializable(*dictionaries: dict):
    """
    Decorator ensures dictionaries are json serializable
    """

    def _ensure_json_serializable(dictionary):
        try:
            json.dumps(dictionary)
        except TypeError as error:
            print(f"{dictionary} is not json serializable. Error: {error}")
            raise

        return dictionary

    return preprocess_args(_ensure_json_serializable, *dictionaries) class Cut(VideoEvent):
    """
    A cut from one piece of sources to another in a video
    """

    def __init__(self, location: TIME_FORMAT, *args, **kwargs):
        super().__init__(location, *args, **kwargs) SUBTITLES_EXTENSION = ".srt"
class Subtitle(NamedTuple):
class SubtitleTrack:
class AudioTrack:
    def __init__(self, subtitles: List[Subtitle], name: str, *, default: bool = False):
    def create(
        cls,
        texts: List[Any],
        name: str,
        *,
        locations: Optional[List[float]] = None,
        durations: Optional[List[float]] = None,
        default: bool = False,
    ) -> "SubtitleTrack":
    def write_to_file(self, output_path: Optional[str] = None):
    def __init__(self, audio_file: str, name: str):
def add_tracks_to_video(
    video_file: str,
    output_path: str,
    *,
    subtitle_tracks: Optional[List[SubtitleTrack]] = None,
    audio_tracks: Optional[List[AudioTrack]] = None,
): class SubtitleTrack:
    name: str
    subtitles: List[Subtitle]
    default: bool

    def __init__(self, subtitles: List[Subtitle], name: str, *, default: bool = False):
        self.subtitles = subtitles
        self.name = name
        self.default = default

    @classmethod
    def create(
        cls,
        texts: List[Any],
        name: str,
        *,
        locations: Optional[List[float]] = None,
        durations: Optional[List[float]] = None,
        default: bool = False,
    ) -> "SubtitleTrack":
        subtitles = []
        if locations:
            start_times, end_times = location.start_end_locations_from_locations(
                locations
            )
        elif durations:
            start_times, end_times = location.start_end_locations_from_intervals(
                durations
            )
        else:
            raise ParameterError(
                f"Must provide either locations or durations for the subtitle track {name}"
            )

        for index, (text, start_time, end_time) in enumerate(
            zip(texts, start_times, end_times)
        ):
            subtitle = Subtitle(str(text), start_time, end_time)
            subtitles.append(subtitle)

        subtitle_track = cls(subtitles, name, default=default)

        return subtitle_track

    @use_temporary_file_fallback("output_path", SUBTITLES_EXTENSION)
    def write_to_file(self, output_path: Optional[str] = None):
        system.touch(output_path)
        sub_rip_file = pysrt.open(output_path, encoding="utf-8")

        for index, sub in enumerate(self.subtitles):
            start_time = pysrt.SubRipTime.from_ordinal(sub.start_time * 1000)
            end_time = pysrt.SubRipTime.from_ordinal(sub.end_time * 1000)
            next_sub = pysrt.SubRipItem(
                index=index, text=sub.text, start=start_time, end=end_time
            )
            sub_rip_file.append(next_sub)

        sub_rip_file.save(output_path, encoding="utf-8")

        return output_path class VideoWriter:
    """
    Class for writing VideoClips and VideoSegments to file

    Parameters
    ----------
    preset
        Sets the time that FFMPEG will spend optimizing compression while writing the video to file.
        Note that this does not impact the quality of the video, only the size of the video file.
        So choose ultrafast when you are in a hurry and file size does not matter.
        Choices are: ultrafast, superfast, veryfast, faster, fast, medium, slow, slower, veryslow, placebo.

    codec
        Video codec to use when writing the music video to file.

    crf
        Constant rate factor (quality) for the music video (0 - 51).

    audio_codec
        Audio codec to use if no audio_file is provided.

    audio_bitrate
        Audio bitrate (kbps) to use if no audio_file is provided.

    ffmpeg_params
        Any additional ffmpeg parameters you would like to pass as a list of terms,
        like ['-option1', 'value1', '-option2', 'value2']
    """

    preset: str
    codec: str
    crf: int
    audio_codec: str
    audio_bitrate: int
    ffmpeg_params: list

    AUDIO_BITRATE = 320
    AUDIO_CODEC = "libmp3lame"
    VIDEO_PRESET = "medium"
    VIDEO_CODEC = "libx264"
    VIDEO_CRF = 18
    VIDEO_EXTENSION = ".mkv"

    def __init__(
        self,
        preset: str = VIDEO_PRESET,
        codec: str = VIDEO_CODEC,
        crf: int = VIDEO_CRF,
        audio_codec: str = AUDIO_CODEC,
        audio_bitrate: int = AUDIO_BITRATE,
        ffmpeg_params: Optional[List[str]] = None,
    ):
        self.preset = preset
        self.codec = codec
        self.crf = crf
        self.audio_codec = audio_codec
        self.audio_bitrate = audio_bitrate
        self.ffmpeg_params = ffmpeg_params or []

    def write_video_clips_to_directory(
        self,
        video_clips: List[VideoClip],
        directory: str,
        *,
        file_extension: str = VIDEO_EXTENSION,
        audio: Union[str, bool] = True,
        **kwargs
    ):
        """
        Writes a list of video segments to files in the specified directory
        """
        for index, segment in enumerate(tqdm(video_clips)):
            output_path = os.path.join(directory, str(index) + file_extension)
            self.write_video_clip_to_file(
                segment,
                output_path,
                audio=audio,
                verbose=False,
                show_progress=False,
                **kwargs
            )

    @use_temporary_file_fallback("output_path", VIDEO_EXTENSION)
    def write_video_clip_to_file(
        self,
        video_clip: VideoClip,
        output_path: Optional[str] = None,
        *,
        audio: Union[str, bool] = True,
        verbose: bool = False,
        show_progress: bool = True,
        **kwargs
    ):
        """
        Writes a video clip to file in the specified directory

        Parameters
        ----------
        video_clip

        output_path

        audio
            Audio for the video clip. Can be True to enable, False to disable, or an external audio file.

        verbose
            Whether output to stdout should include extra information during writing

        show_progress
            Whether to output progress information to stdout

        kwargs
            List of other keyword arguments to pass to moviepy's write_videofile
        """
        # Prepend crf to ffmpeg_params
        ffmpeg_params = ["-crf", str(self.crf)] + self.ffmpeg_params
        audio_bitrate = str(self.audio_bitrate) + "k"

        video_clip.write_videofile(
            output_path,
            audio=audio,
            preset=self.preset,
            codec=self.codec,
            audio_codec=self.audio_codec,
            audio_bitrate=audio_bitrate,
            ffmpeg_params=ffmpeg_params,
            **kwargs,
            verbose=verbose,
            logger=logger if show_progress else None
        )

        return output_path class CompositeVideoClip(moviepy.CompositeVideoClip):
    """
    A wrapper around moviepy's CompositeVideoClip, fixing issues with fps calculation
    """

    def __init__(self, clips, *args, **kwargs):
        super().__init__(clips, *args, **kwargs)

        fpss = [c.fps for c in clips if hasattr(c, "fps") and c.fps is not None]
        if len(fpss) == 0:
            self.fps = None
        else:
            self.fps = max(fpss) class Segment(Filterable, Persistable, ABC):
    """
    A segment of content in a video.
    Simulates a wrapper for moviepy's VideoClip class.

    Attributes
    ----------
    effects
        A list of effects to apply to the segment when composed
    """

    effects: VideoEffectList

    DEFAULT_VIDEO_FPS = 24

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.effects = VideoEffectList()

    def __repr__(self):
        return f"<{self.__class__.__name__}: {self.name}>, duration: {self.duration}>"

    def __copy__(self):
        """
        Override copy to avoid causing conflicts with custom pickling
        """
        cls = self.__class__
        new_segment = cls.__new__(cls)
        new_segment.__dict__.update(self.__dict__)

        return new_segment

    def __deepcopy__(self, memo):
        return self.copy()

    def copy(self) -> "Segment":
        new_segment = super().copy()

        # Deepcopy effects
        new_segment.effects = copy.deepcopy(self.effects)

        return new_segment

    def ipython_display(self, *args, **kwargs):
        """
        Fixes inheritance naming issue with moviepy's ipython_display
        """
        seg_copy = self.copy()
        # Class should also always be set to VideoClip for expected video display
        seg_copy.__class__ = VideoClip().__class__
        return seg_copy.ipython_display(*args, **kwargs)

    @property
    def dimensions(self) -> Dimensions:
        return Dimensions(self.w, self.h)

    @property
    def aspect_ratio(self) -> float:
        return self.w / self.h

    @property
    def resolution(self) -> int:
        return self.w * self.h

    @property
    def duration_time_code(self) -> str:
        return conversion.seconds_to_time_code(self.duration)

    @property
    def first_frame(self) -> LIST_3D:
        return self.get_frame(t=0)

    @property
    def middle_frame(self) -> LIST_3D:
        return self.get_frame(t=self.duration / 2)

    @property
    def last_frame(self) -> LIST_3D:
        return self.get_frame(t=self.duration)

    @property
    def first_last_frames(self) -> List[LIST_3D]:
        return [self.first_frame, self.last_frame]

    @property
    def first_middle_last_frames(self) -> List[LIST_3D]:
        return [self.first_frame, self.middle_frame, self.last_frame]

    def crop_to_aspect_ratio(self, aspect_ratio: float) -> "Segment":
        """
        Returns
        -------
        A new Segment, cropped as necessary to reach specified aspect ratio
        """
        segment = self.copy()

        if segment.aspect_ratio != aspect_ratio:
            # Crop video to match desired aspect ratio
            x1, y1, x2, y2 = video_sizing.crop_coordinates_for_aspect_ratio(
                segment.dimensions, aspect_ratio
            )
            segment = segment.crop(x1=x1, y1=y1, x2=x2, y2=y2)

        return segment

    def crop_scale(self, dimensions: Tuple[int, int]) -> "Segment":
        """
        Returns
        -------
        A new Segment, cropped and/or scaled as necessary to reach specified dimensions
        """
        segment = self.copy()
        dimensions = Dimensions(*dimensions)

        if segment.aspect_ratio != dimensions.aspect_ratio:
            # Crop segment to match aspect ratio
            segment = segment.crop_to_aspect_ratio(dimensions.aspect_ratio)

        if segment.dimensions != dimensions:
            # Resize segment to reach final dimensions
            segment = segment.resize(dimensions)

        return segment

    def apply_effects(self) -> "Segment":
        """
        Composes the segment, applying all effects

        Returns
        -------
        A new segment with all effects applied
        """
        segment = self.copy()

        for effect in self.effects:
            if isinstance(effect, video_effects.FadeIn):
                segment = segment.fadein(effect.duration, effect.rgb_color)
                if segment.audio:
                    segment.audio = segment.audio.audio_fadein(effect.duration)
            elif isinstance(effect, video_effects.FadeOut):
                segment = segment.fadeout(effect.duration, effect.rgb_color)
                if segment.audio:
                    segment.audio = segment.audio.audio_fadeout(effect.duration)

        return segment

    @property
    @abstractmethod
    def name(self) -> str:
        """
        Human-readable name for the segment
        """
        pass

    @abstractmethod
    def trailing_buffer(self, duration) -> "Segment":
        """
        Parameters
        ----------
        duration
            duration of the buffer

        Returns
        -------
        A new segment picking up where this one left off, for use in crossfades
        """
        pass class Dimensions(DimensionsBase):
    __slots__ = ()

    @property
    def aspect_ratio(self) -> float:
        return self.width / self.height

    @property
    def resolution(self) -> int:
        return self.width * self.height
import operator
import os
import mugen.video.effects as video_effects
import mugen.video.sizing as video_sizing
from functools import wraps
from typing import List, Optional, Union
from moviepy.editor import AudioFileClip, VideoClip
from mugen.events import EventList
from mugen.mixins.Persistable import Persistable
from mugen.utilities import location, system
from mugen.utilities.system import use_temporary_file_fallback
from mugen.utilities.validation import ensure_json_serializable
from mugen.video.cuts import Cut
from mugen.video.io import tracks
from mugen.video.io.tracks import SubtitleTrack
from mugen.video.io.VideoWriter import VideoWriter
from mugen.video.moviepy.CompositeVideoClip import CompositeVideoClip
from mugen.video.segments.Segment import Segment
from mugen.video.sizing import Dimensions



SEGMENTS_DIRECTORY = "video_segments"
REJECTED_SEGMENTS_DIRECTORY = "rejected_video_segments"


def requires_video_segments(func):
    """
    Decorator raises Error if there are no video segments
    """

    @wraps(func)
    def _requires_video_segments(self, *args, **kwargs):
        if not self.segments:
            raise ValueError(
                f"MusicVideo's {func} method requires one or more segments."
            )

        return func(self, *args, **kwargs)

    return _requires_video_segments



class MusicVideo(Persistable):