Code Duplication    Length = 56-56 lines in 2 locations

ffmpeg_streaming/ffprobe.py 1 location

@@ 21-76 (lines=56) @@
18
from ._media_property import Size, Bitrate
19
20
21
class Streams:
22
    def __init__(self, streams):
23
        self.streams = streams
24
25
    def video(self, ignore_error=True):
26
        """
27
        @TODO: add documentation
28
        """
29
        return self._get_stream('video', ignore_error)
30
31
    def audio(self, ignore_error=True):
32
        """
33
        @TODO: add documentation
34
        """
35
        return self._get_stream('audio', ignore_error)
36
37
    def first_stream(self):
38
        """
39
        @TODO: add documentation
40
        """
41
        return self.streams[0]
42
43
    def all(self):
44
        """
45
        @TODO: add documentation
46
        """
47
        return self.streams
48
49
    def videos(self):
50
        """
51
        @TODO: add documentation
52
        """
53
        return self._get_streams('video')
54
55
    def audios(self):
56
        """
57
        @TODO: add documentation
58
        """
59
        return self._get_streams('audio')
60
61
    def _get_stream(self, media, ignore_error):
62
        """
63
        @TODO: add documentation
64
        """
65
        media_attr = next((stream for stream in self.streams if stream['codec_type'] == media), None)
66
        if media_attr is None and not ignore_error:
67
            raise ValueError('No ' + str(media) + ' stream found')
68
        return media_attr if media_attr is not None else {}
69
70
    def _get_streams(self, media):
71
        """
72
        @TODO: add documentation
73
        """
74
        for stream in self.streams:
75
            if stream['codec_type'] == media:
76
                yield stream
77
78
79
class FFProbe:

ffmpeg_streaming/_media_streams.py 1 location

@@ 15-70 (lines=56) @@
12
"""
13
14
15
class Streams:
16
    def __init__(self, streams):
17
        self.streams = streams
18
19
    def video(self, ignore_error=True):
20
        """
21
        @TODO: add documentation
22
        """
23
        return self._get_stream('video', ignore_error)    
24
        
25
    def audio(self, ignore_error=True):
26
        """
27
        @TODO: add documentation
28
        """
29
        return self._get_stream('audio', ignore_error)
30
31
    def first_stream(self):
32
        """
33
        @TODO: add documentation
34
        """
35
        return self.streams[0]
36
37
    def all(self):
38
        """
39
        @TODO: add documentation
40
        """
41
        return self.streams
42
43
    def videos(self):
44
        """
45
        @TODO: add documentation
46
        """
47
        return self._get_streams('video')
48
49
    def audios(self):
50
        """
51
        @TODO: add documentation
52
        """
53
        return self._get_streams('audio')
54
55
    def _get_stream(self, media, ignore_error):
56
        """
57
        @TODO: add documentation
58
        """
59
        media_attr = next((stream for stream in self.streams if stream['codec_type'] == media), None)
60
        if media_attr is None and not ignore_error:
61
            raise ValueError('No ' + str(media) + ' stream found')
62
        return media_attr if media_attr is not None else {}
63
64
    def _get_streams(self, media):
65
        """
66
        @TODO: add documentation
67
        """
68
        for stream in self.streams:
69
            if stream['codec_type'] == media:
70
                yield stream
71