decoders/swim/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2018 Mike Jagdis <mjagdis@eris-associates.co.uk>
   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 math
  21 import sigrokdecode as srd
  22
  23 class SamplerateError(Exception):
  24     pass
  25
  26 class Decoder(srd.Decoder):
  27     api_version = 3
  28     id = 'swim'
  29     name = 'SWIM'
  30     longname = 'STM8 SWIM bus'
  31     desc = 'STM8 Single Wire Interface Module (SWIM) protocol.'
  32     license = 'gplv2+'
  33     inputs = ['logic']
  34     outputs = []
  35     tags = ['Debug/trace']
  36     options = (
  37         {'id': 'debug', 'desc': 'Debug', 'default': 'no', 'values': ('yes', 'no') },
  38     )
  39     channels = (
  40         {'id': 'swim', 'name': 'SWIM', 'desc': 'SWIM data line'},
  41     )
  42     annotations = (
  43         ('bit', 'Bit'),
  44         ('enterseq', 'SWIM enter sequence'),
  45         ('start-host', 'Start bit (host)'),
  46         ('start-target', 'Start bit (target)'),
  47         ('parity', 'Parity bit'),
  48         ('ack', 'Acknowledgement'),
  49         ('nack', 'Negative acknowledgement'),
  50         ('byte-write', 'Byte write'),
  51         ('byte-read', 'Byte read'),
  52         ('cmd-unknown', 'Unknown SWIM command'),
  53         ('cmd', 'SWIM command'),
  54         ('bytes', 'Byte count'),
  55         ('address', 'Address'),
  56         ('data-write', 'Data write'),
  57         ('data-read', 'Data read'),
  58         ('debug-msg', 'Debug message'),
  59     )
  60     annotation_rows = (
  61         ('bits', 'Bits', (0,)),
  62         ('framing', 'Framing', (2, 3, 4, 5, 6, 7, 8)),
  63         ('protocol', 'Protocol', (1, 9, 10, 11, 12, 13, 14)),
  64         ('debug', 'Debug', (15,)),
  65     )
  66     binary = (
  67         ('tx', 'Dump of data written to target'),
  68         ('rx', 'Dump of data read from target'),
  69     )
  70
  71     def __init__(self):
  72         # SWIM clock for the target is normally HSI/2 where HSI is 8MHz +- 5%
  73         # although the divisor can be removed by setting the SWIMCLK bit in
  74         # the CLK_SWIMCCR register. There is no standard for the host so we
  75         # will be generous and assume it is using an 8MHz +- 10% oscillator.
  76         # We do not need to be accurate. We just need to avoid treating enter
  77         # sequence pulses as bits. A synchronization frame will cause this
  78         # to be adjusted.
  79         self.HSI = 8000000
  80         self.HSI_min = self.HSI * 0.9
  81         self.HSI_max = self.HSI * 1.1
  82         self.swim_clock = self.HSI_min / 2
  83
  84         self.eseq_edge = [[-1, None], [-1, None]]
  85         self.eseq_pairnum = 0
  86         self.eseq_pairstart = None
  87
  88         self.reset()
  89
  90     def reset(self):
  91         self.bit_edge = [[-1, None], [-1, None]]
  92         self.bit_maxlen = -1
  93         self.bitseq_len = 0
  94         self.bitseq_end = None
  95         self.proto_state = 'CMD'
  96
  97     def metadata(self, key, value):
  98         if key == srd.SRD_CONF_SAMPLERATE:
  99             self.samplerate = value
 100
 101     def adjust_timings(self):
 102         # A low-speed bit is 22 SWIM clocks long.
 103         # There are options to shorten bits to 10 clocks or use HSI rather
 104         # than HSI/2 as the SWIM clock but the longest valid bit should be no
 105         # more than this many samples. This does not need to be accurate.
 106         # It exists simply to prevent bits extending unecessarily far into
 107         # trailing bus-idle periods. This will be adjusted every time we see
 108         # a synchronization frame or start bit in order to show idle periods
 109         # as accurately as possible.
 110         self.bit_reflen = math.ceil(self.samplerate * 22 / self.swim_clock)
 111
 112     def start(self):
 113         self.out_ann = self.register(srd.OUTPUT_ANN)
 114         self.out_binary = self.register(srd.OUTPUT_BINARY)
 115
 116         if not self.samplerate:
 117             raise SamplerateError('Cannot decode without samplerate.')
 118
 119         # A synchronization frame is a low that lasts for more than 64 but no
 120         # more than 128 SWIM clock periods based on the standard SWIM clock.
 121         # Note: we also allow for the possibility that the SWIM clock divisor
 122         # has been disabled here.
 123         self.sync_reflen_min = math.floor(self.samplerate * 64 / self.HSI_max)
 124         self.sync_reflen_max = math.ceil(self.samplerate * 128 / (self.HSI_min / 2))
 125
 126         self.debug = True if self.options['debug'] == 'yes' else False
 127
 128         # The SWIM entry sequence is 4 pulses at 2kHz followed by 4 at 1kHz.
 129         self.eseq_reflen = math.ceil(self.samplerate / 2048)
 130
 131         self.adjust_timings()
 132
 133     def protocol(self):
 134         if self.proto_state == 'CMD':
 135             # Command
 136             if self.bitseq_value == 0x00:
 137                 self.put(self.bitseq_start, self.bitseq_end, self.out_ann, [10, ['system reset', 'SRST', '!']])
 138             elif self.bitseq_value == 0x01:
 139                 self.proto_state = 'N'
 140                 self.put(self.bitseq_start, self.bitseq_end, self.out_ann, [10, ['read on-the-fly', 'ROTF', 'r']])
 141             elif self.bitseq_value == 0x02:
 142                 self.proto_state = 'N'
 143                 self.put(self.bitseq_start, self.bitseq_end, self.out_ann, [10, ['write on-the-fly', 'WOTF', 'w']])
 144             else:
 145                 self.put(self.bitseq_start, self.bitseq_end, self.out_ann, [9, ['unknown', 'UNK']])
 146         elif self.proto_state == 'N':
 147             # Number of bytes
 148             self.proto_byte_count = self.bitseq_value
 149             self.proto_state = '@E'
 150             self.put(self.bitseq_start, self.bitseq_end, self.out_ann, [11, ['byte count 0x%02x' % self.bitseq_value, 'bytes 0x%02x' % self.bitseq_value, '0x%02x' % self.bitseq_value, '%02x' % self.bitseq_value, '%x' % self.bitseq_value]])
 151         elif self.proto_state == '@E':
 152             # Address byte 1
 153             self.proto_addr = self.bitseq_value
 154             self.proto_addr_start = self.bitseq_start
 155             self.proto_state = '@H'
 156         elif self.proto_state == '@H':
 157             # Address byte 2
 158             self.proto_addr = (self.proto_addr << 8) | self.bitseq_value
 159             self.proto_state = '@L'
 160         elif self.proto_state == '@L':
 161             # Address byte 3
 162             self.proto_addr = (self.proto_addr << 8) | self.bitseq_value
 163             self.proto_state = 'D'
 164             self.put(self.proto_addr_start, self.bitseq_end, self.out_ann, [12, ['address 0x%06x' % self.proto_addr, 'addr 0x%06x' % self.proto_addr, '0x%06x' % self.proto_addr, '%06x' %self.proto_addr, '%x' % self.proto_addr]])
 165         else:
 166             if self.proto_byte_count > 0:
 167                 self.proto_byte_count -= 1
 168                 if self.proto_byte_count == 0:
 169                     self.proto_state = 'CMD'
 170
 171             self.put(self.bitseq_start, self.bitseq_end, self.out_ann, [13 + self.bitseq_dir, ['0x%02x' % self.bitseq_value, '%02x' % self.bitseq_value, '%x' % self.bitseq_value]])
 172             self.put(self.bitseq_start, self.bitseq_end, self.out_binary, [0 + self.bitseq_dir, bytes([self.bitseq_value])])
 173             if self.debug:
 174                 self.put(self.bitseq_start, self.bitseq_end, self.out_ann, [15, ['%d more' % self.proto_byte_count, '%d' % self.proto_byte_count]])
 175
 176     def bitseq(self, bitstart, bitend, bit):
 177         if self.bitseq_len == 0:
 178             # Looking for start of a bit sequence (command or byte).
 179             self.bit_reflen = bitend - bitstart
 180             self.bitseq_value = 0
 181             self.bitseq_dir = bit
 182             self.bitseq_len = 1
 183             self.put(bitstart, bitend, self.out_ann, [2 + self.bitseq_dir, ['start', 's']])
 184         elif (self.proto_state == 'CMD' and self.bitseq_len == 4) or (self.proto_state != 'CMD' and self.bitseq_len == 9):
 185             # Parity bit
 186             self.bitseq_end = bitstart
 187             self.bitseq_len += 1
 188
 189             self.put(bitstart, bitend, self.out_ann, [4, ['parity', 'par', 'p']])
 190
 191             # The start bit is not data but was used for parity calculation.
 192             self.bitseq_value &= 0xff
 193             self.put(self.bitseq_start, self.bitseq_end, self.out_ann, [7 + self.bitseq_dir, ['0x%02x' % self.bitseq_value, '%02x' % self.bitseq_value, '%x' % self.bitseq_value]])
 194         elif (self.proto_state == 'CMD' and self.bitseq_len == 5) or (self.proto_state != 'CMD' and self.bitseq_len == 10):
 195             # ACK/NACK bit.
 196             if bit:
 197                 self.put(bitstart, bitend, self.out_ann, [5, ['ack', 'a']])
 198             else:
 199                 self.put(bitstart, bitend, self.out_ann, [6, ['nack', 'n']])
 200
 201             # We only pass data that was ack'd up the stack.
 202             if bit:
 203                 self.protocol()
 204
 205             self.bitseq_len = 0
 206         else:
 207             if self.bitseq_len == 1:
 208                 self.bitseq_start = bitstart
 209             self.bitseq_value = (self.bitseq_value << 1) | bit
 210             self.bitseq_len += 1
 211
 212     def bit(self, start, mid, end):
 213         if mid - start >= end - mid:
 214             self.put(start, end, self.out_ann, [0, ['0']])
 215             bit = 0
 216         else:
 217             self.put(start, end, self.out_ann, [0, ['1']])
 218             bit = 1
 219
 220         self.bitseq(start, end, bit)
 221
 222     def detect_synchronize_frame(self, start, end):
 223         # Strictly speaking, synchronization frames are only recognised when
 224         # SWIM is active. A falling edge on reset disables SWIM and an enter
 225         # sequence is needed to re-enable it. However we do not want to be
 226         # reliant on seeing the NRST pin just for that and we also want to be
 227         # able to decode SWIM even if we just sample parts of the dialogue.
 228         # For this reason we limit ourselves to only recognizing
 229         # synchronization frames that have believable lengths based on our
 230         # knowledge of the range of possible SWIM clocks.
 231         if self.samplenum - self.eseq_edge[1][1] >= self.sync_reflen_min and self.samplenum - self.eseq_edge[1][1] <= self.sync_reflen_max:
 232             self.put(self.eseq_edge[1][1], self.samplenum, self.out_ann, [1, ['synchronization frame', 'synchronization', 'sync', 's']])
 233
 234             # A low that lasts for more than 64 SWIM clock periods causes a
 235             # reset of the SWIM communication state machine and will switch
 236             # the SWIM to low-speed mode (SWIM_CSR.HS is cleared).
 237             self.reset()
 238
 239             # The low SHOULD last 128 SWIM clocks. This is used to
 240             # resynchronize in order to allow for variation in the frequency
 241             # of the internal RC oscillator.
 242             self.swim_clock = 128 * (self.samplerate / (self.samplenum - self.eseq_edge[1][1]))
 243             self.adjust_timings()
 244
 245     def eseq_potential_start(self, start, end):
 246         self.eseq_pairstart = start
 247         self.eseq_reflen = end - start
 248         self.eseq_pairnum = 1
 249
 250     def detect_enter_sequence(self, start, end):
 251         # According to the spec the enter sequence is four pulses at 2kHz
 252         # followed by four at 1kHz. We do not check the frequency but simply
 253         # check the lengths of successive pulses against the first. This means
 254         # we have no need to account for the accuracy (or lack of) of the
 255         # host's oscillator.
 256         if self.eseq_pairnum == 0 or abs(self.eseq_reflen - (end - start)) > 2:
 257             self.eseq_potential_start(start, end)
 258
 259         elif self.eseq_pairnum < 4:
 260             # The next three pulses should be the same length as the first.
 261             self.eseq_pairnum += 1
 262
 263             if self.eseq_pairnum == 4:
 264                 self.eseq_reflen /= 2
 265         else:
 266             # The final four pulses should each be half the length of the
 267             # initial pair. Again, a mismatch causes us to reset and use the
 268             # current pulse as a new potential enter sequence start.
 269             self.eseq_pairnum += 1
 270             if self.eseq_pairnum == 8:
 271                 # Four matching pulses followed by four more that match each
 272                 # other but are half the length of the first 4. SWIM is active!
 273                 self.put(self.eseq_pairstart, end, self.out_ann, [1, ['enter sequence', 'enter seq', 'enter', 'ent', 'e']])
 274                 self.eseq_pairnum = 0
 275
 276     def decode(self):
 277         while True:
 278             if self.bit_maxlen >= 0:
 279                 (swim,) = self.wait()
 280                 self.bit_maxlen -= 1
 281             else:
 282                 (swim,) = self.wait({0: 'e'})
 283
 284             if swim != self.eseq_edge[1][0]:
 285                 if swim == 1 and self.eseq_edge[1][1] is not None:
 286                     self.detect_synchronize_frame(self.eseq_edge[1][1], self.samplenum)
 287                     if self.eseq_edge[0][1] is not None:
 288                         self.detect_enter_sequence(self.eseq_edge[0][1], self.samplenum)
 289                 self.eseq_edge.pop(0)
 290                 self.eseq_edge.append([swim, self.samplenum])
 291
 292             if (swim != self.bit_edge[1][0] and (swim != 1 or self.bit_edge[1][0] != -1)) or self.bit_maxlen == 0:
 293                 if self.bit_maxlen == 0 and self.bit_edge[1][0] == 1:
 294                     swim = -1
 295
 296                 if self.bit_edge[1][0] != 0 and swim == 0:
 297                     self.bit_maxlen = self.bit_reflen
 298
 299                 if self.bit_edge[0][0] == 0 and self.bit_edge[1][0] == 1 and self.samplenum - self.bit_edge[0][1] <= self.bit_reflen + 2:
 300                     self.bit(self.bit_edge[0][1], self.bit_edge[1][1], self.samplenum)
 301
 302                 self.bit_edge.pop(0)
 303                 self.bit_edge.append([swim, self.samplenum])