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