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',
28 'ProbeType', 'Probe', 'ProbeGroup']
30 class Error(Exception):
33 return sr_strerror(self.args[0])
39 def gvariant_to_python(value):
40 type_string = g_variant_get_type_string(value)
41 if type_string == 't':
42 return g_variant_get_uint64(value)
43 if type_string == 'b':
44 return g_variant_get_bool(value)
45 if type_string == 'd':
46 return g_variant_get_double(value)
47 if type_string == 's':
48 return g_variant_get_string(value, None)
49 if type_string == '(tt)':
51 g_variant_get_uint64(g_variant_get_child_value(value, 0)),
52 g_variant_get_uint64(g_variant_get_child_value(value, 1)))
53 raise NotImplementedError(
54 "Can't convert GVariant type '%s' to a Python type." % type_string)
56 def python_to_gvariant(value):
57 if isinstance(value, int):
58 return g_variant_new_uint64(value)
59 if isinstance(value, bool):
60 return g_variant_new_boolean(value)
61 if isinstance(value, float):
62 return g_variant_new_double(value)
63 if isinstance(value, str):
64 return g_variant_new_string(value)
65 if isinstance(value, Fraction):
66 array = new_gvariant_ptr_array(2)
67 gvariant_ptr_array_setitem(array, 0,
68 g_variant_new_uint64(value.numerator))
69 gvariant_ptr_array_setitem(array, 1,
70 g_variant_new_uint64(value.denominator))
71 result = g_variant_new_tuple(array, 2)
72 delete_gvariant_ptr_array(array)
74 raise NotImplementedError(
75 "Can't convert Python '%s' to a GVariant." % type(value))
77 def callback_wrapper(session, callback, device_ptr, packet_ptr):
78 device = session.context._devices[int(device_ptr.this)]
79 packet = Packet(session, packet_ptr)
80 callback(device, packet)
82 class Context(object):
85 context_ptr_ptr = new_sr_context_ptr_ptr()
86 check(sr_init(context_ptr_ptr))
87 self.struct = sr_context_ptr_ptr_value(context_ptr_ptr)
90 self._input_formats = None
91 self._output_formats = None
101 driver_list = sr_driver_list()
102 for i in itertools.count():
103 driver_ptr = sr_dev_driver_ptr_array_getitem(driver_list, i)
105 self._drivers[driver_ptr.name] = Driver(self, driver_ptr)
111 def input_formats(self):
112 if not self._input_formats:
113 self._input_formats = {}
114 input_list = sr_input_list()
115 for i in itertools.count():
116 input_ptr = sr_input_format_ptr_array_getitem(input_list, i)
118 self._input_formats[input_ptr.id] = InputFormat(self, input_ptr)
121 return self._input_formats
124 def output_formats(self):
125 if not self._output_formats:
126 self._output_formats = {}
127 output_list = sr_output_list()
128 for i in itertools.count():
129 output_ptr = sr_output_format_ptr_array_getitem(output_list, i)
131 self._output_formats[output_ptr.id] = OutputFormat(self, output_ptr)
134 return self._output_formats
136 class Driver(object):
138 def __init__(self, context, struct):
139 self.context = context
141 self._initialized = False
145 return self.struct.name
147 def scan(self, **kwargs):
148 if not self._initialized:
149 check(sr_driver_init(self.context.struct, self.struct))
150 self._initialized = True
152 for name, value in kwargs.items():
153 key = getattr(ConfigKey, name.upper())
156 src.data = python_to_gvariant(value)
157 options.append(src.this)
158 option_list = python_to_gslist(options)
159 device_list = sr_driver_scan(self.struct, option_list)
160 g_slist_free(option_list)
161 devices = [Device(self, gpointer_to_sr_dev_inst_ptr(ptr))
162 for ptr in gslist_to_python(device_list)]
163 g_slist_free(device_list)
166 class Device(object):
168 def __new__(cls, driver, struct):
169 address = int(struct.this)
170 if address not in driver.context._devices:
171 device = super(Device, cls).__new__(cls, driver, struct)
172 driver.context._devices[address] = device
173 return driver.context._devices[address]
175 def __init__(self, driver, struct):
179 self._probe_groups = None
181 def __getattr__(self, name):
182 key = getattr(ConfigKey, name.upper())
183 data = new_gvariant_ptr_ptr()
185 check(sr_config_get(self.driver.struct, self.struct, None,
187 except Error as error:
188 if error.errno == SR_ERR_NA:
189 raise NotImplementedError(
190 "Device does not implement %s" % name)
193 value = gvariant_ptr_ptr_value(data)
194 return gvariant_to_python(value)
196 def __setattr__(self, name, value):
198 key = getattr(ConfigKey, name.upper())
199 except AttributeError:
200 super(Device, self).__setattr__(name, value)
202 check(sr_config_set(self.struct, None, key.id, python_to_gvariant(value)))
206 return self.struct.vendor
210 return self.struct.model
214 return self.struct.version
218 if self._probes is None:
220 probe_list = self.struct.probes
222 probe_ptr = void_ptr_to_sr_probe_ptr(probe_list.data)
223 self._probes[probe_ptr.name] = Probe(self, probe_ptr)
224 probe_list = probe_list.next
228 def probe_groups(self):
229 if self._probe_groups is None:
230 self._probe_groups = {}
231 probe_group_list = self.struct.probe_groups
232 while (probe_group_list):
233 probe_group_ptr = void_ptr_to_sr_probe_group_ptr(
234 probe_group_list.data)
235 self._probe_groups[probe_group_ptr.name] = ProbeGroup(self,
237 probe_group_list = probe_group_list.next
238 return self._probe_groups
242 def __init__(self, device, struct):
248 return ProbeType(self.struct.type)
252 return self.struct.enabled
256 return self.struct.name
258 class ProbeGroup(object):
260 def __init__(self, device, struct):
266 return iter(self.probes)
268 def __getattr__(self, name):
269 key = config_key(name)
270 data = new_gvariant_ptr_ptr()
272 check(sr_config_get(self.device.driver.struct, self.device.struct,
273 self.struct, key.id, data))
274 except Error as error:
275 if error.errno == SR_ERR_NA:
276 raise NotImplementedError(
277 "Probe group does not implement %s" % name)
280 value = gvariant_ptr_ptr_value(data)
281 return gvariant_to_python(value)
283 def __setattr__(self, name, value):
285 key = config_key(name)
286 except AttributeError:
287 super(ProbeGroup, self).__setattr__(name, value)
289 check(sr_config_set(self.device.struct, self.struct,
290 key.id, python_to_gvariant(value)))
294 return self.struct.name
298 if self._probes is None:
300 probe_list = self.struct.probes
302 probe_ptr = void_ptr_to_sr_probe_ptr(probe_list.data)
303 self._probes.append(Probe(self, probe_ptr))
304 probe_list = probe_list.next
307 class Session(object):
309 def __init__(self, context):
310 assert context.session is None
311 self.context = context
312 self.struct = sr_session_new()
313 context.session = self
316 check(sr_session_destroy())
318 def add_device(self, device):
319 check(sr_session_dev_add(device.struct))
321 def open_device(self, device):
322 check(sr_dev_open(device.struct))
324 def add_callback(self, callback):
325 wrapper = partial(callback_wrapper, self, callback)
326 check(sr_session_datafeed_python_callback_add(wrapper))
329 check(sr_session_start())
332 check(sr_session_run())
335 check(sr_session_stop())
337 class Packet(object):
339 def __init__(self, session, struct):
340 self.session = session
346 return PacketType(self.struct.type)
350 if self._payload is None:
351 pointer = self.struct.payload
352 if self.type == PacketType.LOGIC:
353 self._payload = Logic(self,
354 void_ptr_to_sr_datafeed_logic_ptr(pointer))
355 elif self.type == PacketType.ANALOG:
356 self._payload = Analog(self,
357 void_ptr_to_sr_datafeed_analog_ptr(pointer))
359 raise NotImplementedError(
360 "No Python mapping for packet type %s" % self.struct.type)
365 def __init__(self, packet, struct):
372 if self._data is None:
373 self._data = cdata(self.struct.data, self.struct.length)
376 class Analog(object):
378 def __init__(self, packet, struct):
384 def num_samples(self):
385 return self.struct.num_samples
389 return Quantity(self.struct.mq)
393 return Unit(self.struct.unit)
397 return QuantityFlag.set_from_mask(self.struct.mqflags)
401 if self._data is None:
402 self._data = float_array.frompointer(self.struct.data)
409 return LogLevel(sr_log_loglevel_get())
413 check(sr_log_loglevel_set(l.id))
417 return sr_log_logdomain_get()
421 check(sr_log_logdomain_set(d))
423 class InputFormat(object):
425 def __init__(self, context, struct):
426 self.context = context
431 return self.struct.id
434 def description(self):
435 return self.struct.description
437 class OutputFormat(object):
439 def __init__(self, context, struct):
440 self.context = context
445 return self.struct.id
448 def description(self):
449 return self.struct.description
451 class EnumValue(object):
455 def __new__(cls, id):
456 if cls not in cls._enum_values:
457 cls._enum_values[cls] = {}
458 if id not in cls._enum_values[cls]:
459 value = super(EnumValue, cls).__new__(cls)
461 cls._enum_values[cls][id] = value
462 return cls._enum_values[cls][id]
464 class LogLevel(EnumValue):
467 class PacketType(EnumValue):
470 class Quantity(EnumValue):
473 class Unit(EnumValue):
476 class QuantityFlag(EnumValue):
479 def set_from_mask(cls, mask):
482 new_mask = mask & (mask - 1)
483 result.add(cls(mask ^ new_mask))
487 class ConfigKey(EnumValue):
490 class ProbeType(EnumValue):
493 for symbol_name in dir(lowlevel):
495 ('SR_LOG_', LogLevel),
496 ('SR_DF_', PacketType),
497 ('SR_MQ_', Quantity),
499 ('SR_MQFLAG_', QuantityFlag),
500 ('SR_CONF_', ConfigKey),
501 ('SR_PROBE_', ProbeType)]:
502 if symbol_name.startswith(prefix):
503 name = symbol_name[len(prefix):]
504 value = getattr(lowlevel, symbol_name)
505 setattr(cls, name, cls(value))