]> sigrok.org Git - libsigrokdecode.git/blobdiff - decoders/numbers_and_state/pd.py
numbers_and_state: introduce decoder (based on vector slicer)
[libsigrokdecode.git] / decoders / numbers_and_state / pd.py
diff --git a/decoders/numbers_and_state/pd.py b/decoders/numbers_and_state/pd.py
new file mode 100644 (file)
index 0000000..b8ac87e
--- /dev/null
@@ -0,0 +1,377 @@
+##
+## This file is part of the libsigrokdecode project.
+##
+## Copyright (C) 2019 Comlab AG
+## Copyright (C) 2020 Gerhard Sittig <gerhard.sittig@gmx.net>
+##
+## 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, see <http://www.gnu.org/licenses/>.
+##
+
+# This implementation started as a "vector slicer", then turned into the
+# "numbers and states" decoder, because users always had the freedom to
+# connect any logic signal to either of the decoder inputs. That's when
+# slicing vectors took second seat, and just was not needed any longer
+# in the strict sense.
+#
+# TODO
+# - Find an appropriate number of input channels, and maximum enum slots.
+# - Re-check correctness of signed integers. Signed fixed point is based
+#   on integers and transparently benefits from fixes and improvements.
+# - Local formatting in individual decoders becomes obsolete when common
+#   support for user selected formatting gets introduced.
+# - There is overlap with the 'parallel' decoder. Ideally the numbers
+#   decoder could stack on top of parallel, but parallel currently is
+#   severely limited in its number of input channels, and dramatically
+#   widening the parallel decoder may be undesirable.
+
+from common.srdhelper import bitpack
+import json
+import sigrokdecode as srd
+import struct
+
+'''
+OUTPUT_PYTHON format:
+
+Packet:
+[<ptype>, <pdata>]
+
+This is a list of <ptype>s and their respective <pdata> values:
+ - 'RAW': The data is a tuple of bit count and bit pattern (a number,
+   assuming unsigned integer presentation of the input data bit pattern).
+ - 'NUMBER': The data is the conversion result of the bit pattern.
+ - 'ENUM': The data is a tuple of the raw number and its mapped text.
+'''
+
+# TODO Better raise the number of channels to 32. This allows access to
+# IEEE754 single precision numbers, and shall cover most busses, _and_
+# remains within most logic analyzers' capabilities, and keeps the UI
+# dialog somewhat managable. What's a good default for the number of
+# enum slots (which translate to annotation rows)? Notice that 2 to the
+# power of the channel count is way out of the question. :)
+_max_channels = 16
+_max_enum_slots = 32
+
+class ChannelError(Exception):
+    pass
+
+class Pin:
+    CLK, BIT_0 = range(2)
+    BIT_N = BIT_0 + _max_channels
+
+class Ann:
+    RAW, NUM = range(2)
+    ENUM_0 = NUM + 1
+    ENUM_OVR = ENUM_0 + _max_enum_slots
+    ENUMS = range(ENUM_0, ENUM_OVR)
+    WARN = ENUM_OVR + 1
+
+    @staticmethod
+    def enum_indices():
+        return [i for i in range(Ann.ENUMS)]
+
+    @staticmethod
+    def get_enum_idx(code):
+        if code in range(_max_enum_slots):
+            return Ann.ENUM_0 + code
+        return Ann.ENUM_OVR
+
+def _channel_decl(count):
+    return tuple([
+        {'id': 'bit{}'.format(i), 'name': 'Bit{}'.format(i), 'desc': 'Bit position {}'.format(i)}
+        for i in range(count)
+    ])
+
+def _enum_cls_decl(count):
+    return tuple([
+        ('enum{}'.format(i), 'Enumeration slot {}'.format(i))
+        for i in range(count)
+    ] + [('enumovr', 'Enumeration overflow')])
+
+def _enum_rows_decl(count):
+    return tuple([
+        ('enums{}'.format(i), 'Enumeration slots {}'.format(i), (Ann.ENUM_0 + i,))
+        for i in range(count)
+    ] + [('enumsovr', 'Enumeration overflows', (Ann.ENUM_OVR,))])
+
+class Decoder(srd.Decoder):
+    api_version = 3
+    id = 'numbers_and_state'
+    name = 'Numbers and State'
+    longname = 'Interpret bit patters as numbers or state enums'
+    desc = 'Interpret bit patterns as different kinds of numbers (integer, float, enum).'
+    license = 'gplv2+'
+    inputs = ['logic']
+    outputs = ['numbers_and_state']
+    tags = ['Encoding', 'Util']
+    optional_channels = (
+        {'id': 'clk', 'name': 'Clock', 'desc': 'Clock'},
+    ) + _channel_decl(_max_channels)
+    options = (
+        {'id': 'clkedge', 'desc': 'Clock edge', 'default': 'rising',
+            'values': ('rising', 'falling', 'either')},
+        {'id': 'count', 'desc': 'Total bits count', 'default': 0},
+        {'id': 'interp', 'desc': 'Interpretation', 'default': 'unsigned',
+            'values': ('unsigned', 'signed', 'fixpoint', 'fixsigned', 'ieee754', 'enum')},
+        {'id': 'fracbits', 'desc': 'Fraction bits count', 'default': 0},
+        {'id': 'mapping', 'desc': 'Enum to text map file',
+            'default': 'enumtext.json'},
+        {'id': 'format', 'desc': 'Number format', 'default': '-',
+            'values': ('-', 'bin', 'oct', 'dec', 'hex')},
+    )
+    annotations = (
+        ('raw', 'Raw pattern'),
+        ('number', 'Number'),
+    ) + _enum_cls_decl(_max_enum_slots) + (
+        ('warning', 'Warning'),
+    )
+    annotation_rows = (
+        ('raws', 'Raw bits', (Ann.RAW,)),
+        ('numbers', 'Numbers', (Ann.NUM,)),
+    ) + _enum_rows_decl(_max_enum_slots) + (
+        ('warnings', 'Warnings', (Ann.WARN,)),
+    )
+
+    def __init__(self):
+        self.reset()
+
+    def reset(self):
+        pass
+
+    def start(self):
+        self.out_ann = self.register(srd.OUTPUT_ANN)
+        self.out_python = self.register(srd.OUTPUT_PYTHON)
+
+    def putg(self, ss, es, cls, data):
+        self.put(ss, es, self.out_ann, [cls, data])
+
+    def putpy(self, ss, es, ptype, pdata):
+        self.put(ss, es, self.out_python, (ptype, pdata))
+
+    def grab_pattern(self, pins):
+        '''Get a bit pattern from potentially incomplete probes' values.'''
+
+        # Pad and trim the input data, to achieve the user specified
+        # total number of bits. Map all unassigned signals to 0 (low).
+        # Return raw number (unsigned integer interpreation).
+        bits = pins + (None,) * self.bitcount
+        bits = bits[:self.bitcount]
+        bits = [b if b in (0, 1) else 0 for b in bits]
+        pattern = bitpack(bits)
+        return pattern
+
+    def handle_pattern(self, ss, es, pattern):
+        fmt = '{{:0{}b}}'.format(self.bitcount)
+        txt = fmt.format(pattern)
+        self.putg(ss, es, Ann.RAW, [txt])
+        self.putpy(ss, es, 'RAW', (self.bitcount, pattern))
+
+        try:
+            value = self.interpreter(ss, es, pattern)
+        except:
+            value = None
+        if value is None:
+            return
+        self.putpy(ss, es, 'NUMBER', value)
+        try:
+            formatted = self.formatter(ss, es, value)
+        except:
+            formatted = None
+        if formatted:
+            self.putg(ss, es, Ann.NUM, formatted)
+            if self.interpreter == self.interp_enum:
+                cls = Ann.get_enum_idx(pattern)
+                self.putg(ss, es, cls, formatted)
+                self.putpy(ss, es, 'ENUM', (value, formatted))
+
+    def interp_unsigned(self, ss, es, pattern):
+        value = pattern
+        return value
+
+    def interp_signed(self, ss, es, pattern):
+        if not 'signmask' in self.interp_state:
+            self.interp_state.update({
+                'signmask': 1 << (self.bitcount - 1),
+                'signfull': 1 << self.bitcount,
+            })
+        is_neg = pattern & self.interp_state['signmask']
+        if is_neg:
+            value = -(self.interp_state['signfull'] - pattern)
+        else:
+            value = pattern
+        return value
+
+    def interp_fixpoint(self, ss, es, pattern):
+        if not 'fixdiv' in self.interp_state:
+            self.interp_state.update({
+                'fixsign': self.options['interp'] == 'fixsigned',
+                'fixdiv': 2 ** self.options['fracbits'],
+            })
+        if self.interp_state['fixsign']:
+            value = self.interp_signed(ss, es, pattern)
+        else:
+            value = self.interp_unsigned(ss, es, pattern)
+        value /= self.interp_state['fixdiv']
+        return value
+
+    def interp_ieee754(self, ss, es, pattern):
+        if not 'ieee_has_16bit' in self.interp_state:
+            self.interp_state.update({
+                'ieee_fmt_int_16': '=H',
+                'ieee_fmt_flt_16': '=e',
+                'ieee_fmt_int_32': '=L',
+                'ieee_fmt_flt_32': '=f',
+                'ieee_fmt_int_64': '=Q',
+                'ieee_fmt_flt_64': '=d',
+            })
+            try:
+                fmt = self.interp_state.update['ieee_fmt_flt_16']
+                has_16bit_support = 8 * struct.calcsize(fmt) == 16
+            except:
+                has_16bit_support = False
+            self.interp_state['ieee_has_16bit'] = has_16bit_support
+        if self.bitcount == 16:
+            if not self.interp_state['ieee_has_16bit']:
+                return None
+            buff = struct.pack(self.interp_state['ieee_fmt_int_16'], pattern)
+            value, = struct.unpack(self.interp_state['ieee_fmt_flt_16'], buff)
+            return value
+        if self.bitcount == 32:
+            buff = struct.pack(self.interp_state['ieee_fmt_int_32'], pattern)
+            value, = struct.unpack(self.interp_state['ieee_fmt_flt_32'], buff)
+            return value
+        if self.bitcount == 64:
+            buff = struct.pack(self.interp_state['ieee_fmt_int_64'], pattern)
+            value, = struct.unpack(self.interp_state['ieee_fmt_flt_64'], buff)
+            return value
+        return None
+
+    def interp_enum(self, ss, es, pattern):
+        if not 'enum_map' in self.interp_state:
+            self.interp_state.update({
+                'enum_fn': self.options['mapping'],
+                'enum_map': {},
+                'enum_have_map': False,
+            })
+            try:
+                fn = self.interp_state['enum_fn']
+                # TODO Optionally try in several locations? Next to the
+                # decoder implementation? Where else? Expect users to
+                # enter absolute paths?
+                with open(fn, 'r') as f:
+                    maptext = f.read()
+                maptable = {}
+                if fn.endswith('.js') or fn.endswith('.json'):
+                    # JSON requires string literals on the LHS, so the
+                    # table is written "in reverse order".
+                    js_table = json.loads(maptext)
+                    for k, v in js_table.items():
+                        maptable[v] = k
+                elif fn.endswith('.py'):
+                    # Expect a specific identifier at the Python module
+                    # level, and assume that it's a dictionary.
+                    py_table = {}
+                    exec(maptext, py_table)
+                    maptable.update(py_table['enumtext'])
+                self.interp_state['enum_map'].update(maptable)
+                self.interp_state['enum_have_map'] = True
+            except:
+                # Silently ignore failure. This happens while the user
+                # is typing the filename, and is non-fatal. If the file
+                # exists and is not readable or not valid or of unknown
+                # format, the worst thing that can happen is that the
+                # decoder implementation keeps using "anonymous" phrases
+                # until a mapping has become available. No harm is done.
+                # This decoder cannot tell intermediate from final file
+                # read attempts, so we cannot raise severity here.
+                pass
+        value = self.interp_state['enum_map'].get(pattern, None)
+        if value is None:
+            value = pattern
+        return value
+
+    def format_native(self, ss, es, value):
+        return ['{}'.format(value),]
+
+    def format_bin(self, ss, es, value):
+        if not self.format_string:
+            self.format_string = '{{:0{}b}}'.format(self.bitcount)
+        return [self.format_string.format(value)]
+
+    def format_oct(self, ss, es, value):
+        if not self.format_string:
+            self.format_string = '{{:0{}o}}'.format((self.bitcount + 3 - 1) // 3)
+        return [self.format_string.format(value)]
+
+    def format_dec(self, ss, es, value):
+        if not self.format_string:
+            self.format_string = '{:d}'
+        return [self.format_string.format(value)]
+
+    def format_hex(self, ss, es, value):
+        if not self.format_string:
+            self.format_string = '{{:0{}x}}'.format((self.bitcount + 4 - 1) // 4)
+        return [self.format_string.format(value)]
+
+    def decode(self):
+        channels = [ch for ch in range(_max_channels) if self.has_channel(ch)]
+        have_clk = Pin.CLK in channels
+        if have_clk:
+            channels.remove(Pin.CLK)
+        if not channels:
+            raise ChannelError("Need at least one bit channel.")
+        if have_clk:
+            clkedge = {
+                'rising': 'r',
+                'falling': 'f',
+                'either': 'e',
+            }.get(self.options['clkedge'])
+            wait_cond = {Pin.CLK: clkedge}
+        else:
+            wait_cond = [{ch: 'e'} for ch in channels]
+
+        bitcount = self.options['count']
+        if not bitcount:
+            bitcount = channels[-1] - Pin.BIT_0 + 1
+        self.bitcount = bitcount
+
+        self.interpreter = {
+            'unsigned': self.interp_unsigned,
+            'signed': self.interp_signed,
+            'fixpoint': self.interp_fixpoint,
+            'fixsigned': self.interp_fixpoint,
+            'ieee754': self.interp_ieee754,
+            'enum': self.interp_enum,
+        }.get(self.options['interp'])
+        self.interp_state = {}
+        self.formatter = {
+            '-': self.format_native,
+            'bin': self.format_bin,
+            'oct': self.format_oct,
+            'dec': self.format_dec,
+            'hex': self.format_hex,
+        }.get(self.options['format'])
+        self.format_string = None
+
+        pins = self.wait()
+        ss = self.samplenum
+        prev_pattern = self.grab_pattern(pins[Pin.BIT_0:])
+        while True:
+            pins = self.wait(wait_cond)
+            es = self.samplenum
+            pattern = self.grab_pattern(pins[Pin.BIT_0:])
+            if pattern == prev_pattern:
+                continue
+            self.handle_pattern(ss, es, prev_pattern)
+            ss = es
+            prev_pattern = pattern