Spaces:
Running
Running
| """Standardized, model-independent names for workflow input devices. | |
| The collector's ``name`` and video ``source_key`` values are compatibility | |
| identifiers: workers and historical batches still use them to locate files. | |
| This module provides a separate UI-facing classification/name so a D435, | |
| DECXIN, or a future camera model follows the same naming rules. | |
| """ | |
| from __future__ import annotations | |
| from collections import defaultdict | |
| from typing import Iterable | |
| DEVICE_LABELS = { | |
| "rgbd_camera": "RGB-D Camera", | |
| "stereo_rgbd_camera": "Stereo RGB-D Camera", | |
| "mono_rgb": "RGB Camera", | |
| "stereo_rgb": "Stereo RGB Camera", | |
| "glove_sensor": "Glove Sensor", | |
| "gripper_device": "UMI Gripper", | |
| } | |
| def _keys(values: Iterable[object] | None) -> list[str]: | |
| if isinstance(values, str): | |
| return [values.strip()] if values.strip() else [] | |
| return [str(value).strip() for value in (values or []) if str(value).strip()] | |
| def _is_left(value: str) -> bool: | |
| low = value.lower() | |
| return low.endswith("_left") or "_left_" in low | |
| def _is_right(value: str) -> bool: | |
| low = value.lower() | |
| return low.endswith("_right") or "_right_" in low | |
| def is_depth_only_key(value: object) -> bool: | |
| """Whether a source key denotes depth only, rather than RGB video.""" | |
| low = str(value or "").strip().lower() | |
| if "depth" not in low: | |
| return False | |
| return not any(token in low for token in ("rgb", "color", "image", "video")) | |
| def camera_profile(kind: object = "", name: object = "", | |
| slots: Iterable[object] | None = None) -> tuple[str, str | None]: | |
| """Return ``(generic device type, lens)`` without depending on a model. | |
| ``mono/stereo`` describes camera layout while ``rgb/depth`` describes | |
| stream modalities. An RGB + depth pair is therefore classified as | |
| RGB-D, even though only the RGB stream is connected to a video input | |
| node. Fisheye remains a lens attribute of Mono RGB. | |
| """ | |
| kind_text = str(kind or "").strip().lower() | |
| name_text = str(name or "").strip().lower() | |
| slot_keys = _keys(slots) | |
| text = " ".join([kind_text, name_text, *[key.lower() for key in slot_keys]]) | |
| # UMI publishes several internal channels (RGB, stereo and tactile data) | |
| # but it is one physical workflow input. Do this before stereo detection | |
| # so its internal left/right camera slots are not exposed as a camera card. | |
| if "gripper" in kind_text or "umi" in name_text: | |
| return "gripper_device", None | |
| # Collectors do not always spell RGB slots with ``rgb``/``color``; | |
| # ``stereo_left`` and ``stereo_right`` are also color video streams. Any | |
| # non-depth-only slot is therefore a color/video slot unless metadata | |
| # explicitly says otherwise. | |
| has_rgb = (any(not is_depth_only_key(key) for key in slot_keys) | |
| or "rgb" in kind_text or "color" in kind_text) | |
| has_depth = any("depth" in key or "depth" in kind_text for key in slot_keys) | |
| has_pair = any(_is_left(key) for key in slot_keys) and any( | |
| _is_right(key) for key in slot_keys | |
| ) | |
| if has_rgb and has_depth and has_pair: | |
| profile = "stereo_rgbd_camera" | |
| elif has_rgb and has_depth: | |
| profile = "rgbd_camera" | |
| elif has_pair or "stereo" in text: | |
| profile = "stereo_rgb" | |
| else: | |
| profile = "mono_rgb" | |
| lens = "fisheye" if "fisheye" in text else None | |
| return profile, lens | |
| def generic_label(profile: str, ordinal: int = 1) -> str: | |
| """Return the fixed, human-facing device category name. | |
| ``ordinal`` is intentionally kept in the signature for compatibility with | |
| callers that still calculate deterministic group numbers. The number is | |
| not part of the UI name: the editable source/device field carries the | |
| concrete physical device identity. | |
| """ | |
| return DEVICE_LABELS.get(profile, "Input Device") | |
| def decorate_device_sources(sources: list[dict]) -> list[dict]: | |
| """Add standardized display metadata while preserving raw identifiers. | |
| Classification is deterministic by source id. ``name`` and | |
| ``source_keys`` are deliberately left unchanged because they are used for | |
| compatibility matching and file resolution. | |
| """ | |
| groups: dict[str, list[dict]] = defaultdict(list) | |
| for source in sources: | |
| input_type = str(source.get("input_type") or "") | |
| if input_type == "glove_sensor": | |
| profile, lens = "glove_sensor", None | |
| elif input_type == "gripper_device": | |
| profile, lens = "gripper_device", None | |
| else: | |
| profile, lens = camera_profile( | |
| source.get("kind"), source.get("name"), | |
| [*(source.get("slots") or []), *(source.get("source_keys") or [])], | |
| ) | |
| source["device_type"] = profile | |
| if lens: | |
| source["lens"] = lens | |
| else: | |
| source.pop("lens", None) | |
| groups[profile].append(source) | |
| for profile, items in groups.items(): | |
| for ordinal, source in enumerate( | |
| sorted(items, key=lambda item: str(item.get("id") or "")), start=1 | |
| ): | |
| display = generic_label(profile, ordinal) | |
| source["display_name"] = display | |
| source["label"] = display | |
| return sources | |
| def display_names_for_sources(sources: Iterable[dict] | None) -> dict[str, str]: | |
| """Build a stream-key → standardized display-name map for the UI.""" | |
| result: dict[str, str] = {} | |
| for source in sources or []: | |
| display = str(source.get("display_name") or "").strip() | |
| if not display: | |
| continue | |
| for key in [*(source.get("source_keys") or []), | |
| *(source.get("depth_keys") or [])]: | |
| key = str(key).strip() | |
| if key: | |
| result[key] = display | |
| return result | |