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