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', 'InputFormat', 'OutputFormat',
  29     'InputFile', 'Output']
  30
  31 class Error(Exception):
  32
  33     def __str__(self):
  34         return sr_strerror(self.args[0])
  35
  36 def check(result):
  37     if result != SR_OK:
  38         raise Error(result)
  39
  40 def gvariant_to_python(value):
  41     type_string = g_variant_get_type_string(value)
  42     if type_string == 't':
  43         return g_variant_get_uint64(value)
  44     if type_string == 'b':
  45         return g_variant_get_bool(value)
  46     if type_string == 'd':
  47         return g_variant_get_double(value)
  48     if type_string == 's':
  49         return g_variant_get_string(value, None)
  50     if type_string == '(tt)':
  51         return Fraction(
  52             g_variant_get_uint64(g_variant_get_child_value(value, 0)),
  53             g_variant_get_uint64(g_variant_get_child_value(value, 1)))
  54     raise NotImplementedError(
  55         "Can't convert GVariant type '%s' to a Python type." % type_string)
  56
  57 def python_to_gvariant(value):
  58     if isinstance(value, int):
  59         return g_variant_new_uint64(value)
  60     if isinstance(value, bool):
  61         return g_variant_new_boolean(value)
  62     if isinstance(value, float):
  63         return g_variant_new_double(value)
  64     if isinstance(value, str):
  65         return g_variant_new_string(value)
  66     if isinstance(value, Fraction):
  67         array = new_gvariant_ptr_array(2)
  68         gvariant_ptr_array_setitem(array, 0,
  69             g_variant_new_uint64(value.numerator))
  70         gvariant_ptr_array_setitem(array, 1,
  71             g_variant_new_uint64(value.denominator))
  72         result = g_variant_new_tuple(array, 2)
  73         delete_gvariant_ptr_array(array)
  74         return result
  75     raise NotImplementedError(
  76         "Can't convert Python '%s' to a GVariant." % type(value))
  77
  78 def callback_wrapper(session, callback, device_ptr, packet_ptr):
  79     device = session.context._devices[int(device_ptr.this)]
  80     packet = Packet(session, packet_ptr)
  81     callback(device, packet)
  82
  83 class Context(object):
  84
  85     def __init__(self):
  86         context_ptr_ptr = new_sr_context_ptr_ptr()
  87         check(sr_init(context_ptr_ptr))
  88         self.struct = sr_context_ptr_ptr_value(context_ptr_ptr)
  89         self._drivers = None
  90         self._devices = {}
  91         self._input_formats = None
  92         self._output_formats = None
  93         self.session = None
  94
  95     def __del__(self):
  96         sr_exit(self.struct)
  97
  98     @property
  99     def drivers(self):
 100         if not self._drivers:
 101             self._drivers = {}
 102             driver_list = sr_driver_list()
 103             for i in itertools.count():
 104                 driver_ptr = sr_dev_driver_ptr_array_getitem(driver_list, i)
 105                 if driver_ptr:
 106                     self._drivers[driver_ptr.name] = Driver(self, driver_ptr)
 107                 else:
 108                     break
 109         return self._drivers
 110
 111     @property
 112     def input_formats(self):
 113         if not self._input_formats:
 114             self._input_formats = {}
 115             input_list = sr_input_list()
 116             for i in itertools.count():
 117                 input_ptr = sr_input_format_ptr_array_getitem(input_list, i)
 118                 if input_ptr:
 119                     self._input_formats[input_ptr.id] = InputFormat(self, input_ptr)
 120                 else:
 121                     break
 122         return self._input_formats
 123
 124     @property
 125     def output_formats(self):
 126         if not self._output_formats:
 127             self._output_formats = {}
 128             output_list = sr_output_list()
 129             for i in itertools.count():
 130                 output_ptr = sr_output_format_ptr_array_getitem(output_list, i)
 131                 if output_ptr:
 132                     self._output_formats[output_ptr.id] = OutputFormat(self, output_ptr)
 133                 else:
 134                     break
 135         return self._output_formats
 136
 137 class Driver(object):
 138
 139     def __init__(self, context, struct):
 140         self.context = context
 141         self.struct = struct
 142         self._initialized = False
 143
 144     @property
 145     def name(self):
 146         return self.struct.name
 147
 148     def scan(self, **kwargs):
 149         if not self._initialized:
 150             check(sr_driver_init(self.context.struct, self.struct))
 151             self._initialized = True
 152         options = []
 153         for name, value in kwargs.items():
 154             key = getattr(ConfigKey, name)
 155             src = sr_config()
 156             src.key = key.id
 157             src.data = python_to_gvariant(value)
 158             options.append(src.this)
 159         option_list = python_to_gslist(options)
 160         device_list = sr_driver_scan(self.struct, option_list)
 161         g_slist_free(option_list)
 162         devices = [HardwareDevice(self, gpointer_to_sr_dev_inst_ptr(ptr))
 163             for ptr in gslist_to_python(device_list)]
 164         g_slist_free(device_list)
 165         return devices
 166
 167 class Device(object):
 168
 169     def __new__(cls, struct, context):
 170         address = int(struct.this)
 171         if address not in context._devices:
 172             device = super(Device, cls).__new__(cls)
 173             device.struct = struct
 174             device.context = context
 175             device._probes = None
 176             device._probe_groups = None
 177             context._devices[address] = device
 178         return context._devices[address]
 179
 180     @property
 181     def vendor(self):
 182         return self.struct.vendor
 183
 184     @property
 185     def model(self):
 186         return self.struct.model
 187
 188     @property
 189     def version(self):
 190         return self.struct.version
 191
 192     @property
 193     def probes(self):
 194         if self._probes is None:
 195             self._probes = {}
 196             probe_list = self.struct.probes
 197             while (probe_list):
 198                 probe_ptr = void_ptr_to_sr_probe_ptr(probe_list.data)
 199                 self._probes[probe_ptr.name] = Probe(self, probe_ptr)
 200                 probe_list = probe_list.next
 201         return self._probes
 202
 203     @property
 204     def probe_groups(self):
 205         if self._probe_groups is None:
 206             self._probe_groups = {}
 207             probe_group_list = self.struct.probe_groups
 208             while (probe_group_list):
 209                 probe_group_ptr = void_ptr_to_sr_probe_group_ptr(
 210                     probe_group_list.data)
 211                 self._probe_groups[probe_group_ptr.name] = ProbeGroup(self,
 212                     probe_group_ptr)
 213                 probe_group_list = probe_group_list.next
 214         return self._probe_groups
 215
 216 class HardwareDevice(Device):
 217
 218     def __new__(cls, driver, struct):
 219         device = Device.__new__(cls, struct, driver.context)
 220         device.driver = driver
 221         return device
 222
 223     def __getattr__(self, name):
 224         key = getattr(ConfigKey, name)
 225         data = new_gvariant_ptr_ptr()
 226         try:
 227             check(sr_config_get(self.driver.struct, self.struct, None,
 228                 key.id, data))
 229         except Error as error:
 230             if error.errno == SR_ERR_NA:
 231                 raise NotImplementedError(
 232                     "Device does not implement %s" % name)
 233             else:
 234                 raise AttributeError
 235         value = gvariant_ptr_ptr_value(data)
 236         return gvariant_to_python(value)
 237
 238     def __setattr__(self, name, value):
 239         try:
 240             key = getattr(ConfigKey, name)
 241         except AttributeError:
 242             super(Device, self).__setattr__(name, value)
 243             return
 244         check(sr_config_set(self.struct, None, key.id, python_to_gvariant(value)))
 245
 246 class Probe(object):
 247
 248     def __init__(self, device, struct):
 249         self.device = device
 250         self.struct = struct
 251
 252     @property
 253     def type(self):
 254         return ProbeType(self.struct.type)
 255
 256     @property
 257     def enabled(self):
 258         return self.struct.enabled
 259
 260     @property
 261     def name(self):
 262         return self.struct.name
 263
 264 class ProbeGroup(object):
 265
 266     def __init__(self, device, struct):
 267         self.device = device
 268         self.struct = struct
 269         self._probes = None
 270
 271     def __iter__(self):
 272         return iter(self.probes)
 273
 274     def __getattr__(self, name):
 275         key = config_key(name)
 276         data = new_gvariant_ptr_ptr()
 277         try:
 278             check(sr_config_get(self.device.driver.struct, self.device.struct,
 279                 self.struct, key.id, data))
 280         except Error as error:
 281             if error.errno == SR_ERR_NA:
 282                 raise NotImplementedError(
 283                     "Probe group does not implement %s" % name)
 284             else:
 285                 raise AttributeError
 286         value = gvariant_ptr_ptr_value(data)
 287         return gvariant_to_python(value)
 288
 289     def __setattr__(self, name, value):
 290         try:
 291             key = config_key(name)
 292         except AttributeError:
 293             super(ProbeGroup, self).__setattr__(name, value)
 294             return
 295         check(sr_config_set(self.device.struct, self.struct,
 296             key.id, python_to_gvariant(value)))
 297
 298     @property
 299     def name(self):
 300         return self.struct.name
 301
 302     @property
 303     def probes(self):
 304         if self._probes is None:
 305             self._probes = []
 306             probe_list = self.struct.probes
 307             while (probe_list):
 308                 probe_ptr = void_ptr_to_sr_probe_ptr(probe_list.data)
 309                 self._probes.append(Probe(self, probe_ptr))
 310                 probe_list = probe_list.next
 311         return self._probes
 312
 313 class Session(object):
 314
 315     def __init__(self, context):
 316         assert context.session is None
 317         self.context = context
 318         self.struct = sr_session_new()
 319         context.session = self
 320
 321     def __del__(self):
 322         check(sr_session_destroy())
 323
 324     def add_device(self, device):
 325         check(sr_session_dev_add(device.struct))
 326
 327     def open_device(self, device):
 328         check(sr_dev_open(device.struct))
 329
 330     def add_callback(self, callback):
 331         wrapper = partial(callback_wrapper, self, callback)
 332         check(sr_session_datafeed_python_callback_add(wrapper))
 333
 334     def start(self):
 335         check(sr_session_start())
 336
 337     def run(self):
 338         check(sr_session_run())
 339
 340     def stop(self):
 341         check(sr_session_stop())
 342
 343 class Packet(object):
 344
 345     def __init__(self, session, struct):
 346         self.session = session
 347         self.struct = struct
 348         self._payload = None
 349
 350     @property
 351     def type(self):
 352         return PacketType(self.struct.type)
 353
 354     @property
 355     def payload(self):
 356         if self._payload is None:
 357             pointer = self.struct.payload
 358             if self.type == PacketType.LOGIC:
 359                 self._payload = Logic(self,
 360                     void_ptr_to_sr_datafeed_logic_ptr(pointer))
 361             elif self.type == PacketType.ANALOG:
 362                 self._payload = Analog(self,
 363                     void_ptr_to_sr_datafeed_analog_ptr(pointer))
 364             else:
 365                 raise NotImplementedError(
 366                     "No Python mapping for packet type %s" % self.struct.type)
 367         return self._payload
 368
 369 class Logic(object):
 370
 371     def __init__(self, packet, struct):
 372         self.packet = packet
 373         self.struct = struct
 374         self._data = None
 375
 376     @property
 377     def data(self):
 378         if self._data is None:
 379             self._data = cdata(self.struct.data, self.struct.length)
 380         return self._data
 381
 382 class Analog(object):
 383
 384     def __init__(self, packet, struct):
 385         self.packet = packet
 386         self.struct = struct
 387         self._data = None
 388
 389     @property
 390     def num_samples(self):
 391         return self.struct.num_samples
 392
 393     @property
 394     def mq(self):
 395         return Quantity(self.struct.mq)
 396
 397     @property
 398     def unit(self):
 399         return Unit(self.struct.unit)
 400
 401     @property
 402     def mqflags(self):
 403         return QuantityFlag.set_from_mask(self.struct.mqflags)
 404
 405     @property
 406     def data(self):
 407         if self._data is None:
 408             self._data = float_array.frompointer(self.struct.data)
 409         return self._data
 410
 411 class Log(object):
 412
 413     @property
 414     def level(self):
 415         return LogLevel(sr_log_loglevel_get())
 416
 417     @level.setter
 418     def level(self, l):
 419         check(sr_log_loglevel_set(l.id))
 420
 421     @property
 422     def domain(self):
 423         return sr_log_logdomain_get()
 424
 425     @domain.setter
 426     def domain(self, d):
 427         check(sr_log_logdomain_set(d))
 428
 429 class InputFormat(object):
 430
 431     def __init__(self, context, struct):
 432         self.context = context
 433         self.struct = struct
 434
 435     @property
 436     def id(self):
 437         return self.struct.id
 438
 439     @property
 440     def description(self):
 441         return self.struct.description
 442
 443     def format_match(self, filename):
 444         return bool(self.struct.call_format_match(filename))
 445
 446 class InputFile(object):
 447
 448     def __init__(self, format, filename, **kwargs):
 449         self.format = format
 450         self.filename = filename
 451         self.struct = sr_input()
 452         self.struct.format = self.format.struct
 453         self.struct.param = g_hash_table_new_full(
 454             g_str_hash_ptr, g_str_equal_ptr, g_free_ptr, g_free_ptr)
 455         for key, value in kwargs.items():
 456             g_hash_table_insert(self.struct.param, g_strdup(key), g_strdup(str(value)))
 457         check(self.format.struct.call_init(self.struct, self.filename))
 458         self.device = InputFileDevice(self)
 459
 460     def load(self):
 461         check(self.format.struct.call_loadfile(self.struct, self.filename))
 462
 463     def __del__(self):
 464         g_hash_table_destroy(self.struct.param)
 465
 466 class InputFileDevice(Device):
 467
 468     def __new__(cls, file):
 469         device = Device.__new__(cls, file.struct.sdi, file.format.context)
 470         device.file = file
 471         return device
 472
 473 class OutputFormat(object):
 474
 475     def __init__(self, context, struct):
 476         self.context = context
 477         self.struct = struct
 478
 479     @property
 480     def id(self):
 481         return self.struct.id
 482
 483     @property
 484     def description(self):
 485         return self.struct.description
 486
 487 class Output(object):
 488
 489     def __init__(self, format, device, param=None):
 490         self.format = format
 491         self.device = device
 492         self.param = param
 493         self.struct = sr_output()
 494         self.struct.format = self.format.struct
 495         self.struct.sdi = self.device.struct
 496         self.struct.param = param
 497         check(self.format.struct.call_init(self.struct))
 498
 499     def receive(self, packet):
 500
 501         output_buf_ptr = new_uint8_ptr_ptr()
 502         output_len_ptr = new_uint64_ptr()
 503         using_obsolete_api = False
 504
 505         if self.format.struct.event and packet.type in (
 506                 PacketType.TRIGGER, PacketType.FRAME_BEGIN,
 507                 PacketType.FRAME_END, PacketType.END):
 508             check(self.format.struct.call_event(self.struct, packet.type.id,
 509                 output_buf_ptr, output_len_ptr))
 510             using_obsolete_api = True
 511         elif self.format.struct.data and packet.type.id == self.format.struct.df_type:
 512             check(self.format.struct.call_data(self.struct,
 513                 packet.payload.struct.data, packet.payload.struct.length,
 514                 output_buf_ptr, output_len_ptr))
 515             using_obsolete_api = True
 516
 517         if using_obsolete_api:
 518             output_buf = uint8_ptr_ptr_value(output_buf_ptr)
 519             output_len = uint64_ptr_value(output_len_ptr)
 520             result = cdata(output_buf, output_len)
 521             g_free(output_buf)
 522             return result
 523
 524         if self.format.struct.receive:
 525             out_ptr = new_gstring_ptr_ptr()
 526             check(self.format.struct.call_receive(self.struct, self.device.struct,
 527                 packet.struct, out_ptr))
 528             out = gstring_ptr_ptr_value(out_ptr)
 529             if out:
 530                 result = out.str
 531                 g_string_free(out, True)
 532                 return result
 533
 534         return None
 535
 536     def __del__(self):
 537         check(self.format.struct.call_cleanup(self.struct))
 538
 539 class ConfigInfo(object):
 540
 541     def __new__(cls, key):
 542         struct = sr_config_info_get(key.id)
 543         if not struct:
 544             return None
 545         obj = super(ConfigInfo, cls).__new__(cls)
 546         obj.key = key
 547         obj.struct = struct
 548         return obj
 549
 550     @property
 551     def datatype(self):
 552         return DataType(self.struct.datatype)
 553
 554     @property
 555     def id(self):
 556         return self.struct.id
 557
 558     @property
 559     def name(self):
 560         return self.struct.name
 561
 562     @property
 563     def description(self):
 564         return self.struct.description
 565
 566 class EnumValue(object):
 567
 568     _enum_values = {}
 569
 570     def __new__(cls, id):
 571         if cls not in cls._enum_values:
 572             cls._enum_values[cls] = {}
 573         if id not in cls._enum_values[cls]:
 574             value = super(EnumValue, cls).__new__(cls)
 575             value.id = id
 576             cls._enum_values[cls][id] = value
 577         return cls._enum_values[cls][id]
 578
 579 class LogLevel(EnumValue):
 580     pass
 581
 582 class PacketType(EnumValue):
 583     pass
 584
 585 class Quantity(EnumValue):
 586     pass
 587
 588 class Unit(EnumValue):
 589     pass
 590
 591 class QuantityFlag(EnumValue):
 592
 593     @classmethod
 594     def set_from_mask(cls, mask):
 595         result = set()
 596         while mask:
 597             new_mask = mask & (mask - 1)
 598             result.add(cls(mask ^ new_mask))
 599             mask = new_mask
 600         return result
 601
 602 class ConfigKey(EnumValue):
 603     pass
 604
 605 class DataType(EnumValue):
 606     pass
 607
 608 class ProbeType(EnumValue):
 609     pass
 610
 611 for symbol_name in dir(lowlevel):
 612     for prefix, cls in [
 613         ('SR_LOG_', LogLevel),
 614         ('SR_DF_', PacketType),
 615         ('SR_MQ_', Quantity),
 616         ('SR_UNIT_', Unit),
 617         ('SR_MQFLAG_', QuantityFlag),
 618         ('SR_CONF_', ConfigKey),
 619         ('SR_T_', DataType),
 620         ('SR_PROBE_', ProbeType)]:
 621         if symbol_name.startswith(prefix):
 622             name = symbol_name[len(prefix):]
 623             value = getattr(lowlevel, symbol_name)
 624             obj = cls(value)
 625             setattr(cls, name, obj)
 626             if cls is ConfigKey:
 627                 obj.info = ConfigInfo(obj)
 628                 if obj.info:
 629                     setattr(cls, obj.info.id, obj)
 630                 else:
 631                     setattr(cls, name.lower(), obj)