bindings/python/sigrok/core/classes.py

   1 ##
   2 ## This file is part of the sigrok 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']
  28
  29 class Error(Exception):
  30
  31     def __str__(self):
  32         return sr_strerror(self.args[0])
  33
  34 def check(result):
  35     if result != SR_OK:
  36         raise Error(result)
  37
  38 def config_key(name):
  39     if not name.lower() == name:
  40         raise AttributeError
  41     key_name = "SR_CONF_" + name.upper()
  42     return getattr(lowlevel, key_name)
  43
  44 def gvariant_to_python(value):
  45     type_string = g_variant_get_type_string(value)
  46     if type_string == 't':
  47         return g_variant_get_uint64(value)
  48     if type_string == 'b':
  49         return g_variant_get_bool(value)
  50     if type_string == 'd':
  51         return g_variant_get_double(value)
  52     if type_string == 's':
  53         return g_variant_get_string(value, None)
  54     if type_string == '(tt)':
  55         return Fraction(
  56             g_variant_get_uint64(g_variant_get_child_value(value, 0)),
  57             g_variant_get_uint64(g_variant_get_child_value(value, 1)))
  58     raise NotImplementedError(
  59         "Can't convert GVariant type '%s' to a Python type." % type_string)
  60
  61 def python_to_gvariant(value):
  62     if isinstance(value, int):
  63         return g_variant_new_uint64(value)
  64     if isinstance(value, bool):
  65         return g_variant_new_boolean(value)
  66     if isinstance(value, float):
  67         return g_variant_new_double(value)
  68     if isinstance(value, str):
  69         return g_variant_new_string(value)
  70     if isinstance(value, Fraction):
  71         array = new_gvariant_ptr_array(2)
  72         gvariant_ptr_array_setitem(array, 0, value.numerator)
  73         gvariant_ptr_array_setitem(array, 1, value.denominator)
  74         result = g_variant_new_tuple(array, 2)
  75         delete_gvariant_ptr_array(array)
  76         return result
  77     raise NotImplementedError(
  78         "Can't convert Python '%s' to a GVariant." % type(value))
  79
  80 def callback_wrapper(session, callback, device_ptr, packet_ptr):
  81     device = session.context._devices[int(device_ptr.this)]
  82     packet = Packet(session, packet_ptr)
  83     callback(device, packet)
  84
  85 class Context(object):
  86
  87     def __init__(self):
  88         context_ptr_ptr = new_sr_context_ptr_ptr()
  89         check(sr_init(context_ptr_ptr))
  90         self.struct = sr_context_ptr_ptr_value(context_ptr_ptr)
  91         self._drivers = None
  92         self._devices = {}
  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 class Driver(object):
 112
 113     def __init__(self, context, struct):
 114         self.context = context
 115         self.struct = struct
 116         self._initialized = False
 117
 118     @property
 119     def name(self):
 120         return self.struct.name
 121
 122     def scan(self):
 123         if not self._initialized:
 124             check(sr_driver_init(self.context.struct, self.struct))
 125             self._initialized = True
 126         devices = []
 127         device_list = sr_driver_scan(self.struct, None)
 128         device_list_item = device_list
 129         while device_list_item:
 130             ptr = device_list_item.data
 131             device_ptr = gpointer_to_sr_dev_inst_ptr(ptr)
 132             devices.append(Device(self, device_ptr))
 133             device_list_item = device_list_item.next
 134         g_slist_free(device_list)
 135         return devices
 136
 137 class Device(object):
 138
 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]
 145
 146     def __init__(self, driver, struct):
 147         self.driver = driver
 148         self.struct = struct
 149
 150     def __getattr__(self, name):
 151         key = config_key(name)
 152         data = new_gvariant_ptr_ptr()
 153         try:
 154             check(sr_config_get(self.driver.struct, key, data, self.struct))
 155         except Error as error:
 156             if error.errno == SR_ERR_NA:
 157                 raise NotImplementedError(
 158                     "Device does not implement %s" % name)
 159             else:
 160                 raise AttributeError
 161         value = gvariant_ptr_ptr_value(data)
 162         return gvariant_to_python(value)
 163
 164     def __setattr__(self, name, value):
 165         try:
 166             key = config_key(name)
 167         except AttributeError:
 168             super(Device, self).__setattr__(name, value)
 169             return
 170         check(sr_config_set(self.struct, key, python_to_gvariant(value)))
 171
 172     @property
 173     def vendor(self):
 174         return self.struct.vendor
 175
 176     @property
 177     def model(self):
 178         return self.struct.model
 179
 180     @property
 181     def version(self):
 182         return self.struct.version
 183
 184 class Session(object):
 185
 186     def __init__(self, context):
 187         assert context.session is None
 188         self.context = context
 189         self.struct = sr_session_new()
 190         context.session = self
 191
 192     def __del__(self):
 193         check(sr_session_destroy())
 194
 195     def add_device(self, device):
 196         check(sr_session_dev_add(device.struct))
 197
 198     def add_callback(self, callback):
 199         wrapper = partial(callback_wrapper, self, callback)
 200         check(sr_session_datafeed_python_callback_add(wrapper))
 201
 202     def start(self):
 203         check(sr_session_start())
 204
 205     def run(self):
 206         check(sr_session_run())
 207
 208     def stop(self):
 209         check(sr_session_stop())
 210
 211 class Packet(object):
 212
 213     def __init__(self, session, struct):
 214         self.session = session
 215         self.struct = struct
 216         self._payload = None
 217
 218     @property
 219     def type(self):
 220         return PacketType(self.struct.type)
 221
 222     @property
 223     def payload(self):
 224         if self._payload is None:
 225             pointer = self.struct.payload
 226             if self.type == PacketType.LOGIC:
 227                 self._payload = Logic(self,
 228                     void_ptr_to_sr_datafeed_logic_ptr(pointer))
 229             elif self.type == PacketType.ANALOG:
 230                 self._payload = Analog(self,
 231                     void_ptr_to_sr_datafeed_analog_ptr(pointer))
 232             else:
 233                 raise NotImplementedError(
 234                     "No Python mapping for packet type %ѕ" % self.struct.type)
 235         return self._payload
 236
 237 class Logic(object):
 238
 239     def __init__(self, packet, struct):
 240         self.packet = packet
 241         self.struct = struct
 242         self._data = None
 243
 244     @property
 245     def data(self):
 246         if self._data is None:
 247             self._data = cdata(self.struct.data, self.struct.length)
 248         return self._data
 249
 250 class Analog(object):
 251
 252     def __init__(self, packet, struct):
 253         self.packet = packet
 254         self.struct = struct
 255         self._data = None
 256
 257     @property
 258     def num_samples(self):
 259         return self.struct.num_samples
 260
 261     @property
 262     def mq(self):
 263         return Quantity(self.struct.mq)
 264
 265     @property
 266     def unit(self):
 267         return Unit(self.struct.unit)
 268
 269     @property
 270     def mqflags(self):
 271         return QuantityFlag.set_from_mask(self.struct.mqflags)
 272
 273     @property
 274     def data(self):
 275         if self._data is None:
 276             self._data = float_array.frompointer(self.struct.data)
 277         return self._data
 278
 279 class Log(object):
 280
 281     @property
 282     def level(self):
 283         return LogLevel(sr_log_loglevel_get())
 284
 285     @level.setter
 286     def level(self, l):
 287         check(sr_log_loglevel_set(l.id))
 288
 289     @property
 290     def domain(self):
 291         return sr_log_logdomain_get()
 292
 293     @domain.setter
 294     def domain(self, d):
 295         check(sr_log_logdomain_set(d))
 296
 297 class EnumValue(object):
 298
 299     _enum_values = {}
 300
 301     def __new__(cls, id):
 302         if cls not in cls._enum_values:
 303             cls._enum_values[cls] = {}
 304         if id not in cls._enum_values[cls]:
 305             value = super(EnumValue, cls).__new__(cls)
 306             value.id = id
 307             cls._enum_values[cls][id] = value
 308         return cls._enum_values[cls][id]
 309
 310 class LogLevel(EnumValue):
 311     pass
 312
 313 class PacketType(EnumValue):
 314     pass
 315
 316 class Quantity(EnumValue):
 317     pass
 318
 319 class Unit(EnumValue):
 320     pass
 321
 322 class QuantityFlag(EnumValue):
 323
 324     @classmethod
 325     def set_from_mask(cls, mask):
 326         result = set()
 327         while mask:
 328             new_mask = mask & (mask - 1)
 329             result.add(cls(mask ^ new_mask))
 330             mask = new_mask
 331         return result
 332
 333 for symbol_name in dir(lowlevel):
 334     for prefix, cls in [
 335         ('SR_LOG_', LogLevel),
 336         ('SR_DF_', PacketType),
 337         ('SR_MQ_', Quantity),
 338         ('SR_UNIT_', Unit),
 339         ('SR_MQFLAG_', QuantityFlag)]:
 340         if symbol_name.startswith(prefix):
 341             name = symbol_name[len(prefix):]
 342             value = getattr(lowlevel, symbol_name)
 343             setattr(cls, name, cls(value))