decoders/ir_sirc/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2020 Tom Flanagan <knio@zkpq.ca>
   5 ##
   6 ## This program is free software; you can redistribute it and/or modify
   7 ## it under the terms of the GNU General Public License as published by
   8 ## the Free Software Foundation; either version 2 of the License, or
   9 ## (at your option) any later version.
  10 ##
  11 ## This program is distributed in the hope that it will be useful,
  12 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 ## GNU General Public License for more details.
  15 ##
  16 ## You should have received a copy of the GNU General Public License
  17 ## along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 ##
  19
  20 import sigrokdecode as srd
  21 from .lists import ADDRESSES
  22
  23 class SamplerateError(Exception):
  24     pass
  25
  26 class SIRCError(Exception):
  27     pass
  28
  29 class SIRCErrorSilent(SIRCError):
  30     pass
  31
  32 class Decoder(srd.Decoder):
  33     api_version = 3
  34     id = 'ir_sirc'
  35     name = 'IR SIRC'
  36     longname = 'Sony IR (SIRC)'
  37     desc = 'Sony infrared remote control protocol (SIRC).'
  38     license = 'gplv2+'
  39     tags = ['IR']
  40     inputs = ['logic']
  41     outputs = []
  42     channels = (
  43         {'id': 'ir', 'name': 'IR', 'desc': 'IR data line'},
  44     )
  45     options = (
  46         {'id': 'polarity', 'desc': 'Polarity', 'default': 'active-low',
  47             'values': ('active-low', 'active-high')},
  48     )
  49     annotations = (
  50         ('bit', 'Bit'),
  51         ('agc', 'AGC'),
  52         ('pause', 'Pause'),
  53         ('start', 'Start'),
  54         ('command', 'Command'),
  55         ('address', 'Address'),
  56         ('extended', 'Extended'),
  57         ('remote', 'Remote'),
  58         ('warning', 'Warning'),
  59     )
  60     annotation_rows = (
  61         ('bits', 'Bits', (0, 1, 2)),
  62         ('fields', 'Fields', (3, 4, 5, 6)),
  63         ('remotes', 'Remotes', (7,)),
  64         ('warnings', 'Warnings', (8,)),
  65     )
  66
  67     def __init__(self):
  68         self.reset()
  69
  70     def reset(self):
  71         pass
  72
  73     def start(self):
  74         self.out_ann = self.register(srd.OUTPUT_ANN)
  75         self.active = self.options['polarity'] == 'active-high'
  76
  77     def metadata(self, key, value):
  78         if key == srd.SRD_CONF_SAMPLERATE:
  79             self.samplerate = value
  80
  81     def tolerance(self, start, end, expected):
  82         microseconds = 1000000 * (end - start) / self.samplerate
  83         tolerance = expected * 0.30
  84         return (expected - tolerance) < microseconds < (expected + tolerance)
  85
  86     def wait(self, *conds, timeout=None):
  87         conds = list(conds)
  88         if timeout is not None:
  89             to = int(self.samplerate * timeout / 1000000)
  90             conds.append({'skip': to})
  91         start = self.samplenum
  92         signals = super(Decoder, self).wait(conds)
  93         end = self.samplenum
  94         return signals, start, end, self.matched
  95
  96     def read_pulse(self, high, time):
  97         e = 'f' if high else 'r'
  98         max_time = int(time * 1.30)
  99         signals, start, end, (edge, timeout) = self.wait({0: e}, timeout=max_time)
 100         if timeout or not self.tolerance(start, end, time):
 101             raise SIRCError('Timeout')
 102         return signals, start, end, (edge, timeout)
 103
 104     def read_bit(self):
 105         e = 'f' if self.active else 'r'
 106         signals, high_start, high_end, (edge, timeout) = self.wait({0:e}, timeout=2000)
 107         if timeout:
 108             raise SIRCError('Bit High Timeout')
 109         if self.tolerance(high_start, high_end, 1200):
 110             bit = 1
 111         elif self.tolerance(high_start, high_end, 600):
 112             bit = 0
 113         else:
 114             raise SIRCError('Bit Low Timeout')
 115         try:
 116             signals, low_start, low_end, matched = self.read_pulse(not self.active, 600)
 117             good = True
 118         except SIRCError:
 119             low_end = high_end + int(600 * self.samplerate / 1000000)
 120             good = False
 121         self.put(high_start, low_end, self.out_ann, [0, [str(bit)]])
 122         return bit, high_start, low_end, good
 123
 124     def read_signal(self):
 125         # Start code
 126         try:
 127             signals, agc_start, agc_end, matched = self.read_pulse(self.active, 2400)
 128             signals, pause_start, pause_end, matched = self.read_pulse(not self.active, 600)
 129         except SIRCError:
 130             raise SIRCErrorSilent('not an SIRC message')
 131         self.put(agc_start, agc_end, self.out_ann, [1, ['AGC', 'A']])
 132         self.put(pause_start, pause_end, self.out_ann, [2, ['Pause', 'P']])
 133         self.put(agc_start, pause_end, self.out_ann, [3, ['Start', 'S']])
 134
 135         # Read bits
 136         bits = []
 137         while True:
 138             bit, start, end, good = self.read_bit()
 139             bits.append((bit, start, end))
 140             if len(bits) > 20:
 141                 raise SIRCError('too many bits')
 142             if not good:
 143                 if len(bits) == 12:
 144                     command = bits[0:7]
 145                     address = bits[7:12]
 146                     extended = []
 147                 elif len(bits) == 15:
 148                     command = bits[0:7]
 149                     address = bits[7:15]
 150                     extended = []
 151                 elif len(bits) == 20:
 152                     command = bits[0:7]
 153                     address = bits[7:12]
 154                     extended = bits[12:20]
 155                 else:
 156                     raise SIRCError('incorrect number of bits: {}'.format(len(bits)))
 157                 break
 158
 159         number = lambda bits:sum(b << i for i, (b, s, e) in enumerate(bits))
 160         command_num = number(command)
 161         address_num = number(address)
 162         command_str = '0x{:02X}'.format(command_num)
 163         address_str = '0x{:02X}'.format(address_num)
 164         self.put(command[0][1], command[-1][2], self.out_ann, [4, ['Command: ' + command_str, 'C:' + command_str]])
 165         self.put(address[0][1], address[-1][2], self.out_ann, [5, ['Address: ' + address_str, 'A:' + address_str]])
 166         extended_num = None
 167         if extended:
 168             extended_num = number(extended)
 169             extended_str = '0x{:02X}'.format(extended_num)
 170             self.put(extended[0][1], extended[-1][2], self.out_ann, [6, ['Extended: ' + extended_str, 'E:' + extended_str]])
 171         return address_num, command_num, extended_num, bits[0][1], bits[-1][2]
 172
 173     def decode(self):
 174         if not self.samplerate:
 175             raise SamplerateError('Cannot decode without samplerate.')
 176
 177         while True:
 178             e = 'h' if self.active else 'l'
 179             signal, start, end, matched = self.wait({0:e})
 180             try:
 181                 address, command, extended, payload_start, payload_end = self.read_signal()
 182                 names, commands = ADDRESSES.get((address, extended), (['Unknown Device: ', 'UNK: '], {}))
 183                 text = commands.get(command, 'Unknown')
 184                 self.put(end, payload_end, self.out_ann, [7, [n + text for n in names]])
 185             except SIRCErrorSilent as e:
 186                 continue
 187             except SIRCError as e:
 188                 self.put(end, self.samplenum, self.out_ann, [8, ['Error: ' + str(e), 'Error', 'E']])
 189                 continue