decoders/sle44xx/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2019 Federico Cerutti <federico@ceres-c.it>
   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 2 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 common.srdhelper import bitpack_lsb
  21 import sigrokdecode as srd
  22
  23 class Pin:
  24     RST, CLK, IO, = range(3)
  25
  26 class Ann:
  27     RESET_SYM, INTR_SYM, START_SYM, STOP_SYM, BIT_SYM, \
  28     ATR_BYTE, CMD_BYTE, OUT_BYTE, PROC_BYTE, \
  29     ATR_DATA, CMD_DATA, OUT_DATA, PROC_DATA, \
  30     = range(13)
  31
  32 class Bin:
  33     BYTES, = range(1)
  34
  35 class Decoder(srd.Decoder):
  36     api_version = 3
  37     id = 'sle44xx'
  38     name = 'SLE 44xx'
  39     longname = 'SLE44xx memory card'
  40     desc = 'SLE 4418/28/32/42 memory card serial protocol'
  41     license = 'gplv2+'
  42     inputs = ['logic']
  43     outputs = []
  44     tags = ['Memory']
  45     channels = (
  46         {'id': 'rst', 'name': 'RST', 'desc': 'Reset line'},
  47         {'id': 'clk', 'name': 'CLK', 'desc': 'Clock line'},
  48         {'id': 'io', 'name': 'I/O', 'desc': 'I/O data line'},
  49     )
  50     annotations = (
  51         ('reset_sym', 'Reset Symbol'),
  52         ('intr_sym', 'Interrupt Symbol'),
  53         ('start_sym', 'Start Symbol'),
  54         ('stop_sym', 'Stop Symbol'),
  55         ('bit_sym', 'Bit Symbol'),
  56         ('atr_byte', 'ATR Byte'),
  57         ('cmd_byte', 'Command Byte'),
  58         ('out_byte', 'Outgoing Byte'),
  59         ('proc_byte', 'Processing Byte'),
  60         ('atr_data', 'ATR data'),
  61         ('cmd_data', 'Command data'),
  62         ('out_data', 'Outgoing data'),
  63         ('proc_data', 'Processing data'),
  64     )
  65     annotation_rows = (
  66         ('symbols', 'Symbols', (Ann.RESET_SYM, Ann.INTR_SYM,
  67             Ann.START_SYM, Ann.STOP_SYM, Ann.BIT_SYM,)),
  68         ('fields', 'Fields', (Ann.ATR_BYTE,
  69             Ann.CMD_BYTE, Ann.OUT_BYTE, Ann.PROC_BYTE,)),
  70         ('operations', 'Operations', (Ann.ATR_DATA,
  71             Ann.CMD_DATA, Ann.OUT_DATA, Ann.PROC_DATA,)),
  72     )
  73     binary = (
  74         ('bytes', 'Bytes'),
  75     )
  76
  77     def __init__(self):
  78         self.reset()
  79
  80     def reset(self):
  81         self.samplerate = None
  82         self.max_addr = 256
  83         self.bits = []
  84         self.atr_bytes = []
  85         self.cmd_bytes = []
  86         self.cmd_proc = None
  87         self.out_len = None
  88         self.out_bytes = []
  89         self.proc_state = None
  90         self.state = None
  91
  92     def metadata(self, key, value):
  93         if key == srd.SRD_CONF_SAMPLERATE:
  94             self.samplerate = value
  95
  96     def start(self):
  97         self.out_ann = self.register(srd.OUTPUT_ANN)
  98         self.out_binary = self.register(srd.OUTPUT_BINARY)
  99
 100     def putx(self, ss, es, cls, data):
 101         self.put(ss, es, self.out_ann, [cls, data,])
 102
 103     def putb(self, ss, es, cls , data):
 104         self.put(ss, es, self.out_binary, [cls, data,])
 105
 106     def snums_to_usecs(self, snum_count):
 107         if not self.samplerate:
 108             return None
 109         snums_per_usec = self.samplerate / 1e6
 110         usecs = snum_count / snums_per_usec
 111         return usecs
 112
 113     def lookup_proto_ann_txt(self, key, variables):
 114         ann = {
 115             'RESET_SYM': [Ann.RESET_SYM, 'Reset', 'R',],
 116             'INTR_SYM': [Ann.INTR_SYM, 'Interrupt', 'Intr', 'I',],
 117             'START_SYM': [Ann.START_SYM, 'Start', 'ST', 'S',],
 118             'STOP_SYM': [Ann.STOP_SYM, 'Stop', 'SP', 'P',],
 119             'BIT_SYM': [Ann.BIT_SYM, '{bit}',],
 120             'ATR_BYTE': [Ann.ATR_BYTE,
 121                 'Answer To Reset: {data:02x}',
 122                 'ATR: {data:02x}',
 123                 '{data:02x}',
 124             ],
 125             'CMD_BYTE': [Ann.CMD_BYTE,
 126                 'Command: {data:02x}',
 127                 'Cmd: {data:02x}',
 128                 '{data:02x}',
 129             ],
 130             'OUT_BYTE': [Ann.OUT_BYTE,
 131                 'Outgoing data: {data:02x}',
 132                 'Data: {data:02x}',
 133                 '{data:02x}',
 134             ],
 135             'PROC_BYTE': [Ann.PROC_BYTE,
 136                 'Internal processing: {data:02x}',
 137                 'Proc: {data:02x}',
 138                 '{data:02x}',
 139             ],
 140             'ATR_DATA': [Ann.ATR_DATA,
 141                 'Answer To Reset: {data}',
 142                 'ATR: {data}',
 143                 '{data}',
 144             ],
 145             'CMD_DATA': [Ann.CMD_DATA,
 146                 'Command: {data}',
 147                 'Cmd: {data}',
 148                 '{data}',
 149             ],
 150             'OUT_DATA': [Ann.OUT_DATA,
 151                 'Outgoing: {data}',
 152                 'Out: {data}',
 153                 '{data}',
 154             ],
 155             'PROC_DATA': [Ann.PROC_DATA,
 156                 'Processing: {data}',
 157                 'Proc: {data}',
 158                 '{data}',
 159             ],
 160         }.get(key, None)
 161         if ann is None:
 162             return None, []
 163         cls, texts = ann[0], ann[1:]
 164         texts = [t.format(**variables) for t in texts]
 165         return cls, texts
 166
 167     def text_for_accu_bytes(self, accu):
 168         if not accu:
 169             return None, None, None, None
 170         ss, es = accu[0][1], accu[-1][2]
 171         data = [a[0] for a in accu]
 172         text = " ".join(['{:02x}'.format(a) for a in data])
 173         return ss, es, data, text
 174
 175     def flush_queued(self):
 176         '''Flush previously accumulated operations details.'''
 177
 178         # Can be called when either the completion of an operation got
 179         # detected (reliably), or when some kind of reset condition was
 180         # met while a potential previously observed operation has not
 181         # been postprocessed yet (best effort). Should not harm when the
 182         # routine gets invoked while no data was collected yet, or was
 183         # flushed already.
 184         # BEWARE! Will void internal state. Should really only get called
 185         # "between operations", NOT between fields of an operation.
 186
 187         if self.atr_bytes:
 188             key = 'ATR_DATA'
 189             ss, es, _, text = self.text_for_accu_bytes(self.atr_bytes)
 190             cls, texts = self.lookup_proto_ann_txt(key, {'data': text})
 191             self.putx(ss, es, cls, texts)
 192
 193         if self.cmd_bytes:
 194             key = 'CMD_DATA'
 195             ss, es, _, text = self.text_for_accu_bytes(self.cmd_bytes)
 196             cls, texts = self.lookup_proto_ann_txt(key, {'data': text})
 197             self.putx(ss, es, cls, texts)
 198
 199         if self.out_bytes:
 200             key = 'OUT_DATA'
 201             ss, es, _, text = self.text_for_accu_bytes(self.out_bytes)
 202             cls, texts = self.lookup_proto_ann_txt(key, {'data': text})
 203             self.putx(ss, es, cls, texts)
 204
 205         if self.proc_state:
 206             key = 'PROC_DATA'
 207             ss = self.proc_state['ss']
 208             es = self.proc_state['es']
 209             clk = self.proc_state['clk']
 210             high = self.proc_state['io1']
 211             text = '{clk} clocks, I/O {high}'.format(clk = clk, high = int(high))
 212             usecs = self.snums_to_usecs(es - ss)
 213             if usecs:
 214                 msecs = usecs / 1000
 215                 text = '{msecs:.2f} ms, {text}'.format(msecs = msecs, text = text)
 216             cls, texts = self.lookup_proto_ann_txt(key, {'data': text})
 217             self.putx(ss, es, cls, texts)
 218
 219         self.atr_bytes = None
 220         self.cmd_bytes = None
 221         self.cmd_proc = None
 222         self.out_len = None
 223         self.out_bytes = None
 224         self.proc_state = None
 225         self.state = None
 226
 227     def handle_reset(self, ss, es, has_clk):
 228         self.flush_queued()
 229         key = '{}_SYM'.format('RESET' if has_clk else 'INTR')
 230         cls, texts = self.lookup_proto_ann_txt(key, {})
 231         self.putx(ss, es, cls, texts)
 232         self.bits = []
 233         self.state = 'ATR' if has_clk else None
 234
 235     def handle_command(self, ss, is_start):
 236         if is_start:
 237             self.flush_queued()
 238         key = '{}_SYM'.format('START' if is_start else 'STOP')
 239         cls, texts = self.lookup_proto_ann_txt(key, {})
 240         self.putx(ss, ss, cls, texts)
 241         self.bits = []
 242         self.state = 'CMD' if is_start else 'DATA'
 243
 244     def command_check(self, ctrl, addr, data):
 245         '''Interpret CTRL/ADDR/DATA command entry.'''
 246
 247         # See the Siemens Datasheet section 2.3 Commands. The abbreviated
 248         # text variants are my guesses, terse for readability at coarser
 249         # zoom levels.
 250         codes_table = {
 251             0x30: {
 252                 'fmt': [
 253                     'read main memory, addr {addr:02x}',
 254                     'RD-M @{addr:02x}',
 255                 ],
 256                 'len': lambda ctrl, addr, data: self.max_addr - addr,
 257             },
 258             0x31: {
 259                 'fmt': [
 260                     'read security memory',
 261                     'RD-S',
 262                 ],
 263                 'len': 4,
 264             },
 265             0x33: {
 266                 'fmt': [
 267                     'compare verification data, addr {addr:02x}, data {data:02x}',
 268                     'CMP-V @{addr:02x} ={data:02x}',
 269                 ],
 270                 'proc': True,
 271             },
 272             0x34: {
 273                 'fmt': [
 274                     'read protection memory, addr {addr:02x}',
 275                     'RD-P @{addr:02x}',
 276                 ],
 277                 'len': 4,
 278             },
 279             0x38: {
 280                 'fmt': [
 281                     'update main memory, addr {addr:02x}, data {data:02x}',
 282                     'WR-M @{addr:02x} ={data:02x}',
 283                 ],
 284                 'proc': True,
 285             },
 286             0x39: {
 287                 'fmt': [
 288                     'update security memory, addr {addr:02x}, data {data:02x}',
 289                     'WR-S @{addr:02x} ={data:02x}',
 290                 ],
 291                 'proc': True,
 292             },
 293             0x3c: {
 294                 'fmt': [
 295                     'write protection memory, addr {addr:02x}, data {data:02x}',
 296                     'WR-P @{addr:02x} ={data:02x}',
 297                 ],
 298                 'proc': True,
 299             },
 300         }
 301         code = codes_table.get(ctrl, {})
 302         dflt_fmt = [
 303             'unknown, ctrl {ctrl:02x}, addr {addr:02x}, data {data:02x}',
 304             'UNK-{ctrl:02x} @{addr:02x}, ={data:02x}',
 305         ]
 306         fmt = code.get('fmt', dflt_fmt)
 307         if not isinstance(fmt, (list, tuple,)):
 308             fmt = [fmt,]
 309         texts = [f.format(ctrl = ctrl, addr = addr, data = data) for f in fmt]
 310         length = code.get('len', None)
 311         if callable(length):
 312             length = length(ctrl, addr, data)
 313         is_proc = code.get('proc', False)
 314         return texts, length, is_proc
 315
 316     def processing_start(self, ss, es, io_high):
 317         self.proc_state = {
 318             'ss': ss or es,
 319             'es': es or ss,
 320             'clk': 0,
 321             'io1': bool(io_high),
 322         }
 323
 324     def processing_update(self, es, clk_inc, io_high):
 325         if es is not None and es > self.proc_state['es']:
 326             self.proc_state['es'] = es
 327         self.proc_state['clk'] += clk_inc
 328         if io_high:
 329             self.proc_state['io1'] = True
 330
 331     def handle_data_byte(self, ss, es, data, bits):
 332         '''Accumulate CMD or OUT data bytes.'''
 333
 334         if self.state == 'ATR':
 335             if not self.atr_bytes:
 336                 self.atr_bytes = []
 337             self.atr_bytes.append([data, ss, es, bits,])
 338             if len(self.atr_bytes) == 4:
 339                 self.flush_queued()
 340             return
 341
 342         if self.state == 'CMD':
 343             if not self.cmd_bytes:
 344                 self.cmd_bytes = []
 345             self.cmd_bytes.append([data, ss, es, bits,])
 346             if len(self.cmd_bytes) == 3:
 347                 ctrl, addr, data = [c[0] for c in self.cmd_bytes]
 348                 texts, length, proc = self.command_check(ctrl, addr, data)
 349                 # Immediately emit the annotation to not lose the text,
 350                 # and to support zoom levels for this specific case.
 351                 ss, es = self.cmd_bytes[0][1], self.cmd_bytes[-1][2]
 352                 cls = Ann.CMD_DATA
 353                 self.putx(ss, es, cls, texts)
 354                 self.cmd_bytes = []
 355                 # Prepare to continue either at OUT or PROC after CMD.
 356                 self.out_len = length
 357                 self.cmd_proc = bool(proc)
 358                 self.state = None
 359             return
 360
 361         if self.state == 'OUT':
 362             if not self.out_bytes:
 363                 self.out_bytes = []
 364             self.out_bytes.append([data, ss, es, bits,])
 365             if self.out_len is not None and len(self.out_bytes) == self.out_len:
 366                 self.flush_queued()
 367             return
 368
 369     def handle_data_bit(self, ss, es, bit):
 370         '''Gather 8 bits of data (or track processing progress).'''
 371
 372         # Switch late from DATA to either OUT or PROC. We can tell the
 373         # type and potentially fixed length at the end of CMD already,
 374         # but a START/STOP condition may void this information. So we
 375         # do the switch at the first data bit after CMD.
 376         # In the OUT case data bytes get accumulated, until either the
 377         # expected byte count is reached, or another CMD starts. In the
 378         # PROC case a high I/O level terminates execution.
 379         if self.state == 'DATA':
 380             if self.out_len:
 381                 self.state = 'OUT'
 382             elif self.cmd_proc:
 383                 self.state = 'PROC'
 384                 self.processing_start(ss or es, es or ss, bit == 1)
 385             else:
 386                 # Implementor's note: Handle unknown situations like
 387                 # outgoing data bytes, for the user's convenience. This
 388                 # will show OUT bytes even if it's just processing CLK
 389                 # cycles with constant or irrelevant I/O bit patterns.
 390                 self.state = 'OUT'
 391         if self.state == 'PROC':
 392             high = bit == 1
 393             if ss is not None:
 394                 self.processing_update(ss, 0, high)
 395             if es is not None:
 396                 self.processing_update(es, 1, high)
 397             if high:
 398                 self.flush_queued()
 399             return
 400
 401         # This routine gets called two times per bit value. Track the
 402         # bit's value and ss timestamp when the bit period starts. And
 403         # update the es timestamp at the end of the bit's validity.
 404         if ss is not None:
 405             self.bits.append([bit, ss, es or ss])
 406             return
 407         if es is None:
 408             # Unexpected invocation. Could be a glitch or invalid input
 409             # data, or an interaction with RESET/START/STOP conditions.
 410             self.bits = []
 411             return
 412         if not self.bits:
 413             return
 414         if bit is not None:
 415             self.bits[-1][0] = bit
 416             # TODO Check for consistent bit level at ss and es when
 417             # the information was available? Is bit data sampled at
 418             # different clock edges depending whether data is sent
 419             # or received?
 420         self.bits[-1][2] = es
 421         # Emit the bit's annotation. See if a byte was received.
 422         bit, ss, es = self.bits[-1]
 423         cls, texts = self.lookup_proto_ann_txt('BIT_SYM', {'bit': bit})
 424         self.putx(ss, es, cls, texts)
 425         if len(self.bits) < 8:
 426             return
 427
 428         # Get the data byte value, and the byte's ss/es. Emit the byte's
 429         # annotation and binary output. Pass the byte to upper layers.
 430         # TODO Vary annotation classes with the byte's position within
 431         # a field? To tell CTRL/ADDR/DATA of a CMD entry apart?
 432         bits = self.bits
 433         self.bits = []
 434         data = bitpack_lsb(bits, 0)
 435         ss = bits[0][1]
 436         es = bits[-1][2]
 437
 438         key = '{}_BYTE'.format(self.state)
 439         cls, texts = self.lookup_proto_ann_txt(key, {'data': data})
 440         if cls:
 441             self.putx(ss, es, cls, texts)
 442         self.putb(ss, es, Bin.BYTES, bytes([data]))
 443
 444         self.handle_data_byte(ss, es, data, bits)
 445
 446     def decode(self):
 447         '''Decoder's main data interpretation loop.'''
 448
 449         # Signal conditions tracked by the protocol decoder:
 450         # - Rising and falling RST edges, which span the width of a
 451         #   high-active RESET pulse. RST has highest priority, no
 452         #   other activity can take place in this period.
 453         # - Rising and falling CLK edges when RST is active. The
 454         #   CLK pulse when RST is asserted will reset the card's
 455         #   address counter. RST alone can terminate memory reads.
 456         # - Rising and falling CLK edges when RST is inactive. This
 457         #   determines the period where BIT values are valid.
 458         # - I/O edges during high CLK. These are START and STOP
 459         #   conditions that tell COMMAND and DATA phases apart.
 460         # - Rise of I/O during internal processing. This expression
 461         #   is an unconditional part of the .wait() condition set. It
 462         #   is assumed that skipping this match in many cases is more
 463         #   efficient than the permanent re-construction of the .wait()
 464         #   condition list in every loop iteration, and preferrable to
 465         #   the maintainance cost of duplicating RST and CLK handling
 466         #   when checking I/O during internal processing.
 467         (
 468             COND_RESET_START, COND_RESET_STOP,
 469             COND_RSTCLK_START, COND_RSTCLK_STOP,
 470             COND_DATA_START, COND_DATA_STOP,
 471             COND_CMD_START, COND_CMD_STOP,
 472             COND_PROC_IOH,
 473         ) = range(9)
 474         conditions = [
 475             {Pin.RST: 'r'},
 476             {Pin.RST: 'f'},
 477             {Pin.RST: 'h', Pin.CLK: 'r'},
 478             {Pin.RST: 'h', Pin.CLK: 'f'},
 479             {Pin.RST: 'l', Pin.CLK: 'r'},
 480             {Pin.RST: 'l', Pin.CLK: 'f'},
 481             {Pin.CLK: 'h', Pin.IO: 'f'},
 482             {Pin.CLK: 'h', Pin.IO: 'r'},
 483             {Pin.RST: 'l', Pin.IO: 'r'},
 484         ]
 485
 486         ss_reset = es_reset = ss_clk = es_clk = None
 487         while True:
 488
 489             is_outgoing = self.state == 'OUT'
 490             is_processing = self.state == 'PROC'
 491             pins = self.wait(conditions)
 492             io = pins[Pin.IO]
 493
 494             # Handle RESET conditions, including an optional CLK pulse
 495             # while RST is asserted.
 496             if self.matched[COND_RESET_START]:
 497                 self.flush_queued()
 498                 ss_reset = self.samplenum
 499                 es_reset = ss_clk = es_clk = None
 500                 continue
 501             if self.matched[COND_RESET_STOP]:
 502                 es_reset = self.samplenum
 503                 self.handle_reset(ss_reset or 0, es_reset, ss_clk and es_clk)
 504                 ss_reset = es_reset = ss_clk = es_clk = None
 505                 continue
 506             if self.matched[COND_RSTCLK_START]:
 507                 ss_clk = self.samplenum
 508                 es_clk = None
 509                 continue
 510             if self.matched[COND_RSTCLK_STOP]:
 511                 es_clk = self.samplenum
 512                 continue
 513
 514             # Handle data bits' validity boundaries. Also covers the
 515             # periodic check for high I/O level and update of details
 516             # during internal processing.
 517             if self.matched[COND_DATA_START]:
 518                 self.handle_data_bit(self.samplenum, None, io)
 519                 continue
 520             if self.matched[COND_DATA_STOP]:
 521                 self.handle_data_bit(None, self.samplenum, None)
 522                 continue
 523
 524             # Additional check for idle I/O during internal processing,
 525             # independent of CLK edges this time. This assures that the
 526             # decoder ends processing intervals as soon as possible, at
 527             # the most precise timestamp.
 528             if is_processing and self.matched[COND_PROC_IOH]:
 529                 self.handle_data_bit(self.samplenum, self.samplenum, io)
 530                 continue
 531
 532             # The START/STOP conditions are only applicable outside of
 533             # "outgoing data" or "internal processing" periods. This is
 534             # what the data sheet specifies.
 535             if not is_outgoing and not is_processing:
 536                 if self.matched[COND_CMD_START]:
 537                     self.handle_command(self.samplenum, True)
 538                     continue
 539                 if self.matched[COND_CMD_STOP]:
 540                     self.handle_command(self.samplenum, False)
 541                     continue