Skip to content

Add parameters in TX device functions to send/request info from specific module (#75) #370

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
ebrahimebrahim merged 1 commit into from
Aug 8, 2025
Merged

Add parameters in TX device functions to send/request info from specific module (#75) #370

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
56 changes: 30 additions & 26 deletions src/openlifu/io/LIFUTXDevice.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,8 @@
ProfileOpts = Literal['active', 'configured', 'all']
TriggerModeOpts = Literal['sequence', 'continuous','single']
DEFAULT_PULSE_WIDTH_US = 20
HW_ID_DATA_LENGTH = 12
TEMPERATURE_DATA_LENGTH = 4

from openlifu.io.LIFUConfig import (
OW_CMD_ASYNC,
Expand Down Expand Up @@ -198,7 +200,7 @@ def close(self):
if self.uart and self.uart.is_connected():
self.uart.disconnect()

def ping(self) -> bool:
def ping(self, module:int=1) -> bool:
"""
Send a ping command to the TX device to verify connectivity.

Expand All @@ -216,7 +218,7 @@ def ping(self) -> bool:

logger.info("Send Ping to Device.")

r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_PING)
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_PING, addr=module)
self.uart.clear_buffer()

if r.packet_type == OW_ERROR:
Expand All @@ -232,7 +234,7 @@ def ping(self) -> bool:
logger.error("Unexpected error during process: %s", e)
raise # Re-raise the exception for the caller to handle

def get_version(self) -> str:
def get_version(self, module:int=1) -> str:
"""
Retrieve the firmware version of the TX device.

Expand All @@ -251,9 +253,9 @@ def get_version(self) -> str:
logger.error("TX Device not connected")
return 'v0.0.0'

r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_VERSION)
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_VERSION, addr=module)
self.uart.clear_buffer()
# r.print_packet()
r.print_packet()
if r.data_len == 3:
ver = f'v{r.data[0]}.{r.data[1]}.{r.data[2]}'
else:
Expand Down Expand Up @@ -316,7 +318,7 @@ def echo(self, echo_data = None) -> tuple[bytes, int]:
logger.error("Unexpected error during echo process: %s", e)
raise # Re-raise the exception for the caller to handle

def toggle_led(self) -> bool:
def toggle_led(self, module:int=1) -> bool:
"""
Toggle the LED on the TX device.

Expand All @@ -332,7 +334,7 @@ def toggle_led(self) -> bool:
logger.error("TX Device not connected")
return False

r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_TOGGLE_LED)
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_TOGGLE_LED, addr=module)
self.uart.clear_buffer()
# r.print_packet()
return True
Expand All @@ -345,7 +347,7 @@ def toggle_led(self) -> bool:
logger.error("Unexpected error during process: %s", e)
raise # Re-raise the exception for the caller to handle

def get_hardware_id(self) -> str:
def get_hardware_id(self, module:int=1) -> str:
"""
Retrieve the hardware ID of the TX device.

Expand All @@ -364,10 +366,10 @@ def get_hardware_id(self) -> str:
logger.error("TX Device not connected")
return None

r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_HWID)
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_HWID, addr=module)
self.uart.clear_buffer()
# r.print_packet()
if r.data_len == 16:
if r.data_len == HW_ID_DATA_LENGTH:
return r.data.hex()
else:
return None
Expand All @@ -379,7 +381,7 @@ def get_hardware_id(self) -> str:
logger.error("Unexpected error during process: %s", e)
raise # Re-raise the exception for the caller to handle

def get_temperature(self) -> float:
def get_temperature(self, module:int=1) -> float:
"""
Retrieve the temperature reading from the TX device.

Expand All @@ -399,12 +401,12 @@ def get_temperature(self) -> float:
return 0

# Send the GET_TEMP command
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_GET_TEMP)
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_GET_TEMP, addr=module)
self.uart.clear_buffer()
# r.print_packet()

# Check if the data length matches a float (4 bytes)
if r.data_len == 4:
if r.data_len == TEMPERATURE_DATA_LENGTH:
# Unpack the float value from the received data (assuming little-endian)
temperature = struct.unpack('<f', r.data)[0]
# Truncate the temperature to 2 decimal places
Expand All @@ -420,7 +422,7 @@ def get_temperature(self) -> float:
logger.error("Unexpected error during process: %s", e)
raise # Re-raise the exception for the caller to handle

def get_ambient_temperature(self) -> float:
def get_ambient_temperature(self, module:int=1) -> float:
"""
Retrieve the ambient temperature reading from the TX device.

Expand All @@ -440,12 +442,12 @@ def get_ambient_temperature(self) -> float:
return 0

# Send the GET_TEMP command
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_GET_AMBIENT)
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_GET_AMBIENT, addr=module)
self.uart.clear_buffer()
# r.print_packet()

# Check if the data length matches a float (4 bytes)
if r.data_len == 4:
if r.data_len == TEMPERATURE_DATA_LENGTH:
# Unpack the float value from the received data (assuming little-endian)
temperature = struct.unpack('<f', r.data)[0]
# Truncate the temperature to 2 decimal places
Expand Down Expand Up @@ -722,7 +724,7 @@ def stop_trigger(self) -> bool:
logger.error("Unexpected error during process: %s", e)
raise # Re-raise the exception for the caller to handle

def soft_reset(self) -> bool:
def soft_reset(self, module:int=1) -> bool:
"""
Perform a soft reset on the TX device.

Expand All @@ -740,7 +742,7 @@ def soft_reset(self) -> bool:
if not self.uart.is_connected():
raise ValueError("TX Device not connected")

r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_RESET)
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_RESET, addr=module)
self.uart.clear_buffer()
# r.print_packet()
if r.packet_type == OW_ERROR:
Expand All @@ -756,7 +758,7 @@ def soft_reset(self) -> bool:
logger.error("Unexpected error during process: %s", e)
raise # Re-raise the exception for the caller to handle

def enter_dfu(self) -> bool:
def enter_dfu(self, module:int=1) -> bool:
"""
Perform a soft reset to enter DFU mode on TX device.

Expand All @@ -774,7 +776,7 @@ def enter_dfu(self) -> bool:
if not self.uart.is_connected():
raise ValueError("TX Device not connected")

r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_DFU)
r = self.uart.send_packet(id=None, packetType=OW_CONTROLLER, command=OW_CMD_DFU, addr=module)
self.uart.clear_buffer()
# r.print_packet()
if r.packet_type == OW_ERROR:
Expand Down Expand Up @@ -970,7 +972,7 @@ def write_register(self, identifier:int, address: int, value: int) -> bool:
logger.error("Unexpected error during process: %s", e)
raise # Re-raise the exception for the caller to handle

def read_register(self, address: int) -> int:
def read_register(self, identifier:int, address: int) -> int:
"""
Read a register value from the TX device.

Expand All @@ -992,7 +994,7 @@ def read_register(self, address: int) -> int:
raise ValueError("TX Device not connected")

# Validate the identifier
if self.identifier < 0:
if identifier < 0:
raise ValueError("TX Chip address NOT SET")

# Pack the address into the required format
Expand All @@ -1007,13 +1009,13 @@ def read_register(self, address: int) -> int:
id=None,
packetType=OW_TX7332,
command=OW_TX7332_RREG,
addr=self.identifier,
addr=identifier,
data=data
)

# Clear UART buffer after sending the packet
self.uart.clear_buffer()

# r.print_packet()
# Check for errors in the response
if r.packet_type == OW_ERROR:
logger.error("Error reading TX register value")
Expand All @@ -1022,14 +1024,16 @@ def read_register(self, address: int) -> int:
# Verify data length and unpack the register value
if r.data_len == 4:
try:
return struct.unpack('<I', r.data)[0]
value = struct.unpack('<I', r.data)[0]
except struct.error as e:
logger.error(f"Error unpacking register value: {e}")
return 0
else:
logger.error(f"Unexpected data length: {r.data_len}")
return 0

logger.info(f"Successfully read value 0x{value:08X} from register 0x{address:04X}")
return value
except ValueError as v:
logger.error("ValueError: %s", v)
raise # Re-raise the exception for the caller to handle
Expand Down Expand Up @@ -1100,7 +1104,7 @@ def write_block(self, identifier: int, start_address: int, reg_values: List[int]

# Clear the UART buffer after sending
self.uart.clear_buffer()

# r.print_packet()
# Check for errors in the response
if r.packet_type == OW_ERROR:
logger.error(f"Error writing TX block at chunk {i}")
Expand Down
Loading