}
class Decoder(srd.Decoder):
- api_version = 2
+ api_version = 3
id = 'onewire_link'
name = '1-Wire link layer'
longname = '1-Wire serial communication bus (link layer)'
license = 'gplv2+'
inputs = ['logic']
outputs = ['onewire_link']
+ tags = ['Embedded/industrial']
channels = (
{'id': 'owr', 'name': 'OWR', 'desc': '1-Wire signal line'},
)
)
annotations = (
('bit', 'Bit'),
- ('warnings', 'Warnings'),
+ ('warning', 'Warning'),
('reset', 'Reset'),
('presence', 'Presence'),
- ('overdrive', 'Overdrive speed notifications'),
+ ('overdrive', 'Overdrive speed notification'),
)
annotation_rows = (
('bits', 'Bits', (0, 2, 3)),
)
def __init__(self):
+ self.reset()
+
+ def reset(self):
self.samplerate = None
- self.samplenum = 0
self.state = 'INITIAL'
self.present = 0
self.bit = 0
self.rise = 0
self.bit_count = -1
+ def putm(self, data):
+ self.put(0, 0, self.out_ann, data)
+
+ def putpfs(self, data):
+ self.put(self.fall, self.samplenum, self.out_python, data)
+
+ def putfs(self, data):
+ self.put(self.fall, self.samplenum, self.out_ann, data)
+
+ def putfr(self, data):
+ self.put(self.fall, self.rise, self.out_ann, data)
+
+ def putprs(self, data):
+ self.put(self.rise, self.samplenum, self.out_python, data)
+
+ def putrs(self, data):
+ self.put(self.rise, self.samplenum, self.out_ann, data)
+
def checks(self):
# Check if samplerate is appropriate.
if self.options['overdrive'] == 'yes':
return
self.samplerate = value
- def decode(self, ss, es, data):
+ def wait_falling_timeout(self, start, t):
+ # Wait until either a falling edge is seen, and/or the specified
+ # number of samples have been skipped (i.e. time has passed).
+ cnt = int((t[self.overdrive] / 1000000.0) * self.samplerate)
+ samples_to_skip = (start + cnt) - self.samplenum
+ samples_to_skip = samples_to_skip if (samples_to_skip > 0) else 0
+ return self.wait([{0: 'f'}, {'skip': samples_to_skip}])
+
+ def decode(self):
if not self.samplerate:
raise SamplerateError('Cannot decode without samplerate.')
- for (self.samplenum, (owr,)) in data:
- if self.samplenum == 0:
- self.checks()
-
+ self.checks()
+ while True:
# State machine.
if self.state == 'INITIAL': # Unknown initial state.
# Wait until we reach the idle high state.
- if owr == 0:
- continue
+ self.wait({0: 'h'})
self.rise = self.samplenum
self.state = 'IDLE'
elif self.state == 'IDLE': # Idle high state.
# Wait for falling edge.
- if owr != 0:
- continue
+ self.wait({0: 'f'})
self.fall = self.samplenum
# Get time since last rising edge.
time = ((self.fall - self.rise) / self.samplerate) * 1000000.0
if self.rise > 0 and \
time < timing['REC']['min'][self.overdrive]:
- self.put(self.fall, self.rise, self.out_ann,
- [1, ['Recovery time not long enough'
+ self.putfr([1, ['Recovery time not long enough'
'Recovery too short',
'REC < ' + str(timing['REC']['min'][self.overdrive])]])
# A reset pulse or slot can start on a falling edge.
# TODO: Check minimum recovery time.
elif self.state == 'LOW': # Reset pulse or slot.
# Wait for rising edge.
- if owr == 0:
- continue
+ self.wait({0: 'r'})
self.rise = self.samplenum
# Detect reset or slot base on timing.
time = ((self.rise - self.fall) / self.samplerate) * 1000000.0
if time >= timing['RSTL']['min'][False]: # Normal reset pulse.
if time > timing['RSTL']['max'][False]:
- self.put(self.fall, self.rise, self.out_ann,
- [1, ['Too long reset pulse might mask interrupt ' +
+ self.putfr([1, ['Too long reset pulse might mask interrupt ' +
'signalling by other devices',
'Reset pulse too long',
'RST > ' + str(timing['RSTL']['max'][False])]])
# Regular reset pulse clears overdrive speed.
if self.overdrive:
- self.put(self.fall, self.rise, self.out_ann,
- [4, ['Exiting overdrive mode', 'Overdrive off']])
+ self.putfr([4, ['Exiting overdrive mode', 'Overdrive off']])
self.overdrive = False
- self.put(self.fall, self.rise, self.out_ann,
- [2, ['Reset', 'Rst', 'R']])
+ self.putfr([2, ['Reset', 'Rst', 'R']])
self.state = 'PRESENCE DETECT HIGH'
elif self.overdrive == True and \
time >= timing['RSTL']['min'][self.overdrive] and \
time < timing['RSTL']['max'][self.overdrive]:
# Overdrive reset pulse.
- self.put(self.fall, self.rise, self.out_ann,
- [2, ['Reset', 'Rst', 'R']])
+ self.putfr([2, ['Reset', 'Rst', 'R']])
self.state = 'PRESENCE DETECT HIGH'
elif time < timing['SLOT']['max'][self.overdrive]:
# Read/write time slot.
if time < timing['LOWR']['min'][self.overdrive]:
- self.put(self.fall, self.rise, self.out_ann,
- [1, ['Low signal not long enough',
+ self.putfr([1, ['Low signal not long enough',
'Low too short',
'LOW < ' + str(timing['LOWR']['min'][self.overdrive])]])
if time < timing['LOWR']['max'][self.overdrive]:
self.state = 'SLOT'
else:
# Timing outside of known states.
- self.put(self.fall, self.rise, self.out_ann,
- [1, ['Erroneous signal', 'Error', 'Err', 'E']])
+ self.putfr([1, ['Erroneous signal', 'Error', 'Err', 'E']])
self.state = 'IDLE'
elif self.state == 'PRESENCE DETECT HIGH': # Wait for slave presence signal.
+ # Wait for a falling edge and/or presence detect signal.
+ self.wait_falling_timeout(self.rise, timing['PDH']['max'])
+
# Calculate time since rising edge.
time = ((self.samplenum - self.rise) / self.samplerate) * 1000000.0
- if owr != 0 and time < timing['PDH']['max'][self.overdrive]:
- continue
- elif owr == 0: # Presence detected.
+
+ if self.matched[0] and not self.matched[1]:
+ # Presence detected.
if time < timing['PDH']['min'][self.overdrive]:
- self.put(self.rise, self.samplenum, self.out_ann,
- [1, ['Presence detect signal is too early',
+ self.putrs([1, ['Presence detect signal is too early',
'Presence detect too early',
'PDH < ' + str(timing['PDH']['min'][self.overdrive])]])
self.fall = self.samplenum
self.state = 'PRESENCE DETECT LOW'
else: # No presence detected.
- self.put(self.rise, self.samplenum, self.out_ann,
- [3, ['Presence: false', 'Presence', 'Pres', 'P']])
- self.put(self.rise, self.samplenum, self.out_python,
- ['RESET/PRESENCE', False])
+ self.putrs([3, ['Presence: false', 'Presence', 'Pres', 'P']])
+ self.putprs(['RESET/PRESENCE', False])
self.state = 'IDLE'
elif self.state == 'PRESENCE DETECT LOW': # Slave presence signalled.
# Wait for end of presence signal (on rising edge).
- if owr == 0:
- continue
+ self.wait({0: 'r'})
# Calculate time since start of presence signal.
time = ((self.samplenum - self.fall) / self.samplerate) * 1000000.0
if time < timing['PDL']['min'][self.overdrive]:
- self.put(self.fall, self.samplenum, self.out_ann,
- [1, ['Presence detect signal is too short',
+ self.putfs([1, ['Presence detect signal is too short',
'Presence detect too short',
'PDL < ' + str(timing['PDL']['min'][self.overdrive])]])
elif time > timing['PDL']['max'][self.overdrive]:
- self.put(self.fall, self.samplenum, self.out_ann,
- [1, ['Presence detect signal is too long',
+ self.putfs([1, ['Presence detect signal is too long',
'Presence detect too long',
'PDL > ' + str(timing['PDL']['max'][self.overdrive])]])
if time > timing['RSTH']['min'][self.overdrive]:
# End states (for additional checks).
if self.state == 'SLOT': # Wait for end of time slot.
- # Calculate time since falling edge.
- time = ((self.samplenum - self.fall) / self.samplerate) * 1000000.0
- if owr != 0 and time < timing['SLOT']['min'][self.overdrive]:
- continue
- elif owr == 0: # Low detected before end of slot.
- # Warn about irregularity.
- self.put(self.fall, self.samplenum, self.out_ann,
- [1, ['Time slot not long enough',
+ # Wait for a falling edge and/or end of timeslot.
+ self.wait_falling_timeout(self.fall, timing['SLOT']['min'])
+
+ if self.matched[0] and not self.matched[1]:
+ # Low detected before end of slot.
+ self.putfs([1, ['Time slot not long enough',
'Slot too short',
'SLOT < ' + str(timing['SLOT']['min'][self.overdrive])]])
# Don't output invalid bit.
self.state = 'LOW'
else: # End of time slot.
# Output bit.
- self.put(self.fall, self.samplenum, self.out_ann,
- [0, ['Bit: %d' % self.bit, '%d' % self.bit]])
- self.put(self.fall, self.samplenum, self.out_python,
- ['BIT', self.bit])
+ self.putfs([0, ['Bit: %d' % self.bit, '%d' % self.bit]])
+ self.putpfs(['BIT', self.bit])
# Save command bits.
if self.bit_count >= 0:
self.command += (self.bit << self.bit_count)
self.state = 'IDLE'
if self.state == 'PRESENCE DETECT':
- # Wait for end of presence detect.
- # Calculate time since falling edge.
- time = ((self.samplenum - self.rise) / self.samplerate) * 1000000.0
- if owr != 0 and time < timing['RSTH']['min'][self.overdrive]:
- continue
- elif owr == 0: # Low detected before end of presence detect.
- # Warn about irregularity.
- self.put(self.fall, self.samplenum, self.out_ann,
- [1, ['Presence detect not long enough',
+ # Wait for a falling edge and/or end of presence detect.
+ self.wait_falling_timeout(self.rise, timing['RSTH']['min'])
+
+ if self.matched[0] and not self.matched[1]:
+ # Low detected before end of presence detect.
+ self.putfs([1, ['Presence detect not long enough',
'Presence detect too short',
'RTSH < ' + str(timing['RSTH']['min'][self.overdrive])]])
# Inform about presence detected.
- self.put(self.rise, self.samplenum, self.out_ann,
- [3, ['Slave presence detected', 'Slave present',
+ self.putrs([3, ['Slave presence detected', 'Slave present',
'Present', 'P']])
- self.put(self.rise, self.samplenum, self.out_python,
- ['RESET/PRESENCE', True])
+ self.putprs(['RESET/PRESENCE', True])
self.fall = self.samplenum
self.state = 'LOW'
else: # End of time slot.
# Inform about presence detected.
- self.put(self.rise, self.samplenum, self.out_ann,
- [3, ['Presence: true', 'Presence', 'Pres', 'P']])
- self.put(self.rise, self.samplenum, self.out_python,
- ['RESET/PRESENCE', True])
+ self.putrs([3, ['Presence: true', 'Presence', 'Pres', 'P']])
+ self.putprs(['RESET/PRESENCE', True])
self.rise = self.samplenum
# Start counting the first 8 bits to get the ROM command.
self.bit_count = 0
projects / libsigrokdecode.git / blobdiff