X-Git-Url: http://sigrok.org/gitweb/?a=blobdiff_plain;f=decoders%2Fir_sirc%2Fpd.py;h=14ba63f0d10a75a0be3289a5111e05566bf8d832;hb=3f77dc2aae971c4ba97ad81c851c63b681074410;hp=10e5f6f62d523cc4f24d70b2518407de571b14b8;hpb=50207d809ae7181e6bb96f1611b599a1f78d27c5;p=libsigrokdecode.git diff --git a/decoders/ir_sirc/pd.py b/decoders/ir_sirc/pd.py index 10e5f6f..14ba63f 100644 --- a/decoders/ir_sirc/pd.py +++ b/decoders/ir_sirc/pd.py @@ -17,7 +17,7 @@ ## along with this program; if not, see . ## -from common.srdhelper import bitpack +from common.srdhelper import bitpack_lsb from .lists import ADDRESSES import sigrokdecode as srd @@ -30,6 +30,14 @@ class SIRCError(Exception): class SIRCErrorSilent(SIRCError): pass +class Ann: + BIT, AGC, PAUSE, START, CMD, ADDR, EXT, REMOTE, WARN = range(9) + +AGC_USEC = 2400 +ONE_USEC = 1200 +ZERO_USEC = 600 +PAUSE_USEC = 600 + class Decoder(srd.Decoder): api_version = 3 id = 'ir_sirc' @@ -59,10 +67,10 @@ class Decoder(srd.Decoder): ('warning', 'Warning'), ) annotation_rows = ( - ('bits', 'Bits', (0, 1, 2)), - ('fields', 'Fields', (3, 4, 5, 6)), - ('remotes', 'Remotes', (7,)), - ('warnings', 'Warnings', (8,)), + ('bits', 'Bits', (Ann.BIT, Ann.AGC, Ann.PAUSE)), + ('fields', 'Fields', (Ann.START, Ann.CMD, Ann.ADDR, Ann.EXT)), + ('remotes', 'Remotes', (Ann.REMOTE,)), + ('warnings', 'Warnings', (Ann.WARN,)), ) def __init__(self): @@ -88,8 +96,7 @@ class Decoder(srd.Decoder): tolerance = expected * 0.30 return (expected - tolerance) < microseconds < (expected + tolerance) - def wait_wrap(self, conds, timeout=None): - conds = list(conds) + def wait_wrap(self, conds, timeout): if timeout is not None: to = int(timeout * self.snum_per_us) conds.append({'skip': to}) @@ -101,41 +108,41 @@ class Decoder(srd.Decoder): def read_pulse(self, high, time): e = 'f' if high else 'r' max_time = int(time * 1.30) - pins, ss, es, (edge, timeout) = self.wait_wrap([{0: e}], max_time) + (ir,), ss, es, (edge, timeout) = self.wait_wrap([{0: e}], max_time) if timeout or not self.tolerance(ss, es, time): raise SIRCError('Timeout') - return pins, ss, es, (edge, timeout) + return ir, ss, es, (edge, timeout) def read_bit(self): e = 'f' if self.active else 'r' _, high_ss, high_es, (edge, timeout) = self.wait_wrap([{0: e}], 2000) if timeout: raise SIRCError('Bit High Timeout') - if self.tolerance(high_ss, high_es, 1200): + if self.tolerance(high_ss, high_es, ONE_USEC): bit = 1 - elif self.tolerance(high_ss, high_es, 600): + elif self.tolerance(high_ss, high_es, ZERO_USEC): bit = 0 else: raise SIRCError('Bit Low Timeout') try: - _, low_ss, low_es, matched = self.read_pulse(not self.active, 600) + _, low_ss, low_es, _ = self.read_pulse(not self.active, PAUSE_USEC) good = True except SIRCError: - low_es = high_es + int(600 * self.snum_per_us) + low_es = high_es + int(PAUSE_USEC * self.snum_per_us) good = False - self.putg(high_ss, low_es, 0, ['{}'.format(bit)]) + self.putg(high_ss, low_es, Ann.BIT, ['{}'.format(bit)]) return bit, high_ss, low_es, good def read_signal(self): # Start code try: - _, agc_ss, agc_es, matched = self.read_pulse(self.active, 2400) - _, pause_ss, pause_es, matched = self.read_pulse(not self.active, 600) + _, agc_ss, agc_es, _ = self.read_pulse(self.active, AGC_USEC) + _, pause_ss, pause_es, _ = self.read_pulse(not self.active, PAUSE_USEC) except SIRCError: raise SIRCErrorSilent('not an SIRC message') - self.putg(agc_ss, agc_es, 1, ['AGC', 'A']) - self.putg(pause_ss, pause_es, 2, ['Pause', 'P']) - self.putg(agc_ss, pause_es, 3, ['Start', 'S']) + self.putg(agc_ss, agc_es, Ann.AGC, ['AGC', 'A']) + self.putg(pause_ss, pause_es, Ann.PAUSE, ['Pause', 'P']) + self.putg(agc_ss, pause_es, Ann.START, ['Start', 'S']) # Read bits bits = [] @@ -158,26 +165,26 @@ class Decoder(srd.Decoder): address = bits[7:12] extended = bits[12:20] else: - raise SIRCError('incorrect number of bits: {}'.format(len(bits))) + raise SIRCError('incorrect bits count {}'.format(len(bits))) break - command_num = bitpack([b[0] for b in command]) - address_num = bitpack([b[0] for b in address]) + command_num = bitpack_lsb(command, 0) + address_num = bitpack_lsb(address, 0) command_str = '0x{:02X}'.format(command_num) address_str = '0x{:02X}'.format(address_num) - self.putg(command[0][1], command[-1][2], 4, [ + self.putg(command[0][1], command[-1][2], Ann.CMD, [ 'Command: {}'.format(command_str), 'C:{}'.format(command_str), ]) - self.putg(address[0][1], address[-1][2], 5, [ + self.putg(address[0][1], address[-1][2], Ann.ADDR, [ 'Address: {}'.format(address_str), 'A:{}'.format(address_str), ]) extended_num = None if extended: - extended_num = bitpack([b[0] for b in extended]) + extended_num = bitpack_lsb(extended, 0) extended_str = '0x{:02X}'.format(extended_num) - self.putg(extended[0][1], extended[-1][2], 6, [ + self.putg(extended[0][1], extended[-1][2], Ann.EXT, [ 'Extended: {}'.format(extended_str), 'E:{}'.format(extended_str), ]) @@ -187,20 +194,22 @@ class Decoder(srd.Decoder): if not self.samplerate: raise SamplerateError('Cannot decode without samplerate.') + unknown = (['Unknown Device: ', 'UNK: '], {}) while True: e = 'h' if self.active else 'l' - signal, ss, es, matched = self.wait_wrap([{0: e}], None) + _, _, frame_ss, _ = self.wait_wrap([{0: e}], None) try: - address, command, extended, payload_ss, payload_es = self.read_signal() - names, commands = ADDRESSES.get((address, extended), (['Unknown Device: ', 'UNK: '], {})) - text = commands.get(command, 'Unknown') - self.putg(es, payload_es, 7, [n + text for n in names]) + addr, cmd, ext, payload_ss, payload_es = self.read_signal() + names, cmds = ADDRESSES.get((addr, ext), unknown) + text = cmds.get(cmd, 'Unknown') + self.putg(frame_ss, payload_es, Ann.REMOTE, [ + n + text for n in names + ]) except SIRCErrorSilent as e: - continue + pass except SIRCError as e: - self.putg(es, self.samplenum, 8, [ + self.putg(frame_ss, self.samplenum, Ann.WARN, [ 'Error: {}'.format(e), 'Error', 'E', ]) - continue