1 3
from mopidy import httpclient
2 3
from mopidy.internal.gi import Gst
3

4

5 3
def calculate_duration(num_samples, sample_rate):
6
    """Determine duration of samples using GStreamer helper for precise
7
    math."""
8 0
    return Gst.util_uint64_scale(num_samples, Gst.SECOND, sample_rate)
9

10

11 3
def create_buffer(data, timestamp=None, duration=None):
12
    """Create a new GStreamer buffer based on provided data.
13

14
    Mainly intended to keep gst imports out of non-audio modules.
15

16
    .. versionchanged:: 2.0
17
        ``capabilites`` argument was removed.
18
    """
19 3
    if not data:
20 3
        raise ValueError("Cannot create buffer without data")
21 3
    buffer_ = Gst.Buffer.new_wrapped(data)
22 3
    if timestamp is not None:
23 3
        buffer_.pts = timestamp
24 3
    if duration is not None:
25 3
        buffer_.duration = duration
26 3
    return buffer_
27

28

29 3
def millisecond_to_clocktime(value):
30
    """Convert a millisecond time to internal GStreamer time."""
31 3
    return value * Gst.MSECOND
32

33

34 3
def clocktime_to_millisecond(value):
35
    """Convert an internal GStreamer time to millisecond time."""
36 0
    return value // Gst.MSECOND
37

38

39 3
def supported_uri_schemes(uri_schemes):
40
    """Determine which URIs we can actually support from provided whitelist.
41

42
    :param uri_schemes: list/set of URIs to check support for.
43
    :type uri_schemes: list or set or URI schemes as strings.
44
    :rtype: set of URI schemes we can support via this GStreamer install.
45
    """
46 3
    supported_schemes = set()
47 3
    registry = Gst.Registry.get()
48

49 3
    for factory in registry.get_feature_list(Gst.ElementFactory):
50 3
        for uri in factory.get_uri_protocols():
51 3
            if uri in uri_schemes:
52 3
                supported_schemes.add(uri)
53

54 3
    return supported_schemes
55

56

57 3
def setup_proxy(element, config):
58
    """Configure a GStreamer element with proxy settings.
59

60
    :param element: element to setup proxy in.
61
    :type element: :class:`Gst.GstElement`
62
    :param config: proxy settings to use.
63
    :type config: :class:`dict`
64
    """
65 3
    if not hasattr(element.props, "proxy") or not config.get("hostname"):
66 3
        return
67

68 0
    element.set_property("proxy", httpclient.format_proxy(config, auth=False))
69 0
    element.set_property("proxy-id", config.get("username"))
70 0
    element.set_property("proxy-pw", config.get("password"))
71

72

73 3
class Signals:
74

75
    """Helper for tracking gobject signal registrations"""
76

77 3
    def __init__(self):
78 3
        self._ids = {}
79

80 3
    def connect(self, element, event, func, *args):
81
        """Connect a function + args to signal event on an element.
82

83
        Each event may only be handled by one callback in this implementation.
84
        """
85 3
        if (element, event) in self._ids:
86 0
            raise AssertionError
87 3
        self._ids[(element, event)] = element.connect(event, func, *args)
88

89 3
    def disconnect(self, element, event):
90
        """Disconnect whatever handler we have for an element+event pair.
91

92
        Does nothing it the handler has already been removed.
93
        """
94 3
        signal_id = self._ids.pop((element, event), None)
95 3
        if signal_id is not None:
96 3
            element.disconnect(signal_id)
97

98 3
    def clear(self):
99
        """Clear all registered signal handlers."""
100 3
        for element, event in list(self._ids):
101 3
            element.disconnect(self._ids.pop((element, event)))

Read our documentation on viewing source code .

Loading