decoders/numbers_and_state/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2019 Comlab AG
   5 ## Copyright (C) 2020 Gerhard Sittig <gerhard.sittig@gmx.net>
   6 ##
   7 ## This program is free software; you can redistribute it and/or modify
   8 ## it under the terms of the GNU General Public License as published by
   9 ## the Free Software Foundation; either version 2 of the License, or
  10 ## (at your option) any later version.
  11 ##
  12 ## This program is distributed in the hope that it will be useful,
  13 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 ## GNU General Public License for more details.
  16 ##
  17 ## You should have received a copy of the GNU General Public License
  18 ## along with this program; if not, see <http://www.gnu.org/licenses/>.
  19 ##
  20
  21 # This implementation started as a "vector slicer", then turned into the
  22 # "numbers and states" decoder, because users always had the freedom to
  23 # connect any logic signal to either of the decoder inputs. That's when
  24 # slicing vectors took second seat, and just was not needed any longer
  25 # in the strict sense.
  26 #
  27 # TODO
  28 # - Find an appropriate number of input channels, and maximum enum slots.
  29 # - Re-check correctness of signed integers. Signed fixed point is based
  30 #   on integers and transparently benefits from fixes and improvements.
  31 # - Local formatting in individual decoders becomes obsolete when common
  32 #   support for user selected formatting gets introduced.
  33 # - There is overlap with the 'parallel' decoder. Ideally the numbers
  34 #   decoder could stack on top of parallel, but parallel currently is
  35 #   severely limited in its number of input channels, and dramatically
  36 #   widening the parallel decoder may be undesirable.
  37
  38 from common.srdhelper import bitpack
  39 import json
  40 import sigrokdecode as srd
  41 import struct
  42
  43 '''
  44 OUTPUT_PYTHON format:
  45
  46 Packet:
  47 [<ptype>, <pdata>]
  48
  49 This is a list of <ptype>s and their respective <pdata> values:
  50  - 'RAW': The data is a tuple of bit count and bit pattern (a number,
  51    assuming unsigned integer presentation of the input data bit pattern).
  52  - 'NUMBER': The data is the conversion result of the bit pattern.
  53  - 'ENUM': The data is a tuple of the raw number and its mapped text.
  54 '''
  55
  56 # TODO Better raise the number of channels to 32. This allows access to
  57 # IEEE754 single precision numbers, and shall cover most busses, _and_
  58 # remains within most logic analyzers' capabilities, and keeps the UI
  59 # dialog somewhat managable. What's a good default for the number of
  60 # enum slots (which translate to annotation rows)? Notice that 2 to the
  61 # power of the channel count is way out of the question. :)
  62 _max_channels = 16
  63 _max_enum_slots = 32
  64
  65 class ChannelError(Exception):
  66     pass
  67
  68 class Pin:
  69     CLK, BIT_0 = range(2)
  70     BIT_N = BIT_0 + _max_channels
  71
  72 class Ann:
  73     RAW, NUM = range(2)
  74     ENUM_0 = NUM + 1
  75     ENUM_OVR = ENUM_0 + _max_enum_slots
  76     ENUMS = range(ENUM_0, ENUM_OVR)
  77     WARN = ENUM_OVR + 1
  78
  79     @staticmethod
  80     def enum_indices():
  81         return [i for i in range(Ann.ENUMS)]
  82
  83     @staticmethod
  84     def get_enum_idx(code):
  85         if code in range(_max_enum_slots):
  86             return Ann.ENUM_0 + code
  87         return Ann.ENUM_OVR
  88
  89 def _channel_decl(count):
  90     return tuple([
  91         {'id': 'bit{}'.format(i), 'name': 'Bit{}'.format(i), 'desc': 'Bit position {}'.format(i)}
  92         for i in range(count)
  93     ])
  94
  95 def _enum_cls_decl(count):
  96     return tuple([
  97         ('enum{}'.format(i), 'Enumeration slot {}'.format(i))
  98         for i in range(count)
  99     ] + [('enumovr', 'Enumeration overflow')])
 100
 101 def _enum_rows_decl(count):
 102     return tuple([
 103         ('enums{}'.format(i), 'Enumeration slots {}'.format(i), (Ann.ENUM_0 + i,))
 104         for i in range(count)
 105     ] + [('enumsovr', 'Enumeration overflows', (Ann.ENUM_OVR,))])
 106
 107 class Decoder(srd.Decoder):
 108     api_version = 3
 109     id = 'numbers_and_state'
 110     name = 'Numbers and State'
 111     longname = 'Interpret bit patters as numbers or state enums'
 112     desc = 'Interpret bit patterns as different kinds of numbers (integer, float, enum).'
 113     license = 'gplv2+'
 114     inputs = ['logic']
 115     outputs = ['numbers_and_state']
 116     tags = ['Encoding', 'Util']
 117     optional_channels = (
 118         {'id': 'clk', 'name': 'Clock', 'desc': 'Clock'},
 119     ) + _channel_decl(_max_channels)
 120     options = (
 121         {'id': 'clkedge', 'desc': 'Clock edge', 'default': 'rising',
 122             'values': ('rising', 'falling', 'either')},
 123         {'id': 'count', 'desc': 'Total bits count', 'default': 0},
 124         {'id': 'interp', 'desc': 'Interpretation', 'default': 'unsigned',
 125             'values': ('unsigned', 'signed', 'fixpoint', 'fixsigned', 'ieee754', 'enum')},
 126         {'id': 'fracbits', 'desc': 'Fraction bits count', 'default': 0},
 127         {'id': 'mapping', 'desc': 'Enum to text map file',
 128             'default': 'enumtext.json'},
 129         {'id': 'format', 'desc': 'Number format', 'default': '-',
 130             'values': ('-', 'bin', 'oct', 'dec', 'hex')},
 131     )
 132     annotations = (
 133         ('raw', 'Raw pattern'),
 134         ('number', 'Number'),
 135     ) + _enum_cls_decl(_max_enum_slots) + (
 136         ('warning', 'Warning'),
 137     )
 138     annotation_rows = (
 139         ('raws', 'Raw bits', (Ann.RAW,)),
 140         ('numbers', 'Numbers', (Ann.NUM,)),
 141     ) + _enum_rows_decl(_max_enum_slots) + (
 142         ('warnings', 'Warnings', (Ann.WARN,)),
 143     )
 144
 145     def __init__(self):
 146         self.reset()
 147
 148     def reset(self):
 149         pass
 150
 151     def start(self):
 152         self.out_ann = self.register(srd.OUTPUT_ANN)
 153         self.out_python = self.register(srd.OUTPUT_PYTHON)
 154
 155     def putg(self, ss, es, cls, data):
 156         self.put(ss, es, self.out_ann, [cls, data])
 157
 158     def putpy(self, ss, es, ptype, pdata):
 159         self.put(ss, es, self.out_python, (ptype, pdata))
 160
 161     def grab_pattern(self, pins):
 162         '''Get a bit pattern from potentially incomplete probes' values.'''
 163
 164         # Pad and trim the input data, to achieve the user specified
 165         # total number of bits. Map all unassigned signals to 0 (low).
 166         # Return raw number (unsigned integer interpreation).
 167         bits = pins + (None,) * self.bitcount
 168         bits = bits[:self.bitcount]
 169         bits = [b if b in (0, 1) else 0 for b in bits]
 170         pattern = bitpack(bits)
 171         return pattern
 172
 173     def handle_pattern(self, ss, es, pattern):
 174         fmt = '{{:0{}b}}'.format(self.bitcount)
 175         txt = fmt.format(pattern)
 176         self.putg(ss, es, Ann.RAW, [txt])
 177         self.putpy(ss, es, 'RAW', (self.bitcount, pattern))
 178
 179         try:
 180             value = self.interpreter(ss, es, pattern)
 181         except:
 182             value = None
 183         if value is None:
 184             return
 185         self.putpy(ss, es, 'NUMBER', value)
 186         try:
 187             formatted = self.formatter(ss, es, value)
 188         except:
 189             formatted = None
 190         if formatted:
 191             self.putg(ss, es, Ann.NUM, formatted)
 192             if self.interpreter == self.interp_enum:
 193                 cls = Ann.get_enum_idx(pattern)
 194                 self.putg(ss, es, cls, formatted)
 195                 self.putpy(ss, es, 'ENUM', (value, formatted))
 196
 197     def interp_unsigned(self, ss, es, pattern):
 198         value = pattern
 199         return value
 200
 201     def interp_signed(self, ss, es, pattern):
 202         if not 'signmask' in self.interp_state:
 203             self.interp_state.update({
 204                 'signmask': 1 << (self.bitcount - 1),
 205                 'signfull': 1 << self.bitcount,
 206             })
 207         is_neg = pattern & self.interp_state['signmask']
 208         if is_neg:
 209             value = -(self.interp_state['signfull'] - pattern)
 210         else:
 211             value = pattern
 212         return value
 213
 214     def interp_fixpoint(self, ss, es, pattern):
 215         if not 'fixdiv' in self.interp_state:
 216             self.interp_state.update({
 217                 'fixsign': self.options['interp'] == 'fixsigned',
 218                 'fixdiv': 2 ** self.options['fracbits'],
 219             })
 220         if self.interp_state['fixsign']:
 221             value = self.interp_signed(ss, es, pattern)
 222         else:
 223             value = self.interp_unsigned(ss, es, pattern)
 224         value /= self.interp_state['fixdiv']
 225         return value
 226
 227     def interp_ieee754(self, ss, es, pattern):
 228         if not 'ieee_has_16bit' in self.interp_state:
 229             self.interp_state.update({
 230                 'ieee_fmt_int_16': '=H',
 231                 'ieee_fmt_flt_16': '=e',
 232                 'ieee_fmt_int_32': '=L',
 233                 'ieee_fmt_flt_32': '=f',
 234                 'ieee_fmt_int_64': '=Q',
 235                 'ieee_fmt_flt_64': '=d',
 236             })
 237             try:
 238                 fmt = self.interp_state.update['ieee_fmt_flt_16']
 239                 has_16bit_support = 8 * struct.calcsize(fmt) == 16
 240             except:
 241                 has_16bit_support = False
 242             self.interp_state['ieee_has_16bit'] = has_16bit_support
 243         if self.bitcount == 16:
 244             if not self.interp_state['ieee_has_16bit']:
 245                 return None
 246             buff = struct.pack(self.interp_state['ieee_fmt_int_16'], pattern)
 247             value, = struct.unpack(self.interp_state['ieee_fmt_flt_16'], buff)
 248             return value
 249         if self.bitcount == 32:
 250             buff = struct.pack(self.interp_state['ieee_fmt_int_32'], pattern)
 251             value, = struct.unpack(self.interp_state['ieee_fmt_flt_32'], buff)
 252             return value
 253         if self.bitcount == 64:
 254             buff = struct.pack(self.interp_state['ieee_fmt_int_64'], pattern)
 255             value, = struct.unpack(self.interp_state['ieee_fmt_flt_64'], buff)
 256             return value
 257         return None
 258
 259     def interp_enum(self, ss, es, pattern):
 260         if not 'enum_map' in self.interp_state:
 261             self.interp_state.update({
 262                 'enum_fn': self.options['mapping'],
 263                 'enum_map': {},
 264                 'enum_have_map': False,
 265             })
 266             try:
 267                 fn = self.interp_state['enum_fn']
 268                 # TODO Optionally try in several locations? Next to the
 269                 # decoder implementation? Where else? Expect users to
 270                 # enter absolute paths?
 271                 with open(fn, 'r') as f:
 272                     maptext = f.read()
 273                 maptable = {}
 274                 if fn.endswith('.js') or fn.endswith('.json'):
 275                     # JSON requires string literals on the LHS, so the
 276                     # table is written "in reverse order".
 277                     js_table = json.loads(maptext)
 278                     for k, v in js_table.items():
 279                         maptable[v] = k
 280                 elif fn.endswith('.py'):
 281                     # Expect a specific identifier at the Python module
 282                     # level, and assume that it's a dictionary.
 283                     py_table = {}
 284                     exec(maptext, py_table)
 285                     maptable.update(py_table['enumtext'])
 286                 self.interp_state['enum_map'].update(maptable)
 287                 self.interp_state['enum_have_map'] = True
 288             except:
 289                 # Silently ignore failure. This happens while the user
 290                 # is typing the filename, and is non-fatal. If the file
 291                 # exists and is not readable or not valid or of unknown
 292                 # format, the worst thing that can happen is that the
 293                 # decoder implementation keeps using "anonymous" phrases
 294                 # until a mapping has become available. No harm is done.
 295                 # This decoder cannot tell intermediate from final file
 296                 # read attempts, so we cannot raise severity here.
 297                 pass
 298         value = self.interp_state['enum_map'].get(pattern, None)
 299         if value is None:
 300             value = pattern
 301         return value
 302
 303     def format_native(self, ss, es, value):
 304         return ['{}'.format(value),]
 305
 306     def format_bin(self, ss, es, value):
 307         if not self.format_string:
 308             self.format_string = '{{:0{}b}}'.format(self.bitcount)
 309         return [self.format_string.format(value)]
 310
 311     def format_oct(self, ss, es, value):
 312         if not self.format_string:
 313             self.format_string = '{{:0{}o}}'.format((self.bitcount + 3 - 1) // 3)
 314         return [self.format_string.format(value)]
 315
 316     def format_dec(self, ss, es, value):
 317         if not self.format_string:
 318             self.format_string = '{:d}'
 319         return [self.format_string.format(value)]
 320
 321     def format_hex(self, ss, es, value):
 322         if not self.format_string:
 323             self.format_string = '{{:0{}x}}'.format((self.bitcount + 4 - 1) // 4)
 324         return [self.format_string.format(value)]
 325
 326     def decode(self):
 327         channels = [ch for ch in range(_max_channels) if self.has_channel(ch)]
 328         have_clk = Pin.CLK in channels
 329         if have_clk:
 330             channels.remove(Pin.CLK)
 331         if not channels:
 332             raise ChannelError("Need at least one bit channel.")
 333         if have_clk:
 334             clkedge = {
 335                 'rising': 'r',
 336                 'falling': 'f',
 337                 'either': 'e',
 338             }.get(self.options['clkedge'])
 339             wait_cond = {Pin.CLK: clkedge}
 340         else:
 341             wait_cond = [{ch: 'e'} for ch in channels]
 342
 343         bitcount = self.options['count']
 344         if not bitcount:
 345             bitcount = channels[-1] - Pin.BIT_0 + 1
 346         self.bitcount = bitcount
 347
 348         self.interpreter = {
 349             'unsigned': self.interp_unsigned,
 350             'signed': self.interp_signed,
 351             'fixpoint': self.interp_fixpoint,
 352             'fixsigned': self.interp_fixpoint,
 353             'ieee754': self.interp_ieee754,
 354             'enum': self.interp_enum,
 355         }.get(self.options['interp'])
 356         self.interp_state = {}
 357         self.formatter = {
 358             '-': self.format_native,
 359             'bin': self.format_bin,
 360             'oct': self.format_oct,
 361             'dec': self.format_dec,
 362             'hex': self.format_hex,
 363         }.get(self.options['format'])
 364         self.format_string = None
 365
 366         pins = self.wait()
 367         ss = self.samplenum
 368         prev_pattern = self.grab_pattern(pins[Pin.BIT_0:])
 369         while True:
 370             pins = self.wait(wait_cond)
 371             es = self.samplenum
 372             pattern = self.grab_pattern(pins[Pin.BIT_0:])
 373             if pattern == prev_pattern:
 374                 continue
 375             self.handle_pattern(ss, es, prev_pattern)
 376             ss = es
 377             prev_pattern = pattern