class Audio:
    """
    Wraps the audio ouput from librosa, providing access to extra features

    Attributes
    ----------
        file
            Loaded audio file

        samples
            Audio samples

        sample_rate
            Audio sample rate

        duration
            Audio duration (seconds)
    """

    file: str
    sample_rate: int
    samples: numpy.ndarray
    duration: float

    def __init__(self, file: str, *, sample_rate: int = 44100):
        """
        Parameters
        ----------
        file
            Audio file to load
        """

        self.file = file
        self.samples, self.sample_rate = librosa.load(file, sr=sample_rate)
        self.duration = librosa.get_duration(filename=self.file)

    def __repr__(self):
        filename = Path(self.file).stem
        return f"<Audio, file: {filename}, duration: {self.duration}>"

    def beats(self, trim: bool = False) -> EventList:
        """
        Gets beat events

        Parameters
        ----------
        trim
            Label weak leading and trailing beats separately

        Returns
        -------
        Detected beat events from the audio
        """
        untrimmed_beats = self._beats()
        untrimmed_beats = EventList(
            [Beat(beat) for beat in untrimmed_beats], end=self.duration
        )

        if not trim:
            beats = untrimmed_beats
        else:
            trimmed_beats = self._beats(trim=True)
            trimmed_leading_beats = [
                beat for beat in untrimmed_beats.locations if beat < trimmed_beats[0]
            ]
            trimmed_trailing_beats = [
                beat for beat in untrimmed_beats.locations if beat > trimmed_beats[-1]
            ]

            # Mark leading & trailing trimmed beats as weak beats
            trimmed_beats = EventList(
                [Beat(beat) for beat in trimmed_beats], end=self.duration
            )
            trimmed_leading_beats = EventList(
                [WeakBeat(beat) for beat in trimmed_leading_beats], end=self.duration
            )
            trimmed_trailing_beats = EventList(
                [WeakBeat(beat) for beat in trimmed_trailing_beats], end=self.duration
            )

            beats = trimmed_leading_beats + trimmed_beats + trimmed_trailing_beats

        return beats

    @lru_cache(maxsize=None)
    def _beats(self, trim: bool = False) -> List[float]:
        """
        Gets beat locations using librosa's beat tracker

        Parameters
        ----------
        trim
            Whether to discard weak beats

        Returns
        -------
        Beat locations
        """
        if trim:
            tempo, beats = librosa.beat.beat_track(
                y=self.samples, sr=self.sample_rate, units="time", trim=True
            )
        else:
            tempo, beats = librosa.beat.beat_track(
                y=self.samples, sr=self.sample_rate, units="time", trim=False
            )

        return beats

    def onsets(self, backtrack: bool = False) -> EventList:
        """
        Gets onset events

        Parameters
        ----------
        backtrack
            Shift onset events back to the nearest local minimum of energy

        Returns
        -------
        Detected onset events from the audio
        """
        if not backtrack:
            onsets = self._onsets()
        else:
            onsets = self._onsets(backtrack=True)

        onsets = EventList([Onset(onset) for onset in onsets], end=self.duration)

        return onsets

    @lru_cache(maxsize=None)
    def _onsets(self, backtrack: bool = False):
        """
        Gets onset locations using librosa's onset detector.

        Parameters
        ----------
        backtrack
            Whether to shift onset events back to the nearest local minimum of energy

        Returns
        -------
        Onset locations
        """
        if backtrack:
            onsets = librosa.beat.onset.onset_detect(
                y=self.samples, sr=self.sample_rate, units="time", backtrack=True
            )
        else:
            onsets = librosa.beat.onset.onset_detect(
                y=self.samples, sr=self.sample_rate, units="time", backtrack=False
            )

        return onsets TIME_FORMAT = Union[float, Tuple[int, float], Tuple[int, int, float], str] 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 MugenError(Exception):
    """
    The root mugen exception class
    """

    pass class ParameterError(MugenError):
    """
    Exception class for mal-formed inputs
    """

    pass class ContextFilter(Filter):
    """
    A Filter which keeps a running memory of objects to pass to the filter function

    Attributes
    ----------
    memory
        A list of objects to use as memory for the given filter

    Requires the filter function to take a 'memory' parameter
    """

    memory: List[Any] = None

    def __init__(self, function, memory: Optional[List[Any]] = None):
        super().__init__(function)

        if memory:
            self.memory = memory

    def __call__(self, *args, **kwargs) -> bool:
        return self.function(*args, memory=self.memory, **kwargs) class Filter:
    """
    A function used to filter an object based its sources

    Attributes
    ----------
    name
        Name of the filter function

    function
        The function to filter with
    """

    name: str
    function: Callable[..., bool]

    def __init__(self, function: Callable[..., bool]):
        self.name = function.__name__
        self.function = function

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

    def __call__(self, *args, **kwargs) -> bool:
        return self.function(*args, **kwargs) def convert_time_to_seconds(*variable_names: str):
    """
    Decorator to convert variables from TIME_FORMAT to seconds
    """
    return preprocess_args(time_to_seconds, *variable_names) 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]) DEFAULT_VIDEO_FILTERS = [
    VideoFilter.not_is_repeat.name,
    VideoFilter.not_has_low_contrast.name,
    VideoFilter.not_has_text.name,
    VideoFilter.not_has_cut.name,
] class VideoFilter(Enum):
    """
    Attributes
    ----------
    has_text
        video segment has detectable text (letters, words, numbers, etc...).
        Supports foreign langauges

    has_cut
        video segment has a detectable cut between shots

    has_low_contrast
        video segment has low contrast (solid color, dark scene, etc...)

    is_repeat
        video segment is a repeat of a video segment already used
    """

    # Content Filters
    has_text = Filter(has_text)
    has_cut = Filter(has_cut)
    has_low_contrast = Filter(has_low_contrast)

    not_has_text = Filter(not_has_text)
    not_has_cut = Filter(not_has_cut)
    not_has_low_contrast = Filter(not_has_low_contrast)

    # Context Filters
    is_repeat = ContextFilter(is_repeat)

    not_is_repeat = ContextFilter(not_is_repeat) 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 MusicVideo(Persistable):
    """
    A video composed of video segments and overlaid audio

    Attributes
    ----------
    writer
        Wrapper for writing VideoClips to video files

    meta
        Json serializable dictionary with extra metadata
    """

    audio_file: Optional[str]
    segments: List[Segment]
    writer: VideoWriter
    _dimensions: Optional[Dimensions]
    aspect_ratio: Optional[float]

    meta: dict

    @ensure_json_serializable("meta")
    def __init__(
        self,
        segments: List[Segment],
        audio_file: Optional[str] = None,
        *,
        dimensions: Optional[Dimensions] = None,
        aspect_ratio: Optional[float] = None,
        **kwargs,
    ):
        """
        Parameters
        ----------
        audio_file
            The audio file to use for the music video.
            If no audio file is provided, the original audio from the video segments will be used

        segments
            Video segments composing the music video

        dimensions
            Width and height for the music video.
            If no dimensions or aspect ratio are specified,
            the largest dimensions among the video segments will be used.

        aspect_ratio
            Aspect ratio for the music video (Overruled by dimensions)
        """
        super().__init__(**kwargs)

        # Required Parameters
        self.audio_file = audio_file
        self.segments = segments
        self._dimensions = dimensions
        self.aspect_ratio = aspect_ratio
        self.writer = VideoWriter()

        # Optional Parameters
        self.meta = {}

    """ PROPERTIES """

    @property
    def duration(self) -> int:
        return sum([segment.duration for segment in self.segments])

    @property
    def dimensions(self) -> Dimensions:
        return self._dimensions or self._calculate_dimensions()

    @dimensions.setter
    def dimensions(self, value):
        self._dimensions = value

    @property
    def cuts(self) -> EventList:
        durations = [segment.duration for segment in self.segments]
        locations = location.locations_from_intervals(durations)
        return EventList(
            [Cut(location) for location in locations[:-1]], end=locations[-1]
        )

    """ METHODS """

    @requires_video_segments
    def _calculate_dimensions(self) -> Dimensions:
        """
        Returns
        -------
        The largest dimensions available for the music video's aspect ratio
        """
        if self.aspect_ratio:
            dimensions = video_sizing.largest_dimensions_for_aspect_ratio(
                [segment.dimensions for segment in self.segments], self.aspect_ratio
            )
        else:
            dimensions = max(
                [segment.dimensions for segment in self.segments],
                key=operator.attrgetter("resolution"),
            )

        return dimensions

    @requires_video_segments
    def compose(self) -> VideoClip:
        """
        Composes the music video into a VideoClip
        """
        segments = [segment.crop_scale(self.dimensions) for segment in self.segments]
        segments = [segment.apply_effects() for segment in segments]

        # Add buffers for crossfaded video segments
        buffered_video_segments = []
        for index, segment in enumerate(segments):
            buffered_video_segments.append(segment)

            if index == len(segments) - 1:
                continue

            next_segment = segments[index + 1]

            for effect in next_segment.effects:
                if isinstance(effect, video_effects.CrossFade):
                    buffer = segment.trailing_buffer(effect.duration)
                    if buffer.audio:
                        buffer = buffer.set_audio(
                            buffer.audio.audio_fadeout(effect.duration)
                        )
                    buffered_video_segments.append(buffer)

        segments = buffered_video_segments

        # Build composite video
        composite_video_segments = [segments[0]]
        for index, segment in enumerate(segments[1:]):
            # Start current segment where previous segment ends in composite video
            previous_segment = composite_video_segments[index]
            segment = segment.set_start(previous_segment.end)

            # Apply any crossfade for the current segment
            for effect in segment.effects:
                if isinstance(effect, video_effects.CrossFade):
                    segment = segment.set_start(previous_segment.end - effect.duration)
                    segment = segment.crossfadein(effect.duration)
                    if segment.audio:
                        segment = segment.set_audio(
                            segment.audio.audio_fadein(effect.duration)
                        )

            composite_video_segments.append(segment)

        music_video = CompositeVideoClip(composite_video_segments)

        if self.audio_file:
            music_video.audio = AudioFileClip(self.audio_file)

        return music_video

    @requires_video_segments
    @use_temporary_file_fallback("output_path", VideoWriter.VIDEO_EXTENSION)
    def write_to_video_file(
        self,
        output_path: Optional[str] = None,
        *,
        audio: Optional[Union[bool, str]] = None,
        add_auxiliary_tracks: bool = True,
        verbose: bool = False,
        show_progress: bool = True,
        **kwargs,
    ):
        """
        writes the music video to a video file

        Parameters
        ----------
        output_path
            Path for the video file

        audio
            Audio for the music video. Can be True to enable, False to disable, an external audio file,
            or None to automatically set the value.

        add_auxiliary_tracks
            Whether or not helpful auxiliary subtitle tracks should be included.

        verbose
            Whether output to stdout should include extra information during writing

        show_progress
            Whether to output progress information to stdout

        Use this method over moviepy's write_videofile to preserve the audio file's codec and bitrate.
        """
        if audio is None:
            audio = self.audio_file or True
        composed_music_video = self.compose()

        if add_auxiliary_tracks:
            temp_output_path = self.writer.write_video_clip_to_file(
                composed_music_video,
                audio=audio,
                verbose=verbose,
                show_progress=show_progress,
                **kwargs,
            )
            # Add helpful subtitle/audio tracks to video file
            self._add_auxiliary_tracks(temp_output_path, output_path)
        else:
            self.writer.write_video_clip_to_file(
                composed_music_video,
                output_path,
                audio=audio,
                verbose=verbose,
                show_progress=show_progress,
                **kwargs,
            )

        return output_path

    def _add_auxiliary_tracks(self, video_file: str, output_path: str):
        """
        Adds metadata subtitle/audio tracks to the music video

        Parameters
        ----------
        video_file
            The temporary music video output file

        output_path
            The final music video output file with added auxiliary tracks
        """
        # Subtitle Tracks
        cuts = self.cuts
        numbers = [index for index, _ in enumerate(self.segments)]
        locations = [round(location, 2) for location in cuts.segment_locations]
        durations = cuts.segment_durations
        rounded_durations = [round(interval, 2) for interval in cuts.segment_durations]

        subtitle_track_segment_numbers = SubtitleTrack.create(
            numbers, "segment_numbers", durations=durations
        )
        subtitle_track_segment_locations = SubtitleTrack.create(
            locations, "segment_locations_seconds", durations=durations
        )
        subtitle_track_segment_durations = SubtitleTrack.create(
            rounded_durations, "segment_durations_seconds", durations=durations
        )

        subtitle_tracks = [
            subtitle_track_segment_numbers,
            subtitle_track_segment_locations,
            subtitle_track_segment_durations,
        ]
        tracks.add_tracks_to_video(
            video_file, output_path, subtitle_tracks=subtitle_tracks
        )

    @requires_video_segments
    def write_video_segments(self, directory: str):
        """
        Saves video_segments or video_segment_rejects to video files in the specified directory

        Parameters
        ----------
        directory
            location to save video segments
        """
        directory = os.path.join(directory, SEGMENTS_DIRECTORY)
        system.recreate_directory(directory)

        video_segments = [
            segment.crop_scale(self.dimensions) for segment in self.segments
        ]

        self.writer.write_video_clips_to_directory(video_segments, directory) class ColorSegment(Segment, ColorClip):
    """
    A segment with a color
    """

    color: str

    @convert_color_to_hex_code(["color"])
    def __init__(
        self,
        color: str,
        duration: float = 1,
        size: Optional[Tuple[int, int]] = (300, 300),
        **kwargs
    ):
        """
        Parameters
        ----------
        color
            hex code of the color, i.e. #000000 for black.
            OR the special inputs 'black' and 'white'.
        """
        super().__init__(
            size, conversion.hex_to_rgb(color), duration=duration, **kwargs
        )

        self.color = color
        self.fps = Segment.DEFAULT_VIDEO_FPS

    @property
    def name(self):
        return self.color

    def trailing_buffer(self, duration) -> "ColorSegment":
        return ColorSegment(self.color, duration, self.size) class VideoSegment(Segment, VideoFileClip):
    """
    A segment with video

    Attributes
    ----------
    source_start_time
        Start time of the video segment in the video file (seconds)
    """

    source_start_time: float
    _streams: List[dict]

    def __init__(self, file: str = None, **kwargs):
        """
        Parameters
        ----------
        file
            path to the video file.
            Supports any extension supported by ffmpeg, in addition to gifs.
        """
        super().__init__(file, **kwargs)

        self.source_start_time = 0
        if not self.fps:
            self.fps = Segment.DEFAULT_VIDEO_FPS
        self._streams = None

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

    def __getstate__(self):
        """
        Custom pickling
        """
        state = self.__dict__.copy()

        # Remove the video segment's audio and reader to allow pickling
        state["reader"] = None
        state["audio"] = None

        return state

    def __setstate__(self, newstate):
        """
        Custom unpickling
        """
        # Recreate the video segment's audio and reader
        newstate["reader"] = FFMPEG_VideoReader(newstate["filename"])
        newstate["audio"] = AudioFileClip(newstate["filename"]).subclip(
            newstate["source_start_time"],
            newstate["source_start_time"] + newstate["duration"],
        )
        self.__dict__.update(newstate)

    """ PROPERTIES """

    @property
    def file(self) -> str:
        return self.filename

    @property
    def name(self) -> str:
        return Path(self.file).stem

    @property
    def source_end_time(self) -> float:
        return self.source_start_time + self.duration

    @property
    def source_start_time_time_code(self) -> str:
        return conversion.seconds_to_time_code(self.source_start_time)

    @property
    def streams(self) -> List[dict]:
        if not self._streams:
            result = system.run_command(
                f"ffprobe -v quiet -print_format json -show_format -show_streams {self.file}".split()
            )
            self._streams = json.loads(result.stdout).get("streams", [])

        return self._streams

    @property
    def video_streams(self) -> List[dict]:
        return [stream for stream in self.streams if stream["codec_type"] == "video"]

    @property
    def audio_streams(self) -> List[dict]:
        return [stream for stream in self.streams if stream["codec_type"] == "audio"]

    @property
    def subtitle_streams(self) -> List[dict]:
        return [stream for stream in self.streams if stream["codec_type"] == "subtitle"]

    @property
    def video_stream(self) -> Optional[dict]:
        """Returns the primary video stream"""
        return self.video_streams[0] if len(self.video_streams) > 0 else None

    @property
    def audio_stream(self) -> Optional[dict]:
        """Returns the primary audio stream"""
        return self.audio_streams[0] if len(self.audio_streams) > 0 else None

    """ METHODS """

    @convert_time_to_seconds(["start_time", "end_time"])
    def subclip(
        self, start_time: TIME_FORMAT = 0, end_time: TIME_FORMAT = None
    ) -> "VideoSegment":
        """
        Parameters
        ----------
        start_time
            Start time of the video segment in the source video file

        end_time
            End time of the video segment in the source video file

        Returns
        -------
        A subclip of the original video segment, starting at 'start_time' and ending at 'end_time'
        """
        subclip = super().subclip(start_time, end_time)

        if start_time < 0:
            # Set relative to end
            start_time = self.duration + start_time

        subclip.source_start_time += start_time

        return subclip

    def trailing_buffer(self, duration) -> "VideoSegment":
        return VideoSegment(self.file).subclip(
            self.source_end_time, self.source_end_time + duration
        )

    def overlaps_segment(self, segment: "VideoSegment") -> bool:
        if not self.file == segment.file:
            return False

        return general.check_if_ranges_overlap(
            self.source_start_time,
            self.source_end_time,
            segment.source_start_time,
            segment.source_end_time,
        )

    def get_subtitle_stream_content(self, stream: int) -> str:
        """Returns the subtitle stream's content"""
        result = system.run_command(
            f"ffmpeg -v quiet -i {self.file} -map 0:s:{stream} -f srt pipe:1".split()
        )
        return result.stdout class SourceSampler:
    """
    A set of content sources for sampling video segments
    """

    sources: SourceList

    def __init__(self, sources: Union[SourceList, list]):
        """
        video_segments

        Parameters
        ----------
        sources
            An arbitrarily nested list of sources. Sources will be flattened internally.
            e.g. [S1, S2, [S3, S4]] -> [S1, S2, S3, S4]
        """
        if not isinstance(sources, SourceList):
            sources = SourceList(sources)

        self.sources = sources.flatten()

    def sample(self, duration: float) -> Segment:
        """
        Randomly samples a segment with the specified duration

        Parameters
        ----------
        duration
            duration of the sample

        Returns
        -------
        A randomly sampled segment with the specified duration
        """
        selected_source = choice(self.sources, p=self.sources.normalized_weights)
        sample = selected_source.sample(duration)

        return sample class VideoSource(Source):
    """
    A video source for sampling video segments
    """

    time_boundaries: List[Tuple[(TIME_FORMAT, TIME_FORMAT)]]

    def __init__(
        self,
        file: str,
        *,
        time_boundaries: Optional[List[Tuple[(TIME_FORMAT, TIME_FORMAT)]]] = None,
        **kwargs,
    ):
        """
        Parameters
        ----------
        file
            video file to sample from

        time_boundaries
            the set of time ranges to sample from in the video.
            For supported formats, see :data:`~mugen.constants.TIME_FORMAT`.
        """
        super().__init__(**kwargs)
        self.segment = VideoSegment(file)
        self.time_boundaries = time_boundaries if time_boundaries else []

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

    @property
    def file(self):
        return self.segment.file

    @property
    def name(self):
        return self.segment.name

    def sample(self, duration: float) -> VideoSegment:
        """
        Randomly samples a video segment with the specified duration.

        Parameters
        ----------
        duration
            duration of the video segment to sample
        """
        if self.time_boundaries:
            # Select a random time boundary to sample from, weighted by duration
            time_ranges = [TimeRange(*boundary) for boundary in self.time_boundaries]
            time_ranges = [
                time_range
                for time_range in time_ranges
                if time_range.duration >= duration
            ]
            total_duration = sum([time_range.duration for time_range in time_ranges])
            time_range_weights = [
                time_range.duration / total_duration for time_range in time_ranges
            ]
            time_range_to_sample = time_ranges[
                choice(len(time_ranges), p=time_range_weights)
            ]
        else:
            time_range_to_sample = TimeRange(0, self.segment.duration)

        start_time = random.uniform(
            time_range_to_sample.start, time_range_to_sample.end - duration
        )
        sampled_clip = self.segment.subclip(start_time, start_time + duration)

        return sampled_clip class VideoSourceList(SourceList):
    """
    A list of VideoSources
    """

    name: Optional[str]

    def __init__(
        self,
        sources=Optional[Union[List[Union[Source, "VideoSourceList"]], str]],
        **kwargs,
    ):
        """
        Parameters
        ----------
        sources
            A list of sources.
            Accepts arbitrarily nested video files, file globs, directories, VideoSources, and VideoSourceLists.
        """
        self.name = None

        if isinstance(sources, str):
            self.name = Path(sources).stem

            # Build list of sources from directory or file glob
            if os.path.isdir(sources):
                sources = self._sources_from_files(system.list_directory_files(sources))
            else:
                sources = self._sources_from_files(globber.glob(sources))
        else:
            # Convert any source files to VideoSources, and any lists, file globs or directories to VideoSourceLists
            sources = self._fill_in_sources(sources)

        super().__init__(sources, **kwargs)

    def list_repr(self):
        """
        Repr for use in lists
        """
        if self.name:
            return f"<{self.__class__.__name__} ({len(self)}): {self.name}, weight: {self.weight}>"

        return super().list_repr()

    @staticmethod
    def _sources_from_files(files: List[str]):
        sources = []

        for file in files:
            try:
                source = VideoSource(file)
            except IOError:
                continue

            sources.append(source)

        return sources

    @staticmethod
    def _fill_in_sources(sources: list):
        for index, source in enumerate(sources):
            if isinstance(source, str):
                try:
                    sources[index] = VideoSource(source)
                except IOError:
                    sources[index] = VideoSourceList(source)
            if isinstance(source, list) and not isinstance(source, VideoSourceList):
                sources[index] = VideoSourceList(source)

        return sources
import copy
import mugen.audio.utilities as audio_utilities
from enum import Enum
from typing import Any, List, Optional, Union
from tqdm import tqdm
from mugen.audio.Audio import Audio
from mugen.constants import TIME_FORMAT
from mugen.events import EventList
from mugen.exceptions import MugenError, ParameterError
from mugen.mixins.Filterable import ContextFilter, Filter
from mugen.utilities.conversion import convert_time_to_seconds
from mugen.utilities.system import use_temporary_file_fallback
from mugen.video.filters import DEFAULT_VIDEO_FILTERS, VideoFilter
from mugen.video.io import tracks
from mugen.video.io.tracks import SubtitleTrack
from mugen.video.MusicVideo import MusicVideo
from mugen.video.segments.ColorSegment import ColorSegment
from mugen.video.segments.VideoSegment import VideoSegment
from mugen.video.sources.SourceSampler import SourceSampler
from mugen.video.sources.VideoSource import VideoSource, VideoSourceList
        audio_file
            audio file to use for the music video

        video_sources
            Source videos to use for the music video.
            Accepts arbitrarily nested video files, directories, VideoSources, and VideoSourceLists.

        video_filters ~
            Video filters that each segment in the music video must pass.
            See :class:`~mugen.video.video_filters.VideoFilter` for a list of supported values.
            Defaults to :data:`~mugen.video.video_filters.VIDEO_FILTERS_DEFAULT`

        exclude_video_filters
            Video filters to exclude from default video_filters.
            Takes precedence over video_filters

        include_video_filters
            Video filters to use in addition to default video_filters.
            Takes precedence over exclude_video_filters

        custom_video_filters ~
            Custom video filters to use in addition to video_filters.
            Allows functions wrapped by :class:`~mugen.mixins.Filterable.Filter` or
            :class:`~mugen.mixins.Filterable.ContextFilter`
        """
        if not audio_file and not duration:
            raise ParameterError(
                "Must provide either the audio file or duration for the music video."
            )

        self.audio = Audio(audio_file) if audio_file else None
        self._duration = duration

        if video_sources:
            self.video_sources = VideoSourceList(video_sources)

        # Assemble list of video filter names
        video_filter_names = (
            video_filters if video_filters is not None else DEFAULT_VIDEO_FILTERS
        )
        if exclude_video_filters:
            for video_filter in exclude_video_filters:
                video_filter_names.remove(video_filter)
        if include_video_filters:
            video_filter_names.extend(include_video_filters)
        custom_video_filters = custom_video_filters or []

        # Compile video filters
        self.video_filters = []
        for filter_name in video_filter_names:
            try:
                self.video_filters.append(VideoFilter[filter_name].value)
            except KeyError as error:
                raise MugenError(f"Unknown video filter '{filter_name}'") from error
        self.video_filters.extend(custom_video_filters)

        self.meta = {self.Meta.REJECTED_SEGMENT_STATS: []}

    @property
    def duration(self):
        return self.audio.duration if self.audio else self._duration

    def generate_from_events(
        self, events: Union[EventList, List[TIME_FORMAT]], show_progress: bool = True
    ) -> MusicVideo:
        """
        Generates a MusicVideo from a list of events

        Parameters
        ----------
        events
            Events corresponding to cuts which occur in the music video.
            Either a list of events or event locations.

        show_progress
            Whether to output progress information to stdout
        """
        if not isinstance(events, EventList):
            events = EventList(events, end=self.duration)

        # Get segment durations from cut locations
        segment_durations = events.segment_durations

        music_video_segments = self._generate_music_video_segments(
            segment_durations, show_progress=show_progress
        )

        # Assemble music video from music video segments and audio
        music_video = MusicVideo(
            music_video_segments, self.audio.file if self.audio else None
        )

        return music_video

    def _generate_music_video_segments(
        self, durations: List[float], *, show_progress: bool = True
    ) -> List[VideoSegment]:
        """
        Generates a list of sampled video segments which pass all trait filters

        Parameters
        ----------
        durations
            durations for each sampled video segment

        show_progress
            Whether to output progress information to stdout

        Returns
        -------
        Sampled video segments
        """
        video_segments = []
        video_segment_sampler = SourceSampler(self.video_sources)

        # Make deep copies of filters
        video_filters = copy.deepcopy(self.video_filters)

        # Set memory for all ContextFilters
        for video_filter in video_filters:

            if isinstance(video_filter, ContextFilter) and video_filter.memory is None: