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         ('fes', 'Frame end sequence'),
  80         ('bss', 'Byte start sequence'),
  81         ('warning', 'Warning'),
  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)