decoders/sae_j1850_vpw/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2016 Anthony Symons <antus@pcmhacking.net>
   5 ## Copyright (C) 2023 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 import sigrokdecode as srd
  22 from common.srdhelper import bitpack_msb
  23
  24 # VPW Timings. From the SAE J1850 1995 rev section 23.406 documentation.
  25 # Ideal, minimum and maximum tolerances.
  26 VPW_SOF = 200
  27 VPW_SOFL = 164
  28 VPW_SOFH = 245  # 240 by the spec, 245 so a 60us 4x sample will pass
  29 VPW_LONG = 128
  30 VPW_LONGL = 97
  31 VPW_LONGH = 170 # 164 by the spec but 170 for low sample rate tolerance.
  32 VPW_SHORT = 64
  33 VPW_SHORTL = 24 # 35 by the spec, 24 to allow down to 6us as measured in practice for 4x @ 1mhz sampling
  34 VPW_SHORTH = 97
  35 VPW_IFS = 240
  36
  37 class SamplerateError(Exception):
  38     pass
  39
  40 (
  41     ANN_SOF, ANN_BIT, ANN_IFS, ANN_BYTE,
  42     ANN_PRIO, ANN_DEST, ANN_SRC, ANN_MODE, ANN_DATA, ANN_CSUM,
  43     ANN_M1_PID,
  44     ANN_WARN,
  45 ) = range(12)
  46
  47 class Decoder(srd.Decoder):
  48     api_version = 3
  49     id = 'sae_j1850_vpw'
  50     name = 'SAE J1850 VPW'
  51     longname = 'SAE J1850 VPW.'
  52     desc = 'SAE J1850 Variable Pulse Width 1x and 4x.'
  53     license = 'gplv2+'
  54     inputs = ['logic']
  55     outputs = []
  56     tags = ['Automotive']
  57     channels = (
  58         {'id': 'data', 'name': 'Data', 'desc': 'Data line'},
  59     )
  60     annotations = (
  61         ('sof', 'SOF'),
  62         ('bit', 'Bit'),
  63         ('ifs', 'EOF/IFS'),
  64         ('byte', 'Byte'),
  65         ('prio', 'Priority'),
  66         ('dest', 'Destination'),
  67         ('src', 'Source'),
  68         ('mode', 'Mode'),
  69         ('data', 'Data'),
  70         ('csum', 'Checksum'),
  71         ('m1_pid', 'Pid'),
  72         ('warn', 'Warning'),
  73     )
  74     annotation_rows = (
  75         ('bits', 'Bits', (ANN_SOF, ANN_BIT, ANN_IFS,)),
  76         ('bytes', 'Bytes', (ANN_BYTE,)),
  77         ('fields', 'Fields', (ANN_PRIO, ANN_DEST, ANN_SRC, ANN_MODE, ANN_DATA, ANN_CSUM,)),
  78         ('values', 'Values', (ANN_M1_PID,)),
  79         ('warns', 'Warnings', (ANN_WARN,)),
  80     )
  81     # TODO Add support for options? Polarity. Glitch length.
  82
  83     def __init__(self):
  84         self.reset()
  85
  86     def reset(self):
  87         self.samplerate = None
  88         self.active = 0 # Signal polarity. Needs to become an option?
  89         self.bits = []
  90         self.fields = {}
  91
  92     def start(self):
  93         self.out_ann = self.register(srd.OUTPUT_ANN)
  94
  95     def metadata(self, key, value):
  96         if key == srd.SRD_CONF_SAMPLERATE:
  97             self.samplerate = value
  98
  99     def putg(self, ss, es, cls, texts):
 100         self.put(ss, es, self.out_ann, [cls, texts])
 101
 102     def invalidate_frame_details(self):
 103         self.bits.clear()
 104         self.fields.clear()
 105
 106     def handle_databytes(self, fields, data):
 107         # TODO Deep inspection of header fields and data values, including
 108         # checksum verification results.
 109         mode = fields.get('mode', None)
 110         if mode is None:
 111             return
 112         if mode == 1:
 113             # An earlier implementation commented that for mode 1 the
 114             # first data byte would be the PID. But example captures
 115             # have no data bytes in packets for that mode. This position
 116             # is taken by the checksum. Is this correct?
 117             pid = data[0] if data else fields.get('csum', None)
 118             if pid is None:
 119                 text = ['PID missing']
 120                 self.putg(ss, es, ANN_WARN, text)
 121             else:
 122                 byte_text = '{:02x}'.format(pid)
 123                 self.putg(ss, es, ANN_M1_PID, [byte_text])
 124
 125     def handle_byte(self, ss, es, b):
 126         # Annotate all raw byte values. Inspect and process the first
 127         # bytes in a frame already. Cease inspection and only accumulate
 128         # all other bytes after the mode. The checksum's position and
 129         # thus the data bytes' span will only be known when EOF or IFS
 130         # were seen. Implementor's note: This method just identifies
 131         # header fields. Processing is left to the .handle_databytes()
 132         # method. Until then validity will have been checked, too (CS).
 133         byte_text = '{:02x}'.format(b)
 134         self.putg(ss, es, ANN_BYTE, [byte_text])
 135
 136         if not 'prio' in self.fields:
 137             self.fields.update({'prio': b})
 138             self.putg(ss, es, ANN_PRIO, [byte_text])
 139             return
 140         if not 'dest' in self.fields:
 141             self.fields.update({'dest': b})
 142             self.putg(ss, es, ANN_DEST, [byte_text])
 143             return
 144         if not 'src' in self.fields:
 145             self.fields.update({'src': b})
 146             self.putg(ss, es, ANN_SRC, [byte_text])
 147             return
 148         if not 'mode' in self.fields:
 149             self.fields.update({'mode': b})
 150             self.putg(ss, es, ANN_MODE, [byte_text])
 151             return
 152         if not 'data' in self.fields:
 153             self.fields.update({'data': [], 'csum': None})
 154         self.fields['data'].append((b, ss, es))
 155
 156     def handle_sof(self, ss, es, speed):
 157         text = ['{speed:d}x SOF', 'S{speed:d}', 'S']
 158         text = [f.format(speed = speed) for f in text]
 159         self.putg(ss, es, ANN_SOF, text)
 160         self.invalidate_frame_details()
 161         self.fields.update({'speed': speed})
 162
 163     def handle_bit(self, ss, es, b):
 164         self.bits.append((b, ss, es))
 165         self.putg(ss, es, ANN_BIT, ['{:d}'.format(b)])
 166         if len(self.bits) < 8:
 167             return
 168         ss, es = self.bits[0][1], self.bits[-1][2]
 169         b = bitpack_msb(self.bits, 0)
 170         self.bits.clear()
 171         self.handle_byte(ss, es, b)
 172
 173     def handle_eof(self, ss, es, is_ifs = False):
 174         # EOF or IFS were seen. Post process the data bytes sequence.
 175         # Separate the checksum from the data bytes. Emit annotations.
 176         # Pass data bytes and header fields to deeper inspection.
 177         data = self.fields.get('data', {})
 178         if not data:
 179             text = ['Short data phase', 'Data']
 180             self.putg(ss, es, ANN_WARN, text)
 181         csum = None
 182         if len(data) >= 1:
 183             csum, ss_csum, es_csum = data.pop()
 184             self.fields.update({'csum': csum})
 185             # TODO Verify checksum's correctness?
 186         if data:
 187             ss_data, es_data = data[0][1], data[-1][2]
 188             text = ' '.join(['{:02x}'.format(b[0]) for b in data])
 189             self.putg(ss_data, es_data, ANN_DATA, [text])
 190         if csum is not None:
 191             text = '{:02x}'.format(csum)
 192             self.putg(ss_csum, es_csum, ANN_CSUM, [text])
 193         text = ['IFS', 'I'] if is_ifs else ['EOF', 'E']
 194         self.putg(ss, es, ANN_IFS, text)
 195         self.handle_databytes(self.fields, data);
 196         self.invalidate_frame_details()
 197
 198     def handle_unknown(self, ss, es):
 199         text = ['Unknown condition', 'Unknown', 'UNK']
 200         self.putg(ss, es, ANN_WARN, text)
 201         self.invalidate_frame_details()
 202
 203     def usecs_to_samples(self, us):
 204         us *= 1e-6
 205         us *= self.samplerate
 206         return int(us)
 207
 208     def samples_to_usecs(self, n):
 209         n /= self.samplerate
 210         n *= 1000.0 * 1000.0
 211         return int(n)
 212
 213     def decode(self):
 214         if not self.samplerate:
 215             raise SamplerateError('Cannot decode without samplerate.')
 216
 217         # Get the distance between edges. Classify the distance
 218         # to derive symbols and data bit values. Prepare waiting
 219         # for an interframe gap as well, while this part of the
 220         # condition is optional (switches in and out at runtime).
 221         conds_edge = {0: 'e'}
 222         conds_edge_only = [conds_edge]
 223         conds_edge_idle = [conds_edge, {'skip': 0}]
 224         conds = conds_edge_only
 225         self.wait(conds)
 226         es = self.samplenum
 227         spd = None
 228         while True:
 229             ss = es
 230             pin, = self.wait(conds)
 231             es = self.samplenum
 232             count = es - ss
 233             t = self.samples_to_usecs(count)
 234
 235             # Synchronization to the next frame. Wait for SOF.
 236             # Silently keep synchronizing until SOF was seen.
 237             if spd is None:
 238                 if not self.matched[0]:
 239                     continue
 240                 if pin != self.active:
 241                     continue
 242
 243                 # Detect the frame's speed from the SOF length. Adjust
 244                 # the expected BIT lengths to the SOF derived speed.
 245                 # Arrange for the additional supervision of EOF/IFS.
 246                 if t in range(VPW_SOFL // 1, VPW_SOFH // 1):
 247                     spd = 1
 248                 elif t in range(VPW_SOFL // 4, VPW_SOFH // 4):
 249                     spd = 4
 250                 else:
 251                     continue
 252                 short_lower, short_upper = VPW_SHORTL // spd, VPW_SHORTH // spd
 253                 long_lower, long_upper = VPW_LONGL // spd, VPW_LONGH // spd
 254                 samples = self.usecs_to_samples(VPW_IFS // spd)
 255                 conds_edge_idle[-1]['skip'] = samples
 256                 conds = conds_edge_idle
 257
 258                 # Emit the SOF annotation. Start collecting DATA.
 259                 self.handle_sof(ss, es, spd)
 260                 continue
 261
 262             # Inside the DATA phase. Get data bits. Handle EOF/IFS.
 263             if len(conds) > 1 and self.matched[1]:
 264                 # TODO The current implementation gets here after a
 265                 # pre-determined minimum wait time. Does not differ
 266                 # between EOF and IFS. An earlier implementation had
 267                 # this developer note: EOF=239-280 IFS=281+
 268                 self.handle_eof(ss, es)
 269                 # Enter the IDLE phase. Wait for the next SOF.
 270                 spd = None
 271                 conds = conds_edge_only
 272                 continue
 273             if t in range(short_lower, short_upper):
 274                 value = 1 if pin == self.active else 0
 275                 self.handle_bit(ss, es, value)
 276                 continue
 277             if t in range(long_lower, long_upper):
 278                 value = 0 if pin == self.active else 1
 279                 self.handle_bit(ss, es, value)
 280                 continue
 281
 282             # Implementation detail: An earlier implementation used to
 283             # ignore everything that was not handled above. This would
 284             # be motivated by the noisy environment the protocol is
 285             # typically used in. This more recent implementation accepts
 286             # short glitches, but by design falls back to synchronization
 287             # to the input stream for other unhandled conditions. This
 288             # wants to improve usability of the decoder, by presenting
 289             # potential issues to the user. The threshold (microseconds
 290             # between edges that are not valid symbols that are handled
 291             # above) is an arbitrary choice.
 292             if t <= 2:
 293                 continue
 294             self.handle_unknown(ss, es)
 295             spd = None
 296             conds = conds_edge_only