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 from common.srdhelper import bitpack
  21 from .lists import ADDRESSES
  22 import sigrokdecode as srd
  23
  24 class SamplerateError(Exception):
  25     pass
  26
  27 class SIRCError(Exception):
  28     pass
  29
  30 class SIRCErrorSilent(SIRCError):
  31     pass
  32
  33 class Decoder(srd.Decoder):
  34     api_version = 3
  35     id = 'ir_sirc'
  36     name = 'IR SIRC'
  37     longname = 'Sony IR (SIRC)'
  38     desc = 'Sony infrared remote control protocol (SIRC).'
  39     license = 'gplv2+'
  40     tags = ['IR']
  41     inputs = ['logic']
  42     outputs = []
  43     channels = (
  44         {'id': 'ir', 'name': 'IR', 'desc': 'IR data line'},
  45     )
  46     options = (
  47         {'id': 'polarity', 'desc': 'Polarity', 'default': 'active-low',
  48             'values': ('active-low', 'active-high')},
  49     )
  50     annotations = (
  51         ('bit', 'Bit'),
  52         ('agc', 'AGC'),
  53         ('pause', 'Pause'),
  54         ('start', 'Start'),
  55         ('command', 'Command'),
  56         ('address', 'Address'),
  57         ('extended', 'Extended'),
  58         ('remote', 'Remote'),
  59         ('warning', 'Warning'),
  60     )
  61     annotation_rows = (
  62         ('bits', 'Bits', (0, 1, 2)),
  63         ('fields', 'Fields', (3, 4, 5, 6)),
  64         ('remotes', 'Remotes', (7,)),
  65         ('warnings', 'Warnings', (8,)),
  66     )
  67
  68     def __init__(self):
  69         self.reset()
  70
  71     def reset(self):
  72         pass
  73
  74     def start(self):
  75         self.out_ann = self.register(srd.OUTPUT_ANN)
  76         self.active = self.options['polarity'] == 'active-high'
  77
  78     def metadata(self, key, value):
  79         if key == srd.SRD_CONF_SAMPLERATE:
  80             self.samplerate = value
  81             self.snum_per_us = self.samplerate / 1e6
  82
  83     def putg(self, ss, es, cls, texts):
  84         self.put(ss, es, self.out_ann, [cls, texts])
  85
  86     def tolerance(self, ss, es, expected):
  87         microseconds = (es - ss) / self.snum_per_us
  88         tolerance = expected * 0.30
  89         return (expected - tolerance) < microseconds < (expected + tolerance)
  90
  91     def wait_wrap(self, conds, timeout=None):
  92         conds = list(conds)
  93         if timeout is not None:
  94             to = int(timeout * self.snum_per_us)
  95             conds.append({'skip': to})
  96         ss = self.samplenum
  97         pins = self.wait(conds)
  98         es = self.samplenum
  99         return pins, ss, es, self.matched
 100
 101     def read_pulse(self, high, time):
 102         e = 'f' if high else 'r'
 103         max_time = int(time * 1.30)
 104         pins, ss, es, (edge, timeout) = self.wait_wrap([{0: e}], max_time)
 105         if timeout or not self.tolerance(ss, es, time):
 106             raise SIRCError('Timeout')
 107         return pins, ss, es, (edge, timeout)
 108
 109     def read_bit(self):
 110         e = 'f' if self.active else 'r'
 111         _, high_ss, high_es, (edge, timeout) = self.wait_wrap([{0: e}], 2000)
 112         if timeout:
 113             raise SIRCError('Bit High Timeout')
 114         if self.tolerance(high_ss, high_es, 1200):
 115             bit = 1
 116         elif self.tolerance(high_ss, high_es, 600):
 117             bit = 0
 118         else:
 119             raise SIRCError('Bit Low Timeout')
 120         try:
 121             _, low_ss, low_es, matched = self.read_pulse(not self.active, 600)
 122             good = True
 123         except SIRCError:
 124             low_es = high_es + int(600 * self.snum_per_us)
 125             good = False
 126         self.putg(high_ss, low_es, 0, ['{}'.format(bit)])
 127         return bit, high_ss, low_es, good
 128
 129     def read_signal(self):
 130         # Start code
 131         try:
 132             _, agc_ss, agc_es, matched = self.read_pulse(self.active, 2400)
 133             _, pause_ss, pause_es, matched = self.read_pulse(not self.active, 600)
 134         except SIRCError:
 135             raise SIRCErrorSilent('not an SIRC message')
 136         self.putg(agc_ss, agc_es, 1, ['AGC', 'A'])
 137         self.putg(pause_ss, pause_es, 2, ['Pause', 'P'])
 138         self.putg(agc_ss, pause_es, 3, ['Start', 'S'])
 139
 140         # Read bits
 141         bits = []
 142         while True:
 143             bit, ss, es, good = self.read_bit()
 144             bits.append((bit, ss, es))
 145             if len(bits) > 20:
 146                 raise SIRCError('too many bits')
 147             if not good:
 148                 if len(bits) == 12:
 149                     command = bits[0:7]
 150                     address = bits[7:12]
 151                     extended = []
 152                 elif len(bits) == 15:
 153                     command = bits[0:7]
 154                     address = bits[7:15]
 155                     extended = []
 156                 elif len(bits) == 20:
 157                     command = bits[0:7]
 158                     address = bits[7:12]
 159                     extended = bits[12:20]
 160                 else:
 161                     raise SIRCError('incorrect number of bits: {}'.format(len(bits)))
 162                 break
 163
 164         command_num = bitpack([b[0] for b in command])
 165         address_num = bitpack([b[0] for b in address])
 166         command_str = '0x{:02X}'.format(command_num)
 167         address_str = '0x{:02X}'.format(address_num)
 168         self.putg(command[0][1], command[-1][2], 4, [
 169             'Command: {}'.format(command_str),
 170             'C:{}'.format(command_str),
 171         ])
 172         self.putg(address[0][1], address[-1][2], 5, [
 173             'Address: {}'.format(address_str),
 174             'A:{}'.format(address_str),
 175         ])
 176         extended_num = None
 177         if extended:
 178             extended_num = bitpack([b[0] for b in extended])
 179             extended_str = '0x{:02X}'.format(extended_num)
 180             self.putg(extended[0][1], extended[-1][2], 6, [
 181                 'Extended: {}'.format(extended_str),
 182                 'E:{}'.format(extended_str),
 183             ])
 184         return address_num, command_num, extended_num, bits[0][1], bits[-1][2]
 185
 186     def decode(self):
 187         if not self.samplerate:
 188             raise SamplerateError('Cannot decode without samplerate.')
 189
 190         while True:
 191             e = 'h' if self.active else 'l'
 192             signal, ss, es, matched = self.wait_wrap([{0: e}], None)
 193             try:
 194                 address, command, extended, payload_ss, payload_es = self.read_signal()
 195                 names, commands = ADDRESSES.get((address, extended), (['Unknown Device: ', 'UNK: '], {}))
 196                 text = commands.get(command, 'Unknown')
 197                 self.putg(es, payload_es, 7, [n + text for n in names])
 198             except SIRCErrorSilent as e:
 199                 continue
 200             except SIRCError as e:
 201                 self.putg(es, self.samplenum, 8, [
 202                     'Error: {}'.format(e),
 203                     'Error',
 204                     'E',
 205                 ])
 206                 continue