decoders/onewire_link/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2017 Kevin Redon <kingkevin@cuvoodoo.info>
   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 class SamplerateError(Exception):
  23     pass
  24
  25 # Timing values in us for the signal at regular and overdrive speed.
  26 timing = {
  27     'RSTL': {
  28         'min': {
  29             False: 480.0,
  30             True: 48.0,
  31         },
  32         'max': {
  33             False: 960.0,
  34             True: 80.0,
  35         },
  36     },
  37     'RSTH': {
  38         'min': {
  39             False: 480.0,
  40             True: 48.0,
  41         },
  42     },
  43     'PDH': {
  44         'min': {
  45             False: 15.0,
  46             True: 2.0,
  47         },
  48         'max': {
  49             False: 60.0,
  50             True: 6.0,
  51         },
  52     },
  53     'PDL': {
  54         'min': {
  55             False: 60.0,
  56             True: 8.0,
  57         },
  58         'max': {
  59             False: 240.0,
  60             True: 24.0,
  61         },
  62     },
  63     'SLOT': {
  64         'min': {
  65             False: 60.0,
  66             True: 6.0,
  67         },
  68         'max': {
  69             False: 120.0,
  70             True: 16.0,
  71         },
  72     },
  73     'REC': {
  74         'min': {
  75             False: 1.0,
  76             True: 1.0,
  77         },
  78     },
  79     'LOWR': {
  80         'min': {
  81             False: 1.0,
  82             True: 1.0,
  83         },
  84         'max': {
  85             False: 15.0,
  86             True: 2.0,
  87         },
  88     },
  89 }
  90
  91 class Decoder(srd.Decoder):
  92     api_version = 2
  93     id = 'onewire_link'
  94     name = '1-Wire link layer'
  95     longname = '1-Wire serial communication bus (link layer)'
  96     desc = 'Bidirectional, half-duplex, asynchronous serial bus.'
  97     license = 'gplv2+'
  98     inputs = ['logic']
  99     outputs = ['onewire_link']
 100     channels = (
 101         {'id': 'owr', 'name': 'OWR', 'desc': '1-Wire signal line'},
 102     )
 103     options = (
 104         {'id': 'overdrive', 'desc': 'Start in overdrive speed',
 105             'default': 'no', 'values': ('yes', 'no')},
 106     )
 107     annotations = (
 108         ('bit', 'Bit'),
 109         ('warnings', 'Warnings'),
 110         ('reset', 'Reset'),
 111         ('presence', 'Presence'),
 112         ('overdrive', 'Overdrive speed notifications'),
 113     )
 114     annotation_rows = (
 115         ('bits', 'Bits', (0, 2, 3)),
 116         ('info', 'Info', (4,)),
 117         ('warnings', 'Warnings', (1,)),
 118     )
 119
 120     def __init__(self):
 121         self.samplerate = None
 122         self.samplenum = 0
 123         self.state = 'INITIAL'
 124         self.present = 0
 125         self.bit = 0
 126         self.bit_count = -1
 127         self.command = 0
 128         self.overdrive = False
 129         self.fall = 0
 130         self.rise = 0
 131
 132     def start(self):
 133         self.out_python = self.register(srd.OUTPUT_PYTHON)
 134         self.out_ann = self.register(srd.OUTPUT_ANN)
 135         self.overdrive = (self.options['overdrive'] == 'yes')
 136         self.fall = 0
 137         self.rise = 0
 138         self.bit_count = -1
 139
 140     def putm(self, data):
 141         self.put(0, 0, self.out_ann, data)
 142
 143     def putpfs(self, data):
 144         self.put(self.fall, self.samplenum, self.out_python, data)
 145
 146     def putfs(self, data):
 147         self.put(self.fall, self.samplenum, self.out_ann, data)
 148
 149     def putfr(self, data):
 150         self.put(self.fall, self.rise, self.out_ann, data)
 151
 152     def putprs(self, data):
 153         self.put(self.rise, self.samplenum, self.out_python, data)
 154
 155     def putrs(self, data):
 156         self.put(self.rise, self.samplenum, self.out_ann, data)
 157
 158     def checks(self):
 159         # Check if samplerate is appropriate.
 160         if self.options['overdrive'] == 'yes':
 161             if self.samplerate < 2000000:
 162                 self.putm([1, ['Sampling rate is too low. Must be above ' +
 163                                '2MHz for proper overdrive mode decoding.']])
 164             elif self.samplerate < 5000000:
 165                 self.putm([1, ['Sampling rate is suggested to be above 5MHz ' +
 166                                'for proper overdrive mode decoding.']])
 167         else:
 168             if self.samplerate < 400000:
 169                 self.putm([1, ['Sampling rate is too low. Must be above ' +
 170                                '400kHz for proper normal mode decoding.']])
 171             elif self.samplerate < 1000000:
 172                 self.putm([1, ['Sampling rate is suggested to be above ' +
 173                                '1MHz for proper normal mode decoding.']])
 174
 175     def metadata(self, key, value):
 176         if key != srd.SRD_CONF_SAMPLERATE:
 177             return
 178         self.samplerate = value
 179
 180     def decode(self, ss, es, data):
 181         if not self.samplerate:
 182             raise SamplerateError('Cannot decode without samplerate.')
 183         for (self.samplenum, (owr,)) in data:
 184             if self.samplenum == 0:
 185                 self.checks()
 186
 187             # State machine.
 188             if self.state == 'INITIAL': # Unknown initial state.
 189                 # Wait until we reach the idle high state.
 190                 if owr == 0:
 191                     continue
 192                 self.rise = self.samplenum
 193                 self.state = 'IDLE'
 194             elif self.state == 'IDLE': # Idle high state.
 195                 # Wait for falling edge.
 196                 if owr != 0:
 197                     continue
 198                 self.fall = self.samplenum
 199                 # Get time since last rising edge.
 200                 time = ((self.fall - self.rise) / self.samplerate) * 1000000.0
 201                 if self.rise > 0 and \
 202                     time < timing['REC']['min'][self.overdrive]:
 203                     self.putfr([1, ['Recovery time not long enough'
 204                         'Recovery too short',
 205                         'REC < ' + str(timing['REC']['min'][self.overdrive])]])
 206                 # A reset pulse or slot can start on a falling edge.
 207                 self.state = 'LOW'
 208                 # TODO: Check minimum recovery time.
 209             elif self.state == 'LOW': # Reset pulse or slot.
 210                 # Wait for rising edge.
 211                 if owr == 0:
 212                     continue
 213                 self.rise = self.samplenum
 214                 # Detect reset or slot base on timing.
 215                 time = ((self.rise - self.fall) / self.samplerate) * 1000000.0
 216                 if time >= timing['RSTL']['min'][False]: # Normal reset pulse.
 217                     if time > timing['RSTL']['max'][False]:
 218                         self.putfr([1, ['Too long reset pulse might mask interrupt ' +
 219                             'signalling by other devices',
 220                             'Reset pulse too long',
 221                             'RST > ' + str(timing['RSTL']['max'][False])]])
 222                     # Regular reset pulse clears overdrive speed.
 223                     if self.overdrive:
 224                         self.putfr([4, ['Exiting overdrive mode', 'Overdrive off']])
 225                     self.overdrive = False
 226                     self.putfr([2, ['Reset', 'Rst', 'R']])
 227                     self.state = 'PRESENCE DETECT HIGH'
 228                 elif self.overdrive == True and \
 229                     time >= timing['RSTL']['min'][self.overdrive] and \
 230                     time < timing['RSTL']['max'][self.overdrive]:
 231                     # Overdrive reset pulse.
 232                     self.putfr([2, ['Reset', 'Rst', 'R']])
 233                     self.state = 'PRESENCE DETECT HIGH'
 234                 elif time < timing['SLOT']['max'][self.overdrive]:
 235                     # Read/write time slot.
 236                     if time < timing['LOWR']['min'][self.overdrive]:
 237                         self.putfr([1, ['Low signal not long enough',
 238                             'Low too short',
 239                             'LOW < ' + str(timing['LOWR']['min'][self.overdrive])]])
 240                     if time < timing['LOWR']['max'][self.overdrive]:
 241                         self.bit = 1 # Short pulse is a 1 bit.
 242                     else:
 243                         self.bit = 0 # Long pulse is a 0 bit.
 244                     # Wait for end of slot.
 245                     self.state = 'SLOT'
 246                 else:
 247                     # Timing outside of known states.
 248                     self.putfr([1, ['Erroneous signal', 'Error', 'Err', 'E']])
 249                     self.state = 'IDLE'
 250             elif self.state == 'PRESENCE DETECT HIGH': # Wait for slave presence signal.
 251                 # Calculate time since rising edge.
 252                 time = ((self.samplenum - self.rise) / self.samplerate) * 1000000.0
 253                 if owr != 0 and time < timing['PDH']['max'][self.overdrive]:
 254                     continue
 255                 elif owr == 0: # Presence detected.
 256                     if time < timing['PDH']['min'][self.overdrive]:
 257                         self.putrs([1, ['Presence detect signal is too early',
 258                             'Presence detect too early',
 259                             'PDH < ' + str(timing['PDH']['min'][self.overdrive])]])
 260                     self.fall = self.samplenum
 261                     self.state = 'PRESENCE DETECT LOW'
 262                 else: # No presence detected.
 263                     self.putrs([3, ['Presence: false', 'Presence', 'Pres', 'P']])
 264                     self.putprs(['RESET/PRESENCE', False])
 265                     self.state = 'IDLE'
 266             elif self.state == 'PRESENCE DETECT LOW': # Slave presence signalled.
 267                 # Wait for end of presence signal (on rising edge).
 268                 if owr == 0:
 269                     continue
 270                 # Calculate time since start of presence signal.
 271                 time = ((self.samplenum - self.fall) / self.samplerate) * 1000000.0
 272                 if time < timing['PDL']['min'][self.overdrive]:
 273                     self.putfs([1, ['Presence detect signal is too short',
 274                         'Presence detect too short',
 275                         'PDL < ' + str(timing['PDL']['min'][self.overdrive])]])
 276                 elif time > timing['PDL']['max'][self.overdrive]:
 277                     self.putfs([1, ['Presence detect signal is too long',
 278                         'Presence detect too long',
 279                         'PDL > ' + str(timing['PDL']['max'][self.overdrive])]])
 280                 if time > timing['RSTH']['min'][self.overdrive]:
 281                     self.rise = self.samplenum
 282                 # Wait for end of presence detect.
 283                 self.state = 'PRESENCE DETECT'
 284
 285             # End states (for additional checks).
 286             if self.state == 'SLOT': # Wait for end of time slot.
 287                 # Calculate time since falling edge.
 288                 time = ((self.samplenum - self.fall) / self.samplerate) * 1000000.0
 289                 if owr != 0 and time < timing['SLOT']['min'][self.overdrive]:
 290                     continue
 291                 elif owr == 0: # Low detected before end of slot.
 292                     # Warn about irregularity.
 293                     self.putfs([1, ['Time slot not long enough',
 294                         'Slot too short',
 295                         'SLOT < ' + str(timing['SLOT']['min'][self.overdrive])]])
 296                     # Don't output invalid bit.
 297                     self.fall = self.samplenum
 298                     self.state = 'LOW'
 299                 else: # End of time slot.
 300                     # Output bit.
 301                     self.putfs([0, ['Bit: %d' % self.bit, '%d' % self.bit]])
 302                     self.putpfs(['BIT', self.bit])
 303                     # Save command bits.
 304                     if self.bit_count >= 0:
 305                         self.command += (self.bit << self.bit_count)
 306                         self.bit_count += 1
 307                     # Check for overdrive ROM command.
 308                     if self.bit_count >= 8:
 309                         if self.command == 0x3c or self.command == 0x69:
 310                             self.overdrive = True
 311                             self.put(self.samplenum, self.samplenum,
 312                                 self.out_ann,
 313                                 [4, ['Entering overdrive mode', 'Overdrive on']])
 314                         self.bit_count = -1
 315                     self.state = 'IDLE'
 316
 317             if self.state == 'PRESENCE DETECT':
 318                 # Wait for end of presence detect.
 319                 # Calculate time since falling edge.
 320                 time = ((self.samplenum - self.rise) / self.samplerate) * 1000000.0
 321                 if owr != 0 and time < timing['RSTH']['min'][self.overdrive]:
 322                     continue
 323                 elif owr == 0: # Low detected before end of presence detect.
 324                     # Warn about irregularity.
 325                     self.putfs([1, ['Presence detect not long enough',
 326                         'Presence detect too short',
 327                         'RTSH < ' + str(timing['RSTH']['min'][self.overdrive])]])
 328                     # Inform about presence detected.
 329                     self.putrs([3, ['Slave presence detected', 'Slave present',
 330                         'Present', 'P']])
 331                     self.putprs(['RESET/PRESENCE', True])
 332                     self.fall = self.samplenum
 333                     self.state = 'LOW'
 334                 else: # End of time slot.
 335                     # Inform about presence detected.
 336                     self.putrs([3, ['Presence: true', 'Presence', 'Pres', 'P']])
 337                     self.putprs(['RESET/PRESENCE', True])
 338                     self.rise = self.samplenum
 339                     # Start counting the first 8 bits to get the ROM command.
 340                     self.bit_count = 0
 341                     self.command = 0
 342                     self.state = 'IDLE'