Skip to content

Fixed _sync_read. #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

Merged
alessiodam merged 3 commits into from
Mar 30, 2026
Merged

Fixed _sync_read. #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
11c5d4c
Fixed _sync_read.
Mar 23, 2026
2bffa43
Fixed signed-word issues.
Mar 23, 2026
99d2c71
Fixed errors pointed out by Copilot.
thvoigtmann Mar 27, 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
18 changes: 9 additions & 9 deletions src/python_st3215/decorators.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,19 +44,19 @@ def wrapper(self: Any, value: int, *args: Any, **kwargs: Any) -> Any:
def encode_signed_word(value: int) -> tuple[int, int]:
"""
Encode a signed 16-bit value to low and high bytes.
Encoding uses most-significant bit as sign bit.

Args:
value: Signed integer (-32768 to 32767)
value: Signed integer (-32767 to 32767)

Returns:
Tuple of (low_byte, high_byte)
"""
if value < -32767 or value > 32767:
raise ValueError
low, high = abs(value) & 0xFF, (abs(value) >> 8) & 0x7F
if value < 0:
raw = 65536 + value
else:
raw = value
low = raw & 0xFF
high = (raw >> 8) & 0xFF
high |= 0x80
return low, high
Comment on lines 44 to 60
Copy link

Copilot AI Mar 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

encode_signed_word will silently produce an incorrect encoding for out-of-range values (e.g., -32768 becomes 0x8000, which decodes back to 0 with the new sign-bit scheme). Since this helper is callable outside the register methods that have @validate_value_range, it should defensively validate the input range (±32767) and raise a clear ValueError if exceeded.

Copilot uses AI. Check for mistakes.


Expand All @@ -65,13 +65,13 @@ def decode_signed_word(raw: int) -> int:
Decode a 16-bit raw value to signed integer.

Args:
raw: Raw 16-bit value (0-65535)
raw: Raw 16-bit value (0-65535) with dedicated sign bit.

Returns:
Signed integer (-32768 to 32767)
Signed integer (-32767 to 32767)
"""
if raw & 0x8000:
return raw - 65536
return -(raw & 0x7FFF)
return raw


Expand Down
4 changes: 2 additions & 2 deletions src/python_st3215/registers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1268,7 +1268,7 @@ def read_current_speed(self) -> Optional[int]:
Returns:
int: Current speed in steps/s. Returns None if read fails.
"""
return read_word(self.servo, 0x3A)
return read_word(self.servo, 0x3A, signed=True)

def sync_read_current_speed(self, servo_ids: list[int]) -> dict[int, Optional[int]]:
"""
Expand All @@ -1295,7 +1295,7 @@ def sync_read_current_speed(self, servo_ids: list[int]) -> dict[int, Optional[in
raw = data[0] | (data[1] << 8)
else:
raw = data[0] | (data[1] << 8)
results[servo_id] = raw
results[servo_id] = decode_signed_word(raw)
else:
results[servo_id] = None
return results
Expand Down
28 changes: 17 additions & 11 deletions src/python_st3215/st3215.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -447,20 +447,26 @@ def _sync_read(
)
parameters = [address, data_length, *servo_ids]
packet = self.send_instruction(0xFE, Instruction.SYNC_READ, parameters)
responses: dict[int, Optional[dict[str, object]]] = {}
for servo_id in servo_ids:
response = self.read_response(packet)
if response:
parsed = self.parse_response(response)
self.logger.debug(
f"Servo {servo_id}: received SYNC READ response {parsed}"
)
responses[servo_id] = parsed
else:
responses: dict[int, Optional[dict[str, object]]] = {
servo_id: None for servo_id in servo_ids
}
rx = self.read_response(packet)
if rx is None:
return responses
b = 0
Comment on lines +453 to +456
Copy link

Copilot AI Mar 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_sync_read is annotated to return a dict, but when read_response() returns None this now returns None. Callers (e.g., sync_read_current_location/speed/...) immediately call .items() on the result, which will raise at runtime and should also fail mypy. Consider always returning a dict (e.g., pre-fill {servo_id: None for servo_id in servo_ids} and return that on timeout) or update the return type and fix all call sites accordingly.

Copilot uses AI. Check for mistakes.
while b+3 < len(rx) and rx[b]==0xFF and rx[b+1]==0xFF:
servo_id = rx[b+2]
paramlen = rx[b+3]
pkglen = paramlen + 4
if paramlen != data_length + 2:
self.logger.warning(
f"Servo {servo_id}: no response received for SYNC READ"
f"Servo {servo_id}: no valid response for SYNC READ"
)
responses[servo_id] = None
break
Comment on lines +461 to +466
Copy link

Copilot AI Mar 30, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In _sync_read, when a packet’s paramlen doesn’t match data_length + 2, the code breaks out of the parsing loop. This will drop any subsequent valid response packets that may already be present in rx (e.g., if one servo returns a malformed packet or there’s a stray/unknown packet in the stream). Consider marking that servo as None and then attempting to re-sync by scanning forward for the next 0xFF 0xFF header (or at least continue parsing remaining bytes when safe) instead of breaking immediately.

Copilot uses AI. Check for mistakes.
if b+pkglen <= len(rx):
responses[servo_id] = self.parse_response(rx[b:b+pkglen])
b += pkglen
Comment on lines +457 to +469
Copy link

Copilot AI Mar 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The loop slices rx[b:b+pkglen] without verifying b + pkglen <= len(rx). If the serial read returns a truncated final packet, parse_response can raise ChecksumError (or mis-parse) and abort the whole sync-read. Add an explicit bounds check before parsing and handle incomplete packets (e.g., log and break / mark remaining requested IDs as None).

Copilot uses AI. Check for mistakes.
Copy link

Copilot AI Mar 30, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If b + pkglen > len(rx) (truncated/partial packet at the end of the read), the code currently increments b and exits the loop without logging or explicitly handling the incomplete packet. This can make partial reads hard to diagnose and may silently return None for that servo. Consider adding an else branch to warn about a truncated packet and break (or attempt a re-read/resync), so callers get clearer diagnostics.

Suggested change
b += pkglen
b += pkglen
else:
self.logger.warning(
"Truncated SYNC READ response for servo %s: expected %d bytes, "
"but only %d bytes remain; stopping parse",
servo_id,
pkglen,
len(rx) - b,
)
break

Copilot uses AI. Check for mistakes.
return responses

def __enter__(self) -> "ST3215":
Expand Down
Loading