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.session = None
  91
  92     def __del__(self):
  93         sr_exit(self.struct)
  94
  95     @property
  96     def drivers(self):
  97         if not self._drivers:
  98             self._drivers = {}
  99             driver_list = sr_driver_list()
 100             for i in itertools.count():
 101                 driver_ptr = sr_dev_driver_ptr_array_getitem(driver_list, i)
 102                 if driver_ptr:
 103                     self._drivers[driver_ptr.name] = Driver(self, driver_ptr)
 104                 else:
 105                     break
 106         return self._drivers
 107
 108 class Driver(object):
 109
 110     def __init__(self, context, struct):
 111         self.context = context
 112         self.struct = struct
 113         self._initialized = False
 114
 115     @property
 116     def name(self):
 117         return self.struct.name
 118
 119     def scan(self, **kwargs):
 120         if not self._initialized:
 121             check(sr_driver_init(self.context.struct, self.struct))
 122             self._initialized = True
 123         options = []
 124         for name, value in kwargs.items():
 125             key = getattr(ConfigKey, name.upper())
 126             src = sr_config()
 127             src.key = key.id
 128             src.data = python_to_gvariant(value)
 129             options.append(src.this)
 130         option_list = python_to_gslist(options)
 131         device_list = sr_driver_scan(self.struct, option_list)
 132         g_slist_free(option_list)
 133         devices = [Device(self, gpointer_to_sr_dev_inst_ptr(ptr))
 134             for ptr in gslist_to_python(device_list)]
 135         g_slist_free(device_list)
 136         return devices
 137
 138 class Device(object):
 139
 140     def __new__(cls, driver, struct):
 141         address = int(struct.this)
 142         if address not in driver.context._devices:
 143             device = super(Device, cls).__new__(cls, driver, struct)
 144             driver.context._devices[address] = device
 145         return driver.context._devices[address]
 146
 147     def __init__(self, driver, struct):
 148         self.driver = driver
 149         self.struct = struct
 150         self._probes = None
 151         self._probe_groups = None
 152
 153     def __getattr__(self, name):
 154         key = getattr(ConfigKey, name.upper())
 155         data = new_gvariant_ptr_ptr()
 156         try:
 157             check(sr_config_get(self.driver.struct, self.struct, None,
 158                 key, data))
 159         except Error as error:
 160             if error.errno == SR_ERR_NA:
 161                 raise NotImplementedError(
 162                     "Device does not implement %s" % name)
 163             else:
 164                 raise AttributeError
 165         value = gvariant_ptr_ptr_value(data)
 166         return gvariant_to_python(value)
 167
 168     def __setattr__(self, name, value):
 169         try:
 170             key = getattr(ConfigKey, name.upper())
 171         except AttributeError:
 172             super(Device, self).__setattr__(name, value)
 173             return
 174         check(sr_config_set(self.struct, None, key, python_to_gvariant(value)))
 175
 176     @property
 177     def vendor(self):
 178         return self.struct.vendor
 179
 180     @property
 181     def model(self):
 182         return self.struct.model
 183
 184     @property
 185     def version(self):
 186         return self.struct.version
 187
 188     @property
 189     def probes(self):
 190         if self._probes is None:
 191             self._probes = {}
 192             probe_list = self.struct.probes
 193             while (probe_list):
 194                 probe_ptr = void_ptr_to_sr_probe_ptr(probe_list.data)
 195                 self._probes[probe_ptr.name] = Probe(self, probe_ptr)
 196                 probe_list = probe_list.next
 197         return self._probes
 198
 199     @property
 200     def probe_groups(self):
 201         if self._probe_groups is None:
 202             self._probe_groups = {}
 203             probe_group_list = self.struct.probe_groups
 204             while (probe_group_list):
 205                 probe_group_ptr = void_ptr_to_sr_probe_group_ptr(
 206                     probe_group_list.data)
 207                 self._probe_groups[probe_group_ptr.name] = ProbeGroup(self,
 208                     probe_group_ptr)
 209                 probe_group_list = probe_group_list.next
 210         return self._probe_groups
 211
 212 class Probe(object):
 213
 214     def __init__(self, device, struct):
 215         self.device = device
 216         self.struct = struct
 217
 218     @property
 219     def type(self):
 220         return ProbeType(self.struct.type)
 221
 222     @property
 223     def enabled(self):
 224         return self.struct.enabled
 225
 226     @property
 227     def name(self):
 228         return self.struct.name
 229
 230 class ProbeGroup(object):
 231
 232     def __init__(self, device, struct):
 233         self.device = device
 234         self.struct = struct
 235         self._probes = None
 236
 237     def __iter__(self):
 238         return iter(self.probes)
 239
 240     def __getattr__(self, name):
 241         key = config_key(name)
 242         data = new_gvariant_ptr_ptr()
 243         try:
 244             check(sr_config_get(self.device.driver.struct, self.device.struct,
 245                 self.struct, key, data))
 246         except Error as error:
 247             if error.errno == SR_ERR_NA:
 248                 raise NotImplementedError(
 249                     "Probe group does not implement %s" % name)
 250             else:
 251                 raise AttributeError
 252         value = gvariant_ptr_ptr_value(data)
 253         return gvariant_to_python(value)
 254
 255     def __setattr__(self, name, value):
 256         try:
 257             key = config_key(name)
 258         except AttributeError:
 259             super(ProbeGroup, self).__setattr__(name, value)
 260             return
 261         check(sr_config_set(self.device.struct, self.struct,
 262             key, python_to_gvariant(value)))
 263
 264     @property
 265     def name(self):
 266         return self.struct.name
 267
 268     @property
 269     def probes(self):
 270         if self._probes is None:
 271             self._probes = []
 272             probe_list = self.struct.probes
 273             while (probe_list):
 274                 probe_ptr = void_ptr_to_sr_probe_ptr(probe_list.data)
 275                 self._probes.append(Probe(self, probe_ptr))
 276                 probe_list = probe_list.next
 277         return self._probes
 278
 279 class Session(object):
 280
 281     def __init__(self, context):
 282         assert context.session is None
 283         self.context = context
 284         self.struct = sr_session_new()
 285         context.session = self
 286
 287     def __del__(self):
 288         check(sr_session_destroy())
 289
 290     def add_device(self, device):
 291         check(sr_session_dev_add(device.struct))
 292
 293     def open_device(self, device):
 294         check(sr_dev_open(device.struct))
 295
 296     def add_callback(self, callback):
 297         wrapper = partial(callback_wrapper, self, callback)
 298         check(sr_session_datafeed_python_callback_add(wrapper))
 299
 300     def start(self):
 301         check(sr_session_start())
 302
 303     def run(self):
 304         check(sr_session_run())
 305
 306     def stop(self):
 307         check(sr_session_stop())
 308
 309 class Packet(object):
 310
 311     def __init__(self, session, struct):
 312         self.session = session
 313         self.struct = struct
 314         self._payload = None
 315
 316     @property
 317     def type(self):
 318         return PacketType(self.struct.type)
 319
 320     @property
 321     def payload(self):
 322         if self._payload is None:
 323             pointer = self.struct.payload
 324             if self.type == PacketType.LOGIC:
 325                 self._payload = Logic(self,
 326                     void_ptr_to_sr_datafeed_logic_ptr(pointer))
 327             elif self.type == PacketType.ANALOG:
 328                 self._payload = Analog(self,
 329                     void_ptr_to_sr_datafeed_analog_ptr(pointer))
 330             else:
 331                 raise NotImplementedError(
 332                     "No Python mapping for packet type %s" % self.struct.type)
 333         return self._payload
 334
 335 class Logic(object):
 336
 337     def __init__(self, packet, struct):
 338         self.packet = packet
 339         self.struct = struct
 340         self._data = None
 341
 342     @property
 343     def data(self):
 344         if self._data is None:
 345             self._data = cdata(self.struct.data, self.struct.length)
 346         return self._data
 347
 348 class Analog(object):
 349
 350     def __init__(self, packet, struct):
 351         self.packet = packet
 352         self.struct = struct
 353         self._data = None
 354
 355     @property
 356     def num_samples(self):
 357         return self.struct.num_samples
 358
 359     @property
 360     def mq(self):
 361         return Quantity(self.struct.mq)
 362
 363     @property
 364     def unit(self):
 365         return Unit(self.struct.unit)
 366
 367     @property
 368     def mqflags(self):
 369         return QuantityFlag.set_from_mask(self.struct.mqflags)
 370
 371     @property
 372     def data(self):
 373         if self._data is None:
 374             self._data = float_array.frompointer(self.struct.data)
 375         return self._data
 376
 377 class Log(object):
 378
 379     @property
 380     def level(self):
 381         return LogLevel(sr_log_loglevel_get())
 382
 383     @level.setter
 384     def level(self, l):
 385         check(sr_log_loglevel_set(l.id))
 386
 387     @property
 388     def domain(self):
 389         return sr_log_logdomain_get()
 390
 391     @domain.setter
 392     def domain(self, d):
 393         check(sr_log_logdomain_set(d))
 394
 395 class EnumValue(object):
 396
 397     _enum_values = {}
 398
 399     def __new__(cls, id):
 400         if cls not in cls._enum_values:
 401             cls._enum_values[cls] = {}
 402         if id not in cls._enum_values[cls]:
 403             value = super(EnumValue, cls).__new__(cls)
 404             value.id = id
 405             cls._enum_values[cls][id] = value
 406         return cls._enum_values[cls][id]
 407
 408 class LogLevel(EnumValue):
 409     pass
 410
 411 class PacketType(EnumValue):
 412     pass
 413
 414 class Quantity(EnumValue):
 415     pass
 416
 417 class Unit(EnumValue):
 418     pass
 419
 420 class QuantityFlag(EnumValue):
 421
 422     @classmethod
 423     def set_from_mask(cls, mask):
 424         result = set()
 425         while mask:
 426             new_mask = mask & (mask - 1)
 427             result.add(cls(mask ^ new_mask))
 428             mask = new_mask
 429         return result
 430
 431 class ConfigKey(EnumValue):
 432     pass
 433
 434 class ProbeType(EnumValue):
 435     pass
 436
 437 for symbol_name in dir(lowlevel):
 438     for prefix, cls in [
 439         ('SR_LOG_', LogLevel),
 440         ('SR_DF_', PacketType),
 441         ('SR_MQ_', Quantity),
 442         ('SR_UNIT_', Unit),
 443         ('SR_MQFLAG_', QuantityFlag),
 444         ('SR_CONF_', ConfigKey),
 445         ('SR_PROBE_', ProbeType)]:
 446         if symbol_name.startswith(prefix):
 447             name = symbol_name[len(prefix):]
 448             value = getattr(lowlevel, symbol_name)
 449             setattr(cls, name, cls(value))