Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_enocean_crc_polynomial(self):
e = Encoding()
msg1 = "aa9a6d201006401009802019e411e8035b"
msg2 = "aa9a6d2010000ffdaaf01019e411e8071b"
# Remove Preamble + SOF + EOF for CRC calculation
msg1 = util.hex2bit("a6d201006401009802019e411e8035")
crc1 = util.hex2bit("35")
msg2 = util.hex2bit("a6d2010000ffdaaf01019e411e8071")
crc2 = util.hex2bit("71")
wsp_checker = WSPChecksum()
calc_crc1 = wsp_checker.calculate(msg1)
calc_crc2 = wsp_checker.calculate(msg2)
self.assertTrue(calc_crc1 == crc1)
self.assertTrue(calc_crc2 == crc2)
def test_enocean_crc8_message(self):
e = Encoding()
received = util.hex2bit("aacbac4cddd5ddd3bddd5ddcc5ddcddd4c2d5d5c2cdddab200000")
preamble, sof, eof = "aa", "9", "b"
decoded, err, state = e.code_enocean(decoding=True, inpt=received)
self.assertEqual(err, 0)
self.assertEqual(state, e.ErrorState.SUCCESS)
self.assertIn(preamble, util.bit2hex(decoded))
self.assertIn(sof, util.bit2hex(decoded))
self.assertIn(eof, util.bit2hex(decoded))
reencoded, errors, state = e.code_enocean(decoding=False, inpt=decoded)
self.assertEqual(errors, 0)
self.assertEqual(state, e.ErrorState.SUCCESS)
redecoded, errors, state = e.code_enocean(decoding=True, inpt=reencoded)
self.assertEqual(errors, 0)
self.assertEqual(state, e.ErrorState.SUCCESS)
def test_find_crc8(self):
messages = ["aabbcc7d", "abcdee24", "dacafe33"]
message_bits = [np.array(msg, dtype=np.uint8) for msg in map(util.hex2bit, messages)]
checksum_engine = ChecksumEngine(message_bits, n_gram_length=8)
result = checksum_engine.find()
self.assertEqual(len(result), 1)
checksum_range = result[0] # type: ChecksumRange
self.assertEqual(checksum_range.length, 8)
self.assertEqual(checksum_range.start, 24)
reference = GenericCRC()
reference.set_polynomial_from_hex("0x07")
self.assertEqual(checksum_range.crc.polynomial, reference.polynomial)
self.assertEqual(checksum_range.message_indices, {0, 1, 2})
def calculate(self, msg: array.array) -> array.array:
"""
Get the checksum for a WSP message. There are three hashes possible:
1) 4 Bit Checksum - For Switch Telegram (RORG=5 or 6 and STATUS = 0x20 or 0x30)
2) 8 Bit Checksum: STATUS bit 2^7 = 0
3) 8 Bit CRC: STATUS bit 2^7 = 1
:param msg: the message without Preamble/SOF and EOF. Message starts with RORG and ends with CRC
"""
try:
if self.mode == self.ChecksumMode.auto:
if msg[0:4] == util.hex2bit("5") or msg[0:4] == util.hex2bit("6"):
# Switch telegram
return self.checksum4(msg)
status = msg[-16:-8]
if status[0]:
return self.crc8(msg[:-8]) # ignore trailing hash
else:
return self.checksum8(msg[:-8]) # ignore trailing hash
elif self.mode == self.ChecksumMode.checksum4:
return self.checksum4(msg)
elif self.mode == self.ChecksumMode.checksum8:
return self.checksum8(msg[:-8])
elif self.mode == self.ChecksumMode.crc8:
return self.crc8(msg[:-8])
elif self.code_substitution == operation:
self.src = self.chain[i + 1][0]
self.dst = self.chain[i + 1][1]
elif self.code_externalprogram == operation:
if self.chain[i + 1] != "":
try:
self.external_decoder, self.external_encoder = self.chain[i + 1].split(";")
except ValueError:
pass
else:
self.external_decoder, self.external_encoder = "", ""
elif self.code_data_whitening == operation:
if self.chain[i + 1].count(';') == 1:
self.data_whitening_sync, self.data_whitening_polynomial = self.chain[i + 1].split(";")
if (len(self.data_whitening_sync) > 0 and len(self.data_whitening_polynomial) > 0):
self.data_whitening_sync = util.hex2bit(self.data_whitening_sync)
self.data_whitening_polynomial = util.hex2bit(self.data_whitening_polynomial)
elif self.code_cut == operation:
if self.chain[i + 1] != "" and self.chain[i + 1].count(';') == 1:
self.cutmode, tmp = self.chain[i + 1].split(";")
self.cutmode = int(self.cutmode)
if self.cutmode < 0 or self.cutmode > 3:
self.cutmode = 0
if self.cutmode == 0 or self.cutmode == 1:
self.cutmark = self.str2bit(tmp)
if len(self.cutmark) == 0: self.cutmark = array.array("B", [True, False, True, False])
else:
try:
self.cutmark = int(tmp)
except ValueError:
self.cutmark = 1
elif self.code_morse == operation:
self.src = self.chain[i + 1][0]
self.dst = self.chain[i + 1][1]
elif self.code_externalprogram == operation:
if self.chain[i + 1] != "":
try:
self.external_decoder, self.external_encoder = self.chain[i + 1].split(";")
except ValueError:
pass
else:
self.external_decoder, self.external_encoder = "", ""
elif self.code_data_whitening == operation:
if self.chain[i + 1].count(';') == 1:
self.data_whitening_sync, self.data_whitening_polynomial = self.chain[i + 1].split(";")
if (len(self.data_whitening_sync) > 0 and len(self.data_whitening_polynomial) > 0):
self.data_whitening_sync = util.hex2bit(self.data_whitening_sync)
self.data_whitening_polynomial = util.hex2bit(self.data_whitening_polynomial)
elif self.code_cut == operation:
if self.chain[i + 1] != "" and self.chain[i + 1].count(';') == 1:
self.cutmode, tmp = self.chain[i + 1].split(";")
self.cutmode = int(self.cutmode)
if self.cutmode < 0 or self.cutmode > 3:
self.cutmode = 0
if self.cutmode == 0 or self.cutmode == 1:
self.cutmark = self.str2bit(tmp)
if len(self.cutmark) == 0: self.cutmark = array.array("B", [True, False, True, False])
else:
try:
self.cutmark = int(tmp)
except ValueError:
self.cutmark = 1
elif self.code_morse == operation:
if self.chain[i + 1] != "" and self.chain[i + 1].count(';') == 2:
def set_polynomial_from_hex(self, hex_str: str):
old = self.polynomial
self.polynomial = array.array("B", [1]) + util.hex2bit(hex_str)
if self.polynomial != old:
self.cache = []
self.__cache_bits = 8
elif self.code_substitution == operation:
self.src = self.chain[i + 1][0]
self.dst = self.chain[i + 1][1]
elif self.code_externalprogram == operation:
if self.chain[i + 1] != "":
try:
self.external_decoder, self.external_encoder = self.chain[i + 1].split(";")
except ValueError:
pass
else:
self.external_decoder, self.external_encoder = "", ""
elif self.code_data_whitening == operation:
if self.chain[i + 1].count(';') == 2:
self.data_whitening_sync, self.data_whitening_polynomial, overwrite_crc = self.chain[i + 1].split(";")
if (len(self.data_whitening_sync) > 0 and len(self.data_whitening_polynomial) > 0 and len(overwrite_crc) > 0):
self.data_whitening_sync = util.hex2bit(self.data_whitening_sync)
self.data_whitening_polynomial = util.hex2bit(self.data_whitening_polynomial)
self.cc1101_overwrite_crc = True if overwrite_crc == "1" else False
elif self.chain[i + 1].count(';') == 1:
self.data_whitening_sync, self.data_whitening_polynomial = self.chain[i + 1].split(";")
if (len(self.data_whitening_sync) > 0 and len(self.data_whitening_polynomial) > 0):
self.data_whitening_sync = util.hex2bit(self.data_whitening_sync)
self.data_whitening_polynomial = util.hex2bit(self.data_whitening_polynomial)
self.cc1101_overwrite_crc = False
elif self.code_cut == operation:
if self.chain[i + 1] != "" and self.chain[i + 1].count(';') == 1:
self.cutmode, tmp = self.chain[i + 1].split(";")
self.cutmode = int(self.cutmode)
if self.cutmode < 0 or self.cutmode > 3:
self.cutmode = 0
if self.cutmode == 0 or self.cutmode == 1:
self.src = self.chain[i + 1][0]
self.dst = self.chain[i + 1][1]
elif self.code_externalprogram == operation:
if self.chain[i + 1] != "":
try:
self.external_decoder, self.external_encoder = self.chain[i + 1].split(";")
except ValueError:
pass
else:
self.external_decoder, self.external_encoder = "", ""
elif self.code_data_whitening == operation:
if self.chain[i + 1].count(';') == 2:
self.data_whitening_sync, self.data_whitening_polynomial, overwrite_crc = self.chain[i + 1].split(";")
if (len(self.data_whitening_sync) > 0 and len(self.data_whitening_polynomial) > 0 and len(overwrite_crc) > 0):
self.data_whitening_sync = util.hex2bit(self.data_whitening_sync)
self.data_whitening_polynomial = util.hex2bit(self.data_whitening_polynomial)
self.cc1101_overwrite_crc = True if overwrite_crc == "1" else False
elif self.chain[i + 1].count(';') == 1:
self.data_whitening_sync, self.data_whitening_polynomial = self.chain[i + 1].split(";")
if (len(self.data_whitening_sync) > 0 and len(self.data_whitening_polynomial) > 0):
self.data_whitening_sync = util.hex2bit(self.data_whitening_sync)
self.data_whitening_polynomial = util.hex2bit(self.data_whitening_polynomial)
self.cc1101_overwrite_crc = False
elif self.code_cut == operation:
if self.chain[i + 1] != "" and self.chain[i + 1].count(';') == 1:
self.cutmode, tmp = self.chain[i + 1].split(";")
self.cutmode = int(self.cutmode)
if self.cutmode < 0 or self.cutmode > 3:
self.cutmode = 0
if self.cutmode == 0 or self.cutmode == 1:
self.cutmark = self.str2bit(tmp)