/usr/share/pyshared/insanity/generators/elements.py is in python-insanity 0.0+git20110920.4750a8e8-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 | # GStreamer QA system
#
# generators/elements.py
#
# Copyright (c) 2008, Edward Hervey <bilboed@bilboed.com>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program; if not, write to the
# Free Software Foundation, Inc., 59 Temple Place - Suite 330,
# Boston, MA 02111-1307, USA.
"""
GstElement-related generators
"""
import gst
from insanity.generator import Generator
#from insanity.log import critical, error, warning, debug, info
class ElementGenerator(Generator):
"""
Expands to a list of gst.ElementFactory names that
match the given list of Class.
If no list of Class is specified, then all available
gst.ElementFactory available on the system is returned.
"""
__args__ = {
"classes":"list of Class to filter against",
"factories":"If set to True, will return objects and not strings"
}
def _generate(self):
def list_compat(la, lb):
for item in la:
if not item in lb:
return False
return True
res = []
classes = self.kwargs.get("classes", [])
retfact = self.kwargs.get("factories", False)
allf = gst.registry_get_default().get_feature_list(gst.TYPE_ELEMENT_FACTORY)
# filter by class
for fact in allf:
if list_compat(classes, fact.get_klass().split('/')):
if retfact:
res.append(fact)
else:
res.append(fact.get_name())
return res
class MuxerGenerator(ElementGenerator):
"""
Expands to a list of muxers
"""
def __init__(self, *args, **kwargs):
kwargs["classes"] = ["Codec", "Muxer"]
ElementGenerator.__init__(self, *args, **kwargs)
class AudioEncoderGenerator(ElementGenerator):
"""
Expands to a list of audio encoder factories
"""
def __init__(self, *args, **kwargs):
kwargs["classes"] = ["Codec", "Encoder", "Audio"]
ElementGenerator.__init__(self, *args, **kwargs)
class VideoEncoderGenerator(ElementGenerator):
"""
Expands to a list of video encoder factories
"""
def __init__(self, *args, **kwargs):
kwargs["classes"] = ["Codec", "Encoder"]
ElementGenerator.__init__(self, *args, **kwargs)
def _generate(self):
res = []
facts = ElementGenerator._generate(self)
# filter those which have Video or Image
for fact in facts:
klasses = fact.get_klass().split('/')
if "Video" in klasses or "audio" in klasses:
res.append(fact)
return res
class EncoderMuxerGenerator(Generator):
"""
Expand to a list of all possible combinations of:
(
* audio encoder
* video encoder
* muxer
)
for a given audio encoder and/or video encoder and/or
muxer.
If one (or all) factory is not specified, then it will
use all available factories of that class on the given
system.
The contents of the tuple are gst.ElementFactory.
If the muxer can handle raw formats, the adequate encoder
field will contain 'identity'.
If the 'single_streams' argument is set to True, then the
returned list will also contain combinations with only one
encoder, in which case that field will contain 'None'.
"""
__args__ = {
"muxer":"Name of the Muxer to filter the results against",
"audio_encoder":"Name of the Audio Encoder to filter the results against",
"video_encoder":"Name of the Video Encoder to filter the results against",
"single_streams":"Also returns single-encoder combinations if True",
"factories":"Returns object and not names if set to True"
}
def _generate(self):
muxername = self.kwargs.get("muxer", None)
aencname = self.kwargs.get("audio_encoder", None)
vencname = self.kwargs.get("video_encoder", None)
singlestreams = self.kwargs.get("single_streams", False)
retfact = self.kwargs.get("factories", False)
muxer = muxername and gst.element_factory_find(muxername)
aenc = aencname and gst.element_factory_find(aencname)
venc = vencname and gst.element_factory_find(vencname)
if muxer:
allmuxers = [muxer]
else:
allmuxers = MuxerGenerator(factories=True).generate()
if aenc:
allaencs = [aenc]
else:
allaencs = AudioEncoderGenerator(factories=True).generate()
if venc:
allvencs = [venc]
else:
allvencs = VideoEncoderGenerator(factories=True).generate()
def can_sink_caps(muxer, ocaps):
pts = muxer.get_static_pad_templates()
sdir = gst.PAD_SINK
sinkcaps = [x.get_caps() for x in pts if x.direction == sdir and not x.get_caps().is_any()]
for caps in sinkcaps:
if not caps.intersect(ocaps).is_empty():
return True
return False
def encoders_muxer_compatible(encoders, muxer):
res = []
for encoder in encoders:
pts = encoder.get_static_pad_templates()
for caps in [x.get_caps() for x in pts if x.direction == gst.PAD_SRC]:
if can_sink_caps(muxer, caps):
res.append(encoder)
break
return res
res = []
# reduce allmuxers to those intersecting with the encoders
for mux in allmuxers:
# get the compatible encoders, without forgetting the
# raw pads
compatvenc = encoders_muxer_compatible(allvencs, mux)
compataenc = encoders_muxer_compatible(allaencs, mux)
# skip muxers than don't accept the specified encoders
if vencname and not gst.element_factory_find(vencname) in compatvenc:
continue
if aencname and not gst.element_factory_find(aencname) in compataenc:
continue
acaps = gst.Caps("audio/x-raw-int;audio/x-raw-float")
vcaps = gst.Caps("video/x-raw-rgb;video/x-raw-yuv")
if not aencname and can_sink_caps(mux, acaps):
compataenc.append(gst.element_factory_find("identity"))
if not vencname and can_sink_caps(mux, vcaps):
compatvenc.append(gst.element_factory_find("identity"))
# and now produce the tuples
for venc in compatvenc:
for aenc in compataenc:
if retfact:
res.append((aenc, venc, mux))
else:
res.append((aenc.get_name(),
venc.get_name(),
mux.get_name()))
if singlestreams:
if not aencname:
for venc in compatvenc:
if retfact:
res.append((None, venc, mux))
else:
res.append((None, venc.get_name(),
mux.get_name()))
if not vencname:
for aenc in compataenc:
if retfact:
res.append((aenc, None, mux))
else:
res.append((aenc.get_name(), None,
mux.get_name()))
return res
|