decoders/flexray/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2019 Stephan Thiele <stephan.thiele@mailbox.org>
   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 import sigrokdecode as srd
  21
  22 # Selection of constants as defined in FlexRay specification 3.0.1 Chapter A.1:
  23 class Const:
  24     cChannelIdleDelimiter = 11
  25     cCrcInitA = 0xFEDCBA
  26     cCrcInitB = 0xABCDEF
  27     cCrcPolynomial = 0x5D6DCB
  28     cCrcSize = 24
  29     cCycleCountMax = 63
  30     cdBSS = 2
  31     cdCAS = 30
  32     cdFES = 2
  33     cdFSS = 1
  34     cHCrcInit = 0x01A
  35     cHCrcPolynomial = 0x385
  36     cHCrcSize = 11
  37     cSamplesPerBit = 8
  38     cSlotIDMax = 2047
  39     cStaticSlotIDMax = 1023
  40     cVotingSamples = 5
  41
  42 class SamplerateError(Exception):
  43     pass
  44
  45 class Decoder(srd.Decoder):
  46     api_version = 3
  47     id = 'flexray'
  48     name = 'FlexRay'
  49     longname = 'FlexRay'
  50     desc = 'Automotive network communications protocol.'
  51     license = 'gplv2+'
  52     inputs = ['logic']
  53     outputs = []
  54     tags = ['Automotive']
  55     channels = (
  56         {'id': 'channel', 'name': 'Channel', 'desc': 'FlexRay bus channel'},
  57     )
  58     options = (
  59         {'id': 'channel_type', 'desc': 'Channel type', 'default': 'A',
  60             'values': ('A', 'B')},
  61         {'id': 'bitrate', 'desc': 'Bitrate (bit/s)', 'default': 10000000,
  62             'values': (10000000, 5000000, 2500000)},
  63     )
  64     annotations = (
  65         ('data', 'FlexRay payload data'),
  66         ('tss', 'Transmission start sequence'),
  67         ('fss', 'Frame start sequence'),
  68         ('reserved-bit', 'Reserved bit'),
  69         ('ppi', 'Payload preamble indicator'),
  70         ('null-frame', 'Nullframe indicator'),
  71         ('sync-frame', 'Full identifier'),
  72         ('startup-frame', 'Startup frame indicator'),
  73         ('id', 'Frame ID'),
  74         ('length', 'Data length'),
  75         ('header-crc', 'Header CRC'),
  76         ('cycle', 'Cycle code'),
  77         ('data-byte', 'Data byte'),
  78         ('frame-crc', 'Frame CRC'),
  79         ('cid-delimiter', 'Channel idle delimiter'),
  80         ('bss', 'Byte start sequence'),
  81         ('warnings', 'Human-readable warnings'),
  82         ('bit', 'Bit'),
  83         ('cid', 'Channel idle delimiter'),
  84         ('dts', 'Dynamic trailing sequence'),
  85         ('cas', 'Collision avoidance symbol'),
  86     )
  87     annotation_rows = (
  88         ('bits', 'Bits', (15, 17)),
  89         ('fields', 'Fields', tuple(range(15)) + (18, 19, 20)),
  90         ('warnings', 'Warnings', (16,)),
  91     )
  92
  93     def __init__(self):
  94         self.reset()
  95
  96     def reset(self):
  97         self.samplerate = None
  98         self.reset_variables()
  99
 100     def start(self):
 101         self.out_ann = self.register(srd.OUTPUT_ANN)
 102
 103     def metadata(self, key, value):
 104         if key == srd.SRD_CONF_SAMPLERATE:
 105             bitrate = float(self.options['bitrate'])
 106             self.samplerate = value
 107             self.bit_width = float(self.samplerate) / bitrate
 108             self.sample_point = (self.bit_width / 100.0) * self.sample_point_percent
 109
 110     # Generic helper for FlexRay bit annotations.
 111     def putg(self, ss, es, data):
 112         left, right = int(self.sample_point), int(self.bit_width - self.sample_point)
 113         self.put(ss - left, es + right, self.out_ann, data)
 114
 115     # Single-FlexRay-bit annotation using the current samplenum.
 116     def putx(self, data):
 117         self.putg(self.samplenum, self.samplenum, data)
 118
 119     # Multi-FlexRay-bit annotation from self.ss_block to current samplenum.
 120     def putb(self, data):
 121         self.putg(self.ss_block, self.samplenum, data)
 122
 123     # Generic CRC algorithm for any bit size and any data length. Used for
 124     # 11-bit header and 24-bit trailer. Not very efficient but at least it
 125     # works for now.
 126     #
 127     # TODO:
 128     # - use precalculated tables to increase performance.
 129     # - Add support for reverse CRC calculations.
 130
 131     @staticmethod
 132     def crc(data, data_len_bits, polynom, crc_len_bits, iv=0, xor=0):
 133         reg = iv ^ xor
 134
 135         for i in range(data_len_bits - 1, -1, -1):
 136             bit = ((reg >> (crc_len_bits - 1)) & 0x1) ^ ((data >> i) & 0x1)
 137             reg <<= 1
 138             if bit:
 139                 reg ^= polynom
 140
 141         mask = (1 << crc_len_bits) - 1
 142         crc = reg & mask
 143
 144         return crc ^ xor
 145
 146     def reset_variables(self):
 147         self.sample_point_percent = 50 # TODO: use vote based sampling
 148         self.state = 'IDLE'
 149         self.tss_start = self.tss_end = self.frame_type = self.dlc = None
 150         self.rawbits = [] # All bits, including byte start sequence bits
 151         self.bits = [] # Only actual FlexRay frame bits (no byte start sequence bits)
 152         self.curbit = 0 # Current bit of FlexRay frame (bit 0 == FSS)
 153         self.last_databit = 999 # Positive value that bitnum+x will never match
 154         self.last_xmit_bit = 999 # Positive value that bitnum+x will never match
 155         self.ss_block = None
 156         self.ss_databytebits = []
 157         self.end_of_frame = False
 158         self.dynamic_frame = False
 159         self.ss_bit0 = None
 160         self.ss_bit1 = None
 161         self.ss_bit2 = None
 162
 163     # Poor man's clock synchronization. Use signal edges which change to
 164     # dominant state in rather simple ways. This naive approach is neither
 165     # aware of the SYNC phase's width nor the specific location of the edge,
 166     # but improves the decoder's reliability when the input signal's bitrate
 167     # does not exactly match the nominal rate.
 168     def dom_edge_seen(self, force=False):
 169         self.dom_edge_snum = self.samplenum
 170         self.dom_edge_bcount = self.curbit
 171
 172     # Determine the position of the next desired bit's sample point.
 173     def get_sample_point(self, bitnum):
 174         samplenum = self.dom_edge_snum
 175         samplenum += self.bit_width * (bitnum - self.dom_edge_bcount)
 176         samplenum += self.sample_point
 177         return int(samplenum)
 178
 179     def is_bss_sequence(self):
 180         # FlexRay uses NRZ encoding and adds a binary 10 sequence before each
 181         # byte. After each 8 data bits, a BSS sequence is added but not after
 182         # frame CRC.
 183
 184         if self.end_of_frame:
 185           return False
 186
 187         if (len(self.rawbits) - 2) % 10 == 0:
 188             return True
 189         elif (len(self.rawbits) - 3) % 10 == 0:
 190             return True
 191
 192         return False
 193
 194     def handle_bit(self, fr_rx):
 195         self.rawbits.append(fr_rx)
 196         self.bits.append(fr_rx)
 197
 198         # Get the index of the current FlexRay frame bit.
 199         bitnum = len(self.bits) - 1
 200
 201         # If this is a byte start sequence remove it from self.bits and ignore it.
 202         if self.is_bss_sequence():
 203             self.bits.pop()
 204
 205             if bitnum > 1:
 206                 self.putx([15, [str(fr_rx)]])
 207             else:
 208                 if len(self.rawbits) == 2:
 209                     self.ss_bit1 = self.samplenum
 210                 elif len(self.rawbits) == 3:
 211                     self.ss_bit2 = self.samplenum
 212
 213             self.curbit += 1 # Increase self.curbit (bitnum is not affected).
 214             return
 215         else:
 216             if bitnum > 1:
 217                 self.putx([17, [str(fr_rx)]])
 218
 219         # Bit 0: Frame start sequence (FSS) bit
 220         if bitnum == 0:
 221             self.ss_bit0 = self.samplenum
 222
 223         # Bit 1: Start of header
 224         elif bitnum == 1:
 225             if self.rawbits[:3] == [1, 1, 0]:
 226                 self.put(self.tss_start, self.tss_end, self.out_ann,
 227                         [1, ['Transmission start sequence', 'TSS']])
 228
 229                 self.putg(self.ss_bit0, self.ss_bit0, [17, [str(self.rawbits[:3][0])]])
 230                 self.putg(self.ss_bit0, self.ss_bit0, [2,  ['FSS', 'Frame start sequence']])
 231                 self.putg(self.ss_bit1, self.ss_bit1, [15, [str(self.rawbits[:3][1])]])
 232                 self.putg(self.ss_bit2, self.ss_bit2, [15, [str(self.rawbits[:3][2])]])
 233                 self.putx([17, [str(fr_rx)]])
 234                 self.putx([3, ['Reserved bit: %d' % fr_rx, 'RB: %d' % fr_rx, 'RB']])
 235             else:
 236                 self.put(self.tss_start, self.tss_end, self.out_ann,
 237                         [20, ['Collision avoidance symbol', 'CAS']])
 238                 self.reset_variables()
 239
 240             # TODO: warning, if sequence is neither [1, 1, 0] nor [1, 1, 1]
 241
 242         # Bit 2: Payload preamble indicator. Must be 0 if null frame indicator is 0.
 243         elif bitnum == 2:
 244             self.putx([4, ['Payload preamble indicator: %d' % fr_rx,
 245                            'PPI: %d' % fr_rx]])
 246
 247         # Bit 3: Null frame indicator (inversed)
 248         elif bitnum == 3:
 249             data_type = 'data frame' if fr_rx else 'null frame'
 250             self.putx([5, ['Null frame indicator: %s' % data_type,
 251                            'NF: %d' % fr_rx, 'NF']])
 252
 253         # Bit 4: Sync frame indicator
 254         # Must be 1 if startup frame indicator is 1.
 255         elif bitnum == 4:
 256             self.putx([6, ['Sync frame indicator: %d' % fr_rx,
 257                            'Sync: %d' % fr_rx, 'Sync']])
 258
 259         # Bit 5: Startup frame indicator
 260         elif bitnum == 5:
 261             self.putx([7, ['Startup frame indicator: %d' % fr_rx,
 262                            'Startup: %d' % fr_rx, 'Startup']])
 263
 264         # Remember start of ID (see below).
 265         elif bitnum == 6:
 266             self.ss_block = self.samplenum
 267
 268         # Bits 6-16: Frame identifier (ID[10..0])
 269         # ID must NOT be 0.
 270         elif bitnum == 16:
 271             self.id = int(''.join(str(d) for d in self.bits[6:]), 2)
 272             self.putb([8, ['Frame ID: %d' % self.id, 'ID: %d' % self.id,
 273                            '%d' % self.id]])
 274
 275         # Remember start of payload length (see below).
 276         elif bitnum == 17:
 277             self.ss_block = self.samplenum
 278
 279         # Bits 17-23: Payload length (Length[7..0])
 280         # Payload length in header is the half of the real payload size.
 281         elif bitnum == 23:
 282             self.payload_length = int(''.join(str(d) for d in self.bits[17:]), 2)
 283             self.putb([9, ['Payload length: %d' % self.payload_length,
 284                            'Length: %d' % self.payload_length,
 285                            '%d' % self.payload_length]])
 286
 287         # Remember start of header CRC (see below).
 288         elif bitnum == 24:
 289             self.ss_block = self.samplenum
 290
 291         # Bits 24-34: Header CRC (11-bit) (HCRC[11..0])
 292         # Calculation of header CRC is equal on both channels.
 293         elif bitnum == 34:
 294             bits = ''.join([str(b) for b in self.bits[4:24]])
 295             header_to_check = int(bits, 2)
 296             expected_crc = self.crc(header_to_check, len(bits),
 297                 Const.cHCrcPolynomial, Const.cHCrcSize, Const.cHCrcInit)
 298             self.header_crc = int(''.join(str(d) for d in self.bits[24:]), 2)
 299
 300             crc_ok = self.header_crc == expected_crc
 301             crc_ann = "OK" if crc_ok else "bad"
 302
 303             self.putb([10, ['Header CRC: 0x%X (%s)' % (self.header_crc, crc_ann),
 304                             '0x%X (%s)' % (self.header_crc, crc_ann),
 305                             '0x%X' % self.header_crc]])
 306
 307         # Remember start of cycle code (see below).
 308         elif bitnum == 35:
 309             self.ss_block = self.samplenum
 310
 311         # Bits 35-40: Cycle code (Cyc[6..0])
 312         # Cycle code. Must be between 0 and 63.
 313         elif bitnum == 40:
 314             self.cycle = int(''.join(str(d) for d in self.bits[35:]), 2)
 315             self.putb([11, ['Cycle: %d' % self.cycle, 'Cyc: %d' % self.cycle,
 316                             '%d' % self.cycle]])
 317             self.last_databit = 41 + 2 * self.payload_length * 8
 318
 319         # Remember all databyte bits, except the very last one.
 320         elif bitnum in range(41, self.last_databit):
 321             self.ss_databytebits.append(self.samplenum)
 322
 323         # Bits 41-X: Data field (0-254 bytes, depending on length)
 324         # The bits within a data byte are transferred MSB-first.
 325         elif bitnum == self.last_databit:
 326             self.ss_databytebits.append(self.samplenum) # Last databyte bit.
 327             for i in range(2 * self.payload_length):
 328                 x = 40 + (8 * i) + 1
 329                 b = int(''.join(str(d) for d in self.bits[x:x + 8]), 2)
 330                 ss = self.ss_databytebits[i * 8]
 331                 es = self.ss_databytebits[((i + 1) * 8) - 1]
 332                 self.putg(ss, es, [12, ['Data byte %d: 0x%02x' % (i, b),
 333                                         'DB%d: 0x%02x' % (i, b), '%02X' % b]])
 334             self.ss_databytebits = []
 335             self.ss_block = self.samplenum # Remember start of trailer CRC.
 336
 337         # Trailer CRC (24-bit) (CRC[11..0])
 338         # Initialization vector of channel A and B are different, so CRCs are
 339         # different for same data.
 340         elif bitnum == self.last_databit + 23:
 341             bits = ''.join([str(b) for b in self.bits[1:-24]])
 342             frame_to_check = int(bits, 2)
 343             iv = Const.cCrcInitA if self.options['channel_type'] == 'A' else Const.cCrcInitB
 344             expected_crc = self.crc(frame_to_check, len(bits),
 345                 Const.cCrcPolynomial, Const.cCrcSize, iv=iv)
 346             self.frame_crc = int(''.join(str(d) for d in self.bits[self.last_databit:]), 2)
 347
 348             crc_ok = self.frame_crc == expected_crc
 349             crc_ann = "OK" if crc_ok else "bad"
 350
 351             self.putb([13, ['Frame CRC: 0x%X (%s)' % (self.frame_crc, crc_ann),
 352                             '0x%X (%s)' % (self.frame_crc, crc_ann),
 353                             '0x%X' % self.frame_crc]])
 354             self.end_of_frame = True
 355
 356         # Remember start of frame end sequence (see below).
 357         elif bitnum == self.last_databit + 24:
 358             self.ss_block = self.samplenum
 359
 360         # Frame end sequence, must be 1 followed by 0.
 361         elif bitnum == self.last_databit + 25:
 362             self.putb([14, ['Frame end sequence', 'FES']])
 363
 364         # Check for DTS
 365         elif bitnum == self.last_databit + 26:
 366             if not fr_rx:
 367                 self.dynamic_frame = True
 368             else:
 369                 self.last_xmit_bit = bitnum
 370             self.ss_block = self.samplenum
 371
 372         # Remember start of channel idle delimiter (see below).
 373         elif bitnum == self.last_xmit_bit:
 374             self.ss_block = self.samplenum
 375
 376         # Channel idle limiter (CID[11..0])
 377         elif bitnum == self.last_xmit_bit + Const.cChannelIdleDelimiter - 1:
 378             self.putb([18, ['Channel idle delimiter', 'CID']])
 379             self.reset_variables()
 380
 381         # DTS if dynamic frame
 382         elif bitnum > self.last_databit + 27:
 383             if self.dynamic_frame:
 384                 if fr_rx:
 385                     if self.last_xmit_bit == 999:
 386                         self.putb([19, ['Dynamic trailing sequence', 'DTS']])
 387                         self.last_xmit_bit = bitnum + 1
 388                         self.ss_block = self.samplenum
 389
 390         self.curbit += 1
 391
 392     def decode(self):
 393         if not self.samplerate:
 394             raise SamplerateError('Cannot decode without samplerate.')
 395
 396         while True:
 397             # State machine.
 398             if self.state == 'IDLE':
 399                 # Wait for a dominant state (logic 0) on the bus.
 400                 (fr_rx,) = self.wait({0: 'l'})
 401                 self.tss_start = self.samplenum
 402                 (fr_rx,) = self.wait({0: 'h'})
 403                 self.tss_end = self.samplenum
 404                 self.dom_edge_seen(force = True)
 405                 self.state = 'GET BITS'
 406             elif self.state == 'GET BITS':
 407                 # Wait until we're in the correct bit/sampling position.
 408                 pos = self.get_sample_point(self.curbit)
 409                 (fr_rx,) = self.wait([{'skip': pos - self.samplenum}, {0: 'f'}])
 410                 if self.matched[1]:
 411                     self.dom_edge_seen()
 412                 if self.matched[0]:
 413                     self.handle_bit(fr_rx)