Skip to content

Refactor: Simplify find_frame logic #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
insalt-glitch wants to merge 4 commits into
base: master
Choose a base branch
from
+22 −30
Open

Refactor: Simplify find_frame logic #7

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e8ba6fa
MAIT: Required buffer length for frame
Apr 3, 2026
cd6c0a2
MAIT: Use buffer.find and check explicitly for match
Apr 3, 2026
a1932db
MAIT: Remove impossible try catch for struct_error
Apr 3, 2026
6052858
MAIT: Remove commented-out code
Apr 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 22 additions & 30 deletions pyACS/acs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -380,37 +380,29 @@ def find_frame(self, buffer):
buffer_post_frame: buffer left after the frame
buffer_pre_frame: buffer preceding the first frame returned (likely unknown frame header)
"""
try:
# Look for registration bytes
i = buffer.index(self.REGISTRATION_BYTES)
# Take care of special case when checksum + pad byte or just checksum = \xff\x00
# It's unlikely that the full packet length is equal to \xff\x00 = 65280
while buffer.find(self.REGISTRATION_BYTES, i + 2, i + 2 + self.REGISTRATION_BYTES_LENGTH) != -1:
i += 2
# Get Length of frame (following 2 bytes, already know it from device file)
# frame_length = unpack_from('!H', buffer, offset=i + self.REGISTRATION_BYTES_LENGTH)
frame_end_index = i + self.frame_length
# Get frame
frame = buffer[i:frame_end_index]
if len(frame) != self.frame_length:
return bytearray(), None, buffer, bytearray()
# Get Checksum
checksum = buffer[frame_end_index:frame_end_index + 2]
if len(checksum) != 2:
return bytearray(), None, buffer, bytearray()
# Check checksum
if not self.valid_frame(frame, checksum):
# Error in frame, remove registration bytes and attempt again
return frame, False, buffer[i+self.REGISTRATION_BYTES_LENGTH:],\
buffer[:i+self.REGISTRATION_BYTES_LENGTH]
# Pad byte is not always present... (only +2 for checksum)
return frame, True, buffer[frame_end_index + 2:], buffer[:i]
except ValueError:
# Look for registration bytes
i = buffer.find(self.REGISTRATION_BYTES)
if i == -1:
# No registration byte found
return bytearray(), None, buffer, bytearray()
except struct_error:
# Buffer is too short to unpack packet length
return bytearray(), None, buffer, bytearray()
return bytearray(), False, buffer, bytearray()
# Take care of special case when checksum + pad byte or just checksum = \xff\x00
# It's unlikely that the full packet length is equal to \xff\x00 = 65280
while buffer.find(self.REGISTRATION_BYTES, i + 2, i + 2 + self.REGISTRATION_BYTES_LENGTH) != -1:
i += 2
frame_end_index = i + self.frame_length
# Make sure buffer is long enough (incl. 2-byte checksum)
if len(buffer) < frame_end_index + 2:
return bytearray(), False, buffer, bytearray()
# Get frame and checksum
frame = buffer[i:frame_end_index]
checksum = buffer[frame_end_index:frame_end_index + 2]
# Check checksum
if not self.valid_frame(frame, checksum):
# Error in frame, remove registration bytes and attempt again
return frame, False, buffer[i+self.REGISTRATION_BYTES_LENGTH:],\
buffer[:i+self.REGISTRATION_BYTES_LENGTH]
# Pad byte is not always present... (only +2 for checksum)
return frame, True, buffer[frame_end_index + 2:], buffer[:i]

@staticmethod
def valid_frame(frame, checksum_received):
Expand Down