bindings/python/sigrok/core/classes.py

   1 ##
   2 ## This file is part of the libsigrok project.
   3 ##
   4 ## Copyright (C) 2013 Martin Ling <martin-sigrok@earth.li>
   5 ##
   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.
  10 ##
  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.
  15 ##
  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/>.
  18 ##
  19
  20 from functools import partial
  21 from fractions import Fraction
  22 from .lowlevel import *
  23 from . import lowlevel
  24 import itertools
  25
  26 __all__ = ['Error', 'Context', 'Driver', 'Device', 'Session', 'Packet', 'Log',
  27     'LogLevel', 'PacketType', 'Quantity', 'Unit', 'QuantityFlag', 'ConfigKey',
  28     'ProbeType', 'Probe', 'ProbeGroup']
  29
  30 class Error(Exception):
  31
  32     def __str__(self):
  33         return sr_strerror(self.args[0])
  34
  35 def check(result):
  36     if result != SR_OK:
  37         raise Error(result)
  38
  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)':
  50         return Fraction(
  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)
  55
  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)
  73         return result
  74     raise NotImplementedError(
  75         "Can't convert Python '%s' to a GVariant." % type(value))
  76
  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)
  81
  82 class Context(object):
  83
  84     def __init__(self):
  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)
  88         self._drivers = None
  89         self._devices = {}
  90         self._input_formats = None
  91         self._output_formats = None
  92         self.session = None
  93
  94     def __del__(self):
  95         sr_exit(self.struct)
  96
  97     @property
  98     def drivers(self):
  99         if not self._drivers:
 100             self._drivers = {}
 101             driver_list = sr_driver_list()
 102             for i in itertools.count():
 103                 driver_ptr = sr_dev_driver_ptr_array_getitem(driver_list, i)
 104                 if driver_ptr:
 105                     self._drivers[driver_ptr.name] = Driver(self, driver_ptr)
 106                 else:
 107                     break
 108         return self._drivers
 109
 110     @property
 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)
 117                 if input_ptr:
 118                     self._input_formats[input_ptr.id] = InputFormat(self, input_ptr)
 119                 else:
 120                     break
 121         return self._input_formats
 122
 123     @property
 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)
 130                 if output_ptr:
 131                     self._output_formats[output_ptr.id] = OutputFormat(self, output_ptr)
 132                 else:
 133                     break
 134         return self._output_formats
 135
 136 class Driver(object):
 137
 138     def __init__(self, context, struct):
 139         self.context = context
 140         self.struct = struct
 141         self._initialized = False
 142
 143     @property
 144     def name(self):
 145         return self.struct.name
 146
 147     def scan(self, **kwargs):
 148         if not self._initialized:
 149             check(sr_driver_init(self.context.struct, self.struct))
 150             self._initialized = True
 151         options = []
 152         for name, value in kwargs.items():
 153             key = getattr(ConfigKey, name.upper())
 154             src = sr_config()
 155             src.key = key.id
 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)
 164         return devices
 165
 166 class Device(object):
 167
 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]
 174
 175     def __init__(self, driver, struct):
 176         self.driver = driver
 177         self.struct = struct
 178         self._probes = None
 179         self._probe_groups = None
 180
 181     def __getattr__(self, name):
 182         key = getattr(ConfigKey, name.upper())
 183         data = new_gvariant_ptr_ptr()
 184         try:
 185             check(sr_config_get(self.driver.struct, self.struct, None,
 186                 key.id, data))
 187         except Error as error:
 188             if error.errno == SR_ERR_NA:
 189                 raise NotImplementedError(
 190                     "Device does not implement %s" % name)
 191             else:
 192                 raise AttributeError
 193         value = gvariant_ptr_ptr_value(data)
 194         return gvariant_to_python(value)
 195
 196     def __setattr__(self, name, value):
 197         try:
 198             key = getattr(ConfigKey, name.upper())
 199         except AttributeError:
 200             super(Device, self).__setattr__(name, value)
 201             return
 202         check(sr_config_set(self.struct, None, key.id, python_to_gvariant(value)))
 203
 204     @property
 205     def vendor(self):
 206         return self.struct.vendor
 207
 208     @property
 209     def model(self):
 210         return self.struct.model
 211
 212     @property
 213     def version(self):
 214         return self.struct.version
 215
 216     @property
 217     def probes(self):
 218         if self._probes is None:
 219             self._probes = {}
 220             probe_list = self.struct.probes
 221             while (probe_list):
 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
 225         return self._probes
 226
 227     @property
 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,
 236                     probe_group_ptr)
 237                 probe_group_list = probe_group_list.next
 238         return self._probe_groups
 239
 240 class Probe(object):
 241
 242     def __init__(self, device, struct):
 243         self.device = device
 244         self.struct = struct
 245
 246     @property
 247     def type(self):
 248         return ProbeType(self.struct.type)
 249
 250     @property
 251     def enabled(self):
 252         return self.struct.enabled
 253
 254     @property
 255     def name(self):
 256         return self.struct.name
 257
 258 class ProbeGroup(object):
 259
 260     def __init__(self, device, struct):
 261         self.device = device
 262         self.struct = struct
 263         self._probes = None
 264
 265     def __iter__(self):
 266         return iter(self.probes)
 267
 268     def __getattr__(self, name):
 269         key = config_key(name)
 270         data = new_gvariant_ptr_ptr()
 271         try:
 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)
 278             else:
 279                 raise AttributeError
 280         value = gvariant_ptr_ptr_value(data)
 281         return gvariant_to_python(value)
 282
 283     def __setattr__(self, name, value):
 284         try:
 285             key = config_key(name)
 286         except AttributeError:
 287             super(ProbeGroup, self).__setattr__(name, value)
 288             return
 289         check(sr_config_set(self.device.struct, self.struct,
 290             key.id, python_to_gvariant(value)))
 291
 292     @property
 293     def name(self):
 294         return self.struct.name
 295
 296     @property
 297     def probes(self):
 298         if self._probes is None:
 299             self._probes = []
 300             probe_list = self.struct.probes
 301             while (probe_list):
 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
 305         return self._probes
 306
 307 class Session(object):
 308
 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
 314
 315     def __del__(self):
 316         check(sr_session_destroy())
 317
 318     def add_device(self, device):
 319         check(sr_session_dev_add(device.struct))
 320
 321     def open_device(self, device):
 322         check(sr_dev_open(device.struct))
 323
 324     def add_callback(self, callback):
 325         wrapper = partial(callback_wrapper, self, callback)
 326         check(sr_session_datafeed_python_callback_add(wrapper))
 327
 328     def start(self):
 329         check(sr_session_start())
 330
 331     def run(self):
 332         check(sr_session_run())
 333
 334     def stop(self):
 335         check(sr_session_stop())
 336
 337 class Packet(object):
 338
 339     def __init__(self, session, struct):
 340         self.session = session
 341         self.struct = struct
 342         self._payload = None
 343
 344     @property
 345     def type(self):
 346         return PacketType(self.struct.type)
 347
 348     @property
 349     def payload(self):
 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))
 358             else:
 359                 raise NotImplementedError(
 360                     "No Python mapping for packet type %s" % self.struct.type)
 361         return self._payload
 362
 363 class Logic(object):
 364
 365     def __init__(self, packet, struct):
 366         self.packet = packet
 367         self.struct = struct
 368         self._data = None
 369
 370     @property
 371     def data(self):
 372         if self._data is None:
 373             self._data = cdata(self.struct.data, self.struct.length)
 374         return self._data
 375
 376 class Analog(object):
 377
 378     def __init__(self, packet, struct):
 379         self.packet = packet
 380         self.struct = struct
 381         self._data = None
 382
 383     @property
 384     def num_samples(self):
 385         return self.struct.num_samples
 386
 387     @property
 388     def mq(self):
 389         return Quantity(self.struct.mq)
 390
 391     @property
 392     def unit(self):
 393         return Unit(self.struct.unit)
 394
 395     @property
 396     def mqflags(self):
 397         return QuantityFlag.set_from_mask(self.struct.mqflags)
 398
 399     @property
 400     def data(self):
 401         if self._data is None:
 402             self._data = float_array.frompointer(self.struct.data)
 403         return self._data
 404
 405 class Log(object):
 406
 407     @property
 408     def level(self):
 409         return LogLevel(sr_log_loglevel_get())
 410
 411     @level.setter
 412     def level(self, l):
 413         check(sr_log_loglevel_set(l.id))
 414
 415     @property
 416     def domain(self):
 417         return sr_log_logdomain_get()
 418
 419     @domain.setter
 420     def domain(self, d):
 421         check(sr_log_logdomain_set(d))
 422
 423 class InputFormat(object):
 424
 425     def __init__(self, context, struct):
 426         self.context = context
 427         self.struct = struct
 428
 429     @property
 430     def id(self):
 431         return self.struct.id
 432
 433     @property
 434     def description(self):
 435         return self.struct.description
 436
 437 class OutputFormat(object):
 438
 439     def __init__(self, context, struct):
 440         self.context = context
 441         self.struct = struct
 442
 443     @property
 444     def id(self):
 445         return self.struct.id
 446
 447     @property
 448     def description(self):
 449         return self.struct.description
 450
 451 class EnumValue(object):
 452
 453     _enum_values = {}
 454
 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)
 460             value.id = id
 461             cls._enum_values[cls][id] = value
 462         return cls._enum_values[cls][id]
 463
 464 class LogLevel(EnumValue):
 465     pass
 466
 467 class PacketType(EnumValue):
 468     pass
 469
 470 class Quantity(EnumValue):
 471     pass
 472
 473 class Unit(EnumValue):
 474     pass
 475
 476 class QuantityFlag(EnumValue):
 477
 478     @classmethod
 479     def set_from_mask(cls, mask):
 480         result = set()
 481         while mask:
 482             new_mask = mask & (mask - 1)
 483             result.add(cls(mask ^ new_mask))
 484             mask = new_mask
 485         return result
 486
 487 class ConfigKey(EnumValue):
 488     pass
 489
 490 class ProbeType(EnumValue):
 491     pass
 492
 493 for symbol_name in dir(lowlevel):
 494     for prefix, cls in [
 495         ('SR_LOG_', LogLevel),
 496         ('SR_DF_', PacketType),
 497         ('SR_MQ_', Quantity),
 498         ('SR_UNIT_', Unit),
 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))