File size: 5,807 Bytes
d646dd7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
af470e1
d646dd7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
af470e1
 
 
 
 
 
d646dd7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
af470e1
 
d646dd7
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
"""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