--- /dev/null
+/*
+ * This file is part of the libsigrok project.
+ *
+ * Copyright (C) 2014 Martin Ling <martin-sigrok@earth.li>
+ *
+ * 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 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+%module classes
+
+%{
+#include <pygobject.h>
+
+PyObject *GLib;
+PyTypeObject *IOChannel;
+PyTypeObject *PollFD;
+
+%}
+
+%init %{
+ pygobject_init(-1, -1, -1);
+ GLib = PyImport_ImportModule("gi.repository.GLib");
+ IOChannel = (PyTypeObject *) PyObject_GetAttrString(GLib, "IOChannel");
+ PollFD = (PyTypeObject *) PyObject_GetAttrString(GLib, "PollFD");
+%}
+
+/* Map file objects to file descriptors. */
+%typecheck(SWIG_TYPECHECK_POINTER) int fd {
+ $1 = (PyObject_AsFileDescriptor($input) != -1);
+}
+
+%typemap(in) int fd {
+ int fd = PyObject_AsFileDescriptor($input);
+ if (fd == -1)
+ SWIG_exception(SWIG_TypeError,
+ "Expected file object or integer file descriptor");
+ else
+ $1 = fd;
+}
+
+/* Map from Glib::Variant to native Python types. */
+%typemap(out) Glib::VariantBase {
+ GValue *value = g_new0(GValue, 1);
+ g_value_init(value, G_TYPE_VARIANT);
+ g_value_set_variant(value, $1.gobj());
+ PyObject *variant = pyg_value_as_pyobject(value, true);
+ $result = PyObject_CallMethod(variant,
+ const_cast<char *>("unpack"),
+ const_cast<char *>(""), NULL);
+ Py_XDECREF(variant);
+ g_free(value);
+}
+
+/* Map from Glib::IOCondition to GLib.IOCondition. */
+%typecheck(SWIG_TYPECHECK_POINTER) Glib::IOCondition {
+ gint flags;
+ $1 = pygobject_check($input, &PyGFlags_Type) &&
+ (pyg_flags_get_value(G_TYPE_IO_CONDITION, $input, &flags) != -1);
+}
+
+%typemap(in) Glib::IOCondition {
+ if (!pygobject_check($input, &PyGFlags_Type))
+ SWIG_exception(SWIG_TypeError, "Expected GLib.IOCondition value");
+ gint flags;
+ if (pyg_flags_get_value(G_TYPE_IO_CONDITION, $input, &flags) == -1)
+ SWIG_exception(SWIG_TypeError, "Not a valid Glib.IOCondition value");
+ $1 = (Glib::IOCondition) flags;
+}
+
+/* And back */
+%typemap(out) Glib::IOCondition {
+ GValue *value = g_new0(GValue, 1);
+ g_value_init(value, G_TYPE_IO_CONDITION);
+ g_value_set_flags(value, &$1);
+ $result = pyg_value_as_pyobject(value, true);
+ g_free(value);
+}
+
+/* Map from GLib.PollFD to Glib::PollFD *. */
+%typecheck(SWIG_TYPECHECK_POINTER) Glib::PollFD {
+ $1 = pygobject_check($input, PollFD);
+}
+
+%typemap(in) Glib::PollFD {
+ if (!pygobject_check($input, PollFD))
+ SWIG_exception(SWIG_TypeError, "Expected GLib.PollFD");
+ PyObject *fd_obj = PyObject_GetAttrString($input, "fd");
+ PyObject *events_obj = PyObject_GetAttrString($input, "events");
+ gint flags;
+ pyg_flags_get_value(G_TYPE_IO_CONDITION, events_obj, &flags);
+ int fd = PyInt_AsLong(fd_obj);
+ Glib::IOCondition events = (Glib::IOCondition) flags;
+ $1 = Glib::PollFD(fd, events);
+}
+
+/* Map from GLib.IOChannel to Glib::IOChannel *. */
+%typecheck(SWIG_TYPECHECK_POINTER) Glib::RefPtr<Glib::IOChannel> {
+ $1 = pygobject_check($input, IOChannel);
+}
+
+%typemap(in) Glib::RefPtr<Glib::IOChannel> {
+ if (!pygobject_check($input, IOChannel))
+ SWIG_exception(SWIG_TypeError, "Expected GLib.IOChannel");
+ $1 = Glib::wrap((GIOChannel *) PyObject_Hash($input), true);
+}
+
+/* Map from callable PyObject to SourceCallbackFunction. */
+%typecheck(SWIG_TYPECHECK_POINTER) sigrok::SourceCallbackFunction {
+ $1 = PyCallable_Check($input);
+}
+
+%typemap(in) sigrok::SourceCallbackFunction {
+ if (!PyCallable_Check($input))
+ SWIG_exception(SWIG_TypeError, "Expected a callable Python object");
+
+ $1 = [=] (Glib::IOCondition revents) {
+ auto gstate = PyGILState_Ensure();
+
+ GValue *value = g_new0(GValue, 1);
+ g_value_init(value, G_TYPE_IO_CONDITION);
+ g_value_set_flags(value, revents);
+ auto revents_obj = pyg_value_as_pyobject(value, true);
+ g_free(value);
+
+ auto arglist = Py_BuildValue("(O)", revents_obj);
+
+ auto result = PyEval_CallObject($input, arglist);
+
+ Py_XDECREF(arglist);
+ Py_XDECREF(revents_obj);
+
+ if (PyErr_Occurred() || !PyBool_Check(result))
+ throw sigrok::Error(SR_ERR);
+
+ bool retval = (result == Py_True);
+
+ Py_XDECREF(result);
+
+ PyGILState_Release(gstate);
+
+ return retval;
+ };
+
+ Py_XINCREF($input);
+}
+
+/* Map from callable PyObject to LogCallbackFunction */
+%typecheck(SWIG_TYPECHECK_POINTER) sigrok::LogCallbackFunction {
+ $1 = PyCallable_Check($input);
+}
+
+%typemap(in) sigrok::LogCallbackFunction {
+ if (!PyCallable_Check($input))
+ SWIG_exception(SWIG_TypeError, "Expected a callable Python object");
+
+ $1 = [=] (const sigrok::LogLevel *loglevel, string message) {
+ auto gstate = PyGILState_Ensure();
+
+ auto log_obj = SWIG_NewPointerObj(
+ SWIG_as_voidptr(loglevel), SWIGTYPE_p_sigrok__LogLevel, 0);
+
+ auto string_obj = PyString_FromString(message.c_str());
+
+ auto arglist = Py_BuildValue("(OO)", log_obj, string_obj);
+
+ auto result = PyEval_CallObject($input, arglist);
+
+ Py_XDECREF(arglist);
+ Py_XDECREF(log_obj);
+ Py_XDECREF(string_obj);
+ Py_XDECREF(result);
+
+ PyGILState_Release(gstate);
+ };
+
+ Py_XINCREF($input);
+}
+
+/* Map from callable PyObject to DatafeedCallbackFunction */
+%typecheck(SWIG_TYPECHECK_POINTER) sigrok::DatafeedCallbackFunction {
+ $1 = PyCallable_Check($input);
+}
+
+%typemap(in) sigrok::DatafeedCallbackFunction {
+ if (!PyCallable_Check($input))
+ SWIG_exception(SWIG_TypeError, "Expected a callable Python object");
+
+ $1 = [=] (std::shared_ptr<sigrok::Device> device,
+ std::shared_ptr<sigrok::Packet> packet) {
+ auto gstate = PyGILState_Ensure();
+
+ auto device_obj = SWIG_NewPointerObj(
+ SWIG_as_voidptr(new std::shared_ptr<sigrok::Device>(device)),
+ SWIGTYPE_p_std__shared_ptrT_sigrok__Device_t, SWIG_POINTER_OWN);
+
+ auto packet_obj = SWIG_NewPointerObj(
+ SWIG_as_voidptr(new std::shared_ptr<sigrok::Packet>(packet)),
+ SWIGTYPE_p_std__shared_ptrT_sigrok__Packet_t, SWIG_POINTER_OWN);
+
+ auto arglist = Py_BuildValue("(OO)", device_obj, packet_obj);
+
+ auto result = PyEval_CallObject($input, arglist);
+
+ Py_XDECREF(arglist);
+ Py_XDECREF(device_obj);
+ Py_XDECREF(packet_obj);
+ Py_XDECREF(result);
+
+ PyGILState_Release(gstate);
+ };
+
+ Py_XINCREF($input);
+}
+
+%{
+
+#include "libsigrok/libsigrok.hpp"
+
+/* Convert from a Python dict to a std::map<std::string, std::string> */
+std::map<std::string, std::string> dict_to_map(PyObject *dict)
+{
+ if (!PyDict_Check(dict))
+ throw sigrok::Error(SR_ERR_ARG);
+
+ std::map<std::string, std::string> output;
+
+ PyObject *py_key, *py_value;
+ Py_ssize_t pos = 0;
+
+ while (PyDict_Next(dict, &pos, &py_key, &py_value)) {
+ if (!PyString_Check(py_key))
+ throw sigrok::Error(SR_ERR_ARG);
+ if (!PyString_Check(py_value))
+ throw sigrok::Error(SR_ERR_ARG);
+ auto key = PyString_AsString(py_key);
+ auto value = PyString_AsString(py_value);
+ output[key] = value;
+ }
+
+ return output;
+}
+
+/* Convert from a Python type to Glib::Variant, according to config key data type. */
+Glib::VariantBase python_to_variant_by_key(PyObject *input, const sigrok::ConfigKey *key)
+{
+ enum sr_datatype type = key->get_data_type()->get_id();
+
+ if (type == SR_T_UINT64 && PyInt_Check(input))
+ return Glib::Variant<guint64>::create(PyInt_AsLong(input));
+ if (type == SR_T_UINT64 && PyLong_Check(input))
+ return Glib::Variant<guint64>::create(PyLong_AsLong(input));
+ else if (type == SR_T_STRING && PyString_Check(input))
+ return Glib::Variant<std::string>::create(PyString_AsString(input));
+ else if (type == SR_T_BOOL && PyBool_Check(input))
+ return Glib::Variant<bool>::create(input == Py_True);
+ else if (type == SR_T_FLOAT && PyFloat_Check(input))
+ return Glib::Variant<double>::create(PyFloat_AsDouble(input));
+ else if (type == SR_T_INT32 && PyInt_Check(input))
+ return Glib::Variant<gint32>::create(PyInt_AsLong(input));
+ else
+ throw sigrok::Error(SR_ERR_ARG);
+}
+
+%}
+
+/* Ignore these methods, we will override them below. */
+%ignore sigrok::Driver::scan;
+%ignore sigrok::InputFormat::open_file;
+%ignore sigrok::OutputFormat::create_output;
+
+%include "../../../swig/classes.i"
+
+/* Support Driver.scan() with keyword arguments. */
+%extend sigrok::Driver
+{
+ std::vector<std::shared_ptr<sigrok::HardwareDevice> > _scan_kwargs(PyObject *dict)
+ {
+ if (!PyDict_Check(dict))
+ throw sigrok::Error(SR_ERR_ARG);
+
+ PyObject *py_key, *py_value;
+ Py_ssize_t pos = 0;
+ std::map<const sigrok::ConfigKey *, Glib::VariantBase> options;
+
+ while (PyDict_Next(dict, &pos, &py_key, &py_value))
+ {
+ if (!PyString_Check(py_key))
+ throw sigrok::Error(SR_ERR_ARG);
+ auto key = sigrok::ConfigKey::get(PyString_AsString(py_key));
+ auto value = python_to_variant_by_key(py_value, key);
+ options[key] = value;
+ }
+
+ return $self->scan(options);
+ }
+}
+
+%pythoncode
+{
+ def _Driver_scan(self, **kwargs):
+ return self._scan_kwargs(kwargs)
+
+ Driver.scan = _Driver_scan
+}
+
+/* Support InputFormat.open_file() with keyword arguments. */
+%extend sigrok::InputFormat
+{
+ std::shared_ptr<sigrok::InputFileDevice> _open_file_kwargs(std::string filename, PyObject *dict)
+ {
+ return $self->open_file(filename, dict_to_map(dict));
+ }
+}
+
+%pythoncode
+{
+ def _InputFormat_open_file(self, filename, **kwargs):
+ return self._open_file_kwargs(filename, kwargs)
+
+ InputFormat.open_file = _InputFormat_open_file
+}
+
+/* Support OutputFormat.create_output() with keyword arguments. */
+%extend sigrok::OutputFormat
+{
+ std::shared_ptr<sigrok::Output> _create_output_kwargs(
+ std::shared_ptr<sigrok::Device> device, PyObject *dict)
+ {
+ return $self->create_output(device, dict_to_map(dict));
+ }
+}
+
+%pythoncode
+{
+ def _OutputFormat_create_output(self, device, **kwargs):
+ return self._create_output_kwargs(device, kwargs)
+
+ OutputFormat.create_output = _OutputFormat_create_output
+}
+
+/* Support config_set() with Python input types. */
+%extend sigrok::Configurable
+{
+ void config_set(const ConfigKey *key, PyObject *input)
+ {
+ $self->config_set(key, python_to_variant_by_key(input, key));
+ }
+}
+++ /dev/null
-##
-## This file is part of the libsigrok project.
-##
-## Copyright (C) 2013 Martin Ling <martin-sigrok@earth.li>
-##
-## 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 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 General Public License for more details.
-##
-## You should have received a copy of the GNU General Public License
-## along with this program. If not, see <http://www.gnu.org/licenses/>.
-##
-
-from functools import partial
-from fractions import Fraction
-from collections import OrderedDict
-from .lowlevel import *
-from . import lowlevel
-import itertools
-
-__all__ = ['Error', 'Context', 'Driver', 'Device', 'Session', 'Packet', 'Log',
- 'LogLevel', 'PacketType', 'Quantity', 'Unit', 'QuantityFlag', 'ConfigKey',
- 'ChannelType', 'Channel', 'ChannelGroup', 'InputFormat', 'OutputFormat',
- 'InputFile', 'Output']
-
-class Error(Exception):
-
- def __str__(self):
- return sr_strerror(self.args[0])
-
-def check(result):
- if result != SR_OK:
- raise Error(result)
-
-def gvariant_to_python(value):
- type_string = g_variant_get_type_string(value)
- if type_string == 't':
- return g_variant_get_uint64(value)
- if type_string == 'b':
- return g_variant_get_bool(value)
- if type_string == 'd':
- return g_variant_get_double(value)
- if type_string == 's':
- return g_variant_get_string(value, None)
- if type_string == '(tt)':
- return Fraction(
- g_variant_get_uint64(g_variant_get_child_value(value, 0)),
- g_variant_get_uint64(g_variant_get_child_value(value, 1)))
- raise NotImplementedError(
- "Can't convert GVariant type '%s' to a Python type." % type_string)
-
-def python_to_gvariant(value):
- if isinstance(value, int):
- return g_variant_new_uint64(value)
- if isinstance(value, bool):
- return g_variant_new_boolean(value)
- if isinstance(value, float):
- return g_variant_new_double(value)
- if isinstance(value, str):
- return g_variant_new_string(value)
- if isinstance(value, Fraction):
- array = new_gvariant_ptr_array(2)
- gvariant_ptr_array_setitem(array, 0,
- g_variant_new_uint64(value.numerator))
- gvariant_ptr_array_setitem(array, 1,
- g_variant_new_uint64(value.denominator))
- result = g_variant_new_tuple(array, 2)
- delete_gvariant_ptr_array(array)
- return result
- raise NotImplementedError(
- "Can't convert Python '%s' to a GVariant." % type(value))
-
-def callback_wrapper(session, callback, device_ptr, packet_ptr):
- device = session.context._devices[int(device_ptr.this)]
- packet = Packet(session, packet_ptr)
- callback(device, packet)
-
-class Context(object):
-
- def __init__(self):
- context_ptr_ptr = new_sr_context_ptr_ptr()
- check(sr_init(context_ptr_ptr))
- self.struct = sr_context_ptr_ptr_value(context_ptr_ptr)
- self._drivers = None
- self._devices = {}
- self._input_formats = None
- self._output_formats = None
- self.session = None
-
- def __del__(self):
- sr_exit(self.struct)
-
- @property
- def drivers(self):
- if not self._drivers:
- self._drivers = {}
- driver_list = sr_driver_list()
- for i in itertools.count():
- driver_ptr = sr_dev_driver_ptr_array_getitem(driver_list, i)
- if driver_ptr:
- self._drivers[driver_ptr.name] = Driver(self, driver_ptr)
- else:
- break
- return self._drivers
-
- @property
- def input_formats(self):
- if not self._input_formats:
- self._input_formats = OrderedDict()
- input_list = sr_input_list()
- for i in itertools.count():
- input_ptr = sr_input_format_ptr_array_getitem(input_list, i)
- if input_ptr:
- self._input_formats[input_ptr.id] = InputFormat(self, input_ptr)
- else:
- break
- return self._input_formats
-
- @property
- def output_formats(self):
- if not self._output_formats:
- self._output_formats = {}
- output_list = sr_output_list()
- for i in itertools.count():
- output_ptr = sr_output_format_ptr_array_getitem(output_list, i)
- if output_ptr:
- self._output_formats[output_ptr.id] = OutputFormat(self, output_ptr)
- else:
- break
- return self._output_formats
-
-class Driver(object):
-
- def __init__(self, context, struct):
- self.context = context
- self.struct = struct
- self._initialized = False
-
- @property
- def name(self):
- return self.struct.name
-
- def scan(self, **kwargs):
- if not self._initialized:
- check(sr_driver_init(self.context.struct, self.struct))
- self._initialized = True
- options = []
- for name, value in kwargs.items():
- key = getattr(ConfigKey, name)
- src = sr_config()
- src.key = key.id
- src.data = python_to_gvariant(value)
- options.append(src.this)
- option_list = python_to_gslist(options)
- device_list = sr_driver_scan(self.struct, option_list)
- g_slist_free(option_list)
- devices = [HardwareDevice(self, gpointer_to_sr_dev_inst_ptr(ptr))
- for ptr in gslist_to_python(device_list)]
- g_slist_free(device_list)
- return devices
-
-class Device(object):
-
- def __new__(cls, struct, context):
- address = int(struct.this)
- if address not in context._devices:
- device = super(Device, cls).__new__(cls)
- device.struct = struct
- device.context = context
- device._channels = None
- device._channel_groups = None
- context._devices[address] = device
- return context._devices[address]
-
- @property
- def vendor(self):
- return self.struct.vendor
-
- @property
- def model(self):
- return self.struct.model
-
- @property
- def version(self):
- return self.struct.version
-
- @property
- def channels(self):
- if self._channels is None:
- self._channels = {}
- channel_list = self.struct.channels
- while (channel_list):
- channel_ptr = void_ptr_to_sr_channel_ptr(channel_list.data)
- self._channels[channel_ptr.name] = Channel(self, channel_ptr)
- channel_list = channel_list.next
- return self._channels
-
- @property
- def channel_groups(self):
- if self._channel_groups is None:
- self._channel_groups = {}
- channel_group_list = self.struct.channel_groups
- while (channel_group_list):
- channel_group_ptr = void_ptr_to_sr_channel_group_ptr(
- channel_group_list.data)
- self._channel_groups[channel_group_ptr.name] = ChannelGroup(self,
- channel_group_ptr)
- channel_group_list = channel_group_list.next
- return self._channel_groups
-
-class HardwareDevice(Device):
-
- def __new__(cls, driver, struct):
- device = Device.__new__(cls, struct, driver.context)
- device.driver = driver
- return device
-
- def __getattr__(self, name):
- key = getattr(ConfigKey, name)
- data = new_gvariant_ptr_ptr()
- try:
- check(sr_config_get(self.driver.struct, self.struct, None,
- key.id, data))
- except Error as error:
- if error.errno == SR_ERR_NA:
- raise NotImplementedError(
- "Device does not implement %s" % name)
- else:
- raise AttributeError
- value = gvariant_ptr_ptr_value(data)
- return gvariant_to_python(value)
-
- def __setattr__(self, name, value):
- try:
- key = getattr(ConfigKey, name)
- except AttributeError:
- super(Device, self).__setattr__(name, value)
- return
- check(sr_config_set(self.struct, None, key.id, python_to_gvariant(value)))
-
-class Channel(object):
-
- def __init__(self, device, struct):
- self.device = device
- self.struct = struct
-
- @property
- def type(self):
- return ChannelType(self.struct.type)
-
- @property
- def enabled(self):
- return self.struct.enabled
-
- @property
- def name(self):
- return self.struct.name
-
-class ChannelGroup(object):
-
- def __init__(self, device, struct):
- self.device = device
- self.struct = struct
- self._channels = None
-
- def __iter__(self):
- return iter(self.channels)
-
- def __getattr__(self, name):
- key = config_key(name)
- data = new_gvariant_ptr_ptr()
- try:
- check(sr_config_get(self.device.driver.struct, self.device.struct,
- self.struct, key.id, data))
- except Error as error:
- if error.errno == SR_ERR_NA:
- raise NotImplementedError(
- "Channel group does not implement %s" % name)
- else:
- raise AttributeError
- value = gvariant_ptr_ptr_value(data)
- return gvariant_to_python(value)
-
- def __setattr__(self, name, value):
- try:
- key = config_key(name)
- except AttributeError:
- super(ChannelGroup, self).__setattr__(name, value)
- return
- check(sr_config_set(self.device.struct, self.struct,
- key.id, python_to_gvariant(value)))
-
- @property
- def name(self):
- return self.struct.name
-
- @property
- def channels(self):
- if self._channels is None:
- self._channels = []
- channel_list = self.struct.channels
- while (channel_list):
- channel_ptr = void_ptr_to_sr_channel_ptr(channel_list.data)
- self._channels.append(Channel(self, channel_ptr))
- channel_list = channel_list.next
- return self._channels
-
-class Session(object):
-
- def __init__(self, context):
- assert context.session is None
- self.context = context
- self.struct = sr_session_new()
- context.session = self
-
- def __del__(self):
- check(sr_session_destroy())
-
- def add_device(self, device):
- check(sr_session_dev_add(device.struct))
-
- def open_device(self, device):
- check(sr_dev_open(device.struct))
-
- def add_callback(self, callback):
- wrapper = partial(callback_wrapper, self, callback)
- check(sr_session_datafeed_python_callback_add(wrapper))
-
- def start(self):
- check(sr_session_start())
-
- def run(self):
- check(sr_session_run())
-
- def stop(self):
- check(sr_session_stop())
-
-class Packet(object):
-
- def __init__(self, session, struct):
- self.session = session
- self.struct = struct
- self._payload = None
-
- @property
- def type(self):
- return PacketType(self.struct.type)
-
- @property
- def payload(self):
- if self._payload is None:
- pointer = self.struct.payload
- if self.type == PacketType.LOGIC:
- self._payload = Logic(self,
- void_ptr_to_sr_datafeed_logic_ptr(pointer))
- elif self.type == PacketType.ANALOG:
- self._payload = Analog(self,
- void_ptr_to_sr_datafeed_analog_ptr(pointer))
- else:
- raise NotImplementedError(
- "No Python mapping for packet type %s" % self.struct.type)
- return self._payload
-
-class Logic(object):
-
- def __init__(self, packet, struct):
- self.packet = packet
- self.struct = struct
- self._data = None
-
- @property
- def data(self):
- if self._data is None:
- self._data = cdata(self.struct.data, self.struct.length)
- return self._data
-
-class Analog(object):
-
- def __init__(self, packet, struct):
- self.packet = packet
- self.struct = struct
- self._data = None
-
- @property
- def num_samples(self):
- return self.struct.num_samples
-
- @property
- def mq(self):
- return Quantity(self.struct.mq)
-
- @property
- def unit(self):
- return Unit(self.struct.unit)
-
- @property
- def mqflags(self):
- return QuantityFlag.set_from_mask(self.struct.mqflags)
-
- @property
- def data(self):
- if self._data is None:
- self._data = float_array.frompointer(self.struct.data)
- return self._data
-
-class Log(object):
-
- @property
- def level(self):
- return LogLevel(sr_log_loglevel_get())
-
- @level.setter
- def level(self, l):
- check(sr_log_loglevel_set(l.id))
-
- @property
- def domain(self):
- return sr_log_logdomain_get()
-
- @domain.setter
- def domain(self, d):
- check(sr_log_logdomain_set(d))
-
-class InputFormat(object):
-
- def __init__(self, context, struct):
- self.context = context
- self.struct = struct
-
- @property
- def id(self):
- return self.struct.id
-
- @property
- def description(self):
- return self.struct.description
-
- def format_match(self, filename):
- return bool(self.struct.call_format_match(filename))
-
-class InputFile(object):
-
- def __init__(self, format, filename, **kwargs):
- self.format = format
- self.filename = filename
- self.struct = sr_input()
- self.struct.format = self.format.struct
- self.struct.param = g_hash_table_new_full(
- g_str_hash_ptr, g_str_equal_ptr, g_free_ptr, g_free_ptr)
- for key, value in kwargs.items():
- g_hash_table_insert(self.struct.param, g_strdup(key), g_strdup(str(value)))
- check(self.format.struct.call_init(self.struct, self.filename))
- self.device = InputFileDevice(self)
-
- def load(self):
- check(self.format.struct.call_loadfile(self.struct, self.filename))
-
- def __del__(self):
- g_hash_table_destroy(self.struct.param)
-
-class InputFileDevice(Device):
-
- def __new__(cls, file):
- device = Device.__new__(cls, file.struct.sdi, file.format.context)
- device.file = file
- return device
-
-class OutputFormat(object):
-
- def __init__(self, context, struct):
- self.context = context
- self.struct = struct
-
- @property
- def id(self):
- return self.struct.id
-
- @property
- def description(self):
- return self.struct.description
-
-class Output(object):
-
- def __init__(self, format, device, param=None):
- self.format = format
- self.device = device
- self.param = param
- self.struct = sr_output()
- self.struct.format = self.format.struct
- self.struct.sdi = self.device.struct
- self.struct.param = param
- check(self.format.struct.call_init(self.struct))
-
- def receive(self, packet):
-
- output_buf_ptr = new_uint8_ptr_ptr()
- output_len_ptr = new_uint64_ptr()
- using_obsolete_api = False
-
- if self.format.struct.event and packet.type in (
- PacketType.TRIGGER, PacketType.FRAME_BEGIN,
- PacketType.FRAME_END, PacketType.END):
- check(self.format.struct.call_event(self.struct, packet.type.id,
- output_buf_ptr, output_len_ptr))
- using_obsolete_api = True
- elif self.format.struct.data and packet.type.id == self.format.struct.df_type:
- check(self.format.struct.call_data(self.struct,
- packet.payload.struct.data, packet.payload.struct.length,
- output_buf_ptr, output_len_ptr))
- using_obsolete_api = True
-
- if using_obsolete_api:
- output_buf = uint8_ptr_ptr_value(output_buf_ptr)
- output_len = uint64_ptr_value(output_len_ptr)
- result = cdata(output_buf, output_len)
- g_free(output_buf)
- return result
-
- if self.format.struct.receive:
- out_ptr = new_gstring_ptr_ptr()
- check(self.format.struct.call_receive(self.struct, self.device.struct,
- packet.struct, out_ptr))
- out = gstring_ptr_ptr_value(out_ptr)
- if out:
- result = out.str
- g_string_free(out, True)
- return result
-
- return None
-
- def __del__(self):
- check(self.format.struct.call_cleanup(self.struct))
-
-class ConfigInfo(object):
-
- def __new__(cls, key):
- struct = sr_config_info_get(key.id)
- if not struct:
- return None
- obj = super(ConfigInfo, cls).__new__(cls)
- obj.key = key
- obj.struct = struct
- return obj
-
- @property
- def datatype(self):
- return DataType(self.struct.datatype)
-
- @property
- def id(self):
- return self.struct.id
-
- @property
- def name(self):
- return self.struct.name
-
- @property
- def description(self):
- return self.struct.description
-
-class EnumValue(object):
-
- _enum_values = {}
-
- def __new__(cls, id):
- if cls not in cls._enum_values:
- cls._enum_values[cls] = {}
- if id not in cls._enum_values[cls]:
- value = super(EnumValue, cls).__new__(cls)
- value.id = id
- cls._enum_values[cls][id] = value
- return cls._enum_values[cls][id]
-
-class LogLevel(EnumValue):
- pass
-
-class PacketType(EnumValue):
- pass
-
-class Quantity(EnumValue):
- pass
-
-class Unit(EnumValue):
- pass
-
-class QuantityFlag(EnumValue):
-
- @classmethod
- def set_from_mask(cls, mask):
- result = set()
- while mask:
- new_mask = mask & (mask - 1)
- result.add(cls(mask ^ new_mask))
- mask = new_mask
- return result
-
-class ConfigKey(EnumValue):
- pass
-
-class DataType(EnumValue):
- pass
-
-class ChannelType(EnumValue):
- pass
-
-for symbol_name in dir(lowlevel):
- for prefix, cls in [
- ('SR_LOG_', LogLevel),
- ('SR_DF_', PacketType),
- ('SR_MQ_', Quantity),
- ('SR_UNIT_', Unit),
- ('SR_MQFLAG_', QuantityFlag),
- ('SR_CONF_', ConfigKey),
- ('SR_T_', DataType),
- ('SR_CHANNEL_', ChannelType)]:
- if symbol_name.startswith(prefix):
- name = symbol_name[len(prefix):]
- value = getattr(lowlevel, symbol_name)
- obj = cls(value)
- setattr(cls, name, obj)
- if cls is ConfigKey:
- obj.info = ConfigInfo(obj)
- if obj.info:
- setattr(cls, obj.info.id, obj)
- else:
- setattr(cls, name.lower(), obj)
projects / libsigrok.git / commitdiff