1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22 from flumotion.worker.checks import check
23
24 import gst
25
26 from video010 import *
27
29 """
30 Probe the given device node as a TV card.
31 Return a deferred firing a human-readable device name, a list of channel
32 names (Tuner/Composite/...), and a list of norms (PAL/NTSC/SECAM/...).
33
34 @rtype: L{twisted.internet.defer.Deferred}
35 """
36 result = messages.Result()
37
38 def get_name_channels_norms(element):
39 deviceName = element.get_property('device-name')
40 channels = [channel.label for channel in element.list_channels()]
41 norms = [norm.label for norm in element.list_norms()]
42 return (deviceName, channels, norms)
43
44 pipeline = 'v4lsrc name=source device=%s ! fakesink' % device
45 d = do_element_check(pipeline, 'source', get_name_channels_norms)
46
47 d.addCallback(check.callbackResult, result)
48 d.addErrback(check.errbackNotFoundResult, result, id, device)
49 d.addErrback(check.errbackResult, result, id, device)
50
51 return d
52
54 """
55 Probe the given device node as a webcam.
56
57 The result is either:
58 - succesful, with a None value: no device found
59 - succesful, with a tuple:
60 - device name
61 - dict of mime, format, width, height, fps pair
62 - failed
63
64 @rtype: L{flumotion.common.messages.Result}
65 """
66 result = messages.Result()
67
68
69
70 def get_device_name(element):
71 name = element.get_property('device-name')
72 caps = element.get_pad("src").get_negotiated_caps()
73 log.debug('check', 'negotiated caps: %s' % caps.to_string())
74 s = caps[0]
75 num = s['framerate'].num
76 denom = s['framerate'].denom
77
78 d = {
79 'mime': s.get_name(),
80 'width': s['width'],
81 'height': s['height'],
82 'framerate': (num, denom),
83 }
84
85 if s.get_name() == 'video/x-raw-yuv':
86 d['format'] = s['format'].fourcc
87 return (name, d)
88
89
90
91
92
93
94
95 autoprobe = "autoprobe-fps=false"
96
97 autoprobe = ""
98
99 pipeline = 'v4lsrc name=source device=%s %s ! fakesink' % (device,
100 autoprobe)
101 d = do_element_check(pipeline, 'source', get_device_name,
102 state=gst.STATE_PAUSED, set_state_deferred=True)
103
104 d.addCallback(check.callbackResult, result)
105 d.addErrback(check.errbackNotFoundResult, result, id, device)
106 d.addErrback(check.errbackResult, result, id, device)
107
108 return d
109
110
111
113 """
114 Probe the given GStreamer element factory with the given device for
115 audio mixer tracks.
116 Return a deferred firing a result.
117
118 The result is either:
119 - succesful, with a None value: no device found
120 - succesful, with a human-readable device name and a list of mixer
121 track labels.
122 - failed
123
124 @rtype: L{twisted.internet.defer.Deferred}
125 """
126 result = messages.Result()
127
128 def get_tracks(element):
129
130 if not element.implements_interface(gst.interfaces.Mixer):
131 msg = 'Cannot get mixer tracks from the device. '\
132 'Check permissions on the mixer device.'
133 log.debug('checks', "returning failure: %s" % msg)
134 raise check.CheckProcError(msg)
135
136 return (element.get_property('device-name'),
137 [track.label for track in element.list_tracks()])
138
139 pipeline = '%s name=source device=%s ! audio/x-raw-int,channels=%d ! fakesink' % (source_factory, device, channels)
140 d = do_element_check(pipeline, 'source', get_tracks)
141
142 d.addCallback(check.callbackResult, result)
143 d.addErrback(check.errbackNotFoundResult, result, id, device)
144 d.addErrback(check.errbackResult, result, id, device)
145
146 return d
147