Mercurial > pygpibtoolkit / file revision

# -*- coding: utf-8 -*-
# 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 2 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, write to the Free Software Foundation, Inc.,
# 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
""" Copyright (c) 2007-2018 David Douard (Paris, FRANCE).
http://www.logilab.org/project/pygpibtoolkit -- mailto:david.douard@sdfa3.org

Several utility functions for GPIB data conversions
"""
import struct
import re
import numpy


########################################
# internal binary types decoders

def decode_float(s):
    """
    Decode a Float from the HP binary representation.
    """
    if len(s) == 4:
        i1, i2, e = struct.unpack('>hbb', s)
        i2 = i2 * 2**(-23)
        return (i1/32768. + i2)*2**e
    else:
        i1, i2, i3, i4, e = struct.unpack('>hhhbb', s)
        if i2 < 0:
            i2 = (i2+32768.)*2**(-15)
        else:
            i2 = i2*2**(-15)
        if i3 < 0:
            i3 = (i3+32768.)*2**(-30)
        else:
            i3 = i3*2**(-15)
        i4 = i4 * 2**(-38)
        return (i1+i2+i3+i4)*2**(e-15)


def decode_string(s):
    """
    Decode a string from the HP binay representation.
    """
    nb = s[0]
    s = s[1:nb+2]
    assert s[-1] == 0
    s = s[:-1]
    return ''.join(chr(c & 0x7F) for c in s)


###
# Datablock useful functions
def format_datablock_header(header, head_struct, columns=80):
    """
    Pretty print a data block (trace, state or coord)
    """
    bool_re = re.compile(
        r'((?P<before>.*) )?(?P<flag>\w+/\w+)( (?P<after>.*))?')

    todisp = []
    for row in head_struct:
        pname = row[0]
        key = row[1]
        typ = row[2]
        if typ is None:
            continue
        val = header.get(key, "N/A")
        if isinstance(val, str):
            val = repr(val)
        elif typ is bool and isinstance(val, typ):
            m = bool_re.match(key)
            if m:
                d = m.groupdict()
                key = ""
                if d['before']:
                    key += d['before']
                if d['after']:
                    key += d['after']
                key = key.capitalize()
                val = d['flag'].split('/')[not val]
            else:
                val = str(val)
        else:
            val = str(val)
        todisp.append((key + ":", val))
    maxk = max([len(k) for k, v in todisp])
    maxv = max([len(v) for k, v in todisp])
    fmt = "%%-%ds %%-%ds" % (maxk, maxv)
    w = maxk + maxv + 4
    ncols = columns // w
    if ncols:
        nrows = len(todisp) // ncols
    else:
        nrows = len(todisp)
        ncols = 1
    res = []
    for i in range(nrows):
        res.append(("| ".join(
            fmt % todisp[j * nrows + i] for j in range(ncols))).rstrip())
    return '\n'.join(res)


def decode_datablock_header(data, header_struct, idx=0):
    d = data
    if d[idx:idx + 1] == b'#':
        # we have a preliminary header here...
        typ = d[idx:idx + 2]
        assert typ == b"#A"
        idx += 2
        totlen = struct.unpack('>h', d[idx:idx+2])[0]
        idx += 2
        print('  header at %s for %s' % (idx, totlen))
    tt = 0
    header = {}
    for i, (cmd, nam, dtype, fmt, nbytes) in enumerate(header_struct):
        if dtype is None:
            idx += nbytes
            continue
        elif dtype == str:
            val = decode_string(d[idx:])
        else:
            if fmt:
                v = struct.unpack('>'+fmt, d[idx: idx+nbytes])[0]
                if isinstance(dtype, dict):
                    val = dtype.get(int(v), "N/A")
                else:
                    val = dtype(v)
            else:
                val = dtype(d[idx: idx+nbytes])
        header[nam] = val
        idx += nbytes
    return header, idx


def read_datablock_trace(data, idx, nelts):
    assert len(data[idx:]) >= (nelts*4), "data[idx:] is too small (%s for %s)" % (len(data[idx:]), (nelts*4))
    resu = []
    for i in range(nelts):
        resu.append(decode_float(data[idx: idx + 4]))
        idx += 4
    return numpy.array(resu, dtype=float)