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