first pass at wrapping GstProbe

Original commit message from CVS:
first pass at wrapping GstProbe
This commit is contained in:
Thomas Vander Stichele 2005-01-10 17:29:47 +00:00
parent 6782bfbc7b
commit b67833de2a
7 changed files with 164 additions and 3 deletions

View file

@ -1,3 +1,12 @@
2005-01-10 Thomas Vander Stichele <thomas at apestaart dot org>
* configure.ac:
* gst/arg-types.py:
* gst/gst-types.defs:
* gst/gst.override:
* testsuite/test_probe.py:
first pass at wrapping GstProbe
2005-01-01 Johan Dahlin <johan@gnome.org> 2005-01-01 Johan Dahlin <johan@gnome.org>
* gst/Makefile.am: * gst/Makefile.am:

View file

@ -26,7 +26,7 @@ dnl required versions of other packages
AC_SUBST(PYGTK_REQ, 2.4.0) AC_SUBST(PYGTK_REQ, 2.4.0)
AC_SUBST(GLIB_REQ, 2.0.0) AC_SUBST(GLIB_REQ, 2.0.0)
AC_SUBST(GTK_REQ, 2.0.0) AC_SUBST(GTK_REQ, 2.0.0)
AC_SUBST(GST_REQ, 0.8.0) AC_SUBST(GST_REQ, 0.8.8.1)
AC_DISABLE_STATIC AC_DISABLE_STATIC
AC_PROG_LIBTOOL AC_PROG_LIBTOOL

View file

@ -102,8 +102,7 @@ class XmlDocArg(XmlNodeArg):
"xwrap":"libxml_xmlDocPtrWrap"} "xwrap":"libxml_xmlDocPtrWrap"}
arg = GstDataPtrArg() matcher.register('GstData*', GstDataPtrArg())
matcher.register('GstData*', arg)
matcher.register('GstClockTime', UInt64Arg()) matcher.register('GstClockTime', UInt64Arg())
matcher.register('GstClockTimeDiff', Int64Arg()) matcher.register('GstClockTimeDiff', Int64Arg())
matcher.register('xmlNodePtr', XmlNodeArg()) matcher.register('xmlNodePtr', XmlNodeArg())

View file

@ -181,6 +181,12 @@
(gtype-id "GST_TYPE_CAPS") (gtype-id "GST_TYPE_CAPS")
) )
(define-boxed Probe
(in-module "Gst")
(c-name "GstProbe")
(gtype-id "GST_TYPE_PROBE")
)
; Defined in arg-types.py ; Defined in arg-types.py
;(define-boxed Data ;(define-boxed Data
; (in-module "Gst") ; (in-module "Gst")

View file

@ -34,6 +34,7 @@ headers
#include <gst/gsttypefind.h> #include <gst/gsttypefind.h>
#include <gst/gsttag.h> #include <gst/gsttag.h>
#include <gst/gsttaginterface.h> #include <gst/gsttaginterface.h>
#include <gst/gstprobe.h>
extern gboolean pygst_data_from_pyobject (PyObject *object, GstData **data); extern gboolean pygst_data_from_pyobject (PyObject *object, GstData **data);
PyObject *PyGstExc_LinkError = NULL; PyObject *PyGstExc_LinkError = NULL;
@ -108,6 +109,7 @@ ignore
gst_tag_list_copy_value gst_tag_list_copy_value
gst_trace_read_tsc gst_trace_read_tsc
%% %%
define GstTagList.keys noargs define GstTagList.keys noargs
void void
tag_foreach_func_dict (const GstTagList *list, tag_foreach_func_dict (const GstTagList *list,
@ -616,3 +618,114 @@ _wrap_gst_tag_setter_get_list(PyGObject *self)
/* pyg_boxed_new handles NULL checking */ /* pyg_boxed_new handles NULL checking */
return pyg_boxed_new(GST_TYPE_TAG_LIST, ret, TRUE, TRUE); return pyg_boxed_new(GST_TYPE_TAG_LIST, ret, TRUE, TRUE);
} }
%%
override gst_probe_new args
static gboolean
probe_handler_marshal(GstProbe *probe, GstData **data, gpointer user_data)
{
PyGILState_STATE state;
PyObject *callback, *args;
PyObject *ret;
gboolean res;
g_return_val_if_fail(user_data != NULL, FALSE);
state = pyg_gil_state_ensure();
callback = PyTuple_GetItem((PyObject *)user_data, 0);
args = Py_BuildValue("(NNO)",
pyg_boxed_new(GST_TYPE_PROBE, probe, TRUE, TRUE),
pyg_boxed_new(GST_TYPE_DATA, *data, TRUE, TRUE),
PyTuple_GetItem((PyObject *)user_data, 1));
ret = PyObject_CallObject(callback, args);
if (!ret) {
PyErr_Print();
res = FALSE;
} else {
res = PyObject_IsTrue(ret);
Py_DECREF(ret);
}
pyg_gil_state_release(state);
return res;
}
static int
_wrap_gst_probe_new(PyGBoxed *self, PyObject *args, PyObject *kwargs)
{
PyObject *first, *callback, *cbargs = NULL, *data;
gboolean single_shot;
gint len;
len = PyTuple_Size(args);
self->gtype = GST_TYPE_PROBE;
self->free_on_dealloc = FALSE;
if (len < 2) {
PyErr_SetString(PyExc_TypeError,
"Probe requires at least 2 args");
return -1;
}
first = PySequence_GetSlice(args, 0, 2);
if (!PyArg_ParseTuple(first, "iO:Probe", &single_shot, &callback)) {
Py_DECREF(first);
return -1;
}
Py_DECREF(first);
if (!PyCallable_Check(callback)) {
PyErr_SetString(PyExc_TypeError, "second argument not callable");
return -1;
}
cbargs = PySequence_GetSlice(args, 2, len);
if (cbargs == NULL)
return -1;
data = Py_BuildValue("(ON)", callback, cbargs);
if (data == NULL)
return -1;
self->boxed = gst_probe_new(single_shot, probe_handler_marshal, data);
return 0;
}
%%
override gst_probe_perform object
static PyObject *
_wrap_gst_probe_perform(PyGBoxed *self, PyObject *args, PyObject *kwargs)
{
PyObject *py_data;
GstData *data = NULL;
gint len;
len = PyTuple_Size(args);
if (len != 1) {
PyErr_SetString(PyExc_TypeError,
"perform requires 1 arg");
return NULL;
}
if (!PyArg_ParseTuple(args, "O:perform", &py_data)) {
return NULL;
}
/* FIXME: GstBuffer and GstEvent are not really "subclasses" so
we hardcode checks for them here by hand. Ugh. */
if (pyg_boxed_check(py_data, GST_TYPE_EVENT))
data = pyg_boxed_get(py_data, GstEvent);
if (pyg_boxed_check(py_data, GST_TYPE_BUFFER))
data = pyg_boxed_get(py_data, GstBuffer);
if (pyg_boxed_check(py_data, GST_TYPE_DATA))
data = pyg_boxed_get(py_data, GstData);
if (!data) {
PyErr_SetString(PyExc_TypeError, "arg 1 must be GstData");
return NULL;
}
return PyBool_FromLong(gst_probe_perform(self->boxed, &data));
}

View file

@ -25,6 +25,7 @@ tests = \
test_interface.py \ test_interface.py \
test_pad.py \ test_pad.py \
test_pipeline.py \ test_pipeline.py \
test_probe.py \
test_registry.py \ test_registry.py \
test_struct.py \ test_struct.py \
test_xml.py test_xml.py

33
testsuite/test_probe.py Normal file
View file

@ -0,0 +1,33 @@
# -*- Mode: Python; test-case-name: testsuite.test_probe -*-
# vi:si:et:sw=4:sts=4:ts=4
import sys
from common import gst, unittest
class ProbeTest(unittest.TestCase):
def testWrongNumber(self):
self.assertRaises(TypeError, gst.Probe, True)
def testWrongType(self):
# bool is int type
self.assertRaises(TypeError, gst.Probe, "noint", lambda x: "x")
# second arg should be callable
self.assertRaises(TypeError, gst.Probe, True, "nocallable")
def testPerformNoData(self):
probe = gst.Probe(True, self._probe_callback, "yeeha")
self.assertRaises(TypeError, probe.perform, None)
self.assertRaises(TypeError, probe.perform, "nodata")
def testPerform(self):
probe = gst.Probe(True, self._probe_callback, "yeeha")
buffer = gst.Buffer()
probe.perform(buffer)
self.assertEqual(self._probe_result, "yeeha")
def _probe_callback(self, probe, data, result):
self._probe_result = result
return True
if __name__ == "__main__":
unittest.main()