X-Git-Url: https://sigrok.org/gitweb/?a=blobdiff_plain;f=decoders%2Fir_sirc%2Fpd.py;h=10e5f6f62d523cc4f24d70b2518407de571b14b8;hb=50207d809ae7181e6bb96f1611b599a1f78d27c5;hp=b713c56acdc7f3f3fb2a4d23403c0409c987dda5;hpb=c449c5832bb560ca543b9d6c962c405d1a227312;p=libsigrokdecode.git diff --git a/decoders/ir_sirc/pd.py b/decoders/ir_sirc/pd.py index b713c56..10e5f6f 100644 --- a/decoders/ir_sirc/pd.py +++ b/decoders/ir_sirc/pd.py @@ -17,8 +17,9 @@ ## along with this program; if not, see . ## -import sigrokdecode as srd +from common.srdhelper import bitpack from .lists import ADDRESSES +import sigrokdecode as srd class SamplerateError(Exception): pass @@ -77,66 +78,70 @@ class Decoder(srd.Decoder): def metadata(self, key, value): if key == srd.SRD_CONF_SAMPLERATE: self.samplerate = value + self.snum_per_us = self.samplerate / 1e6 + + def putg(self, ss, es, cls, texts): + self.put(ss, es, self.out_ann, [cls, texts]) - def tolerance(self, start, end, expected): - microseconds = 1000000 * (end - start) / self.samplerate + def tolerance(self, ss, es, expected): + microseconds = (es - ss) / self.snum_per_us tolerance = expected * 0.30 return (expected - tolerance) < microseconds < (expected + tolerance) - def wait(self, *conds, timeout=None): + def wait_wrap(self, conds, timeout=None): conds = list(conds) if timeout is not None: - to = int(self.samplerate * timeout / 1000000) + to = int(timeout * self.snum_per_us) conds.append({'skip': to}) - start = self.samplenum - signals = super(Decoder, self).wait(conds) - end = self.samplenum - return signals, start, end, self.matched + ss = self.samplenum + pins = self.wait(conds) + es = self.samplenum + return pins, ss, es, self.matched def read_pulse(self, high, time): e = 'f' if high else 'r' max_time = int(time * 1.30) - signals, start, end, (edge, timeout) = self.wait({0: e}, timeout=max_time) - if timeout or not self.tolerance(start, end, time): + pins, ss, es, (edge, timeout) = self.wait_wrap([{0: e}], max_time) + if timeout or not self.tolerance(ss, es, time): raise SIRCError('Timeout') - return signals, start, end, (edge, timeout) + return pins, ss, es, (edge, timeout) def read_bit(self): e = 'f' if self.active else 'r' - signals, high_start, high_end, (edge, timeout) = self.wait({0:e}, timeout=2000) + _, high_ss, high_es, (edge, timeout) = self.wait_wrap([{0: e}], 2000) if timeout: raise SIRCError('Bit High Timeout') - if self.tolerance(high_start, high_end, 1200): + if self.tolerance(high_ss, high_es, 1200): bit = 1 - elif self.tolerance(high_start, high_end, 600): + elif self.tolerance(high_ss, high_es, 600): bit = 0 else: raise SIRCError('Bit Low Timeout') try: - signals, low_start, low_end, matched = self.read_pulse(not self.active, 600) + _, low_ss, low_es, matched = self.read_pulse(not self.active, 600) good = True except SIRCError: - low_end = high_end + int(600 * self.samplerate / 1000000) + low_es = high_es + int(600 * self.snum_per_us) good = False - self.put(high_start, low_end, self.out_ann, [0, [str(bit)]]) - return bit, high_start, low_end, good + self.putg(high_ss, low_es, 0, ['{}'.format(bit)]) + return bit, high_ss, low_es, good def read_signal(self): # Start code try: - signals, agc_start, agc_end, matched = self.read_pulse(self.active, 2400) - signals, pause_start, pause_end, matched = self.read_pulse(not self.active, 600) + _, agc_ss, agc_es, matched = self.read_pulse(self.active, 2400) + _, pause_ss, pause_es, matched = self.read_pulse(not self.active, 600) except SIRCError: raise SIRCErrorSilent('not an SIRC message') - self.put(agc_start, agc_end, self.out_ann, [1, ['AGC', 'A']]) - self.put(pause_start, pause_end, self.out_ann, [2, ['Pause', 'P']]) - self.put(agc_start, pause_end, self.out_ann, [3, ['Start', 'S']]) + 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']) # Read bits bits = [] while True: - bit, start, end, good = self.read_bit() - bits.append((bit, start, end)) + bit, ss, es, good = self.read_bit() + bits.append((bit, ss, es)) if len(bits) > 20: raise SIRCError('too many bits') if not good: @@ -156,18 +161,26 @@ class Decoder(srd.Decoder): raise SIRCError('incorrect number of bits: {}'.format(len(bits))) break - number = lambda bits:sum(b << i for i, (b, s, e) in enumerate(bits)) - command_num = number(command) - address_num = number(address) + command_num = bitpack([b[0] for b in command]) + address_num = bitpack([b[0] for b in address]) command_str = '0x{:02X}'.format(command_num) address_str = '0x{:02X}'.format(address_num) - self.put(command[0][1], command[-1][2], self.out_ann, [4, ['Command: ' + command_str, 'C:' + command_str]]) - self.put(address[0][1], address[-1][2], self.out_ann, [5, ['Address: ' + address_str, 'A:' + address_str]]) + self.putg(command[0][1], command[-1][2], 4, [ + 'Command: {}'.format(command_str), + 'C:{}'.format(command_str), + ]) + self.putg(address[0][1], address[-1][2], 5, [ + 'Address: {}'.format(address_str), + 'A:{}'.format(address_str), + ]) extended_num = None if extended: - extended_num = number(extended) + extended_num = bitpack([b[0] for b in extended]) extended_str = '0x{:02X}'.format(extended_num) - self.put(extended[0][1], extended[-1][2], self.out_ann, [6, ['Extended: ' + extended_str, 'E:' + extended_str]]) + self.putg(extended[0][1], extended[-1][2], 6, [ + 'Extended: {}'.format(extended_str), + 'E:{}'.format(extended_str), + ]) return address_num, command_num, extended_num, bits[0][1], bits[-1][2] def decode(self): @@ -176,14 +189,18 @@ class Decoder(srd.Decoder): while True: e = 'h' if self.active else 'l' - signal, start, end, matched = self.wait({0:e}) + signal, ss, es, matched = self.wait_wrap([{0: e}], None) try: - address, command, extended, payload_start, payload_end = self.read_signal() + 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.put(end, payload_end, self.out_ann, [7, [n + text for n in names]]) + self.putg(es, payload_es, 7, [n + text for n in names]) except SIRCErrorSilent as e: continue except SIRCError as e: - self.put(end, self.samplenum, self.out_ann, [8, ['Error: ' + str(e), 'Error', 'E']]) + self.putg(es, self.samplenum, 8, [ + 'Error: {}'.format(e), + 'Error', + 'E', + ]) continue