Skip to content

Add a fallback to USB 2.0 for Luxonis head cameras #101

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

Draft
hello-ola wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Add a fallback to USB 2.0 for Luxonis head cameras #101

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
17 changes: 14 additions & 3 deletions stretch4_body/subsystem/cameras/adapters/luxonis_camera_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,18 @@ def clear_device_cache():
except Exception as e:
print(f"Warning: Failed to clear device cache: {e}")

def _open_luxonis_device(name_or_device_id, product_name=None):
try:
return dai.Device(maxUsbSpeed=dai.UsbSpeed.SUPER_PLUS, nameOrDeviceId=name_or_device_id)
except RuntimeError as e:
device_label = product_name if product_name is not None else name_or_device_id
logging.warning(
"Failed to open Luxonis device %s at SUPER_PLUS. Retrying at HIGH. Error: %s",
device_label,
e,
)
return dai.Device(maxUsbSpeed=dai.UsbSpeed.HIGH, nameOrDeviceId=name_or_device_id)

def get_device_port_by_product_name(product_name:str):
"""When multiple Luxonis cameras are connected, we should query their serial number to use them."""
devices = dai.Device.getAllAvailableDevices()
Expand All @@ -69,7 +81,7 @@ def get_device_port_by_product_name(product_name:str):
if cached_device and current_time - cached_device.get("timestamp", 0) <= CACHE_EXPIRY_SECONDS:
continue

with dai.Device(maxUsbSpeed=dai.UsbSpeed.SUPER_PLUS, nameOrDeviceId=info.deviceId) as device:
with _open_luxonis_device(info.deviceId) as device:
actual_product_name = device.getProductName()
print(f"Found device {actual_product_name}. Looking for {product_name=}")

Expand Down Expand Up @@ -163,7 +175,7 @@ def create_camera_node(pipeline: dai.Pipeline, camera_config: RGBCameraConfig) -
@staticmethod
def create_pipeline(luxonis_device_product_name:str) -> tuple[dai.Pipeline, dai.Device]:
device_port = get_device_port_by_product_name(luxonis_device_product_name)
device = dai.Device(maxUsbSpeed=dai.UsbSpeed.SUPER_PLUS, nameOrDeviceId=device_port)
device = _open_luxonis_device(device_port, luxonis_device_product_name)
pipeline = dai.Pipeline(defaultDevice=device)

print("DeviceID:", device.getDeviceInfo().getDeviceId())
Expand Down Expand Up @@ -519,4 +531,3 @@ def _get_luxonis_diagnostics_message(product_name: str) -> str:
msg_lines.append("To bring it back, please try restarting the robot.")

return "\n".join(msg_lines)