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_lsb
  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):
 100         if timeout is not None:
 101             to = int(timeout * self.snum_per_us)
 102             conds.append({'skip': to})
 103         ss = self.samplenum
 104         pins = self.wait(conds)
 105         es = self.samplenum
 106         return pins, ss, es, self.matched
 107
 108     def read_pulse(self, high, time):
 109         e = 'f' if high else 'r'
 110         max_time = int(time * 1.30)
 111         (ir,), ss, es, (edge, timeout) = self.wait_wrap([{0: e}], max_time)
 112         if timeout or not self.tolerance(ss, es, time):
 113             raise SIRCError('Timeout')
 114         return ir, ss, es, (edge, timeout)
 115
 116     def read_bit(self):
 117         e = 'f' if self.active else 'r'
 118         _, high_ss, high_es, (edge, timeout) = self.wait_wrap([{0: e}], 2000)
 119         if timeout:
 120             raise SIRCError('Bit High Timeout')
 121         if self.tolerance(high_ss, high_es, ONE_USEC):
 122             bit = 1
 123         elif self.tolerance(high_ss, high_es, ZERO_USEC):
 124             bit = 0
 125         else:
 126             raise SIRCError('Bit Low Timeout')
 127         try:
 128             _, low_ss, low_es, _ = self.read_pulse(not self.active, PAUSE_USEC)
 129             good = True
 130         except SIRCError:
 131             low_es = high_es + int(PAUSE_USEC * self.snum_per_us)
 132             good = False
 133         self.putg(high_ss, low_es, Ann.BIT, ['{}'.format(bit)])
 134         return bit, high_ss, low_es, good
 135
 136     def read_signal(self):
 137         # Start code
 138         try:
 139             _, agc_ss, agc_es, _ = self.read_pulse(self.active, AGC_USEC)
 140             _, pause_ss, pause_es, _ = self.read_pulse(not self.active, PAUSE_USEC)
 141         except SIRCError:
 142             raise SIRCErrorSilent('not an SIRC message')
 143         self.putg(agc_ss, agc_es, Ann.AGC, ['AGC', 'A'])
 144         self.putg(pause_ss, pause_es, Ann.PAUSE, ['Pause', 'P'])
 145         self.putg(agc_ss, pause_es, Ann.START, ['Start', 'S'])
 146
 147         # Read bits
 148         bits = []
 149         while True:
 150             bit, ss, es, good = self.read_bit()
 151             bits.append((bit, ss, es))
 152             if len(bits) > 20:
 153                 raise SIRCError('too many bits')
 154             if not good:
 155                 if len(bits) == 12:
 156                     command = bits[0:7]
 157                     address = bits[7:12]
 158                     extended = []
 159                 elif len(bits) == 15:
 160                     command = bits[0:7]
 161                     address = bits[7:15]
 162                     extended = []
 163                 elif len(bits) == 20:
 164                     command = bits[0:7]
 165                     address = bits[7:12]
 166                     extended = bits[12:20]
 167                 else:
 168                     raise SIRCError('incorrect bits count {}'.format(len(bits)))
 169                 break
 170
 171         command_num = bitpack_lsb(command, 0)
 172         address_num = bitpack_lsb(address, 0)
 173         command_str = '0x{:02X}'.format(command_num)
 174         address_str = '0x{:02X}'.format(address_num)
 175         self.putg(command[0][1], command[-1][2], Ann.CMD, [
 176             'Command: {}'.format(command_str),
 177             'C:{}'.format(command_str),
 178         ])
 179         self.putg(address[0][1], address[-1][2], Ann.ADDR, [
 180             'Address: {}'.format(address_str),
 181             'A:{}'.format(address_str),
 182         ])
 183         extended_num = None
 184         if extended:
 185             extended_num = bitpack_lsb(extended, 0)
 186             extended_str = '0x{:02X}'.format(extended_num)
 187             self.putg(extended[0][1], extended[-1][2], Ann.EXT, [
 188                 'Extended: {}'.format(extended_str),
 189                 'E:{}'.format(extended_str),
 190             ])
 191         return address_num, command_num, extended_num, bits[0][1], bits[-1][2]
 192
 193     def decode(self):
 194         if not self.samplerate:
 195             raise SamplerateError('Cannot decode without samplerate.')
 196
 197         unknown = (['Unknown Device: ', 'UNK: '], {})
 198         while True:
 199             e = 'h' if self.active else 'l'
 200             _, _, frame_ss, _ = self.wait_wrap([{0: e}], None)
 201             try:
 202                 addr, cmd, ext, payload_ss, payload_es = self.read_signal()
 203                 names, cmds = ADDRESSES.get((addr, ext), unknown)
 204                 text = cmds.get(cmd, 'Unknown')
 205                 self.putg(frame_ss, payload_es, Ann.REMOTE, [
 206                     n + text for n in names
 207                 ])
 208             except SIRCErrorSilent as e:
 209                 pass
 210             except SIRCError as e:
 211                 self.putg(frame_ss, self.samplenum, Ann.WARN, [
 212                     'Error: {}'.format(e),
 213                     'Error',
 214                     'E',
 215                 ])