gstreamer/gi/overrides/Gst.py
Havard Graff af4ade3743 meson: use new python module
This patch makes the tests pass running uninstalled and installed, with
python2 and python3 on linux, windows and osx.

The main gist is to use the new python-module to do the lifting done
by pythondetector, and with that add support for python2 and windows.
2018-10-28 17:53:44 +01:00

639 lines
18 KiB
Python

# -*- Mode: Python; py-indent-offset: 4 -*-
# vim: tabstop=4 shiftwidth=4 expandtab
#
# Gst.py
#
# Copyright (C) 2012 Thibault Saunier <thibault.saunier@collabora.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., 51 Franklin Street, Fifth Floor,
# Boston, MA 02110-1301, USA.
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3, or (at your option)
# any later version.
import sys
import inspect
from ..overrides import override
from ..module import get_introspection_module
from gi.repository import GLib
if sys.version_info >= (3, 0):
_basestring = str
_callable = lambda c: hasattr(c, '__call__')
long = int
else:
_basestring = basestring
_callable = callable
Gst = get_introspection_module('Gst')
__all__ = []
if Gst._version == '0.10':
import warnings
warn_msg = "You have imported the Gst 0.10 module. Because Gst 0.10 \
was not designed for use with introspection some of the \
interfaces and API will fail. As such this is not supported \
by the GStreamer development team and we encourage you to \
port your app to Gst 1 or greater. gst-python is the recomended \
python module to use with Gst 0.10"
warnings.warn(warn_msg, RuntimeWarning)
class Bin(Gst.Bin):
def __init__(self, name=None):
Gst.Bin.__init__(self, name=name)
def add(self, *args):
for arg in args:
if not Gst.Bin.add(self, arg):
raise AddError(arg)
Bin = override(Bin)
__all__.append('Bin')
class Caps(Gst.Caps):
def __nonzero__(self):
return not self.is_empty()
def __new__(cls, *args):
if not args:
return Caps.new_empty()
elif len(args) > 1:
raise TypeError("wrong arguments when creating GstCaps object")
elif isinstance(args[0], str):
return Caps.from_string(args[0])
elif isinstance(args[0], Caps):
return args[0].copy()
elif isinstance(args[0], Structure):
res = Caps.new_empty()
res.append_structure(args[0])
return res
elif isinstance(args[0], (list, tuple)):
res = Caps.new_empty()
for e in args[0]:
res.append_structure(e)
return res
raise TypeError("wrong arguments when creating GstCaps object")
def __init__(self, *args, **kwargs):
return super(Caps, self).__init__()
def __str__(self):
return self.to_string()
def __getitem__(self, index):
if index >= self.get_size():
raise IndexError('structure index out of range')
return self.get_structure(index)
def __len__(self):
return self.get_size()
Caps = override(Caps)
__all__.append('Caps')
class Pad(Gst.Pad):
def __init__(self, *args, **kwargs):
self._real_chain_func = None
self._real_event_func = None
self._real_query_func = None
super(Gst.Pad, self).__init__(*args, **kwargs)
def _chain_override(self, pad, parent, buf):
return self._real_chain_func(pad, buf)
def _event_override(self, pad, parent, event):
return self._real_event_func(pad, event)
def _query_override(self, pad, parent, query):
query.mini_object.refcount -= 1
try:
res = self._real_query_func(pad, query)
except TypeError:
try:
res = self._real_query_func(pad, parent, query)
except TypeError:
raise TypeError("Invalid query method %s, 2 or 3 arguments required"
% self._real_query_func)
query.mini_object.refcount += 1
return res
def set_chain_function(self, func):
self._real_chain_func = func
self.set_chain_function_full(self._chain_override, None)
def set_event_function(self, func):
self._real_event_func = func
self.set_event_function_full(self._event_override, None)
def set_query_function(self, func):
self._real_query_func = func
self.set_query_function_full(self._query_override, None)
def set_query_function_full(self, func, udata):
self._real_query_func = func
self._real_set_query_function_full(self._query_override, None)
def query_caps(self, filter=None):
return Gst.Pad.query_caps(self, filter)
def link(self, pad):
ret = Gst.Pad.link(self, pad)
if ret != Gst.PadLinkReturn.OK:
raise LinkError(ret)
return ret
Pad._real_set_query_function_full = Gst.Pad.set_query_function_full
Pad = override(Pad)
__all__.append('Pad')
class GhostPad(Gst.GhostPad):
def __init__(self, name, target=None, direction=None):
if direction is None:
if target is None:
raise TypeError('you must pass at least one of target '
'and direction')
direction = target.props.direction
Gst.GhostPad.__init__(self, name=name, direction=direction)
self.construct()
if target is not None:
self.set_target(target)
def query_caps(self, filter=None):
return Gst.GhostPad.query_caps(self, filter)
GhostPad = override(GhostPad)
__all__.append('GhostPad')
class IteratorError(Exception):
pass
__all__.append('IteratorError')
class AddError(Exception):
pass
__all__.append('AddError')
class LinkError(Exception):
pass
__all__.append('LinkError')
class Iterator(Gst.Iterator):
def __iter__(self):
while True:
result, value = self.next()
if result == Gst.IteratorResult.DONE:
break
if result != Gst.IteratorResult.OK:
raise IteratorError(result)
yield value
Iterator = override(Iterator)
__all__.append('Iterator')
class ElementFactory(Gst.ElementFactory):
# ElementFactory
def get_longname(self):
return self.get_metadata("long-name")
def get_description(self):
return self.get_metadata("description")
def get_klass(self):
return self.get_metadata("klass")
@classmethod
def make(cls, factory_name, instance_name=None):
return Gst.ElementFactory.make(factory_name, instance_name)
class Pipeline(Gst.Pipeline):
def __init__(self, name=None):
Gst.Pipeline.__init__(self, name=name)
Pipeline = override(Pipeline)
__all__.append('Pipeline')
class Structure(Gst.Structure):
def __new__(cls, *args, **kwargs):
if not args:
if kwargs:
raise TypeError("wrong arguments when creating GstStructure, first argument"
" must be the structure name.")
return Structure.new_empty()
elif len(args) > 1:
raise TypeError("wrong arguments when creating GstStructure object")
elif isinstance(args[0], str):
if not kwargs:
return Structure.from_string(args[0])[0]
struct = Structure.new_empty(args[0])
for k, v in kwargs.items():
struct[k] = v
return struct
elif isinstance(args[0], Structure):
return args[0].copy()
raise TypeError("wrong arguments when creating GstStructure object")
def __init__(self, *args, **kwargs):
pass
def __getitem__(self, key):
return self.get_value(key)
def keys(self):
keys = set()
def foreach(fid, value, unused1, udata):
keys.add(GLib.quark_to_string(fid))
return True
self.foreach(foreach, None, None)
return keys
def __setitem__(self, key, value):
return self.set_value(key, value)
def __str__(self):
return self.to_string()
Structure = override(Structure)
__all__.append('Structure')
ElementFactory = override(ElementFactory)
__all__.append('ElementFactory')
class Fraction(Gst.Fraction):
def __init__(self, num, denom=1):
def __gcd(a, b):
while b != 0:
tmp = a
a = b
b = tmp % b
return abs(a)
def __simplify():
num = self.num
denom = self.denom
if num < 0:
num = -num
denom = -denom
# Compute greatest common divisor
gcd = __gcd(num, denom)
if gcd != 0:
num /= gcd
denom /= gcd
self.num = num
self.denom = denom
self.num = num
self.denom = denom
__simplify()
self.type = "fraction"
def __repr__(self):
return '<Gst.Fraction %s>' % (str(self))
def __value__(self):
return self.num / self.denom
def __eq__(self, other):
if isinstance(other, Fraction):
return self.num * other.denom == other.num * self.denom
return False
def __ne__(self, other):
return not self.__eq__(other)
def __mul__(self, other):
if isinstance(other, Fraction):
return Fraction(self.num * other.num,
self.denom * other.denom)
elif isinstance(other, int):
return Fraction(self.num * other, self.denom)
raise TypeError("%s is not supported, use Gst.Fraction or int." %
(type(other)))
__rmul__ = __mul__
def __truediv__(self, other):
if isinstance(other, Fraction):
return Fraction(self.num * other.denom,
self.denom * other.num)
elif isinstance(other, int):
return Fraction(self.num, self.denom * other)
return TypeError("%s is not supported, use Gst.Fraction or int." %
(type(other)))
__div__ = __truediv__
def __rtruediv__(self, other):
if isinstance(other, int):
return Fraction(self.denom * other, self.num)
return TypeError("%s is not an int." % (type(other)))
__rdiv__ = __rtruediv__
def __float__(self):
return float(self.num) / float(self.denom)
def __str__(self):
return '%d/%d' % (self.num, self.denom)
Fraction = override(Fraction)
__all__.append('Fraction')
class IntRange(Gst.IntRange):
def __init__(self, r):
if not isinstance(r, range):
raise TypeError("%s is not a range." % (type(r)))
if (r.start >= r.stop):
raise TypeError("Range start must be smaller then stop")
if r.start % r.step != 0:
raise TypeError("Range start must be a multiple of the step")
if r.stop % r.step != 0:
raise TypeError("Range stop must be a multiple of the step")
self.range = r
def __repr__(self):
return '<Gst.IntRange [%d,%d,%d]>' % (self.range.start,
self.range.stop, self.range.step)
def __str__(self):
if self.range.step == 1:
return '[%d,%d]' % (self.range.start, self.range.stop)
else:
return '[%d,%d,%d]' % (self.range.start, self.range.stop,
self.range.step)
def __eq__(self, other):
if isinstance(other, range):
return self.range == other
elif isinstance(other, IntRange):
return self.range == other.range
return False
if sys.version_info >= (3, 0):
IntRange = override(IntRange)
__all__.append('IntRange')
class Int64Range(Gst.Int64Range):
def __init__(self, r):
if not isinstance(r, range):
raise TypeError("%s is not a range." % (type(r)))
if (r.start >= r.stop):
raise TypeError("Range start must be smaller then stop")
if r.start % r.step != 0:
raise TypeError("Range start must be a multiple of the step")
if r.stop % r.step != 0:
raise TypeError("Range stop must be a multiple of the step")
self.range = r
def __repr__(self):
return '<Gst.Int64Range [%d,%d,%d]>' % (self.range.start,
self.range.stop, self.range.step)
def __str__(self):
if self.range.step == 1:
return '(int64)[%d,%d]' % (self.range.start, self.range.stop)
else:
return '(int64)[%d,%d,%d]' % (self.range.start, self.range.stop,
self.range.step)
def __eq__(self, other):
if isinstance(other, range):
return self.range == other
elif isinstance(other, IntRange):
return self.range == other.range
return False
class Bitmask(Gst.Bitmask):
def __init__(self, v):
if not isinstance(v, long) and not isinstance(v, int):
raise TypeError("%s is not an int or long." % (type(v)))
self.v = long(v)
def __str__(self):
return hex(self.v)
def __eq__(self, other):
return self.v == other
Bitmask = override(Bitmask)
__all__.append('Bitmask')
if sys.version_info >= (3, 0):
Int64Range = override(Int64Range)
__all__.append('Int64Range')
class DoubleRange(Gst.DoubleRange):
def __init__(self, start, stop):
self.start = float(start)
self.stop = float(stop)
if (start >= stop):
raise TypeError("Range start must be smaller then stop")
def __repr__(self):
return '<Gst.DoubleRange [%s,%s]>' % (str(self.start), str(self.stop))
def __str__(self):
return '(double)[%s,%s]' % (str(self.range.start), str(self.range.stop))
DoubleRange = override(DoubleRange)
__all__.append('DoubleRange')
class FractionRange(Gst.FractionRange):
def __init__(self, start, stop):
if not isinstance(start, Gst.Fraction):
raise TypeError("%s is not a Gst.Fraction." % (type(start)))
if not isinstance(stop, Gst.Fraction):
raise TypeError("%s is not a Gst.Fraction." % (type(stop)))
if (float(start) >= float(stop)):
raise TypeError("Range start must be smaller then stop")
self.start = start
self.stop = stop
def __repr__(self):
return '<Gst.FractionRange [%s,%s]>' % (str(self.start),
str(self.stop))
def __str__(self):
return '(fraction)[%s,%s]' % (str(self.start), str(self.stop))
FractionRange = override(FractionRange)
__all__.append('FractionRange')
class ValueArray(Gst.ValueArray):
def __init__(self, array):
self.array = list(array)
def __getitem__(self, index):
return self.array[index]
def __setitem__(self, index, value):
self.array[index] = value
def __len__(self):
return len(self.array)
def __str__(self):
return '<' + ','.join(map(str,self.array)) + '>'
def __repr__(self):
return '<Gst.ValueArray %s>' % (str(self))
ValueArray = override(ValueArray)
__all__.append('ValueArray')
class ValueList(Gst.ValueList):
def __init__(self, array):
self.array = list(array)
def __getitem__(self, index):
return self.array[index]
def __setitem__(self, index, value):
self.array[index] = value
def __len__(self):
return len(self.array)
def __str__(self):
return '{' + ','.join(map(str,self.array)) + '}'
def __repr__(self):
return '<Gst.ValueList %s>' % (str(self))
ValueList = override(ValueList)
__all__.append('ValueList')
def TIME_ARGS(time):
if time == Gst.CLOCK_TIME_NONE:
return "CLOCK_TIME_NONE"
return "%u:%02u:%02u.%09u" % (time / (Gst.SECOND * 60 * 60),
(time / (Gst.SECOND * 60)) % 60,
(time / Gst.SECOND) % 60,
time % Gst.SECOND)
__all__.append('TIME_ARGS')
import _gi_gst
_gi_gst
# maybe more python and less C some day if core turns a bit more introspection
# and binding friendly in the debug area
Gst.trace = _gi_gst.trace
Gst.log = _gi_gst.log
Gst.debug = _gi_gst.debug
Gst.info = _gi_gst.info
Gst.warning = _gi_gst.warning
Gst.error = _gi_gst.error
Gst.fixme = _gi_gst.fixme
Gst.memdump = _gi_gst.memdump
# Make sure PyGst is not usable if GStreamer has not been initialized
class NotInitialized(Exception):
pass
__all__.append('NotInitialized')
def fake_method(*args):
raise NotInitialized("Please call Gst.init(argv) before using GStreamer")
real_functions = [o for o in inspect.getmembers(Gst) if isinstance(o[1], type(Gst.init))]
class_methods = []
for cname_klass in [o for o in inspect.getmembers(Gst) if isinstance(o[1], type(Gst.Element)) or isinstance(o[1], type(Gst.Caps))]:
class_methods.append((cname_klass,
[(o, cname_klass[1].__dict__[o])
for o in cname_klass[1].__dict__
if isinstance(cname_klass[1].__dict__[o], type(Gst.init))]))
def init_pygst():
for fname, function in real_functions:
if fname not in ["init", "init_check", "deinit"]:
setattr(Gst, fname, function)
for cname_class, methods in class_methods:
for mname, method in methods:
setattr(cname_class[1], mname, method)
def deinit_pygst():
for fname, func in real_functions:
if fname not in ["init", "init_check", "deinit"]:
setattr(Gst, fname, fake_method)
for cname_class, methods in class_methods:
for mname, method in methods:
setattr(cname_class[1], mname, fake_method)
real_init = Gst.init
def init(argv):
init_pygst()
return real_init(argv)
Gst.init = init
real_init_check = Gst.init_check
def init_check(argv):
init_pygst()
return real_init_check(argv)
Gst.init_check = init_check
real_deinit = Gst.deinit
def deinit():
deinit_pygst()
return real_deinit()
Gst.deinit = deinit
if not Gst.is_initialized():
deinit_pygst()