libsigrok/bindings/python/sigrok/core/classes.py

##
## This file is part of the libsigrok project.
##
## Copyright (C) 2013 Martin Ling <martin-sigrok@earth.li>
##
## This program is free software: you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program.  If not, see <http://www.gnu.org/licenses/>.
##

from functools import partial
from fractions import Fraction
from collections import OrderedDict
from .lowlevel import *
from . import lowlevel
import itertools

__all__ = ['Error', 'Context', 'Driver', 'Device', 'Session', 'Packet', 'Log',
    'LogLevel', 'PacketType', 'Quantity', 'Unit', 'QuantityFlag', 'ConfigKey',
    'ProbeType', 'Probe', 'ChannelGroup', 'InputFormat', 'OutputFormat',
    'InputFile', 'Output']

class Error(Exception):

    def __str__(self):
        return sr_strerror(self.args[0])

def check(result):
    if result != SR_OK:
        raise Error(result)

def gvariant_to_python(value):
    type_string = g_variant_get_type_string(value)
    if type_string == 't':
        return g_variant_get_uint64(value)
    if type_string == 'b':
        return g_variant_get_bool(value)
    if type_string == 'd':
        return g_variant_get_double(value)
    if type_string == 's':
        return g_variant_get_string(value, None)
    if type_string == '(tt)':
        return Fraction(
            g_variant_get_uint64(g_variant_get_child_value(value, 0)),
            g_variant_get_uint64(g_variant_get_child_value(value, 1)))
    raise NotImplementedError(
        "Can't convert GVariant type '%s' to a Python type." % type_string)

def python_to_gvariant(value):
    if isinstance(value, int):
        return g_variant_new_uint64(value)
    if isinstance(value, bool):
        return g_variant_new_boolean(value)
    if isinstance(value, float):
        return g_variant_new_double(value)
    if isinstance(value, str):
        return g_variant_new_string(value)
    if isinstance(value, Fraction):
        array = new_gvariant_ptr_array(2)
        gvariant_ptr_array_setitem(array, 0,
            g_variant_new_uint64(value.numerator))
        gvariant_ptr_array_setitem(array, 1,
            g_variant_new_uint64(value.denominator))
        result = g_variant_new_tuple(array, 2)
        delete_gvariant_ptr_array(array)
        return result
    raise NotImplementedError(
        "Can't convert Python '%s' to a GVariant." % type(value))

def callback_wrapper(session, callback, device_ptr, packet_ptr):
    device = session.context._devices[int(device_ptr.this)]
    packet = Packet(session, packet_ptr)
    callback(device, packet)

class Context(object):

    def __init__(self):
        context_ptr_ptr = new_sr_context_ptr_ptr()
        check(sr_init(context_ptr_ptr))
        self.struct = sr_context_ptr_ptr_value(context_ptr_ptr)
        self._drivers = None
        self._devices = {}
        self._input_formats = None
        self._output_formats = None
        self.session = None

    def __del__(self):
        sr_exit(self.struct)

    @property
    def drivers(self):
        if not self._drivers:
            self._drivers = {}
            driver_list = sr_driver_list()
            for i in itertools.count():
                driver_ptr = sr_dev_driver_ptr_array_getitem(driver_list, i)
                if driver_ptr:
                    self._drivers[driver_ptr.name] = Driver(self, driver_ptr)
                else:
                    break
        return self._drivers

    @property
    def input_formats(self):
        if not self._input_formats:
            self._input_formats = OrderedDict()
            input_list = sr_input_list()
            for i in itertools.count():
                input_ptr = sr_input_format_ptr_array_getitem(input_list, i)
                if input_ptr:
                    self._input_formats[input_ptr.id] = InputFormat(self, input_ptr)
                else:
                    break
        return self._input_formats

    @property
    def output_formats(self):
        if not self._output_formats:
            self._output_formats = {}
            output_list = sr_output_list()
            for i in itertools.count():
                output_ptr = sr_output_format_ptr_array_getitem(output_list, i)
                if output_ptr:
                    self._output_formats[output_ptr.id] = OutputFormat(self, output_ptr)
                else:
                    break
        return self._output_formats

class Driver(object):

    def __init__(self, context, struct):
        self.context = context
        self.struct = struct
        self._initialized = False

    @property
    def name(self):
        return self.struct.name

    def scan(self, **kwargs):
        if not self._initialized:
            check(sr_driver_init(self.context.struct, self.struct))
            self._initialized = True
        options = []
        for name, value in kwargs.items():
            key = getattr(ConfigKey, name)
            src = sr_config()
            src.key = key.id
            src.data = python_to_gvariant(value)
            options.append(src.this)
        option_list = python_to_gslist(options)
        device_list = sr_driver_scan(self.struct, option_list)
        g_slist_free(option_list)
        devices = [HardwareDevice(self, gpointer_to_sr_dev_inst_ptr(ptr))
            for ptr in gslist_to_python(device_list)]
        g_slist_free(device_list)
        return devices

class Device(object):

    def __new__(cls, struct, context):
        address = int(struct.this)
        if address not in context._devices:
            device = super(Device, cls).__new__(cls)
            device.struct = struct
            device.context = context
            device._probes = None
            device._channel_groups = None
            context._devices[address] = device
        return context._devices[address]

    @property
    def vendor(self):
        return self.struct.vendor

    @property
    def model(self):
        return self.struct.model

    @property
    def version(self):
        return self.struct.version

    @property
    def probes(self):
        if self._probes is None:
            self._probes = {}
            probe_list = self.struct.probes
            while (probe_list):
                probe_ptr = void_ptr_to_sr_channel_ptr(probe_list.data)
                self._probes[probe_ptr.name] = Probe(self, probe_ptr)
                probe_list = probe_list.next
        return self._probes

    @property
    def channel_groups(self):
        if self._channel_groups is None:
            self._channel_groups = {}
            channel_group_list = self.struct.channel_groups
            while (channel_group_list):
                channel_group_ptr = void_ptr_to_sr_channel_group_ptr(
                    channel_group_list.data)
                self._channel_groups[channel_group_ptr.name] = ChannelGroup(self,
                    channel_group_ptr)
                channel_group_list = channel_group_list.next
        return self._channel_groups

class HardwareDevice(Device):

    def __new__(cls, driver, struct):
        device = Device.__new__(cls, struct, driver.context)
        device.driver = driver
        return device

    def __getattr__(self, name):
        key = getattr(ConfigKey, name)
        data = new_gvariant_ptr_ptr()
        try:
            check(sr_config_get(self.driver.struct, self.struct, None,
                key.id, data))
        except Error as error:
            if error.errno == SR_ERR_NA:
                raise NotImplementedError(
                    "Device does not implement %s" % name)
            else:
                raise AttributeError
        value = gvariant_ptr_ptr_value(data)
        return gvariant_to_python(value)

    def __setattr__(self, name, value):
        try:
            key = getattr(ConfigKey, name)
        except AttributeError:
            super(Device, self).__setattr__(name, value)
            return
        check(sr_config_set(self.struct, None, key.id, python_to_gvariant(value)))

class Probe(object):

    def __init__(self, device, struct):
        self.device = device
        self.struct = struct

    @property
    def type(self):
        return ProbeType(self.struct.type)

    @property
    def enabled(self):
        return self.struct.enabled

    @property
    def name(self):
        return self.struct.name

class ChannelGroup(object):

    def __init__(self, device, struct):
        self.device = device
        self.struct = struct
        self._channels = None

    def __iter__(self):
        return iter(self.channels)

    def __getattr__(self, name):
        key = config_key(name)
        data = new_gvariant_ptr_ptr()
        try:
            check(sr_config_get(self.device.driver.struct, self.device.struct,
                self.struct, key.id, data))
        except Error as error:
            if error.errno == SR_ERR_NA:
                raise NotImplementedError(
                    "Channel group does not implement %s" % name)
            else:
                raise AttributeError
        value = gvariant_ptr_ptr_value(data)
        return gvariant_to_python(value)

    def __setattr__(self, name, value):
        try:
            key = config_key(name)
        except AttributeError:
            super(ChannelGroup, self).__setattr__(name, value)
            return
        check(sr_config_set(self.device.struct, self.struct,
            key.id, python_to_gvariant(value)))

    @property
    def name(self):
        return self.struct.name

    @property
    def channels(self):
        if self._channels is None:
            self._channels = []
            channel_list = self.struct.channels
            while (channel_list):
                channel_ptr = void_ptr_to_sr_channel_ptr(channel_list.data)
                self._channels.append(Probe(self, probe_ptr))
                channel_list = channel_list.next
        return self._channels

class Session(object):

    def __init__(self, context):
        assert context.session is None
        self.context = context
        self.struct = sr_session_new()
        context.session = self

    def __del__(self):
        check(sr_session_destroy())

    def add_device(self, device):
        check(sr_session_dev_add(device.struct))

    def open_device(self, device):
        check(sr_dev_open(device.struct))

    def add_callback(self, callback):
        wrapper = partial(callback_wrapper, self, callback)
        check(sr_session_datafeed_python_callback_add(wrapper))

    def start(self):
        check(sr_session_start())

    def run(self):
        check(sr_session_run())

    def stop(self):
        check(sr_session_stop())

class Packet(object):

    def __init__(self, session, struct):
        self.session = session
        self.struct = struct
        self._payload = None

    @property
    def type(self):
        return PacketType(self.struct.type)

    @property
    def payload(self):
        if self._payload is None:
            pointer = self.struct.payload
            if self.type == PacketType.LOGIC:
                self._payload = Logic(self,
                    void_ptr_to_sr_datafeed_logic_ptr(pointer))
            elif self.type == PacketType.ANALOG:
                self._payload = Analog(self,
                    void_ptr_to_sr_datafeed_analog_ptr(pointer))
            else:
                raise NotImplementedError(
                    "No Python mapping for packet type %s" % self.struct.type)
        return self._payload

class Logic(object):

    def __init__(self, packet, struct):
        self.packet = packet
        self.struct = struct
        self._data = None

    @property
    def data(self):
        if self._data is None:
            self._data = cdata(self.struct.data, self.struct.length)
        return self._data

class Analog(object):

    def __init__(self, packet, struct):
        self.packet = packet
        self.struct = struct
        self._data = None

    @property
    def num_samples(self):
        return self.struct.num_samples

    @property
    def mq(self):
        return Quantity(self.struct.mq)

    @property
    def unit(self):
        return Unit(self.struct.unit)

    @property
    def mqflags(self):
        return QuantityFlag.set_from_mask(self.struct.mqflags)

    @property
    def data(self):
        if self._data is None:
            self._data = float_array.frompointer(self.struct.data)
        return self._data

class Log(object):

    @property
    def level(self):
        return LogLevel(sr_log_loglevel_get())

    @level.setter
    def level(self, l):
        check(sr_log_loglevel_set(l.id))

    @property
    def domain(self):
        return sr_log_logdomain_get()

    @domain.setter
    def domain(self, d):
        check(sr_log_logdomain_set(d))

class InputFormat(object):

    def __init__(self, context, struct):
        self.context = context
        self.struct = struct

    @property
    def id(self):
        return self.struct.id

    @property
    def description(self):
        return self.struct.description

    def format_match(self, filename):
        return bool(self.struct.call_format_match(filename))

class InputFile(object):

    def __init__(self, format, filename, **kwargs):
        self.format = format
        self.filename = filename
        self.struct = sr_input()
        self.struct.format = self.format.struct
        self.struct.param = g_hash_table_new_full(
            g_str_hash_ptr, g_str_equal_ptr, g_free_ptr, g_free_ptr)
        for key, value in kwargs.items():
            g_hash_table_insert(self.struct.param, g_strdup(key), g_strdup(str(value)))
        check(self.format.struct.call_init(self.struct, self.filename))
        self.device = InputFileDevice(self)

    def load(self):
        check(self.format.struct.call_loadfile(self.struct, self.filename))

    def __del__(self):
        g_hash_table_destroy(self.struct.param)

class InputFileDevice(Device):

    def __new__(cls, file):
        device = Device.__new__(cls, file.struct.sdi, file.format.context)
        device.file = file
        return device

class OutputFormat(object):

    def __init__(self, context, struct):
        self.context = context
        self.struct = struct

    @property
    def id(self):
        return self.struct.id

    @property
    def description(self):
        return self.struct.description

class Output(object):

    def __init__(self, format, device, param=None):
        self.format = format
        self.device = device
        self.param = param
        self.struct = sr_output()
        self.struct.format = self.format.struct
        self.struct.sdi = self.device.struct
        self.struct.param = param
        check(self.format.struct.call_init(self.struct))

    def receive(self, packet):

        output_buf_ptr = new_uint8_ptr_ptr()
        output_len_ptr = new_uint64_ptr()
        using_obsolete_api = False

        if self.format.struct.event and packet.type in (
                PacketType.TRIGGER, PacketType.FRAME_BEGIN,
                PacketType.FRAME_END, PacketType.END):
            check(self.format.struct.call_event(self.struct, packet.type.id,
                output_buf_ptr, output_len_ptr))
            using_obsolete_api = True
        elif self.format.struct.data and packet.type.id == self.format.struct.df_type:
            check(self.format.struct.call_data(self.struct,
                packet.payload.struct.data, packet.payload.struct.length,
                output_buf_ptr, output_len_ptr))
            using_obsolete_api = True

        if using_obsolete_api:
            output_buf = uint8_ptr_ptr_value(output_buf_ptr)
            output_len = uint64_ptr_value(output_len_ptr)
            result = cdata(output_buf, output_len)
            g_free(output_buf)
            return result

        if self.format.struct.receive:
            out_ptr = new_gstring_ptr_ptr()
            check(self.format.struct.call_receive(self.struct, self.device.struct,
                packet.struct, out_ptr))
            out = gstring_ptr_ptr_value(out_ptr)
            if out:
                result = out.str
                g_string_free(out, True)
                return result

        return None

    def __del__(self):
        check(self.format.struct.call_cleanup(self.struct))

class ConfigInfo(object):

    def __new__(cls, key):
        struct = sr_config_info_get(key.id)
        if not struct:
            return None
        obj = super(ConfigInfo, cls).__new__(cls)
        obj.key = key
        obj.struct = struct
        return obj

    @property
    def datatype(self):
        return DataType(self.struct.datatype)

    @property
    def id(self):
        return self.struct.id

    @property
    def name(self):
        return self.struct.name

    @property
    def description(self):
        return self.struct.description

class EnumValue(object):

    _enum_values = {}

    def __new__(cls, id):
        if cls not in cls._enum_values:
            cls._enum_values[cls] = {}
        if id not in cls._enum_values[cls]:
            value = super(EnumValue, cls).__new__(cls)
            value.id = id
            cls._enum_values[cls][id] = value
        return cls._enum_values[cls][id]

class LogLevel(EnumValue):
    pass

class PacketType(EnumValue):
    pass

class Quantity(EnumValue):
    pass

class Unit(EnumValue):
    pass

class QuantityFlag(EnumValue):

    @classmethod
    def set_from_mask(cls, mask):
        result = set()
        while mask:
            new_mask = mask & (mask - 1)
            result.add(cls(mask ^ new_mask))
            mask = new_mask
        return result

class ConfigKey(EnumValue):
    pass

class DataType(EnumValue):
    pass

class ProbeType(EnumValue):
    pass

for symbol_name in dir(lowlevel):
    for prefix, cls in [
        ('SR_LOG_', LogLevel),
        ('SR_DF_', PacketType),
        ('SR_MQ_', Quantity),
        ('SR_UNIT_', Unit),
        ('SR_MQFLAG_', QuantityFlag),
        ('SR_CONF_', ConfigKey),
        ('SR_T_', DataType),
        ('SR_PROBE_', ProbeType)]:
        if symbol_name.startswith(prefix):
            name = symbol_name[len(prefix):]
            value = getattr(lowlevel, symbol_name)
            obj = cls(value)
            setattr(cls, name, obj)
            if cls is ConfigKey:
                obj.info = ConfigInfo(obj)
                if obj.info:
                    setattr(cls, obj.info.id, obj)
                else:
                    setattr(cls, name.lower(), obj)