Crash_Override/FreeDATA
1
0
Fork 0
mirror of https://github.com/DJ2LS/FreeDATA synced 2024-05-14 08:04:33 +00:00
Code Issues Projects Releases Packages Wiki Activity

WIP ARQ - final crc check

Browse source
This commit is contained in:
DJ2LS 2023-12-13 23:07:06 +01:00
parent 463c8c3829
commit e357c6a1dc
2 changed files with 18 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
modem/arq_session_irs.py
View file

@ -2,6 +2,7 @@ import threading
import data_frame_factory
import queue
import arq_session
import helpers
class ARQSessionIRS(arq_session.ARQSession):
@ -52,6 +53,9 @@ class ARQSessionIRS(arq_session.ARQSession):
def _all_data_received(self):
return self.received_bytes == len(self.received_data)
def _final_crc_check(self):
return self.received_crc == helpers.get_crc_32(bytes(self.received_data)).hex()
def handshake_session(self):
if self.state in [self.STATE_CONN_REQ_RECEIVED, self.STATE_WAITING_INFO]:
self.send_open_ack()
@ -86,8 +90,13 @@ class ARQSessionIRS(arq_session.ARQSession):
retries -= 1
if self._all_data_received():
self.set_state(self.STATE_ENDED)
else:
if self._final_crc_check():
self.set_state(self.STATE_ENDED)
else:
self.logger.warning("CRC check failed.")
self.set_state(self.STATE_FAILED)
else:
self.set_state(self.STATE_FAILED)

11
modem/helpers.py
View file

@ -49,6 +49,9 @@ def get_crc_8(data: str) -> bytes:
Returns:
CRC-8 (CCITT) of the provided data as bytes
"""
if not isinstance(data, (bytes)) or isinstance(data, (bytearray)):
data = bytes(data,"utf-8")
crc_algorithm = crcengine.new("crc8-ccitt") # load crc8 library
crc_data = crc_algorithm(data)
crc_data = crc_data.to_bytes(1, byteorder="big")
@ -68,7 +71,7 @@ def get_crc_16(data: str) -> bytes:
Returns:
CRC-16 (CCITT) of the provided data as bytes
"""
if not isinstance(data, (bytes)):
if not isinstance(data, (bytes)) or isinstance(data, (bytearray)):
data = bytes(data,"utf-8")
crc_algorithm = crcengine.new("crc16-ccitt-false") # load crc16 library
@ -88,7 +91,7 @@ def get_crc_24(data: str) -> bytes:
Returns:
CRC-24 (OpenPGP) of the provided data as bytes
"""
if not isinstance(data, (bytes)):
if not isinstance(data, (bytes)) or isinstance(data, (bytearray)):
data = bytes(data,'utf-8')
crc_algorithm = crcengine.create(
0x864CFB,
@ -115,8 +118,8 @@ def get_crc_32(data: str) -> bytes:
Returns:
CRC-32 of the provided data as bytes
"""
if not isinstance(data, (bytes)):
data = bytes(data,"utf-8")
if not isinstance(data, (bytes)) or isinstance(data, (bytearray)):
data = bytes(data, "utf-8")
crc_algorithm = crcengine.new("crc32") # load crc32 library
return crc_algorithm(data).to_bytes(4, byteorder="big")