Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
39 changes: 36 additions & 3 deletions src/inspect_robots/_setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -321,8 +321,14 @@ def _camera_view_state(
record.by_id is None or (record.serial is not None and record.serial in duplicated)
for record in inventory
)
return active_is_by_id, advertise_path_toggle
return bool(by_id_rows) or not by_path_rows, len(by_path_rows) > len(by_id_rows)
else:
active_is_by_id = bool(by_id_rows) or not by_path_rows
advertise_path_toggle = len(by_path_rows) > len(by_id_rows)

if by_id_rows == by_path_rows:
advertise_path_toggle = False

return active_is_by_id, advertise_path_toggle


def _identify_by_replug(
Expand Down Expand Up @@ -528,6 +534,7 @@ def _prompt_device_slot(
out: IO[str],
identify: Callable[[bool], str | None],
camera_role: str | None = None,
rescan_inventory: Callable[[], list[_CameraNode]] | None = None,
) -> tuple[str | None, bool]:
"""Prompt for one slot and return its device plus active listing state."""
warned_current = False
Expand Down Expand Up @@ -557,6 +564,15 @@ def _prompt_device_slot(
selected = identify(active_is_by_id)
if selected is None:
continue
if kind == "v4l2" and rescan_inventory is not None:
new_inventory = rescan_inventory()
if new_inventory:
inventory.clear()
inventory.extend(new_inventory)
by_id_devices.clear()
by_id_devices.extend(_camera_rows(inventory, by_id_dir, by_id=True))
by_path_devices.clear()
by_path_devices.extend(_camera_rows(inventory, by_path_dir, by_id=False))
elif not entered and current is not None:
if (
kind == "v4l2"
Expand Down Expand Up @@ -635,11 +651,26 @@ def _prompt_device_slot(
)
continue

def physical_id(dev_path: str) -> str:
resolved = str(Path(dev_path).resolve(strict=False))
for rec in inventory:
if dev_path in (rec.by_id, rec.by_path, rec.node) or resolved == str(
Path(rec.node).resolve(strict=False)
):
return rec.camera or rec.node
return resolved

sel_id = physical_id(selected) if (kind == "v4l2" and inventory) else selected
other = next(
(
(assigned_label, device)
for assigned_kind, assigned_label, device in assigned.values()
if assigned_kind == kind and device == selected
if assigned_kind == kind
and (
physical_id(device) == sel_id
if (kind == "v4l2" and inventory)
else device == selected
)
),
None,
)
Expand Down Expand Up @@ -735,6 +766,7 @@ def _camera_section(
by_path_dir=by_path_dir,
),
camera_role=role,
rescan_inventory=rescan_inventory,
)
if selected is not None:
assignments[key] = selected
Expand Down Expand Up @@ -897,6 +929,7 @@ def _identify(_prefer_by_id: bool) -> str | None:
input_fn=input_fn,
out=out,
identify=identify,
rescan_inventory=rescan_inventory if slot.kind == "v4l2" else None,
)
assignments.pop(slot.arg, None)
assigned_devices.pop(slot.arg, None)
Expand Down
59 changes: 59 additions & 0 deletions tests/test_setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
SUGGESTED,
_camera_inventory,
_camera_rows,
_camera_view_state,
_CameraNode,
_can_serial,
_duplicated_serials,
Expand Down Expand Up @@ -3813,3 +3814,61 @@ def test_suggest_can_pinning_pinned_names_or_no_assigned_kernel_name_are_silent(
_suggest_can_pinning(order_net, slots, {"left_channel": "can9"}, out=unassigned_out)
assert pinned_out.getvalue() == ""
assert unassigned_out.getvalue() == ""


def test_camera_view_state_suppresses_toggle_when_rows_are_identical() -> None:
inv = [
_CameraNode(
node="/dev/video0",
camera="/sys/devices/pci/usb1/1-1",
serial="SN1",
by_id="/dev/v4l/by-id/cam1",
by_path="/dev/v4l/by-path/cam1",
)
]
# When by_id_rows and by_path_rows match exactly, toggle 'p' should be suppressed
active_is_by_id, advertise_toggle = _camera_view_state(
inv, ["/dev/v4l/by-id/cam1"], ["/dev/v4l/by-id/cam1"]
)
assert active_is_by_id is True
assert advertise_toggle is False


def test_prompt_device_slot_detects_duplicate_across_by_id_and_by_path() -> None:
inventory = [
_CameraNode(
node="/dev/video0",
camera="/sys/devices/pci0000:00/0000:00:14.0/usb1/1-1",
serial="SN123",
by_id="/dev/v4l/by-id/usb-Cam_123-video-index0",
by_path="/dev/v4l/by-path/pci-0000:00:14.0-usb-0:1:1.0-video-index0",
)
]
by_id_dir = Path("/dev/v4l/by-id")
by_path_dir = Path("/dev/v4l/by-path")
assigned = {"top_cam_device": ("v4l2", "top", "/dev/v4l/by-id/usb-Cam_123-video-index0")}

# User attempts to assign the by-path path of the SAME camera to the left camera role
by_path_val = "/dev/v4l/by-path/pci-0000:00:14.0-usb-0:1:1.0-video-index0"
input_fn, _prompts = _scripted_input([by_path_val, "n", "s"])
out = io.StringIO()

res, _ = _prompt_device_slot(
"left camera",
"v4l2",
["/dev/v4l/by-id/usb-Cam_123-video-index0"],
[by_path_val],
True,
by_id_dir,
by_path_dir,
None,
assigned,
True,
inventory,
input_fn=input_fn,
out=out,
identify=lambda _b: None,
camera_role="left",
)
assert res is None
assert "already assigned to the top camera" in out.getvalue()
Loading