react-viewer: firmware update proposal + one-click update

wrappers/rest-api/app/api/endpoints/firmware.py

@@ -84,3 +84,22 @@ async def update_firmware_from_file(
     except Exception:
         logging.exception("Unexpected error updating firmware for %s", device_id)
         raise HTTPException(status_code=500, detail="Unexpected error while updating firmware")
+
+
+@router.post("/{device_id}/firmware/update_from_recommended", response_model=dict)
+async def update_firmware_from_recommended(
+    device_id: str,
+    rs_manager: RealSenseManager = Depends(get_realsense_manager),
+):
+    """Download the device's recommended firmware (from the online versions DB) and flash it.
+
+    One-click alternative to update_from_file: no upload — the backend fetches the .bin
+    and reuses the same DFU flow (and Socket.IO progress events).
+    """
+    try:
+        return await run_in_threadpool(rs_manager.update_firmware_from_recommended, device_id)
+    except RealSenseError as e:
+        raise HTTPException(status_code=e.status_code, detail=e.detail)
+    except Exception:
+        logging.exception("Unexpected error updating firmware from recommended for %s", device_id)
+        raise HTTPException(status_code=500, detail="Unexpected error while updating firmware")

wrappers/rest-api/app/services/firmware.py

@@ -0,0 +1,157 @@
+# License: Apache 2.0. See LICENSE file in root directory.
+# Copyright(c) 2026 RealSense, Inc. All Rights Reserved.
+
+"""Firmware version comparison, online versions-DB lookup, and image download.
+
+Pure logic with no pyrealsense2 dependency, so tests can import it directly without
+loading the SDK. The SDK no longer bundles firmware, so the recommended version is
+looked up from the online versions DB (mirrors the C++ viewer's server_versions_db_url),
+and the image is downloaded into memory and flashed from there (like the C++ viewer's
+download_to_bytes_vector) — no temp files.
+"""
+
+import json
+import logging
+import platform
+import threading
+import urllib.parse
+import urllib.request
+from typing import Any, Dict, Optional
+
+from app.core.errors import RealSenseError
+
+FW_STATUS_UNKNOWN = "unknown"
+FW_STATUS_OUTDATED = "outdated"
+FW_STATUS_UP_TO_DATE = "up_to_date"
+
+SERVER_VERSIONS_DB_URL = "https://librealsense.realsenseai.com/Releases/rs_versions_db.json"
+_MAX_FW_DOWNLOAD_BYTES = 64 * 1024 * 1024
+
+# The versions DB is fetched once per process (no TTL — re-downloaded on backend restart,
+# like the C++ viewer). Guarded by a lock since request-handler threads read/populate it.
+_versions_db_lock = threading.Lock()
+_versions_db_cache: Dict[str, Any] = {"entries": None}
+
+
+def _parse_fw_version(v: Optional[str]) -> Optional[tuple]:
+    """Parse a dotted firmware version ("5.17.0.10") into an int tuple, or None."""
+    if not v:
+        return None
+    try:
+        return tuple(int(p) for p in v.split("."))
+    except ValueError:
+        return None
+
+
+def firmware_update_status(current: Optional[str], recommended: Optional[str]) -> str:
+    """Compare current vs recommended FW versions numerically.
+
+    Returns FW_STATUS_OUTDATED when current < recommended, FW_STATUS_UP_TO_DATE when
+    current >= recommended, and FW_STATUS_UNKNOWN when either can't be parsed.
+    """
+    cur = _parse_fw_version(current)
+    rec = _parse_fw_version(recommended)
+    if cur is None or rec is None:
+        return FW_STATUS_UNKNOWN
+    return FW_STATUS_OUTDATED if cur < rec else FW_STATUS_UP_TO_DATE
+
+
+def _strip_intel_prefix(name):
+    """Drop the legacy 'Intel ' vendor prefix. Newer SDKs report names without it while
+    the DB still carries it, so comparisons must be prefix-agnostic (mirrors the C++
+    versions_db_manager::strip_intel_prefix)."""
+    prefix = "Intel "
+    return name[len(prefix):] if name.startswith(prefix) else name
+
+
+def _device_name_matches(db_name, device_name):
+    """True if a DB device_name pattern matches the reported name, '*' = trailing wildcard.
+    Mirrors versions_db_manager::is_device_name_equal (prefix-stripped, compare up to '*')."""
+    db = _strip_intel_prefix(db_name or "")
+    cmp = _strip_intel_prefix(device_name or "")
+    star = db.find("*")
+    if star == -1:
+        return db == cmp
+    return db[:star] == cmp[:star]
+
+
+def _pick_recommended_fw(entries, device_name, host_platform):
+    """Pick the best RECOMMENDED FIRMWARE (version, link) for a device from DB entries.
+
+    Matches each entry's `device_name` against the device (prefix-agnostic, '*' wildcard);
+    the most specific pattern (longest literal) wins. Entry `platform` must be '*' or
+    match the host. Returns (None, None) when nothing matches.
+    """
+    if not entries or not device_name:
+        return None, None
+    host = (host_platform or "").lower()
+    best = None  # (specificity, version, link)
+    for e in entries:
+        if e.get("component") != "FIRMWARE" or e.get("policy_type") != "RECOMMENDED":
+            continue
+        pattern = e.get("device_name", "")
+        if not _device_name_matches(pattern, device_name):
+            continue
+        plat = (e.get("platform") or "*").lower()
+        if plat != "*" and plat not in host and host not in plat:
+            continue
+        specificity = len(pattern.replace("*", ""))
+        if best is None or specificity > best[0]:
+            best = (specificity, e.get("version"), e.get("link"))
+    return (best[1], best[2]) if best else (None, None)
+
+
+def _fetch_versions_db():
+    """Return the DB 'versions' list (cached for the process lifetime), or None on failure."""
+    with _versions_db_lock:
+        if _versions_db_cache["entries"] is not None:
+            return _versions_db_cache["entries"]
+        try:
+            with urllib.request.urlopen(SERVER_VERSIONS_DB_URL, timeout=10) as resp:
+                data = json.loads(resp.read().decode("utf-8"))
+            entries = data.get("versions", [])
+            _versions_db_cache["entries"] = entries
+            return entries
+        except Exception as e:  # network down, timeout, bad JSON — degrade gracefully
+            logging.warning("Could not fetch firmware versions DB: %s", e)
+            return None
+
+
+def recommended_firmware(device_name):
+    """Recommended FIRMWARE (version, link) for a device from the online DB, or (None, None)."""
+    return _pick_recommended_fw(_fetch_versions_db(), device_name, platform.system())
+
+
+def download_firmware(url: str, on_progress=None) -> bytes:
+    """Download a firmware .bin into memory (no disk cache). Raises RealSenseError on failure.
+
+    Restricts to http(s) URLs (the link comes from the versions DB; reject anything else
+    to avoid file:// / relative / other-scheme fetches). Streams in chunks and reports
+    0..1 progress via on_progress(fraction) when Content-Length is available.
+    """
+    parsed = urllib.parse.urlparse(url)
+    if parsed.scheme not in ("http", "https") or not parsed.netloc:
+        raise RealSenseError(status_code=400, detail="Refusing to download firmware from a non-http(s) URL")
+    try:
+        buf = bytearray()
+        with urllib.request.urlopen(url, timeout=30) as resp:
+            total = int(resp.headers.get("Content-Length") or 0)
+            while True:
+                chunk = resp.read(256 * 1024)
+                if not chunk:
+                    break
+                buf.extend(chunk)
+                if len(buf) > _MAX_FW_DOWNLOAD_BYTES:
+                    raise RealSenseError(status_code=413, detail="Firmware image exceeds size limit")
+                if on_progress and total:
+                    on_progress(min(len(buf) / total, 1.0))
+        data = bytes(buf)
+    except RealSenseError:
+        raise
+    except Exception as e:
+        raise RealSenseError(status_code=502, detail=f"Failed to download firmware: {e}")
+    if not data:
+        raise RealSenseError(status_code=502, detail="Downloaded firmware image is empty")
+    if on_progress:
+        on_progress(1.0)
+    return data

wrappers/rest-api/app/services/rs_manager.py

@@ -2,6 +2,7 @@
 # Copyright(c) 2026 RealSense, Inc. All Rights Reserved.
 
 import asyncio
+import os
 import platform
 import struct
 import threading
@@ -27,12 +28,11 @@
 from datetime import datetime
 
 from app.services.metadata_socket_server import MetadataSocketServer
+from app.services import firmware as fw
 
 
 _IS_WINDOWS = platform.system() == "Windows"
 
-FW_STATUS_UNKNOWN = "unknown"
-
 # Recent color frames retained per device for texturing the 3D point cloud.
 # Five frames at 30 fps covers the typical depth/color SDK-latency offset
 # (~one capture interval); the picker selects the entry whose timestamp is
@@ -210,9 +210,11 @@ def _info(key, default=None):
             name=_info(rs.camera_info.name, "Unknown Device"),
             serial_number=device_id,
             firmware_version=_info(rs.camera_info.firmware_version),
+            # Recommended version comes from the online versions DB, fetched lazily on the
+            # firmware-status check (the SDK no longer bundles/reports it); unknown until then.
             recommended_firmware_version=None,
-            firmware_status=FW_STATUS_UNKNOWN,
-            firmware_file_available=False,
+            firmware_status=fw.FW_STATUS_UNKNOWN,
+            firmware_file_available=False,  # no bundled firmware — user supplies the image
             physical_port=_info(rs.camera_info.physical_port),
             usb_type=_info(rs.camera_info.usb_type_descriptor),
             product_id=_info(rs.camera_info.product_id),
@@ -303,14 +305,42 @@ def get_device(self, device_id: str, force_refresh: bool = False) -> DeviceInfo:
     def get_firmware_status(self, device_id: str) -> Dict[str, Any]:
         """Return firmware status metadata for a device."""
         device = self.get_device(device_id)
+        recommended, link = fw.recommended_firmware(device.name)
         return {
             "device_id": device_id,
             "current": device.firmware_version,
-            "recommended": None,
-            "status": device.firmware_status or FW_STATUS_UNKNOWN,
-            "file_available": False,
+            "recommended": recommended,
+            "status": fw.firmware_update_status(device.firmware_version, recommended),
+            "file_available": False,  # no bundled image — user downloads the .bin (see link)
+            "link": link,
         }
 
+    def update_firmware_from_recommended(self, device_id: str) -> Dict[str, Any]:
+        """Download the device's recommended firmware image and flash it.
+
+        Looks up the recommended .bin from the online versions DB, downloads it into
+        memory, and reuses the standard DFU flash flow — so Socket.IO
+        progress/success/failure events are emitted exactly as for a user-supplied file.
+        """
+        device = self.get_device(device_id)
+        _recommended, link = fw.recommended_firmware(device.name)
+        if not link:
+            raise RealSenseError(status_code=404, detail="No recommended firmware available for this device")
+        # Emit download progress under the "downloading" phase so the UI can show it
+        # before the (separate) install phase begins.
+        _holder, on_download = self._make_fw_progress_callback(device_id, phase="downloading")
+        try:
+            fw_bytes = fw.download_firmware(link, on_progress=on_download)
+        except Exception as exc:
+            # Surface download failures on the same channel as flash failures so the
+            # progress modal shows the error instead of hanging.
+            self._emit_socket_event(
+                f"firmware_update_failed_{device_id}",
+                {"device_id": device_id, "error": str(exc)},
+            )
+            raise
+        return self.update_firmware_from_bytes(device_id, fw_bytes)
+
     @staticmethod
     def _is_update_device(dev: rs.device) -> bool:
         """True if the device exposes the DFU update interface.
@@ -357,7 +387,7 @@ def update_firmware_from_bytes(self, device_id: str, fw_bytes: bytes) -> Dict[st
             # Always emit a starting progress so the UI doesn't stay at 0% forever
             self._emit_socket_event(
                 f"firmware_progress_{device_id}",
-                {"device_id": device_id, "progress": 0.0},
+                {"device_id": device_id, "progress": 0.0, "phase": "installing"},
             )
 
             try:
@@ -400,7 +430,7 @@ def update_firmware_from_bytes(self, device_id: str, fw_bytes: bytes) -> Dict[st
 
             self._emit_socket_event(
                 f"firmware_progress_{device_id}",
-                {"device_id": device_id, "progress": 1.0},
+                {"device_id": device_id, "progress": 1.0, "phase": "installing"},
             )
             self._emit_socket_event(
                 f"firmware_update_success_{device_id}",
@@ -493,11 +523,12 @@ def _resolve_firmware_update_id(target_dev: rs.device, device_id: str) -> str:
         return firmware_update_id
 
     def _make_fw_progress_callback(
-        self, device_id: str,
+        self, device_id: str, phase: str = "installing",
     ) -> Tuple[Dict[str, float], Callable[[float], None]]:
         """Build the on_progress callback and a holder dict that records the latest value.
 
         Rate-limits emissions to ~10/sec but always lets the final 100% through.
+        `phase` ("downloading" | "installing") lets the UI label what's happening.
         """
         progress_holder = {"value": 0.0}
         last_emit_ts = {"value": 0.0}
@@ -510,7 +541,7 @@ def _on_progress(p: float) -> None:
             last_emit_ts["value"] = now
             self._emit_socket_event(
                 f"firmware_progress_{device_id}",
-                {"device_id": device_id, "progress": float(p)},
+                {"device_id": device_id, "progress": float(p), "phase": phase},
             )
 
         return progress_holder, _on_progress

wrappers/rest-api/tests/test_firmware_status.py

@@ -0,0 +1,82 @@
+# License: Apache 2.0. See LICENSE file in root directory.
+# Copyright(c) 2026 RealSense, Inc. All Rights Reserved.
+
+from app.services.firmware import (
+    firmware_update_status,
+    _pick_recommended_fw,
+    FW_STATUS_OUTDATED,
+    FW_STATUS_UP_TO_DATE,
+    FW_STATUS_UNKNOWN,
+)
+
+_DB = [
+    {"device_name": "Intel RealSense D4*", "policy_type": "RECOMMENDED", "component": "FIRMWARE",
+     "version": "5.17.0.10", "platform": "*", "link": "d4x.bin"},
+    {"device_name": "Intel RealSense D455", "policy_type": "RECOMMENDED", "component": "FIRMWARE",
+     "version": "5.17.3.10", "platform": "*", "link": "d455.bin"},
+    {"device_name": "Intel RealSense D4*", "policy_type": "RECOMMENDED", "component": "LIBREALSENSE",
+     "version": "2.58.2", "platform": "*", "link": "sw"},
+    {"device_name": "Intel RealSense D457", "policy_type": "REQUIRED", "component": "FIRMWARE",
+     "version": "9.9.9.9", "platform": "*", "link": "x"},
+]
+
+
+def test_current_below_recommended_is_outdated():
+    assert firmware_update_status("5.16.0.1", "5.17.0.9") == FW_STATUS_OUTDATED
+
+
+def test_current_equal_recommended_is_up_to_date():
+    assert firmware_update_status("5.17.0.9", "5.17.0.9") == FW_STATUS_UP_TO_DATE
+
+
+def test_current_above_recommended_is_up_to_date():
+    # numeric compare, not lexical: 10 > 9
+    assert firmware_update_status("5.17.0.10", "5.17.0.9") == FW_STATUS_UP_TO_DATE
+
+
+def test_missing_recommended_is_unknown():
+    assert firmware_update_status("5.17.0.9", None) == FW_STATUS_UNKNOWN
+    assert firmware_update_status("5.17.0.9", "") == FW_STATUS_UNKNOWN
+
+
+def test_missing_current_is_unknown():
+    assert firmware_update_status(None, "5.17.0.9") == FW_STATUS_UNKNOWN
+
+
+def test_non_numeric_is_unknown():
+    assert firmware_update_status("abc", "5.17.0.9") == FW_STATUS_UNKNOWN
+
+
+def test_pick_prefers_specific_over_wildcard():
+    # D455 has an exact entry (5.17.3.10) that must win over the D4* wildcard (5.17.0.10)
+    ver, link = _pick_recommended_fw(_DB, "Intel RealSense D455", "Windows")
+    assert (ver, link) == ("5.17.3.10", "d455.bin")
+
+
+def test_pick_falls_back_to_wildcard():
+    ver, link = _pick_recommended_fw(_DB, "Intel RealSense D435", "Windows")
+    assert (ver, link) == ("5.17.0.10", "d4x.bin")
+
+
+def test_pick_ignores_non_firmware_and_non_recommended():
+    # LIBREALSENSE component and REQUIRED policy must never be returned as FW recommendation
+    ver, _ = _pick_recommended_fw(_DB, "Intel RealSense D457", "Windows")
+    assert ver == "5.17.0.10"  # matches D4* FIRMWARE/RECOMMENDED, not the REQUIRED 9.9.9.9
+
+
+def test_pick_matches_prefixless_reported_name():
+    # SDK reports "RealSense D455" (no "Intel " prefix); DB uses "Intel RealSense D455".
+    # The prefix must be stripped on both sides or the proposal never fires.
+    ver, link = _pick_recommended_fw(_DB, "RealSense D455", "Windows")
+    assert (ver, link) == ("5.17.3.10", "d455.bin")
+
+
+def test_pick_prefixless_falls_back_to_wildcard():
+    ver, _ = _pick_recommended_fw(_DB, "RealSense D435", "Windows")
+    assert ver == "5.17.0.10"
+
+
+def test_pick_no_match_returns_none():
+    assert _pick_recommended_fw(_DB, "Some Other Camera", "Windows") == (None, None)
+    assert _pick_recommended_fw([], "Intel RealSense D455", "Windows") == (None, None)
+    assert _pick_recommended_fw(None, "Intel RealSense D455", "Windows") == (None, None)