## ## This file is part of the libsigrokdecode project. ## ## Copyright (C) 2020 Tom Flanagan ## ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by ## the Free Software Foundation; either version 2 of the License, or ## (at your option) any later version. ## ## This program is distributed in the hope that it will be useful, ## but WITHOUT ANY WARRANTY; without even the implied warranty of ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ## GNU General Public License for more details. ## ## You should have received a copy of the GNU General Public License ## along with this program; if not, see . ## from common.srdhelper import bitpack from .lists import ADDRESSES import sigrokdecode as srd class SamplerateError(Exception): pass class SIRCError(Exception): pass 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' name = 'IR SIRC' longname = 'Sony IR (SIRC)' desc = 'Sony infrared remote control protocol (SIRC).' license = 'gplv2+' tags = ['IR'] inputs = ['logic'] outputs = [] channels = ( {'id': 'ir', 'name': 'IR', 'desc': 'IR data line'}, ) options = ( {'id': 'polarity', 'desc': 'Polarity', 'default': 'active-low', 'values': ('active-low', 'active-high')}, ) annotations = ( ('bit', 'Bit'), ('agc', 'AGC'), ('pause', 'Pause'), ('start', 'Start'), ('command', 'Command'), ('address', 'Address'), ('extended', 'Extended'), ('remote', 'Remote'), ('warning', 'Warning'), ) annotation_rows = ( ('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): self.reset() def reset(self): pass def start(self): self.out_ann = self.register(srd.OUTPUT_ANN) self.active = self.options['polarity'] == 'active-high' 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, ss, es, expected): microseconds = (es - ss) / self.snum_per_us tolerance = expected * 0.30 return (expected - tolerance) < microseconds < (expected + tolerance) def wait_wrap(self, conds, timeout): if timeout is not None: to = int(timeout * self.snum_per_us) conds.append({'skip': to}) 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) (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 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, ONE_USEC): bit = 1 elif self.tolerance(high_ss, high_es, ZERO_USEC): bit = 0 else: raise SIRCError('Bit Low Timeout') try: _, low_ss, low_es, _ = self.read_pulse(not self.active, PAUSE_USEC) good = True except SIRCError: low_es = high_es + int(PAUSE_USEC * self.snum_per_us) good = False 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, _ = 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, 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 = [] while True: bit, ss, es, good = self.read_bit() bits.append((bit, ss, es)) if len(bits) > 20: raise SIRCError('too many bits') if not good: if len(bits) == 12: command = bits[0:7] address = bits[7:12] extended = [] elif len(bits) == 15: command = bits[0:7] address = bits[7:15] extended = [] elif len(bits) == 20: command = bits[0:7] address = bits[7:12] extended = bits[12:20] else: 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_str = '0x{:02X}'.format(command_num) address_str = '0x{:02X}'.format(address_num) 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], 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_str = '0x{:02X}'.format(extended_num) self.putg(extended[0][1], extended[-1][2], Ann.EXT, [ 'Extended: {}'.format(extended_str), 'E:{}'.format(extended_str), ]) return address_num, command_num, extended_num, bits[0][1], bits[-1][2] def decode(self): if not self.samplerate: raise SamplerateError('Cannot decode without samplerate.') unknown = (['Unknown Device: ', 'UNK: '], {}) while True: e = 'h' if self.active else 'l' _, _, frame_ss, _ = self.wait_wrap([{0: e}], None) try: 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: pass except SIRCError as e: self.putg(frame_ss, self.samplenum, Ann.WARN, [ 'Error: {}'.format(e), 'Error', 'E', ])