## along with this program; if not, see <http://www.gnu.org/licenses/>.
##
-from common.srdhelper import bitpack
+from common.srdhelper import bitpack_lsb
from .lists import ADDRESSES
import sigrokdecode as srd
class SIRCErrorSilent(SIRCError):
pass
+class Ann:
+ BIT, AGC, PAUSE, START, CMD, ADDR, EXT, REMOTE, WARN = range(9)
+
+AGC_USEC = 2400
+ONE_USEC = 1200
+ZERO_USEC = 600
+PAUSE_USEC = 600
+
class Decoder(srd.Decoder):
api_version = 3
id = 'ir_sirc'
('warning', 'Warning'),
)
annotation_rows = (
- ('bits', 'Bits', (0, 1, 2)),
- ('fields', 'Fields', (3, 4, 5, 6)),
- ('remotes', 'Remotes', (7,)),
- ('warnings', 'Warnings', (8,)),
+ ('bits', 'Bits', (Ann.BIT, Ann.AGC, Ann.PAUSE)),
+ ('fields', 'Fields', (Ann.START, Ann.CMD, Ann.ADDR, Ann.EXT)),
+ ('remotes', 'Remotes', (Ann.REMOTE,)),
+ ('warnings', 'Warnings', (Ann.WARN,)),
)
def __init__(self):
tolerance = expected * 0.30
return (expected - tolerance) < microseconds < (expected + tolerance)
- def wait_wrap(self, conds, timeout=None):
- conds = list(conds)
+ def wait_wrap(self, conds, timeout):
if timeout is not None:
to = int(timeout * self.snum_per_us)
conds.append({'skip': to})
def read_pulse(self, high, time):
e = 'f' if high else 'r'
max_time = int(time * 1.30)
- pins, ss, es, (edge, timeout) = self.wait_wrap([{0: e}], max_time)
+ (ir,), ss, es, (edge, timeout) = self.wait_wrap([{0: e}], max_time)
if timeout or not self.tolerance(ss, es, time):
raise SIRCError('Timeout')
- return pins, ss, es, (edge, timeout)
+ return ir, ss, es, (edge, timeout)
def read_bit(self):
e = 'f' if self.active else 'r'
_, high_ss, high_es, (edge, timeout) = self.wait_wrap([{0: e}], 2000)
if timeout:
raise SIRCError('Bit High Timeout')
- if self.tolerance(high_ss, high_es, 1200):
+ if self.tolerance(high_ss, high_es, ONE_USEC):
bit = 1
- elif self.tolerance(high_ss, high_es, 600):
+ elif self.tolerance(high_ss, high_es, ZERO_USEC):
bit = 0
else:
raise SIRCError('Bit Low Timeout')
try:
- _, low_ss, low_es, matched = self.read_pulse(not self.active, 600)
+ _, low_ss, low_es, _ = self.read_pulse(not self.active, PAUSE_USEC)
good = True
except SIRCError:
- low_es = high_es + int(600 * self.snum_per_us)
+ low_es = high_es + int(PAUSE_USEC * self.snum_per_us)
good = False
- self.putg(high_ss, low_es, 0, ['{}'.format(bit)])
+ self.putg(high_ss, low_es, Ann.BIT, ['{}'.format(bit)])
return bit, high_ss, low_es, good
def read_signal(self):
# Start code
try:
- _, agc_ss, agc_es, matched = self.read_pulse(self.active, 2400)
- _, pause_ss, pause_es, matched = self.read_pulse(not self.active, 600)
+ _, agc_ss, agc_es, _ = self.read_pulse(self.active, AGC_USEC)
+ _, pause_ss, pause_es, _ = self.read_pulse(not self.active, PAUSE_USEC)
except SIRCError:
raise SIRCErrorSilent('not an SIRC message')
- self.putg(agc_ss, agc_es, 1, ['AGC', 'A'])
- self.putg(pause_ss, pause_es, 2, ['Pause', 'P'])
- self.putg(agc_ss, pause_es, 3, ['Start', 'S'])
+ self.putg(agc_ss, agc_es, Ann.AGC, ['AGC', 'A'])
+ self.putg(pause_ss, pause_es, Ann.PAUSE, ['Pause', 'P'])
+ self.putg(agc_ss, pause_es, Ann.START, ['Start', 'S'])
# Read bits
bits = []
address = bits[7:12]
extended = bits[12:20]
else:
- raise SIRCError('incorrect number of bits: {}'.format(len(bits)))
+ raise SIRCError('incorrect bits count {}'.format(len(bits)))
break
- command_num = bitpack([b[0] for b in command])
- address_num = bitpack([b[0] for b in address])
+ command_num = bitpack_lsb(command, 0)
+ address_num = bitpack_lsb(address, 0)
command_str = '0x{:02X}'.format(command_num)
address_str = '0x{:02X}'.format(address_num)
- self.putg(command[0][1], command[-1][2], 4, [
+ self.putg(command[0][1], command[-1][2], Ann.CMD, [
'Command: {}'.format(command_str),
'C:{}'.format(command_str),
])
- self.putg(address[0][1], address[-1][2], 5, [
+ self.putg(address[0][1], address[-1][2], Ann.ADDR, [
'Address: {}'.format(address_str),
'A:{}'.format(address_str),
])
extended_num = None
if extended:
- extended_num = bitpack([b[0] for b in extended])
+ extended_num = bitpack_lsb(extended, 0)
extended_str = '0x{:02X}'.format(extended_num)
- self.putg(extended[0][1], extended[-1][2], 6, [
+ self.putg(extended[0][1], extended[-1][2], Ann.EXT, [
'Extended: {}'.format(extended_str),
'E:{}'.format(extended_str),
])
if not self.samplerate:
raise SamplerateError('Cannot decode without samplerate.')
+ unknown = (['Unknown Device: ', 'UNK: '], {})
while True:
e = 'h' if self.active else 'l'
- signal, ss, es, matched = self.wait_wrap([{0: e}], None)
+ _, _, frame_ss, _ = self.wait_wrap([{0: e}], None)
try:
- address, command, extended, payload_ss, payload_es = self.read_signal()
- names, commands = ADDRESSES.get((address, extended), (['Unknown Device: ', 'UNK: '], {}))
- text = commands.get(command, 'Unknown')
- self.putg(es, payload_es, 7, [n + text for n in names])
+ addr, cmd, ext, payload_ss, payload_es = self.read_signal()
+ names, cmds = ADDRESSES.get((addr, ext), unknown)
+ text = cmds.get(cmd, 'Unknown')
+ self.putg(frame_ss, payload_es, Ann.REMOTE, [
+ n + text for n in names
+ ])
except SIRCErrorSilent as e:
- continue
+ pass
except SIRCError as e:
- self.putg(es, self.samplenum, 8, [
+ self.putg(frame_ss, self.samplenum, Ann.WARN, [
'Error: {}'.format(e),
'Error',
'E',
])
- continue