2 ## This file is part of the libsigrok project.
4 ## Copyright (C) 2013 Martin Ling <martin-sigrok@earth.li>
6 ## This program is free software: you can redistribute it and/or modify
7 ## it under the terms of the GNU General Public License as published by
8 ## the Free Software Foundation, either version 3 of the License, or
9 ## (at your option) any later version.
11 ## This program is distributed in the hope that it will be useful,
12 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
13 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 ## GNU General Public License for more details.
16 ## You should have received a copy of the GNU General Public License
17 ## along with this program. If not, see <http://www.gnu.org/licenses/>.
20 from functools import partial
21 from fractions import Fraction
22 from .lowlevel import *
23 from . import lowlevel
26 __all__ = ['Error', 'Context', 'Driver', 'Device', 'Session', 'Packet', 'Log',
27 'LogLevel', 'PacketType', 'Quantity', 'Unit', 'QuantityFlag', 'ConfigKey']
29 class Error(Exception):
32 return sr_strerror(self.args[0])
38 def gvariant_to_python(value):
39 type_string = g_variant_get_type_string(value)
40 if type_string == 't':
41 return g_variant_get_uint64(value)
42 if type_string == 'b':
43 return g_variant_get_bool(value)
44 if type_string == 'd':
45 return g_variant_get_double(value)
46 if type_string == 's':
47 return g_variant_get_string(value, None)
48 if type_string == '(tt)':
50 g_variant_get_uint64(g_variant_get_child_value(value, 0)),
51 g_variant_get_uint64(g_variant_get_child_value(value, 1)))
52 raise NotImplementedError(
53 "Can't convert GVariant type '%s' to a Python type." % type_string)
55 def python_to_gvariant(value):
56 if isinstance(value, int):
57 return g_variant_new_uint64(value)
58 if isinstance(value, bool):
59 return g_variant_new_boolean(value)
60 if isinstance(value, float):
61 return g_variant_new_double(value)
62 if isinstance(value, str):
63 return g_variant_new_string(value)
64 if isinstance(value, Fraction):
65 array = new_gvariant_ptr_array(2)
66 gvariant_ptr_array_setitem(array, 0,
67 g_variant_new_uint64(value.numerator))
68 gvariant_ptr_array_setitem(array, 1,
69 g_variant_new_uint64(value.denominator))
70 result = g_variant_new_tuple(array, 2)
71 delete_gvariant_ptr_array(array)
73 raise NotImplementedError(
74 "Can't convert Python '%s' to a GVariant." % type(value))
76 def callback_wrapper(session, callback, device_ptr, packet_ptr):
77 device = session.context._devices[int(device_ptr.this)]
78 packet = Packet(session, packet_ptr)
79 callback(device, packet)
81 class Context(object):
84 context_ptr_ptr = new_sr_context_ptr_ptr()
85 check(sr_init(context_ptr_ptr))
86 self.struct = sr_context_ptr_ptr_value(context_ptr_ptr)
98 driver_list = sr_driver_list()
99 for i in itertools.count():
100 driver_ptr = sr_dev_driver_ptr_array_getitem(driver_list, i)
102 self._drivers[driver_ptr.name] = Driver(self, driver_ptr)
107 class Driver(object):
109 def __init__(self, context, struct):
110 self.context = context
112 self._initialized = False
116 return self.struct.name
118 def scan(self, **kwargs):
119 if not self._initialized:
120 check(sr_driver_init(self.context.struct, self.struct))
121 self._initialized = True
123 for name, value in kwargs.items():
124 key = getattr(ConfigKey, name.upper())
127 src.data = python_to_gvariant(value)
128 options.append(src.this)
129 option_list = python_to_gslist(options)
130 device_list = sr_driver_scan(self.struct, option_list)
131 g_slist_free(option_list)
132 devices = [Device(self, gpointer_to_sr_dev_inst_ptr(ptr))
133 for ptr in gslist_to_python(device_list)]
134 g_slist_free(device_list)
137 class Device(object):
139 def __new__(cls, driver, struct):
140 address = int(struct.this)
141 if address not in driver.context._devices:
142 device = super(Device, cls).__new__(cls, driver, struct)
143 driver.context._devices[address] = device
144 return driver.context._devices[address]
146 def __init__(self, driver, struct):
150 def __getattr__(self, name):
151 key = getattr(ConfigKey, name.upper())
152 data = new_gvariant_ptr_ptr()
154 check(sr_config_get(self.driver.struct,
155 key.id, data, self.struct))
156 except Error as error:
157 if error.errno == SR_ERR_NA:
158 raise NotImplementedError(
159 "Device does not implement %s" % name)
162 value = gvariant_ptr_ptr_value(data)
163 return gvariant_to_python(value)
165 def __setattr__(self, name, value):
167 key = getattr(ConfigKey, name.upper())
168 except AttributeError:
169 super(Device, self).__setattr__(name, value)
171 check(sr_config_set(self.struct,
172 key.id, python_to_gvariant(value)))
176 return self.struct.vendor
180 return self.struct.model
184 return self.struct.version
186 class Session(object):
188 def __init__(self, context):
189 assert context.session is None
190 self.context = context
191 self.struct = sr_session_new()
192 context.session = self
195 check(sr_session_destroy())
197 def add_device(self, device):
198 check(sr_session_dev_add(device.struct))
200 def add_callback(self, callback):
201 wrapper = partial(callback_wrapper, self, callback)
202 check(sr_session_datafeed_python_callback_add(wrapper))
205 check(sr_session_start())
208 check(sr_session_run())
211 check(sr_session_stop())
213 class Packet(object):
215 def __init__(self, session, struct):
216 self.session = session
222 return PacketType(self.struct.type)
226 if self._payload is None:
227 pointer = self.struct.payload
228 if self.type == PacketType.LOGIC:
229 self._payload = Logic(self,
230 void_ptr_to_sr_datafeed_logic_ptr(pointer))
231 elif self.type == PacketType.ANALOG:
232 self._payload = Analog(self,
233 void_ptr_to_sr_datafeed_analog_ptr(pointer))
235 raise NotImplementedError(
236 "No Python mapping for packet type %ѕ" % self.struct.type)
241 def __init__(self, packet, struct):
248 if self._data is None:
249 self._data = cdata(self.struct.data, self.struct.length)
252 class Analog(object):
254 def __init__(self, packet, struct):
260 def num_samples(self):
261 return self.struct.num_samples
265 return Quantity(self.struct.mq)
269 return Unit(self.struct.unit)
273 return QuantityFlag.set_from_mask(self.struct.mqflags)
277 if self._data is None:
278 self._data = float_array.frompointer(self.struct.data)
285 return LogLevel(sr_log_loglevel_get())
289 check(sr_log_loglevel_set(l.id))
293 return sr_log_logdomain_get()
297 check(sr_log_logdomain_set(d))
299 class EnumValue(object):
303 def __new__(cls, id):
304 if cls not in cls._enum_values:
305 cls._enum_values[cls] = {}
306 if id not in cls._enum_values[cls]:
307 value = super(EnumValue, cls).__new__(cls)
309 cls._enum_values[cls][id] = value
310 return cls._enum_values[cls][id]
312 class LogLevel(EnumValue):
315 class PacketType(EnumValue):
318 class Quantity(EnumValue):
321 class Unit(EnumValue):
324 class QuantityFlag(EnumValue):
327 def set_from_mask(cls, mask):
330 new_mask = mask & (mask - 1)
331 result.add(cls(mask ^ new_mask))
335 class ConfigKey(EnumValue):
338 for symbol_name in dir(lowlevel):
340 ('SR_LOG_', LogLevel),
341 ('SR_DF_', PacketType),
342 ('SR_MQ_', Quantity),
344 ('SR_MQFLAG_', QuantityFlag),
345 ('SR_CONF_', ConfigKey)]:
346 if symbol_name.startswith(prefix):
347 name = symbol_name[len(prefix):]
348 value = getattr(lowlevel, symbol_name)
349 setattr(cls, name, cls(value))