From 25ccffd8146ae497b62a6c1ca2ec08ba4ff67c07 Mon Sep 17 00:00:00 2001 From: ykcheungab Date: Mon, 5 Oct 2026 09:30:04 +0800 Subject: [PATCH] Discover sensor bundles from the SDK catalog, not a product list The picker listed a hard-coded product list and treated any *.json beside sensor.json as a configuration, so a folder of stray JSON could be offered as a sensor. Discovery now scans the shipped root and a writable user root beside it, keeps the SDK's verdict for every entry (listing rejected ones with their reason) and flags duplicate profile_ids instead of shadowing them; verified by shared/tests and apps/studio/tests. --- apps/studio/config/locales/en_US.json | 5 +- apps/studio/config/locales/zh_CN.json | 5 +- .../src/orisys_studio/ui/sensor_dialog.py | 35 ++ .../src/orisys_studio/ui/sensor_picker.py | 121 +++- .../tests/test_sensor_catalog_picker.py | 129 ++++ apps/studio/tests/test_sensor_catalog_root.py | 271 +++++++++ shared/src/orisys_app_shared/paths.py | 28 + .../src/orisys_app_shared/sensors/__init__.py | 44 +- .../sensors/bundle_catalog.py | 567 ++++++++++++++++++ .../src/orisys_app_shared/sensors/catalog.py | 59 +- shared/tests/test_sensor_bundle_catalog.py | 292 +++++++++ 11 files changed, 1513 insertions(+), 43 deletions(-) create mode 100644 apps/studio/tests/test_sensor_catalog_picker.py create mode 100644 apps/studio/tests/test_sensor_catalog_root.py create mode 100644 shared/src/orisys_app_shared/sensors/bundle_catalog.py create mode 100644 shared/tests/test_sensor_bundle_catalog.py diff --git a/apps/studio/config/locales/en_US.json b/apps/studio/config/locales/en_US.json index bf2cf6f..22a5f78 100644 --- a/apps/studio/config/locales/en_US.json +++ b/apps/studio/config/locales/en_US.json @@ -179,7 +179,10 @@ "export_none": "No coefficients exported yet", "export_file": "file {name} ({size} KB, {time})", "export_identity": "file identity type {sensor_type} · SN {sn} · config {config}", - "thumb_placeholder": "Product model" + "thumb_placeholder": "Product model", + "family_unidentified": "Unidentified", + "catalog_profile": "Profile {profile_id} · family {family} · from {origin} · {state}", + "catalog_blocked": "Cannot start: {reason}" }, "panels": { "arrow": "2D image", diff --git a/apps/studio/config/locales/zh_CN.json b/apps/studio/config/locales/zh_CN.json index 9047c1a..ef73faf 100644 --- a/apps/studio/config/locales/zh_CN.json +++ b/apps/studio/config/locales/zh_CN.json @@ -179,7 +179,10 @@ "export_none": "还没有导出过系数", "export_file": "文件 {name}（{size} KB，{time}）", "export_identity": "文件身份 类型 {sensor_type} · SN {sn} · 配置 {config}", - "thumb_placeholder": "产品型号" + "thumb_placeholder": "产品型号", + "family_unidentified": "未识别", + "catalog_profile": "配置档案 {profile_id} · 家族 {family} · 来源 {origin} · 状态 {state}", + "catalog_blocked": "无法启动：{reason}" }, "panels": { "arrow": "2D 图像", diff --git a/apps/studio/src/orisys_studio/ui/sensor_dialog.py b/apps/studio/src/orisys_studio/ui/sensor_dialog.py index bbd392f..b487bb3 100644 --- a/apps/studio/src/orisys_studio/ui/sensor_dialog.py +++ b/apps/studio/src/orisys_studio/ui/sensor_dialog.py @@ -59,6 +59,12 @@ class SensorDialog(QDialog): self.axes_label.setObjectName("panelMeta") self.config_label = ShrinkableLabel() self.config_label.setObjectName("panelMeta") + #: What the catalog says the selected configuration *is*: family, profile, origin and + #: the SDK's verdict — plus, when it must not be started, why. The picker can only fit + #: a label on a 30 px rail; the reason needs this room. + self.profile_label = ShrinkableLabel() + self.profile_label.setObjectName("panelMeta") + self.profile_label.setWordWrap(True) for label in (self.version_label, self.axes_label): root.addWidget(label) config_row = QHBoxLayout() @@ -69,6 +75,7 @@ class SensorDialog(QDialog): config_row.addWidget(self.btn_reveal) config_row.addStretch(1) root.addLayout(config_row) + root.addWidget(self.profile_label) root.addWidget(self._section_rule()) self.export_title = QLabel() @@ -145,6 +152,7 @@ class SensorDialog(QDialog): f"{Path(config_ref).name if config_ref else t('status.config_none')}" ) self.btn_reveal.setEnabled(bool(config_ref)) + self._refresh_profile() export = str(getattr(window, "_last_calibration_export", "") or "") if not export: @@ -179,6 +187,33 @@ class SensorDialog(QDialog): ) self.btn_apply.setEnabled(True) + def _refresh_profile(self) -> None: + """State the selected catalog entry, and the reason when it cannot be started. + + Read from the picker rather than re-derived here, so the window and the dialog can never + say two different things about one bundle. + """ + read_entry = getattr(self.picker, "selected_entry", None) + entry = read_entry() if callable(read_entry) else None + if entry is None: + self.profile_label.setText("") + self.profile_label.setToolTip("") + return + line = t( + "sensor.catalog_profile", + family=entry.product_family or t("sensor.family_unidentified"), + profile_id=entry.profile_id or entry.folder, + origin=entry.origin, + state=entry.state, + ) + blocked = entry.start_blocked_reason + if blocked: + # In the label, not only the tooltip: a refused configuration has to say why without + # being hovered over, and this is the surface that has room for the sentence. + line = f"{line}\n{t('sensor.catalog_blocked', reason=blocked)}" + self.profile_label.setText(line) + self.profile_label.setToolTip(entry.tooltip()) + def retranslate(self) -> None: self.setWindowTitle(t("sensor.dialog_title")) self.calib_title.setText(t("sensor.calib_section")) diff --git a/apps/studio/src/orisys_studio/ui/sensor_picker.py b/apps/studio/src/orisys_studio/ui/sensor_picker.py index 887868d..b68fa53 100644 --- a/apps/studio/src/orisys_studio/ui/sensor_picker.py +++ b/apps/studio/src/orisys_studio/ui/sensor_picker.py @@ -5,12 +5,26 @@ selected product, then the product/sensor choice and the config file that belongs to it. The order matters — picking ``box`` / ``mini`` / ``finger`` is what decides which configs exist, so the two sit next to each other with their own labels and tooltips instead of a bare "配置" combo. + +Both combos are filled from the configuration catalog +(:mod:`orisys_app_shared.sensors`), not from a hard-coded product list: the type +combo offers the families bundles actually exist for, and the config combo offers +the catalog's entries for the chosen family — each labelled with its profile, +origin and the SDK's own verdict. An entry the SDK rejects stays listed, because +that is how its reason reaches the operator, and +:meth:`SensorPickerWidget.start_blocked_reason` states why the current selection +must not be started (see also :meth:`selected_entry`). """ from __future__ import annotations from orisys_app_shared.i18n import t -from orisys_app_shared.sensors import list_configs_for_type, list_sensor_types +from orisys_app_shared.sensors import ( + FAMILY_UNIDENTIFIED, + CatalogEntry, + SensorCatalog, + scan_sensor_catalog, +) from PySide6.QtCore import Qt, Signal from PySide6.QtWidgets import ( QComboBox, @@ -26,6 +40,10 @@ TYPE_WIDTH = 76 CONFIG_WIDTH = 120 BADGE_WIDTH = 46 +#: The badge for a family group that is not a product at all. Short on purpose: the rail is +#: sized for product names, and "?" states "not a product model" better than a truncated word. +UNIDENTIFIED_BADGE = "?" + class SensorPickerWidget(QWidget): sensor_type_changed = Signal(str) @@ -37,6 +55,11 @@ class SensorPickerWidget(QWidget): layout.setContentsMargins(0, 0, 0, 0) layout.setSpacing(6) + #: The catalog this picker lists. Rescanned by :meth:`reload_catalog`. + self._catalog: SensorCatalog = scan_sensor_catalog() + #: The entries behind the config items, by their selection key. + self._entries_by_key: dict[str, CatalogEntry] = {} + #: The product badge that used to sit above the combos. Kept as an #: attribute: it states which sensor the whole window is configured for. self.thumb_label = QLabel() @@ -54,9 +77,10 @@ class SensorPickerWidget(QWidget): self.type_combo = QComboBox() self.type_combo.setMinimumWidth(TYPE_WIDTH) self.type_combo.setProperty("rail", "true") - self.type_combo.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed) - for name in list_sensor_types(): - self.type_combo.addItem(name, name) + self.type_combo.setSizePolicy( + QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed + ) + self._reload_families() self.type_combo.currentIndexChanged.connect(self._on_type_index_changed) self.config_label = QLabel() @@ -64,7 +88,9 @@ class SensorPickerWidget(QWidget): self.config_combo = QComboBox() self.config_combo.setMinimumWidth(CONFIG_WIDTH) self.config_combo.setProperty("rail", "true") - self.config_combo.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed) + self.config_combo.setSizePolicy( + QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed + ) self.config_combo.currentIndexChanged.connect(self._on_config_index_changed) layout.addWidget(self.thumb_label) @@ -77,6 +103,7 @@ class SensorPickerWidget(QWidget): self._refresh_thumb() self.retranslate() + # ------------------------------------------------------------------ selection def sensor_type(self) -> str: data = self.type_combo.currentData() return str(data or self.type_combo.currentText() or "box") @@ -87,6 +114,26 @@ class SensorPickerWidget(QWidget): return str(self.config_combo.currentText() or "") return str(data) + def selected_entry(self) -> CatalogEntry | None: + """The catalog entry behind the current configuration, when it is one.""" + return self._entries_by_key.get(self.config_name()) + + def start_blocked_reason(self) -> str | None: + """Why the selected configuration must not be started; ``None`` when it may be. + + The verdict is the catalog's: it is non-``None`` for an entry the SDK's validation + rejected and for one whose ``profile_id`` another bundle also declares. The entry stays + listed and explainable either way. + """ + return self._catalog.start_blocked(self.config_name()) + + def reload_catalog(self) -> None: + """Rescan both roots, so a bundle added or removed since start-up appears.""" + self._catalog = scan_sensor_catalog() + self._reload_families() + self._reload_configs() + self._refresh_thumb() + def set_sensor_type(self, sensor_type: str) -> None: wanted = str(sensor_type or "box") idx = self.type_combo.findData(wanted) @@ -110,24 +157,60 @@ class SensorPickerWidget(QWidget): self.config_combo.blockSignals(True) self.config_combo.setCurrentIndex(max(0, idx)) self.config_combo.blockSignals(False) + self._refresh_config_tooltip() def set_picker_enabled(self, enabled: bool) -> None: """Enable/disable both pickers (used while a session is running).""" self.type_combo.setEnabled(bool(enabled)) self.config_combo.setEnabled(bool(enabled)) + # ------------------------------------------------------------------ contents + def _reload_families(self) -> None: + """Offer the families bundles exist for, the unidentified group last.""" + current = self.type_combo.currentData() + self.type_combo.blockSignals(True) + self.type_combo.clear() + for family in self._catalog.families(): + self.type_combo.addItem(_family_label(family), family) + if current is not None: + idx = self.type_combo.findData(current) + if idx >= 0: + self.type_combo.setCurrentIndex(idx) + self.type_combo.blockSignals(False) + def _reload_configs(self) -> None: + family = self.sensor_type() + entries = self._catalog.entries_for_family(family) + self._entries_by_key = {entry.key: entry for entry in entries} self.config_combo.blockSignals(True) self.config_combo.clear() - for display, key in list_configs_for_type(self.sensor_type()): - self.config_combo.addItem(display, key) - if self.config_combo.count() == 0: - self.config_combo.addItem(self.sensor_type(), self.sensor_type()) + for entry in entries: + self.config_combo.addItem(_entry_label(entry), entry.key) + self.config_combo.setItemData( + self.config_combo.count() - 1, + entry.tooltip(), + Qt.ItemDataRole.ToolTipRole, + ) + if not entries: + # No bundle for this family at all: keep the single fallback entry the old catalog + # offered, so a preference (or a future Run) still has something to name. + self.config_combo.addItem(family, family) self.config_combo.blockSignals(False) + self._refresh_config_tooltip() + + def _refresh_config_tooltip(self) -> None: + entry = self.selected_entry() + self.config_combo.setToolTip( + entry.tooltip() if entry is not None else t("sensor.config_tip") + ) def _refresh_thumb(self) -> None: - self.thumb_label.setText(self.sensor_type().upper()) + family = self.sensor_type() + self.thumb_label.setText( + UNIDENTIFIED_BADGE if family == FAMILY_UNIDENTIFIED else family.upper() + ) + # ------------------------------------------------------------------ signals def _on_type_index_changed(self, _index: int) -> None: self._reload_configs() self._refresh_thumb() @@ -135,6 +218,7 @@ class SensorPickerWidget(QWidget): self.config_changed.emit(self.config_name()) def _on_config_index_changed(self, _index: int) -> None: + self._refresh_config_tooltip() self.config_changed.emit(self.config_name()) def retranslate(self) -> None: @@ -143,6 +227,21 @@ class SensorPickerWidget(QWidget): self.type_label.setToolTip(t("sensor.type_tip")) self.config_label.setToolTip(t("sensor.config_tip")) self.type_combo.setToolTip(t("sensor.type_tip")) - self.config_combo.setToolTip(t("sensor.config_tip")) self.thumb_label.setToolTip(t("sensor.thumb_placeholder")) self._refresh_thumb() + self._refresh_config_tooltip() + + +def _family_label(family: str) -> str: + """A family as the operator reads it; the unidentified group is not a product name.""" + return t("sensor.family_unidentified") if family == FAMILY_UNIDENTIFIED else family + + +def _entry_label(entry: CatalogEntry) -> str: + """One combo item: the catalog's label with its family group named in this locale. + + The catalog's own label is the data form (and is what a log or a test should compare); the + internal marker for "no product family" is replaced here, where translating belongs, so an + operator never reads a sentinel. + """ + return entry.label().replace(FAMILY_UNIDENTIFIED, t("sensor.family_unidentified")) diff --git a/apps/studio/tests/test_sensor_catalog_picker.py b/apps/studio/tests/test_sensor_catalog_picker.py new file mode 100644 index 0000000..5f96d60 --- /dev/null +++ b/apps/studio/tests/test_sensor_catalog_picker.py @@ -0,0 +1,129 @@ +"""The picker lists what the catalog returns, including what it refuses. + +This is the selector side of the catalog contract: an entry the SDK rejects must still be +listed (that is how its reason reaches the operator), a duplicate ``profile_id`` must be listed +twice with both sides named, and ``start_blocked_reason()`` must be the answer a caller can +refuse to start on. The catalog itself is tested in ``shared/tests``; here it is a fixture tree +pointed at the widget, so the assertions are about the combo contents rather than the +filesystem. +""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from pathlib import Path + +import pytest +from orisys_app_shared.sensors import FAMILY_UNIDENTIFIED, scan_sensor_catalog +from orisys_studio.ui import sensor_picker as picker_module +from orisys_studio.ui.sensor_picker import SensorPickerWidget +from PySide6.QtCore import Qt + +CONFIG = { + "name": "fixture", + "camera": { + "resolution_raw": [640, 480], + "fps": 30, + "resolution_stitch": [100, 100], + }, + "tracker": {"algorithm": "dis"}, +} + + +def _write(path: Path, payload: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2), encoding="utf-8") + + +def _descriptor(profile_id: str) -> dict[str, object]: + return { + "schema_version": 1, + "profile_id": profile_id, + "display_name": "Mini lab", + "product_family": "mini", + "processing_backend": "single_camera", + "sensor_json": "sensor.json", + } + + +def _make_picker(qt_app, tmp_path, monkeypatch, build: Callable[[Path], None]): + """A picker whose catalog is a fixture tree instead of the installed one.""" + root = tmp_path / "user" + build(root) + catalog = scan_sensor_catalog( + shipped_root=tmp_path / "no-shipped-root", user_root=root + ) + monkeypatch.setattr(picker_module, "scan_sensor_catalog", lambda: catalog) + return SensorPickerWidget() + + +@pytest.fixture +def mixed_picker(qt_app, tmp_path, monkeypatch): + """Two bundles sharing a profile_id, plus a folder holding only a calibration.""" + + def build(root: Path) -> None: + for name in ("mini-a", "mini-b"): + _write(root / name / "sensor.json", CONFIG) + _write(root / name / "manifest.json", _descriptor("orisys.mini.same-r1")) + _write(root / "stray" / "force_calibration.json", {"schema_version": 2}) + + widget = _make_picker(qt_app, tmp_path, monkeypatch, build) + yield widget + widget.deleteLater() + + +def test_a_calibration_folder_is_listed_as_invalid_with_its_reason(mixed_picker): + picker = mixed_picker + assert picker.type_combo.findData(FAMILY_UNIDENTIFIED) >= 0 + picker.set_sensor_type(FAMILY_UNIDENTIFIED) + assert picker.config_combo.count() == 1 + entry = picker.selected_entry() + assert entry is not None and not entry.valid + # The internal "no product family" marker is named in the operator's language, not leaked. + assert FAMILY_UNIDENTIFIED not in picker.config_combo.itemText(0) + tooltip = picker.config_combo.itemData(0, Qt.ItemDataRole.ToolTipRole) + assert "force_calibration.json" in tooltip + assert "sensor.json or a manifest.json descriptor" in tooltip + assert picker.start_blocked_reason() == entry.start_blocked_reason + + +def test_a_shared_profile_id_is_listed_twice_and_can_be_refused(mixed_picker): + picker = mixed_picker + picker.set_sensor_type("mini") + assert picker.config_combo.count() == 2, "one of the duplicates was shadowed" + + first = picker.selected_entry() + picker.config_combo.setCurrentIndex(1) + second = picker.selected_entry() + assert first is not None and second is not None + assert {first.profile_id, second.profile_id} == {"orisys.mini.same-r1"} + assert {first.folder, second.folder} == {"mini-a", "mini-b"} + # Each side names the other, and the verdict follows the *selected* entry. + for index, other_ref in ((0, "user/mini-b"), (1, "user/mini-a")): + picker.config_combo.setCurrentIndex(index) + entry = picker.selected_entry() + assert entry is not None and entry.duplicates == (other_ref,) + reason = picker.start_blocked_reason() + assert reason is not None and other_ref in reason + # Both stay selectable: the list shows the ambiguity instead of resolving it silently. + assert picker.config_combo.count() == 2 + + +def test_a_single_valid_entry_is_listed_with_its_identity_and_not_blocked( + qt_app, tmp_path, monkeypatch +): + def build(root: Path) -> None: + _write(root / "mini-lab" / "sensor.json", CONFIG) + _write(root / "mini-lab" / "manifest.json", _descriptor("orisys.mini.lab-r1")) + + picker = _make_picker(qt_app, tmp_path, monkeypatch, build) + assert picker.type_combo.findData("mini") >= 0 + picker.set_sensor_type("mini") + label = picker.config_combo.itemText(0) + assert "orisys.mini.lab-r1" in label + assert "mini" in label and "user" in label and "degraded" in label + assert picker.start_blocked_reason() is None + entry = picker.selected_entry() + assert entry is not None and entry.profile_id == "orisys.mini.lab-r1" + picker.deleteLater() diff --git a/apps/studio/tests/test_sensor_catalog_root.py b/apps/studio/tests/test_sensor_catalog_root.py new file mode 100644 index 0000000..a2edf02 --- /dev/null +++ b/apps/studio/tests/test_sensor_catalog_root.py @@ -0,0 +1,271 @@ +"""The catalog against the configuration the studio actually ships, and the rename contract. + +The first half asserts what ``apps/studio/config/sensors`` contains *today*, entry by entry, so +a change to the shipped tree shows up here instead of only on a machine that happens to have +the right folders. + +``finger`` is the one shipped bundle that carries a descriptor, so its identity is the profile +the descriptor declares instead of one derived from its folder name. Its declared asset paths +are asserted to be the files in the tree, because a rename that silently resolved elsewhere is +exactly the failure the descriptor exists to prevent. + +The second half is the rename contract: a bundle keeps its identity when its folder is renamed. +That is true because every descriptor path is relative to the bundle root, so it belongs to +discovery — the catalog is where a renamed bundle is compared with the one it was copied from. +""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path + +import pytest +from orisys.sensors import DescriptorError, resolve_bundle +from orisys.unroll import scaled_map_cache_path +from orisys_app_shared.acquisition.finger_worker import resolve_finger_export_dir +from orisys_app_shared.paths import sensors_dir +from orisys_app_shared.sensors import ( + ORIGIN_SHIPPED, + ROOT_KIND_FALLBACK, + ROOT_KIND_SIBLING, + scan_sensor_catalog, +) + +#: The identity the shipped finger descriptor declares. A copy of the bundle keeps it. +FINGER_PROFILE_ID = "orisys.finger.r01" + +#: What the shipped root provides today, as the SDK resolves it. +EXPECTED_SHIPPED: dict[str, tuple[str, str, str, bool, str]] = { + # folder: (family, profile_id, backend, valid, state) + "box": ("box", "orisys.box.legacy", "compound_eye", True, "degraded"), + "finger": ("finger", FINGER_PROFILE_ID, "uv_unroll", True, "degraded"), + "finger-mems": ( + "finger-mems", + "orisys.finger-mems.legacy", + "compound_eye", + True, + "degraded", + ), + "mini": ("mini", "orisys.mini.legacy", "compound_eye", True, "degraded"), +} + + +@pytest.fixture(scope="module") +def shipped(): + return scan_sensor_catalog() + + +def test_the_shipped_root_is_the_one_the_app_resolves(shipped): + assert shipped.shipped_root == sensors_dir() + assert shipped.roots[0].exists is True + + +def test_the_shipped_root_offers_exactly_its_four_bundles(shipped): + refs = {entry.ref for entry in shipped.entries_for_origin(ORIGIN_SHIPPED)} + assert refs == { + "shipped/box", + "shipped/finger", + "shipped/finger-mems", + "shipped/mini", + }, sorted(refs) + + +def test_every_shipped_entry_reports_its_family_profile_and_verdict(shipped): + for folder, (family, profile_id, backend, valid, state) in EXPECTED_SHIPPED.items(): + entry = shipped.by_ref(f"shipped/{folder}") + assert entry is not None, folder + assert entry.product_family == family, folder + assert entry.profile_id == profile_id, folder + assert entry.processing_backend == backend, folder + assert entry.valid is valid, f"{folder}: {entry.reason}" + assert entry.state == state, f"{folder}: {entry.reason}" + assert entry.origin == ORIGIN_SHIPPED + # The verdict is the SDK's, not a second opinion computed here. + assert (entry.start_blocked_reason is None) == ( + valid and not entry.duplicate + ), folder + + +def test_the_shipped_families_keep_the_picker_order(shipped): + assert shipped.families() == ["box", "mini", "finger", "finger-mems"] + + +def test_box_mini_and_finger_mems_are_legacy_sensor_json_bundles(shipped): + for folder in ("box", "mini", "finger-mems"): + bundle = sensors_dir() / folder + assert (bundle / "sensor.json").is_file(), folder + assert not (bundle / "manifest.json").exists(), folder + entry = shipped.by_ref(f"shipped/{folder}") + assert entry.profile is not None and entry.profile.sensor_json_path.is_file(), ( + folder + ) + + +def test_the_finger_bundle_ships_a_descriptor_naming_the_files_it_has(shipped): + """The shipped ``finger/manifest.json`` is a *descriptor*, not the legacy shape. + + Its identity is declared rather than inferred from the folder name, and every path it + declares is relative to the bundle root. The declared files are asserted to exist, so + renaming a shipped map or scene without updating the descriptor fails here. + """ + bundle = sensors_dir() / "finger" + manifest = json.loads((bundle / "manifest.json").read_text(encoding="utf-8")) + assert manifest["schema_version"] == 1, "the descriptor contract is version 1" + assert manifest["profile_id"] == FINGER_PROFILE_ID + assert manifest["display_name"] + assert manifest["product_family"] == "finger" + assert manifest["processing_backend"] == "uv_unroll" + assert manifest["sensor_json"] == "sensor.json" + + declared = [ + manifest["sensor_json"], + *manifest["computational_assets"].values(), + *manifest["assets"].values(), + ] + assert declared, "a descriptor that names no file describes nothing" + for relative in declared: + assert not Path(relative).is_absolute(), relative + assert ".." not in Path(relative).parts, relative + assert (bundle / relative).is_file(), f"declared but absent: {relative}" + + entry = shipped.by_ref("shipped/finger") + assert entry.profile is not None + assert entry.profile.sensor_json_path == bundle / "sensor.json" + assert entry.valid and entry.start_blocked_reason is None, entry.reason + # The verdict is the SDK's, and the undeclared-asset failure is gone for good. + assert "assets.computational" not in entry.reason, entry.reason + + +def test_the_declared_assets_are_the_files_the_running_chain_loads(shipped): + """The descriptor's asset paths and the live chain agree, file name included. + + The finger chain builds its UV map cache name for itself, from the export directory and + ``camera.source_scale`` (the SDK's ``scaled_map_cache_path``). That derivation and the + descriptor are two statements of the same fact, so they are compared here: a bundle whose + map was renamed would otherwise keep resolving to the name the chain still asks for. + """ + bundle = sensors_dir() / "finger" + profile = shipped.by_ref("shipped/finger").profile + assert profile is not None + + export_dir = resolve_finger_export_dir(profile.sensor_json_path) + assert export_dir == bundle / "assets" / "30fps-0703" + chain_map = Path( + scaled_map_cache_path( + str(export_dir / "unroll_map_uv.npz"), profile.config["camera"]["source_scale"] + ) + ) + declared_map = profile.computational_assets["unroll_map"] + assert declared_map == chain_map + # The scale suffix is the configuration's (0.42), never a neighbouring variant. + assert declared_map.name == "unroll_map_uv_s0.42.npz" + assert declared_map.is_file() + assert not (export_dir / "unroll_map_uv_s0.4.npz").exists() + + declared_scene = profile.assets["scene"] + assert declared_scene == bundle / "scene" / "finger-display-1.0.0.oryscene" + assert declared_scene.is_file() + assert profile.capabilities.visualization_3d.available + + +def test_a_calibration_file_is_never_offered_as_a_configuration(shipped): + """``box/`` holds ``sensor.json`` and ``force_calibration.json``: one entry, not two.""" + box = sensors_dir() / "box" + assert (box / "force_calibration.json").is_file(), "the premise of this test" + entries = shipped.entries_for_family("box") + assert len(entries) == 1, [entry.key for entry in entries] + assert Path(entries[0].key).name == "sensor.json" + for entry in shipped.entries: + assert Path(entry.key).name != "force_calibration.json" + assert "force_calibration" not in entry.folder + + +def test_the_user_root_is_absent_today_and_that_is_not_an_error(shipped): + """Nothing is offered from a user root that does not exist, and no error is raised. + + W2 will import bundles into that root; when it does, this assertion is the one to revisit — + the point being that an absent root is empty, not broken. + """ + assert shipped.user_root_kind in (ROOT_KIND_SIBLING, ROOT_KIND_FALLBACK) + assert shipped.user_root_exists is False, shipped.user_root + assert shipped.entries_for_origin(shipped.roots[1].origin) == () + # The shipped root is never the place user bundles go. + assert shipped.shipped_root != shipped.user_root + assert shipped.shipped_root not in shipped.user_root.parents + + +# ------------------------------------------------------------------ rename contract +def _copy_finger(destination: Path, *, descriptor: bool = True) -> Path: + """Copy the shipped finger bundle, optionally stripping its descriptor.""" + shutil.copytree(sensors_dir() / "finger", destination) + if not descriptor: + (destination / "manifest.json").unlink() + return destination + + +def test_a_renamed_bundle_keeps_its_identity(tmp_path): + """The descriptor's relative paths are what make moving or renaming a bundle safe.""" + root = tmp_path / "bundles" + first = _copy_finger(root / "finger-a") + second = _copy_finger(root / "somewhere-else-entirely") + assert first.name != second.name + + catalog = scan_sensor_catalog( + shipped_root=tmp_path / "no-shipped-root", user_root=root + ) + a = catalog.by_ref("user/finger-a") + b = catalog.by_ref("user/somewhere-else-entirely") + assert a is not None and b is not None + assert a.valid and b.valid, (a.reason, b.reason) + + assert a.profile_id == b.profile_id == FINGER_PROFILE_ID + assert a.product_family == b.product_family == "finger" + assert a.processing_backend == b.processing_backend == "uv_unroll" + assert a.display_name == b.display_name + assert a.profile.processing_fingerprint == b.profile.processing_fingerprint + # The reproducible snapshot is bundle-relative, so the folder name cannot appear in it. + assert a.profile.inventory() == b.profile.inventory() + inventory = a.profile.inventory() + assert inventory["sensor_config"] == "sensor.json" + assert inventory["computational_assets"]["unroll_map"]["path"] == ( + "assets/30fps-0703/unroll_map_uv_s0.42.npz" + ) + assert inventory["visualization_assets"]["scene"] == ( + "scene/finger-display-1.0.0.oryscene" + ) + assert str(tmp_path) not in json.dumps(inventory) + + # Every resolved path stays inside the copy it was resolved from. + for entry, copy in ((a, first), (b, second)): + inside = copy.resolve() + assert entry.profile.bundle_root == inside + assert entry.profile.sensor_json_path.is_relative_to(inside) + for role, path in entry.profile.computational_assets.items(): + assert path.is_file(), role + assert path.is_relative_to(inside), path + for role, path in entry.profile.assets.items(): + assert path.is_relative_to(inside), path + + # One profile, two bundles: the catalog names the twin instead of picking one silently. + assert a.key != b.key + assert a.duplicates == ("user/somewhere-else-entirely",) + assert b.duplicates == ("user/finger-a",) + assert a.start_blocked_reason is not None + + +def test_the_same_copy_without_a_descriptor_is_refused_by_its_folder_name(tmp_path): + """The negative control: rename a *legacy* bundle and it is no longer identifiable.""" + plain = _copy_finger(tmp_path / "finger-a", descriptor=False) + with pytest.raises(DescriptorError) as failure: + resolve_bundle(plain) + assert "not a built-in bundle" in str(failure.value) + + catalog = scan_sensor_catalog( + shipped_root=tmp_path / "no-shipped-root", user_root=tmp_path + ) + entry = catalog.by_ref("user/finger-a") + assert entry is not None, "a bundle that cannot be resolved is still listed" + assert not entry.valid + assert "not a built-in bundle" in entry.reason + assert entry.start_blocked_reason == entry.reason diff --git a/shared/src/orisys_app_shared/paths.py b/shared/src/orisys_app_shared/paths.py index 709a444..7c05bb6 100644 --- a/shared/src/orisys_app_shared/paths.py +++ b/shared/src/orisys_app_shared/paths.py @@ -92,6 +92,34 @@ def sensors_dir() -> Path: return config_dir() / "sensors" +def user_sensors_dir() -> Path: + """The writable user bundle root: a **sibling** of the shipped one. + + The app ships as a portable folder, so the natural place for a bundle the + operator adds is beside the shipped ``config/sensors`` rather than inside it: + an upgrade that replaces the shipped tree must not delete their work, and a + zip that is unpacked somewhere else must carry it along. + + This is only the preferred location. Whether it can be written — and where + the catalog goes when it cannot — is decided by + :func:`orisys_app_shared.sensors.bundle_catalog.resolve_user_root`, which + reports the root actually in use. Never place user bundles in the shipped + root, and never in Documents. + """ + return config_dir() / "sensors-user" + + +def fallback_user_sensors_dir() -> Path: + """The per-user bundle root used when the portable sibling is not writable. + + ``%LOCALAPPDATA%\\OrisysStudio\\config\\sensors-user`` on Windows, and the + equivalent cache root elsewhere — the same derived-data area + :func:`derived_data_dir` already owns, so a read-only installation medium + still has somewhere writable to keep the operator's bundles. + """ + return derived_data_dir() / "config" / "sensors-user" + + def locales_dir() -> Path: return config_dir() / "locales" diff --git a/shared/src/orisys_app_shared/sensors/__init__.py b/shared/src/orisys_app_shared/sensors/__init__.py index 302e2d5..dcb81f1 100644 --- a/shared/src/orisys_app_shared/sensors/__init__.py +++ b/shared/src/orisys_app_shared/sensors/__init__.py @@ -1,5 +1,45 @@ -"""Sensor catalog helpers.""" +"""Sensor catalog helpers. +The catalog answers three questions the picker and the resolver both need: which configuration +bundles exist, what each one is (family, profile, origin and the SDK's verdict), and where a +user's own bundles go on this machine. +""" + +from orisys_app_shared.sensors.bundle_catalog import ( + FAMILY_UNIDENTIFIED, + KNOWN_FAMILY_ORDER, + ORIGIN_FALLBACK, + ORIGIN_SHIPPED, + ORIGIN_USER, + ROOT_KIND_EXPLICIT, + ROOT_KIND_FALLBACK, + ROOT_KIND_SIBLING, + STATE_UNREADABLE, + CatalogEntry, + CatalogRoot, + SensorCatalog, + UserRootDecision, + resolve_user_root, + scan_sensor_catalog, +) from orisys_app_shared.sensors.catalog import list_configs_for_type, list_sensor_types -__all__ = ["list_configs_for_type", "list_sensor_types"] +__all__ = [ + "FAMILY_UNIDENTIFIED", + "KNOWN_FAMILY_ORDER", + "ORIGIN_FALLBACK", + "ORIGIN_SHIPPED", + "ORIGIN_USER", + "ROOT_KIND_EXPLICIT", + "ROOT_KIND_FALLBACK", + "ROOT_KIND_SIBLING", + "STATE_UNREADABLE", + "CatalogEntry", + "CatalogRoot", + "SensorCatalog", + "UserRootDecision", + "list_configs_for_type", + "list_sensor_types", + "resolve_user_root", + "scan_sensor_catalog", +] diff --git a/shared/src/orisys_app_shared/sensors/bundle_catalog.py b/shared/src/orisys_app_shared/sensors/bundle_catalog.py new file mode 100644 index 0000000..a62a587 --- /dev/null +++ b/shared/src/orisys_app_shared/sensors/bundle_catalog.py @@ -0,0 +1,567 @@ +"""The sensor configuration catalog: which bundles exist, and what each one is. + +Studio used to offer a hard-coded product list and to treat any ``*.json`` beside a +``sensor.json`` as another configuration. Both are replaced here by discovery over the two +roots the app actually has: + +* the **shipped** root — ``config/sensors``, resolved exactly as + :func:`orisys_app_shared.paths.sensors_dir` resolves it today; +* a **writable user** root *beside* it — ``config/sensors-user``, because the app ships as a + portable folder and the operator's own bundles belong next to it, never inside the shipped + tree and never in Documents. When that directory cannot be written (a read-only install + medium, a locked-down install directory) the OS per-user area is used instead and the + catalog says which root is in use. + +Discovery is the SDK's, in this order: a parseable ``manifest.json`` descriptor, then the +compatibility adapter for a legacy ``sensor.json`` bundle, then an entry retained as +**invalid** carrying the SDK's own explanation. This module never decides a processing +backend and never reads a folder name as evidence of one; it reports what +:func:`orisys.sensors.resolve_bundle` decided. + +A folder holding only an arbitrary JSON document — ``force_calibration.json`` — is not a +configuration. It is still listed, marked invalid, and its reason says so, because silently +hiding it is how the same file ends up offered as a sensor config. + +Two bundles that declare one ``profile_id`` are both listed and each names the other. Nothing +is shadowed, and the ambiguity is available to a caller that must refuse to start. +""" + +from __future__ import annotations + +import contextlib +import json +import os +import uuid +from dataclasses import dataclass, replace +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from orisys_app_shared.paths import ( + fallback_user_sensors_dir, + sensors_dir, + user_sensors_dir, +) + +if TYPE_CHECKING: # pragma: no cover - typing only + from orisys.sensors import ResolvedSensorProfile + +#: Which root an entry came from. +ORIGIN_SHIPPED = "shipped" +ORIGIN_USER = "user" +ORIGIN_FALLBACK = "fallback" + +#: Which physical location the writable user root resolved to. +ROOT_KIND_SIBLING = "sibling" +ROOT_KIND_FALLBACK = "fallback" +ROOT_KIND_EXPLICIT = "explicit" + +#: Family bucket for an entry that cannot be attributed to a product family. It is deliberately +#: not a product name: a folder of stray JSON has no product, and inventing one is how a +#: calibration file becomes a sensor. +FAMILY_UNIDENTIFIED = "__unidentified__" + +#: The order the picker offers families in. The families the shipped tree has always provided +#: come first, in the order the picker has always used them, so replacing the hard-coded product +#: list does not reshuffle the picker for the products that exist today. This is presentation +#: order only — which families are offered comes from what is installed. +KNOWN_FAMILY_ORDER: tuple[str, ...] = ("box", "mini", "finger", "finger-mems") + +#: ``CatalogEntry.state`` for a folder no bundle could be read from at all. Every other state is +#: the SDK's own :attr:`ValidationReport.state` (``ok`` / ``degraded`` / ``failed``). +STATE_UNREADABLE = "invalid" + +_WRITE_PROBE_PREFIX = ".orisys-write-probe" + + +def _sdk(): + """The SDK surface this module needs, imported on first use. + + ``orisys_app_shared.paths`` already reaches for the SDK lazily, so a shared module stays + importable without it; this keeps that behaviour instead of making the SDK a hard import of + the shared sensor package. + """ + from orisys import sensors + + return sensors + + +def _sort_key(path: Path) -> str: + return path.name.lower() + + +@dataclass(frozen=True) +class UserRootDecision: + """Where the writable user bundles live, and why.""" + + path: Path + #: ``sibling`` (the portable ``config/sensors-user``), ``fallback`` (the OS per-user area) + #: or ``explicit`` (named by the caller, as a test does). + kind: str + #: Whether a file can actually be created there right now. + writable: bool + reason: str + + def __str__(self) -> str: + return self.reason + + +@dataclass(frozen=True) +class CatalogRoot: + """One directory the catalog scans.""" + + origin: str + path: Path + exists: bool + + def __str__(self) -> str: + return f"{self.origin}: {self.path}" + ("" if self.exists else " (absent)") + + +@dataclass(frozen=True) +class CatalogEntry: + """One bundle folder, as the catalog sees it. + + ``valid`` is the SDK's verdict (its validation report is usable), never a judgement made + here, so an entry the SDK rejects is shown with the reason instead of being dropped. + """ + + #: Selection key: the configuration file the app already passes around, or the bundle folder + #: when no ``sensor.json`` could be found. Unique per entry. + key: str + folder: str + bundle_path: Path + origin: str + display_name: str + product_family: str + profile_id: str + sensor_json_path: Path | None + processing_backend: str + valid: bool + state: str + reason: str = "" + errors: tuple[str, ...] = () + warnings: tuple[str, ...] = () + diagnostics: tuple[str, ...] = () + #: ``origin/folder`` of every other bundle declaring this entry's ``profile_id``. + duplicates: tuple[str, ...] = () + profile: ResolvedSensorProfile | None = None + + @property + def ref(self) -> str: + """``origin/folder`` — readable, and unique across the two roots.""" + return f"{self.origin}/{self.folder}" + + @property + def family(self) -> str: + """The family group this entry is listed under; never empty.""" + return self.product_family or FAMILY_UNIDENTIFIED + + @property + def duplicate(self) -> bool: + return bool(self.duplicates) + + @property + def start_blocked_reason(self) -> str | None: + """Why no session may be started with this entry, or ``None`` when one may. + + The verdict belongs to the SDK: a profile its validation rejected is not startable, and + two bundles declaring one ``profile_id`` make choosing between them ambiguous. The entry + stays listed and explainable either way — that is the point of returning a reason. + """ + if not self.valid: + return self.reason or f"the SDK rejected this profile ({self.state})" + if self.duplicates: + return ( + f"profile_id {self.profile_id!r} is also provided by " + f"{', '.join(self.duplicates)}; selecting between them is ambiguous" + ) + return None + + def label(self) -> str: + """One line for a picker item: family, profile, origin and validity.""" + return f"{self.family} · {self.profile_id or self.folder} · {self.origin} · {self.state}" + + def tooltip(self) -> str: + """The facts behind the label, including why the entry is not startable.""" + lines = [ + f"{self.display_name} ({self.ref})", + f"profile_id: {self.profile_id or 'unknown'}", + f"product family: {self.product_family or 'unknown'}", + f"processing backend: {self.processing_backend or 'unknown'}", + f"origin: {self.origin} ({self.bundle_path})", + f"state: {self.state}", + ] + blocked = self.start_blocked_reason + if blocked: + lines.append(f"cannot start: {blocked}") + # The reason already quotes the errors it was built from; repeating them here would + # bury the one line that explains the entry. + lines.extend( + f"error: {text}" for text in self.errors if text not in self.reason + ) + lines.extend(f"warning: {text}" for text in self.warnings) + return "\n".join(lines) + + +@dataclass(frozen=True) +class SensorCatalog: + """Every configuration bundle the roots provide, with the roots themselves.""" + + roots: tuple[CatalogRoot, ...] + entries: tuple[CatalogEntry, ...] + user_root_decision: UserRootDecision + + # ------------------------------------------------------------------ roots + @property + def shipped_root(self) -> Path: + return self.roots[0].path + + @property + def user_root(self) -> Path: + return self.roots[1].path + + @property + def user_root_exists(self) -> bool: + return self.roots[1].exists + + @property + def user_root_kind(self) -> str: + return self.user_root_decision.kind + + @property + def user_root_reason(self) -> str: + return self.user_root_decision.reason + + # ------------------------------------------------------------------ entries + def entries_for_origin(self, origin: str) -> tuple[CatalogEntry, ...]: + wanted = str(origin or "") + return tuple(entry for entry in self.entries if entry.origin == wanted) + + def entries_for_family(self, family: str) -> tuple[CatalogEntry, ...]: + """The entries one family group lists; :data:`FAMILY_UNIDENTIFIED` is a group too.""" + wanted = str(family or "") + if wanted == FAMILY_UNIDENTIFIED: + return tuple(entry for entry in self.entries if not entry.product_family) + return tuple(entry for entry in self.entries if entry.product_family == wanted) + + def families(self) -> list[str]: + """The family groups to offer, in the picker's order, unidentified last.""" + present: list[str] = [] + for entry in self.entries: + if entry.family not in present: + present.append(entry.family) + known = [name for name in KNOWN_FAMILY_ORDER if name in present] + others = sorted( + name + for name in present + if name not in KNOWN_FAMILY_ORDER and name != FAMILY_UNIDENTIFIED + ) + tail = [FAMILY_UNIDENTIFIED] if FAMILY_UNIDENTIFIED in present else [] + return known + others + tail + + def invalid_entries(self) -> tuple[CatalogEntry, ...]: + return tuple(entry for entry in self.entries if not entry.valid) + + def by_key(self, key: str) -> CatalogEntry | None: + wanted = str(key or "") + for entry in self.entries: + if entry.key == wanted: + return entry + return None + + def by_ref(self, ref: str) -> CatalogEntry | None: + wanted = str(ref or "") + for entry in self.entries: + if entry.ref == wanted: + return entry + return None + + def start_blocked(self, key_or_entry: CatalogEntry | str) -> str | None: + """Why the named entry must not be started; ``None`` when it may be. + + This is the whole answer a caller needs to refuse to start: it is ``None`` only for an + entry the SDK validated *and* that no other bundle duplicates. + """ + entry = ( + key_or_entry + if isinstance(key_or_entry, CatalogEntry) + else self.by_key(str(key_or_entry)) + ) + return None if entry is None else entry.start_blocked_reason + + def summary_lines(self) -> list[str]: + """A short block for a log header or a diagnostics panel.""" + lines = [ + f"shipped root: {self.shipped_root}" + + ("" if self.roots[0].exists else " (absent)"), + f"user root: {self.user_root} ({self.user_root_kind})" + + ("" if self.user_root_exists else " (absent)"), + ] + for entry in self.entries: + blocked = entry.start_blocked_reason + lines.append( + f" {entry.ref}: {entry.label()}" + (f" - {blocked}" if blocked else "") + ) + return lines + + +def resolve_user_root() -> UserRootDecision: + """The writable user bundle root, preferring the portable sibling. + + Returns the decision rather than just a path, because which root is in use is a fact the + app has to be able to report: a user who cannot find their bundles needs to be told where + they went, not shown a path the app silently chose. + """ + sibling = user_sensors_dir() + if _is_writable(sibling): + return UserRootDecision( + path=sibling, + kind=ROOT_KIND_SIBLING, + writable=True, + reason=f"{sibling} (sibling of the shipped root; writable)", + ) + fallback = fallback_user_sensors_dir() + return UserRootDecision( + path=fallback, + kind=ROOT_KIND_FALLBACK, + writable=_is_writable(fallback), + reason=f"{sibling} is not writable; using {fallback}", + ) + + +def _is_writable(target: Path) -> bool: + """Whether a file can be created in ``target``, or beside it when it does not exist. + + A real write is the only honest test on Windows, where a read-only directory still answers + ``os.access(..., os.W_OK)``. The directory itself is never created: reporting where user + bundles *would* go must not have the side effect of creating that place, and a catalogue + scan must not change the tree it is describing. + """ + directory = target if target.is_dir() else target.parent + if not directory.is_dir(): + return False + probe = directory / f"{_WRITE_PROBE_PREFIX}-{os.getpid()}-{uuid.uuid4().hex[:8]}" + try: + probe.write_text("", encoding="utf-8") + except OSError: + return False + # A cleanup failure is not a writability finding: the file was created, which is the answer, + # and one empty probe file left behind is harmless. + with contextlib.suppress(OSError): + probe.unlink() + return True + + +def scan_sensor_catalog( + *, + shipped_root: str | Path | None = None, + user_root: str | Path | None = None, +) -> SensorCatalog: + """Discover every bundle under the shipped and writable user roots. + + ``shipped_root`` defaults to :func:`orisys_app_shared.paths.sensors_dir` and ``user_root`` + to whatever :func:`resolve_user_root` decided. Both are parameters so a test — or a caller + with an unusual layout — can point the catalog at another tree without touching the real + one. A root that does not exist contributes no entries and is not an error. + """ + shipped = Path(shipped_root) if shipped_root is not None else sensors_dir() + if user_root is None: + decision = resolve_user_root() + user = decision.path + user_origin = ( + ORIGIN_FALLBACK if decision.kind == ROOT_KIND_FALLBACK else ORIGIN_USER + ) + else: + user = Path(user_root) + decision = UserRootDecision( + path=user, + kind=ROOT_KIND_EXPLICIT, + writable=_is_writable(user), + reason=f"{user} (named by the caller)", + ) + user_origin = ORIGIN_USER + + roots = ( + CatalogRoot(origin=ORIGIN_SHIPPED, path=shipped, exists=shipped.is_dir()), + CatalogRoot(origin=user_origin, path=user, exists=user.is_dir()), + ) + entries: list[CatalogEntry] = [] + for root in roots: + if root.exists: + entries.extend(_scan_root(root)) + return SensorCatalog( + roots=roots, + entries=tuple(_mark_duplicates(entries)), + user_root_decision=decision, + ) + + +def _scan_root(root: CatalogRoot) -> list[CatalogEntry]: + """Every bundle folder directly under one root, in a stable order.""" + try: + children = sorted( + (path for path in root.path.iterdir() if path.is_dir()), key=_sort_key + ) + except OSError: + # An unreadable root lists nothing; the root itself is still reported, so the reason a + # family is missing remains visible above the (empty) entry list. + return [] + found: list[CatalogEntry] = [] + for folder in children: + entry = _entry_for(folder, root.origin) + if entry is not None: + found.append(entry) + return found + + +def _entry_for(folder: Path, origin: str) -> CatalogEntry | None: + """One child folder, or ``None`` when it is not a bundle candidate at all. + + A candidate holds at least one top-level JSON document, because both a descriptor and a + configuration are JSON. Whether it *is* a usable configuration is then the SDK's answer, + never this module's. + """ + json_files = _json_files(folder) + if not json_files: + return None + try: + profile = _sdk().resolve_bundle(folder) + except Exception as exc: # noqa: BLE001 - one broken bundle must not hide the others + return _unreadable_entry(folder, origin, exc) + return _resolved_entry(folder, origin, profile) + + +def _resolved_entry( + folder: Path, origin: str, profile: ResolvedSensorProfile +) -> CatalogEntry: + """An entry the SDK resolved, valid or rejected, with its verdict kept whole.""" + validation = profile.validation + errors = tuple(str(issue) for issue in validation.errors) + return CatalogEntry( + key=str(profile.sensor_json_path), + folder=folder.name, + bundle_path=folder.resolve(), + origin=origin, + display_name=profile.display_name, + product_family=profile.product_family, + profile_id=profile.profile_id, + sensor_json_path=profile.sensor_json_path, + processing_backend=profile.processing_backend, + valid=validation.usable, + state=validation.state, + reason="; ".join(errors), + errors=errors, + warnings=tuple(str(issue) for issue in validation.warnings), + diagnostics=tuple(profile.diagnostics), + profile=profile, + ) + + +def _unreadable_entry(folder: Path, origin: str, exc: Exception) -> CatalogEntry: + """An entry no bundle could be read from, with the SDK's explanation as its reason.""" + display_name, family, profile_id = _best_effort_identity(folder) + sensor_json = folder / "sensor.json" + has_config = sensor_json.is_file() + return CatalogEntry( + key=str(sensor_json if has_config else folder), + folder=folder.name, + bundle_path=folder.resolve(), + origin=origin, + display_name=display_name, + product_family=family, + profile_id=profile_id, + sensor_json_path=sensor_json if has_config else None, + processing_backend="", + valid=False, + state=STATE_UNREADABLE, + reason=_unreadable_reason(folder, exc), + ) + + +def _unreadable_reason(folder: Path, exc: Exception) -> str: + """The SDK's own explanation, prefixed when the folder holds no configuration at all. + + "Not a built-in bundle" is true of a folder of stray JSON but does not say *why* it is not + a configuration. Naming the files that are there does, and it is the sentence that stops + ``force_calibration.json`` from being offered as a sensor. + """ + manifest = folder / _sdk().DESCRIPTOR_FILENAME + note = "" + if not manifest.is_file() and not (folder / "sensor.json").is_file(): + names = ", ".join(path.name for path in _json_files(folder)) + note = ( + f"only {names} present; a sensor configuration needs sensor.json or a " + "manifest.json descriptor, so this folder is not offered as one. " + ) + return f"{folder.name}: {note}the SDK rejected it: {exc}" + + +def _best_effort_identity(folder: Path) -> tuple[str, str, str]: + """``(display_name, family, profile_id)`` a broken bundle still declares. + + A bundle whose resolution failed should not reduce to a bare folder name when the fields + that *are* readable identify it. Only ``sensor.json`` and ``manifest.json`` are consulted: + a calibration document's keys are not a profile's identity. The entry stays invalid. + """ + manifest = folder / _sdk().DESCRIPTOR_FILENAME + sensor_json = folder / "sensor.json" + data: dict[str, Any] = {} + for path in (sensor_json, manifest): + parsed = _read_json_object(path) + if parsed: + data.update(parsed) + built_in = _sdk().BUILT_IN_PROFILES.get(folder.name, {}) + display = _text(data.get("display_name")) or folder.name + family = _text(data.get("product_family")) or _text(built_in.get("product_family")) + return display, family, _text(data.get("profile_id")) + + +def _text(value: Any) -> str: + """A non-empty stripped string, or ``""``.""" + return value.strip() if isinstance(value, str) and value.strip() else "" + + +def _read_json_object(path: Path) -> dict[str, Any] | None: + """A JSON object, or ``None`` when the file is unreadable or is not an object.""" + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + return data if isinstance(data, dict) else None + + +def _json_files(folder: Path) -> tuple[Path, ...]: + """The top-level JSON documents in ``folder``, sorted by name.""" + try: + found = [ + path + for path in folder.iterdir() + if path.is_file() and path.suffix.lower() == ".json" + ] + except OSError: + return () + return tuple(sorted(found, key=_sort_key)) + + +def _mark_duplicates(entries: list[CatalogEntry]) -> list[CatalogEntry]: + """Flag every entry sharing a ``profile_id`` with another, naming the others. + + Nothing is hidden or shadowed: both sides stay listed and each says which other bundles + declare the same profile, so a caller can refuse an ambiguous selection instead of + silently starting one of two profiles that claim to be the same thing. + """ + by_profile: dict[str, list[CatalogEntry]] = {} + for entry in entries: + if entry.profile_id: + by_profile.setdefault(entry.profile_id, []).append(entry) + if not any(len(group) > 1 for group in by_profile.values()): + return entries + marked: list[CatalogEntry] = [] + for entry in entries: + others = tuple( + other.ref + for other in by_profile.get(entry.profile_id, []) + if other.key != entry.key + ) + marked.append(replace(entry, duplicates=others) if others else entry) + return marked diff --git a/shared/src/orisys_app_shared/sensors/catalog.py b/shared/src/orisys_app_shared/sensors/catalog.py index 27b641a..6e96dac 100644 --- a/shared/src/orisys_app_shared/sensors/catalog.py +++ b/shared/src/orisys_app_shared/sensors/catalog.py @@ -1,43 +1,46 @@ -"""List local editable sensor bundles under config/sensors.""" +"""List the sensor bundles the app offers, over both the shipped and user roots. + +This used to be a hard-coded product list that made every folder under ``config/sensors`` a +product whether or not a bundle was there. The offered set now comes from +:mod:`orisys_app_shared.sensors.bundle_catalog`, which asks the SDK what each folder is: a +descriptor bundle, a legacy ``sensor.json`` bundle, or an entry that stays listed as invalid +with the SDK's own reason. + +The two functions below stay the picker's compatibility surface. The picker itself uses the +catalog object directly, so it can show each entry's family, profile, origin and validity. +""" from __future__ import annotations -from orisys_app_shared.paths import sensors_dir - -#: Product types the UI offers. -#: -#: ``finger-mems`` is a deprecated product line that is deliberately **not** part of the SDK -#: (there is no MEMS read path there), so it is carried studio-side only: its bundle copies -#: ``box``'s ``sensor.json`` without a ``force_calibration.json``, which means N units are -#: unavailable and the app reports raw for it — the honest answer for an uncalibrated device. -#: It runs on the box/mini pipeline, because the worker routing is a binary ``== "finger"`` -#: test (see ``session_controller.worker_for``). -#: -#: NOTE: the copied ``camera``/``stitcher`` numbers belong to the box prototype. If the real -#: MEMS-finger device stitches differently, live frames are refused with -#: ``field_domain_mismatch`` — correct them from the device, not by guessing. -PRODUCT_TYPES = ("box", "mini", "finger", "finger-mems") +from orisys_app_shared.sensors.bundle_catalog import ( + FAMILY_UNIDENTIFIED, + scan_sensor_catalog, +) def list_sensor_types() -> list[str]: - return list(PRODUCT_TYPES) + """The family groups the installed bundles provide, unidentified last. + + A family is offered when a bundle exists for it — including a bundle the SDK rejects, which + must stay visible with its reason rather than vanish from the picker. + """ + return scan_sensor_catalog().families() def list_configs_for_type(sensor_type: str) -> list[tuple[str, str]]: - """Return ``(display_name, config_key)`` entries for one sensor type. + """Return ``(display_name, config_key)`` entries for one family. - Prefers ``config/sensors//sensor.json`` when present; otherwise a - single default entry named after the type (for prefs / future Run). + The key is the configuration file path the app already passes around (``sensor.json``, or + the bundle folder for an entry that has none), so a selection keeps its meaning. When no + bundle exists at all, a single default entry named after the type is offered — for + preferences and a future Run — exactly as before. """ name = str(sensor_type or "").strip() if not name: return [] - root = sensors_dir() / name - sensor_json = root / "sensor.json" - if sensor_json.is_file(): - return [(f"{name}/sensor.json", str(sensor_json))] - # Also accept extra *.json next to sensor.json later; for now only sensor.json. - extras = sorted(p for p in root.glob("*.json") if p.name != "manifest.json") - if extras: - return [(f"{name}/{p.name}", str(p)) for p in extras] + entries = scan_sensor_catalog().entries_for_family(name) + if entries: + return [(entry.label(), entry.key) for entry in entries] + if name == FAMILY_UNIDENTIFIED: + return [] return [(name, name)] diff --git a/shared/tests/test_sensor_bundle_catalog.py b/shared/tests/test_sensor_bundle_catalog.py new file mode 100644 index 0000000..911be2b --- /dev/null +++ b/shared/tests/test_sensor_bundle_catalog.py @@ -0,0 +1,292 @@ +"""The sensor configuration catalog: discovery, validity and duplicate profile ids. + +The catalog replaced two rules that were wrong in the same direction — a hard-coded product +list, and "any ``*.json`` beside a ``sensor.json`` is another configuration". These tests build +a tree containing exactly the shapes those rules mishandled: a descriptor bundle, a legacy +``sensor.json`` bundle, a folder holding only ``force_calibration.json``, a malformed +``manifest.json``, a bundle whose declared asset is missing, and two bundles sharing one +``profile_id``. + +The tree is built here, in the test's own temporary directory: a committed fixture would be a +second copy of the bundle format that could drift from the real one without anyone noticing. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest +from orisys_app_shared.sensors import ( + FAMILY_UNIDENTIFIED, + ORIGIN_FALLBACK, + ORIGIN_SHIPPED, + ORIGIN_USER, + ROOT_KIND_FALLBACK, + ROOT_KIND_SIBLING, + STATE_UNREADABLE, + resolve_user_root, + scan_sensor_catalog, +) + +CAMERA_OK = {"resolution_raw": [640, 480], "fps": 30, "resolution_stitch": [100, 100]} +TRACKER_OK = {"algorithm": "dis"} + + +def _config( + *, declared_type: str | None = None, stitcher: bool = False +) -> dict[str, Any]: + data: dict[str, Any] = { + "name": "fixture", + "camera": CAMERA_OK, + "tracker": TRACKER_OK, + } + if declared_type: + data["type"] = declared_type + if stitcher: + data["stitcher"] = {"n": [1, 1]} + return data + + +def _descriptor(profile_id: str, **overrides: Any) -> dict[str, Any]: + data: dict[str, Any] = { + "schema_version": 1, + "profile_id": profile_id, + "display_name": profile_id, + "product_family": "mini", + "processing_backend": "single_camera", + "sensor_json": "sensor.json", + } + data.update(overrides) + return data + + +def _write(path: Path, payload: object) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + text = payload if isinstance(payload, str) else json.dumps(payload, indent=2) + path.write_text(text, encoding="utf-8") + return path + + +@pytest.fixture +def trees(tmp_path): + """One shipped root and one user root covering every shape the catalog must handle.""" + shipped = tmp_path / "shipped" + user = tmp_path / "user" + # A legacy bundle: a built-in folder name, sensor.json, and no manifest. The config carries + # a stitcher block, which is what the SDK's own rule reads to reach compound_eye. + _write(shipped / "box" / "sensor.json", _config(stitcher=True)) + # Arbitrary JSON only: not a configuration. + _write(shipped / "calib-only" / "force_calibration.json", {"axes": ["fn", "fx"]}) + # A descriptor bundle. + _write(user / "mini-lab" / "sensor.json", _config()) + _write(user / "mini-lab" / "manifest.json", _descriptor("orisys.mini.lab-r1")) + # A manifest that is not JSON at all. + _write( + user / "broken-manifest" / "manifest.json", + '{"schema_version": 1, "profile_id":', + ) + # A descriptor whose declared computational asset is absent. + _write(user / "missing-assets" / "sensor.json", _config()) + _write( + user / "missing-assets" / "manifest.json", + _descriptor( + "orisys.finger.missing-r1", + product_family="finger", + processing_backend="uv_unroll", + computational_assets={"unroll_map": "assets/not-here.npz"}, + ), + ) + # Two bundles declaring one profile_id. + for name in ("dup-a", "dup-b"): + _write(user / name / "sensor.json", _config()) + _write(user / name / "manifest.json", _descriptor("orisys.mini.duplicate-r1")) + # No JSON document at all: a folder, not a bundle. + _write(user / "notes" / "readme.txt", "not a bundle") + return scan_sensor_catalog(shipped_root=shipped, user_root=user) + + +def test_every_bundle_shape_is_discovered_with_its_origin(trees): + refs = {entry.ref for entry in trees.entries} + assert refs == { + "shipped/box", + "shipped/calib-only", + "user/mini-lab", + "user/broken-manifest", + "user/missing-assets", + "user/dup-a", + "user/dup-b", + }, f"a folder with no JSON document is not a bundle candidate: {sorted(refs)}" + assert trees.by_ref("shipped/box").origin == ORIGIN_SHIPPED + assert trees.by_ref("user/mini-lab").origin == ORIGIN_USER + + +def test_a_calibration_document_is_not_offered_as_a_configuration(trees): + entry = trees.by_ref("shipped/calib-only") + assert entry is not None, "the folder is retained, so its reason can be read" + assert not entry.valid + assert "force_calibration.json" in entry.reason + assert "not offered as one" in entry.reason + # Nothing about it may be presented as a product identity. + assert entry.product_family == "" + assert entry.family == FAMILY_UNIDENTIFIED + assert entry.profile_id == "" + assert entry.start_blocked_reason == entry.reason + + +def test_a_malformed_manifest_is_invalid_with_the_sdks_own_reason(trees): + entry = trees.by_ref("user/broken-manifest") + assert entry is not None and not entry.valid + assert entry.state == STATE_UNREADABLE + assert "unreadable descriptor" in entry.reason, entry.reason + + +def test_the_two_invalid_shapes_carry_distinct_reasons(trees): + calibration_only = trees.by_ref("shipped/calib-only").reason + malformed = trees.by_ref("user/broken-manifest").reason + assert calibration_only != malformed + assert "force_calibration.json" in calibration_only + assert "force_calibration.json" not in malformed + assert "unreadable descriptor" in malformed + assert "unreadable descriptor" not in calibration_only + + +def test_a_descriptor_and_a_legacy_bundle_both_resolve(trees): + descriptor = trees.by_ref("user/mini-lab") + assert descriptor.valid and descriptor.profile is not None + assert descriptor.profile_id == "orisys.mini.lab-r1" + assert descriptor.product_family == "mini" + assert descriptor.processing_backend == "single_camera" + assert descriptor.sensor_json_path == descriptor.bundle_path / "sensor.json" + assert descriptor.start_blocked_reason is None + + legacy = trees.by_ref("shipped/box") + assert legacy.valid and legacy.profile is not None + # The identity is the SDK's built-in one for that folder ... + assert legacy.profile_id == "orisys.box.legacy" + assert legacy.product_family == "box" + # ... and the backend is the configuration's (the stitcher block), decided by the SDK. + assert legacy.processing_backend == "compound_eye" + assert legacy.sensor_json_path == legacy.bundle_path / "sensor.json" + + +def test_a_legacy_config_that_contradicts_the_builtin_table_is_listed_with_the_sdks_reason( + tmp_path, +): + """A folder name identifies a legacy bundle; it must not decide its backend. + + The configuration wins — the SDK says so and this asserts it — but the SDK then validates + the backend requirements against the *descriptor's* pre-override backend, so a legacy + bundle whose config contradicts the built-in table fails a check that does not apply to the + backend it actually uses. That is an SDK-side finding, reported rather than worked around + here: the catalog's job is to pass the verdict on, not to second-guess it. If this test ever + fails because the entry became valid, the SDK fixed it: delete the ``not entry.valid`` + assertion and record where. + """ + shipped = tmp_path / "shipped" + _write(shipped / "box" / "sensor.json", _config(declared_type="single_camera")) + + entry = scan_sensor_catalog( + shipped_root=shipped, user_root=tmp_path / "user" + ).by_ref("shipped/box") + assert entry.profile is not None + assert entry.profile.processing_backend == "single_camera", ( + "the config decides the backend" + ) + assert entry.product_family == "box", "the folder name decides the legacy identity" + assert not entry.valid + assert "stitcher" in entry.reason + assert entry.start_blocked_reason == entry.reason + + +def test_a_shared_profile_id_is_flagged_and_both_entries_stay_listed(trees): + first = trees.by_ref("user/dup-a") + second = trees.by_ref("user/dup-b") + assert first.valid and second.valid, "sharing a profile_id is not itself a failure" + assert first.profile_id == second.profile_id == "orisys.mini.duplicate-r1" + assert first.duplicates == ("user/dup-b",) + assert second.duplicates == ("user/dup-a",) + assert first.duplicate and second.duplicate + assert "user/dup-b" in first.start_blocked_reason + assert "user/dup-a" in second.start_blocked_reason + # Nothing is shadowed: both are still offered under their family, and by key. + offered = {entry.key for entry in trees.entries_for_family("mini")} + assert {first.key, second.key} <= offered + assert trees.start_blocked(first.key) == first.start_blocked_reason + assert trees.start_blocked(second.key) == second.start_blocked_reason + + +def test_a_missing_computational_asset_is_reported_and_keeps_its_identity(trees): + entry = trees.by_ref("user/missing-assets") + assert not entry.valid + assert entry.state == "failed", "the SDK's own state, not the catalog's" + assert entry.profile_id == "orisys.finger.missing-r1" + assert entry.product_family == "finger" + assert "unroll_map" in entry.reason + assert entry.start_blocked_reason + + +def test_an_entry_the_sdk_rejects_is_never_reported_as_startable(trees): + for entry in trees.entries: + expected = entry.valid and not entry.duplicate + assert (entry.start_blocked_reason is None) == expected, entry.ref + + +def test_the_label_states_family_profile_origin_and_validity(trees): + entry = trees.by_ref("user/missing-assets") + label = entry.label() + assert "finger" in label + assert entry.profile_id in label + assert ORIGIN_USER in label + assert entry.state in label + + +def test_the_root_decision_names_the_root_in_use(): + decision = resolve_user_root() + assert decision.kind in (ROOT_KIND_SIBLING, ROOT_KIND_FALLBACK) + assert str(decision.path) in decision.reason + + +def test_an_unwritable_sibling_falls_back_to_the_per_user_root(tmp_path, monkeypatch): + from orisys_app_shared.sensors import bundle_catalog + + sibling = tmp_path / "no" / "such" / "sensors-user" + fallback = tmp_path / "per-user" + monkeypatch.setattr(bundle_catalog, "user_sensors_dir", lambda: sibling) + monkeypatch.setattr(bundle_catalog, "fallback_user_sensors_dir", lambda: fallback) + + decision = bundle_catalog.resolve_user_root() + assert decision.kind == ROOT_KIND_FALLBACK + assert decision.path == fallback + assert decision.writable is True + assert str(sibling) in decision.reason + + +def test_entries_under_the_fallback_root_keep_that_origin(tmp_path, monkeypatch): + from orisys_app_shared.sensors import bundle_catalog + + shipped = tmp_path / "shipped" + fallback = tmp_path / "per-user" + _write(fallback / "mini-lab" / "sensor.json", _config()) + _write(fallback / "mini-lab" / "manifest.json", _descriptor("orisys.mini.lab-r1")) + monkeypatch.setattr( + bundle_catalog, "user_sensors_dir", lambda: tmp_path / "no" / "such" + ) + monkeypatch.setattr(bundle_catalog, "fallback_user_sensors_dir", lambda: fallback) + + catalog = bundle_catalog.scan_sensor_catalog(shipped_root=shipped) + assert [entry.ref for entry in catalog.entries_for_origin(ORIGIN_FALLBACK)] == [ + "fallback/mini-lab" + ] + assert catalog.user_root_kind == ROOT_KIND_FALLBACK + + +def test_an_absent_root_is_not_an_error(tmp_path): + catalog = scan_sensor_catalog( + shipped_root=tmp_path / "gone", user_root=tmp_path / "also-gone" + ) + assert catalog.entries == () + assert catalog.roots[0].exists is False + assert catalog.user_root_exists is False + assert catalog.families() == []