desc = 'HDMI Consumer Electronics Control (CEC) protocol.'
license = 'gplv2+'
inputs = ['logic']
- outputs = ['cec']
+ outputs = []
+ tags = ['Display', 'PC']
channels = (
{'id': 'cec', 'name': 'CEC', 'desc': 'CEC bus data'},
)
('eom-1', 'Message continued'),
('nack', 'ACK not set'),
('ack', 'ACK set'),
- ('bits', 'Bits'),
- ('bytes', 'Bytes'),
- ('frames', 'Frames'),
- ('sections', 'Sections'),
- ('warnings', 'Warnings')
+ ('bit', 'Bit'),
+ ('byte', 'Byte'),
+ ('frame', 'Frame'),
+ ('section', 'Section'),
+ ('warning', 'Warning')
)
annotation_rows = (
('bits', 'Bits', (0, 1, 2, 3, 4, 5)),
def precalculate(self):
# Restrict max length of ACK/NACK labels to 2 BIT pulses.
- bit_time = timing[Pulse.ZERO]['total']['min']
- bit_time = bit_time * 2
+ bit_time = timing[Pulse.ZERO]['total']['min'] * 2
self.max_ack_len_samples = round((bit_time / 1000) * self.samplerate)
def reset(self):
self.samplerate = value
self.precalculate()
- def set_stat(self, stat):
- self.stat = stat
-
def handle_frame(self, is_nack):
if self.fall_start is None or self.fall_end is None:
return
i = 0
- str = ''
+ string = ''
while i < len(self.cmd_bytes):
- str += '{:02x}'.format(self.cmd_bytes[i]['val'])
+ string += '{:02x}'.format(self.cmd_bytes[i]['val'])
if i != (len(self.cmd_bytes) - 1):
- str += ':'
+ string += ':'
i += 1
- self.put(self.frame_start, self.frame_end, self.out_ann, [7, [str]])
+ self.put(self.frame_start, self.frame_end, self.out_ann, [7, [string]])
i = 0
operands = 0
- str = ''
+ string = ''
while i < len(self.cmd_bytes):
if i == 0: # Parse header
(src, dst) = decode_header(self.cmd_bytes[i]['val'])
- str = 'HDR: ' + src + ', ' + dst
+ string = 'HDR: ' + src + ', ' + dst
elif i == 1: # Parse opcode
- str += ' | OPC: ' + decode_opcode(self.cmd_bytes[i]['val'])
+ string += ' | OPC: ' + opcodes.get(self.cmd_bytes[i]['val'], 'Invalid')
else: # Parse operands
if operands == 0:
- str += ' | OPS: '
+ string += ' | OPS: '
operands += 1
- str += '0x{:02x}'.format(self.cmd_bytes[i]['val'])
+ string += '0x{:02x}'.format(self.cmd_bytes[i]['val'])
if i != len(self.cmd_bytes) - 1:
- str += ', '
+ string += ', '
i += 1
# Header only commands are PINGS
if i == 1:
- if self.eom:
- str += ' | OPC: PING'
- else:
- str += ' | OPC: NONE. Aborted cmd'
+ string += ' | OPC: PING' if self.eom else ' | OPC: NONE. Aborted cmd'
# Add extra information (ack of the command from the destination)
- if is_nack:
- str += ' | R: NACK'
- else:
- str += ' | R: ACK'
+ string += ' | R: NACK' if is_nack else ' | R: ACK'
- self.put(self.frame_start, self.frame_end, self.out_ann, [8, [str]])
+ self.put(self.frame_start, self.frame_end, self.out_ann, [8, [string]])
def process(self):
zero_time = ((self.rise - self.fall_start) / self.samplerate) * 1000.0
# VALIDATION: Invalid pulse
if pulse == Pulse.INVALID:
- self.set_stat(Stat.WAIT_START)
+ self.stat = Stat.WAIT_START
self.put(self.fall_start, self.fall_end, self.out_ann, [9, ['Invalid pulse: Wrong timing']])
return
# VALIDATION: If waiting for ACK or EOM, only BIT pulses (0/1) are expected
if (self.stat == Stat.WAIT_ACK or self.stat == Stat.WAIT_EOM) and pulse == Pulse.START:
self.put(self.fall_start, self.fall_end, self.out_ann, [9, ['Expected BIT: START received)']])
- self.set_stat(Stat.WAIT_START)
+ self.stat = Stat.WAIT_START
# VALIDATION: ACK bit pulse remains high till the next frame (if any): Validate only min time of the low period
if self.stat == Stat.WAIT_ACK and pulse != Pulse.START:
if total_time < timing[pulse]['total']['min']:
pulse = Pulse.INVALID
self.put(self.fall_start, self.fall_end, self.out_ann, [9, ['ACK pulse below minimun time']])
- self.set_stat(Stat.WAIT_START)
+ self.stat = Stat.WAIT_START
return
# VALIDATION / PING FRAME DETECTION: Initiator doesn't sets the EOM = 1 but stops sending when ack doesn't arrive
self.put(self.frame_start, self.samplenum, self.out_ann, [9, ['ERROR: Incomplete byte received']])
# Set wait start so we receive next frame
- self.set_stat(Stat.WAIT_START)
+ self.stat = Stat.WAIT_START
# VALIDATION: Check timing of the BIT (0/1) pulse in any other case (not waiting for ACK)
if self.stat != Stat.WAIT_ACK and pulse != Pulse.START:
if total_time < timing[pulse]['total']['min'] or total_time > timing[pulse]['total']['max']:
self.put(self.fall_start, self.fall_end, self.out_ann, [9, ['Bit pulse exceeds total pulse timespan']])
pulse = Pulse.INVALID
- self.set_stat(Stat.WAIT_START)
+ self.stat = Stat.WAIT_START
return
if pulse == Pulse.ZERO:
# STATE: WAIT START
if self.stat == Stat.WAIT_START:
- self.set_stat(Stat.GET_BITS)
+ self.stat = Stat.GET_BITS
self.reset_frame_vars()
self.put(self.fall_start, self.fall_end, self.out_ann, [0, ['ST']])
if self.bit_count == 8:
self.bit_count = 0
self.byte_count += 1
- self.set_stat(Stat.WAIT_EOM)
+ self.stat = Stat.WAIT_EOM
self.put(self.byte_start, self.samplenum, self.out_ann, [6, ['0x{:02x}'.format(self.byte)]])
self.cmd_bytes.append({'st': self.byte_start, 'ed': self.samplenum, 'val': self.byte})
self.eom = bit
self.frame_end = self.fall_end
- if self.eom:
- self.put(self.fall_start, self.fall_end, self.out_ann, [2, ['EOM=Y']])
- else:
- self.put(self.fall_start, self.fall_end, self.out_ann, [1, ['EOM=N']])
+ a = [2, ['EOM=Y']] if self.eom else [1, ['EOM=N']]
+ self.put(self.fall_start, self.fall_end, self.out_ann, a)
- self.set_stat(Stat.WAIT_ACK)
+ self.stat = Stat.WAIT_ACK
# STATE: WAIT ACK
elif self.stat == Stat.WAIT_ACK:
# After ACK bit, wait for new datagram or continue reading current
# one based on EOM value.
if self.eom or self.is_nack:
- self.set_stat(Stat.WAIT_START)
+ self.stat = Stat.WAIT_START
self.handle_frame(self.is_nack)
else:
- self.set_stat(Stat.GET_BITS)
+ self.stat = Stat.GET_BITS
def start(self):
self.out_ann = self.register(srd.OUTPUT_ANN)
projects / libsigrokdecode.git / blobdiff