Tasmota/lib/IRremoteESP8266-2.6.5/tools/auto_analyse_raw_data.py

433 lines
16 KiB
Python
Executable File

#!/usr/bin/python
"""Attempt an automatic analysis of IRremoteESP8266's Raw data output.
Makes suggestions on key values and tried to break down the message
into likely chunks."""
#
# Copyright 2018 David Conran
import argparse
import sys
class RawIRMessage():
"""Basic analyse functions & structure for raw IR messages."""
# pylint: disable=too-many-instance-attributes
def __init__(self, margin, timings, output=sys.stdout, verbose=True):
self.hdr_mark = None
self.hdr_space = None
self.bit_mark = None
self.zero_space = None
self.one_space = None
self.gaps = []
self.margin = margin
self.marks = []
self.mark_buckets = {}
self.spaces = []
self.space_buckets = {}
self.output = output
self.verbose = verbose
if len(timings) <= 3:
raise ValueError("Too few message timings supplied.")
self.timings = timings
self._generate_timing_candidates()
self._calc_values()
def _generate_timing_candidates(self):
"""Determine the likely values from the given data."""
count = 0
for usecs in self.timings:
count = count + 1
if count % 2:
self.marks.append(usecs)
else:
self.spaces.append(usecs)
self.marks, self.mark_buckets = self.reduce_list(self.marks)
self.spaces, self.space_buckets = self.reduce_list(self.spaces)
def reduce_list(self, items):
"""Reduce a list of numbers into buckets that are at least margin apart."""
result = []
last = -1
buckets = {}
for item in sorted(items, reverse=True):
if last == -1 or item < last - self.margin:
result.append(item)
last = item
buckets[last] = [item]
else:
buckets[last].append(item)
return result, buckets
def _usec_compare(self, seen, expected):
"""Compare two usec values and see if they match within a
subtractive margin."""
return expected - self.margin < seen <= expected
def _usec_compares(self, usecs, expecteds):
"""Compare a usec value to a list of values and return True
if they are within a subtractive margin."""
for expected in expecteds:
if self._usec_compare(usecs, expected):
return True
return False
def display_binary(self, binary_str):
"""Display common representations of the suppied binary string."""
num = int(binary_str, 2)
bits = len(binary_str)
rev_binary_str = binary_str[::-1]
rev_num = int(rev_binary_str, 2)
self.output.write("\n Bits: %d\n"
" Hex: %s (MSB first)\n"
" %s (LSB first)\n"
" Dec: %s (MSB first)\n"
" %s (LSB first)\n"
" Bin: 0b%s (MSB first)\n"
" 0b%s (LSB first)\n" %
(bits, ("0x{0:0%dX}" % (bits / 4)).format(num),
("0x{0:0%dX}" % (bits / 4)).format(rev_num), num,
rev_num, binary_str, rev_binary_str))
def add_data_code(self, bin_str, footer=True):
"""Add the common "data" sequence of code to send the bulk of a message."""
# pylint: disable=no-self-use
code = []
code.append(" // Data")
code.append(" // e.g. data = 0x%X, nbits = %d" % (int(bin_str, 2),
len(bin_str)))
code.append(" sendData(kBitMark, kOneSpace, kBitMark, kZeroSpace, data, "
"nbits, true);")
if footer:
code.append(" // Footer")
code.append(" mark(kBitMark);")
return code
def _calc_values(self):
"""Calculate the values which describe the standard timings
for the protocol."""
if self.verbose:
self.output.write("Potential Mark Candidates:\n"
"%s\n"
"Potential Space Candidates:\n"
"%s\n" % (str(self.marks), str(self.spaces)))
# Largest mark is likely the kHdrMark
self.hdr_mark = self.marks[0]
# The bit mark is likely to be the smallest mark.
self.bit_mark = self.marks[-1]
if self.is_space_encoded() and len(self.spaces) >= 3:
if self.verbose and len(self.marks) > 2:
self.output.write("DANGER: Unexpected and unused mark timings!")
# We should have 3 space candidates at least.
# They should be: zero_space (smallest), one_space, & hdr_space (largest)
spaces = list(self.spaces)
self.zero_space = spaces.pop()
self.one_space = spaces.pop()
self.hdr_space = spaces.pop()
# Rest are probably message gaps
self.gaps = spaces
def is_space_encoded(self):
"""Make an educated guess if the message is space encoded."""
return len(self.spaces) > len(self.marks)
def is_hdr_mark(self, usec):
"""Is usec the header mark?"""
return self._usec_compare(usec, self.hdr_mark)
def is_hdr_space(self, usec):
"""Is usec the header space?"""
return self._usec_compare(usec, self.hdr_space)
def is_bit_mark(self, usec):
"""Is usec the bit mark?"""
return self._usec_compare(usec, self.bit_mark)
def is_one_space(self, usec):
"""Is usec the one space?"""
return self._usec_compare(usec, self.one_space)
def is_zero_space(self, usec):
"""Is usec the zero_space?"""
return self._usec_compare(usec, self.zero_space)
def is_gap(self, usec):
"""Is usec the a space gap?"""
return self._usec_compares(usec, self.gaps)
def avg_list(items):
"""Return the average of a list of numbers."""
if items:
return int(sum(items) / len(items))
return 0
def add_bit(so_far, bit, output=sys.stdout):
"""Add a bit to the end of the bits collected so far."""
if bit == "reset":
return ""
output.write(str(bit)) # This effectively displays in LSB first order.
return so_far + str(bit) # Storing it in MSB first order.
def convert_rawdata(data_str):
"""Parse a C++ rawdata declaration into a list of values."""
start = data_str.find('{')
end = data_str.find('}')
if end == -1:
end = len(data_str)
if start > end:
raise ValueError("Raw Data not parsible due to parentheses placement.")
data_str = data_str[start + 1:end]
results = []
for timing in [x.strip() for x in data_str.split(',')]:
try:
results.append(int(timing))
except ValueError:
raise ValueError(
"Raw Data contains a non-numeric value of '%s'." % timing)
return results
def dump_constants(message, defines, output=sys.stdout):
"""Dump the key constants and generate the C++ #defines."""
hdr_mark = avg_list(message.mark_buckets[message.hdr_mark])
bit_mark = avg_list(message.mark_buckets[message.bit_mark])
hdr_space = avg_list(message.space_buckets[message.hdr_space])
one_space = avg_list(message.space_buckets[message.one_space])
zero_space = avg_list(message.space_buckets[message.zero_space])
output.write("Guessing key value:\n"
"kHdrMark = %d\n"
"kHdrSpace = %d\n"
"kBitMark = %d\n"
"kOneSpace = %d\n"
"kZeroSpace = %d\n" % (hdr_mark, hdr_space, bit_mark, one_space,
zero_space))
defines.append("const uint16_t kHdrMark = %d;" % hdr_mark)
defines.append("const uint16_t kBitMark = %d;" % bit_mark)
defines.append("const uint16_t kHdrSpace = %d;" % hdr_space)
defines.append("const uint16_t kOneSpace = %d;" % one_space)
defines.append("const uint16_t kZeroSpace = %d;" % zero_space)
avg_gaps = [avg_list(message.space_buckets[x]) for x in message.gaps]
if len(message.gaps) == 1:
output.write("kSpaceGap = %d\n" % avg_gaps[0])
defines.append("const uint16_t kSpaceGap = %d;" % avg_gaps[0])
else:
count = 0
for gap in avg_gaps:
# We probably (still) have a gap in the protocol.
count = count + 1
output.write("kSpaceGap%d = %d\n" % (count, gap))
defines.append("const uint16_t kSpaceGap%d = %d;" % (count, gap))
def parse_and_report(rawdata_str, margin, gen_code=False, output=sys.stdout):
"""Analyse the rawdata c++ definition of a IR message."""
defines = []
function_code = []
# Parse the input.
rawdata = convert_rawdata(rawdata_str)
output.write("Found %d timing entries.\n" % len(rawdata))
message = RawIRMessage(margin, rawdata, output)
output.write("\nGuessing encoding type:\n")
if message.is_space_encoded():
output.write("Looks like it uses space encoding. Yay!\n\n")
dump_constants(message, defines, output)
else:
output.write("Sorry, it looks like it is Mark encoded. "
"I can't do that yet. Exiting.\n")
sys.exit(1)
total_bits = decode_data(message, defines, function_code, output)
if gen_code:
generate_irsend_code(defines, function_code, total_bits, output)
def decode_data(message, defines, function_code, output=sys.stdout):
"""Decode the data sequence with the given values in mind."""
# pylint: disable=too-many-branches,too-many-statements
# Now we have likely candidates for the key values, go through the original
# sequence and break it up and indicate accordingly.
output.write("\nDecoding protocol based on analysis so far:\n\n")
state = ""
count = 1
total_bits = ""
binary_value = add_bit("", "reset")
function_code.extend([
"// Function should be safe up to 64 bits.",
"void IRsend::sendXyz(const uint64_t data, const uint16_t"
" nbits, const uint16_t repeat) {",
" enableIROut(38); // A guess. Most common frequency.",
" for (uint16_t r = 0; r <= repeat; r++) {"
])
for usec in message.timings:
if (message.is_hdr_mark(usec) and count % 2 and
not message.is_bit_mark(usec)):
state = "HM"
if binary_value:
message.display_binary(binary_value)
function_code.extend(message.add_data_code(binary_value, False))
total_bits = total_bits + binary_value
binary_value = add_bit(binary_value, "reset")
output.write("kHdrMark+")
function_code.extend([" // Header", " mark(kHdrMark);"])
elif message.is_hdr_space(usec) and not message.is_one_space(usec):
if state != "HM":
if binary_value:
message.display_binary(binary_value)
total_bits = total_bits + binary_value
function_code.extend(message.add_data_code(binary_value))
binary_value = add_bit(binary_value, "reset")
output.write("UNEXPECTED->")
state = "HS"
output.write("kHdrSpace+")
function_code.append(" space(kHdrSpace);")
elif message.is_bit_mark(usec) and count % 2:
if state not in ("HS", "BS"):
output.write("kBitMark(UNEXPECTED)")
state = "BM"
elif message.is_zero_space(usec):
if state != "BM":
output.write("kZeroSpace(UNEXPECTED)")
state = "BS"
binary_value = add_bit(binary_value, 0, output)
elif message.is_one_space(usec):
if state != "BM":
output.write("kOneSpace(UNEXPECTED)")
state = "BS"
binary_value = add_bit(binary_value, 1, output)
elif message.is_gap(usec):
if state != "BM":
output.write("UNEXPECTED->")
state = "GS"
output.write("GAP(%d)" % usec)
if binary_value:
message.display_binary(binary_value)
function_code.extend(message.add_data_code(binary_value))
else:
function_code.extend([" // Gap", " mark(kBitMark);"])
function_code.append(" space(kSpaceGap);")
total_bits = total_bits + binary_value
binary_value = add_bit(binary_value, "reset")
else:
output.write("UNKNOWN(%d)" % usec)
state = "UNK"
count = count + 1
if binary_value:
message.display_binary(binary_value)
function_code.extend(message.add_data_code(binary_value))
function_code.extend([
" space(100000); // A 100% made up guess of the gap"
" between messages.", " }", "}"
])
total_bits = total_bits + binary_value
output.write("\nTotal Nr. of suspected bits: %d\n" % len(total_bits))
defines.append("const uint16_t kXyzBits = %d;" % len(total_bits))
if len(total_bits) > 64:
defines.append("const uint16_t kXyzStateLength = %d;" %
(len(total_bits) / 8))
return total_bits
def generate_irsend_code(defines, normal, bits_str, output=sys.stdout):
"""Output the estimated C++ code to reproduce the IR message."""
output.write("\nGenerating a VERY rough code outline:\n\n"
"// WARNING: This probably isn't directly usable."
" It's a guide only.\n")
for line in defines:
output.write("%s\n" % line)
if len(bits_str) > 64: # Will it fit in a uint64_t?
output.write("// DANGER: More than 64 bits detected. A uint64_t for "
"'data' won't work!\n")
# Display the "normal" version's code incase there are some
# oddities in it.
for line in normal:
output.write("%s\n" % line)
if len(bits_str) > 64: # Will it fit in a uint64_t?
output.write("\n\n// Alternative >64 bit Function\n"
"void IRsend::sendXyz(uint8_t data[], uint16_t nbytes,"
" uint16_t repeat) {\n"
" // nbytes should typically be kXyzStateLength\n"
" // data should typically be:\n"
" // uint8_t data[kXyzStateLength] = {0x%s};\n"
" // data[] is assumed to be in MSB order for this code.\n"
" for (uint16_t r = 0; r <= repeat; r++) {\n"
" sendGeneric(kHdrMark, kHdrSpace,\n"
" kBitMark, kOneSpace,\n"
" kBitMark, kZeroSpace,\n"
" kBitMark,\n"
" 100000, // 100%% made-up guess at the"
" message gap.\n"
" data, nbytes,\n"
" 38000, // Complete guess of the modulation"
" frequency.\n"
" true, 0, 50);\n"
" }\n"
"}\n" % ", 0x".join("%02X" % int(bits_str[i:i + 8], 2)
for i in range(0, len(bits_str), 8)))
def main():
"""Parse the commandline arguments and call the method."""
arg_parser = argparse.ArgumentParser(
description="Read an IRremoteESP8266 rawData declaration and tries to "
"analyse it.",
formatter_class=argparse.ArgumentDefaultsHelpFormatter)
arg_parser.add_argument(
"-g",
"--code",
action="store_true",
default=False,
dest="gen_code",
help="Generate a C++ code outline to aid making an IRsend function.")
arg_group = arg_parser.add_mutually_exclusive_group(required=True)
arg_group.add_argument(
"rawdata",
help="A rawData line from IRrecvDumpV2. e.g. 'uint16_t rawbuf[37] = {"
"7930, 3952, 494, 1482, 520, 1482, 494, 1508, 494, 520, 494, 1482, 494, "
"520, 494, 1482, 494, 1482, 494, 3978, 494, 520, 494, 520, 494, 520, "
"494, 520, 520, 520, 494, 520, 494, 520, 494, 520, 494};'",
nargs="?")
arg_group.add_argument(
"-f", "--file", help="Read in a rawData line from the file.")
arg_parser.add_argument(
"-r",
"--range",
type=int,
help="Max number of micro-seconds difference between values to consider"
" it the same value.",
dest="margin",
default=200)
arg_group.add_argument(
"--stdin",
help="Read in a rawData line from STDIN.",
action="store_true",
default=False)
arg_options = arg_parser.parse_args()
if arg_options.stdin:
data = sys.stdin.read()
elif arg_options.file:
with open(arg_options.file) as input_file:
data = input_file.read()
else:
data = arg_options.rawdata
parse_and_report(data, arg_options.margin, arg_options.gen_code)
if __name__ == '__main__':
main()