From 16101a787fe8da09c8b43296e3695b6e849c1675 Mon Sep 17 00:00:00 2001 From: Brandon Schneider Date: Sat, 30 May 2026 19:57:07 -0500 Subject: [PATCH] feat(infra): device capability probe with framebuffer fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit device_capability_probe.py: classify every device into a compute tier. Tiers (highest to lowest): GPU_CUDA — NVIDIA discrete + CUDA (NVENC, Ray GPU worker) GPU_VAAPI — AMD/Intel discrete + VA-API (hardware encode) GPU_APU — AMD integrated, yuvj420p, bandwidth-optimized CPU_FFMPEG — Software encode only (libx264) FRAMEBUFFER — /dev/fb0 DMA backplane (8.29 MB/frame at 1080p) ESP32 — MCU, Q0_16 scalar in FreeRTOS idle hook RELAY — Network only, no compute OFFLINE — Unreachable Features: - Multi-GPU DRM render node scanning (card0=AMD, card1=NVIDIA) - APU vs dGPU classification via device name + VRAM heuristics - Framebuffer detection with /sys/class/graphics/fb0 resolution - Ray scheduling helpers (get_ray_placement_strategy) - Cluster probe via SSH - JSON + human-readable output --- .../shim/device_capability_probe.py | 540 ++++++++++++++++++ 1 file changed, 540 insertions(+) create mode 100644 4-Infrastructure/shim/device_capability_probe.py diff --git a/4-Infrastructure/shim/device_capability_probe.py b/4-Infrastructure/shim/device_capability_probe.py new file mode 100644 index 00000000..aa792f4c --- /dev/null +++ b/4-Infrastructure/shim/device_capability_probe.py @@ -0,0 +1,540 @@ +#!/usr/bin/env python3 +""" +Device Capability Probe — classify every device in the cluster into a compute tier. + +Scans DRM render nodes, FFmpeg encoders, framebuffer devices, and network +connectivity to assign each device the cheapest compute role it can handle. + +Hierarchy (highest to lowest capability): + GPU_CUDA — NVIDIA discrete with CUDA (Ray GPU worker, NVENC) + GPU_VAAPI — AMD/Intel with VA-API (Ray worker, hardware encode) + GPU_APU — AMD integrated, shared memory (yuvj420p, bandwidth-optimized) + CPU_FFMPEG — No GPU, but FFmpeg with libx264 (Ray CPU worker) + FRAMEBUFFER — Has /dev/fb0 only (DMA backplane, 8.29 MB/frame at 1080p) + ESP32 — MCU with Q0_16 scalar compute (FreeRTOS idle hook) + RELAY — Network only, no compute (forward frames) + OFFLINE — Unreachable + +Usage: + from device_capability_probe import probe_device, probe_cluster + caps = probe_device() # local device + cluster_caps = probe_cluster(nodes) # all nodes via SSH/kubectl + +Matches: VCNHardwareCapabilities, MathOptimizationLoad from vcn_compute_substrate +""" + +from __future__ import annotations + +import os +import json +import struct +import subprocess +from enum import IntEnum +from pathlib import Path +from dataclasses import dataclass, field, asdict +from typing import List, Optional, Tuple + + +class ComputeTier(IntEnum): + """Compute capability tiers, ordered by decreasing capability.""" + GPU_CUDA = 6 # NVIDIA discrete + CUDA (NVENC, NVDEC) + GPU_VAAPI = 5 # AMD/Intel discrete + VA-API hardware encode + GPU_APU = 4 # AMD integrated, shared memory, bandwidth-optimized + CPU_FFMPEG = 3 # Software encode only (libx264/libx265) + FRAMEBUFFER = 2 # /dev/fb0 DMA backplane only + ESP32 = 1 # MCU, Q0_16 scalar in idle hook + RELAY = 0 # Network only, no compute + OFFLINE = -1 # Unreachable + + +@dataclass +class GPUDevice: + """A single GPU/render node detected on the system.""" + render_node: str # e.g. "/dev/dri/renderD128" + card_id: int # e.g. 0 + vendor_id: str # e.g. "0x10de" + device_id: str # e.g. "0x2783" + vendor_name: str # e.g. "nvidia", "amd", "intel" + device_name: str # e.g. "NVIDIA GeForce RTX 4070 SUPER" + is_discrete: bool # True for dGPU, False for iGPU/APU + vram_mb: int = 0 # VRAM in MB (0 for shared memory) + vaapi_profiles: List[str] = field(default_factory=list) + + +@dataclass +class FramebufferDevice: + """A framebuffer device detected on the system.""" + device_path: str # e.g. "/dev/fb0" + width: int = 0 + height: int = 0 + bpp: int = 32 # bits per pixel + capacity_bytes: int = 0 # width * height * (bpp // 8) + + +@dataclass +class DeviceCapabilities: + """Complete capability profile for a single device/node.""" + hostname: str + tier: ComputeTier = ComputeTier.RELAY + gpus: List[GPUDevice] = field(default_factory=list) + framebuffer: Optional[FramebufferDevice] = None + has_ffmpeg: bool = False + ffmpeg_encoders: List[str] = field(default_factory=list) + preferred_encoder: str = "libx264" + preferred_pixel_format: str = "yuv420p" + max_resolution: Tuple[int, int] = (1920, 1080) + ray_resources: dict = field(default_factory=dict) + os_arch: str = "x86_64" + total_memory_mb: int = 0 + # VCN alignment + packing_density: str = "macroblock_1x" + color_range: str = "tv" + encoder_opts: List[str] = field(default_factory=list) + + +# ── Vendor ID mapping ──────────────────────────────────────────────────────── + +VENDOR_MAP = { + "0x10de": "nvidia", + "0x1002": "amd", + "0x8086": "intel", +} + +# Discrete GPU device ID ranges (approximate) +NVIDIA_DISCRETE_IDS = {"0x2", "0x1"} # 0x2xxx = Ada, 0x1xxx = Ampere/Turing +AMD_DISCRETE_MARKERS = {"radeon", "rx ", "vega", "navi", "rdna"} +AMD_APU_MARKERS = {"graphics", "integrated", "apu", "drm", "ryzen", "radeon graphics"} + + +def _run(cmd: List[str], timeout: int = 5) -> Optional[str]: + """Run a command, return stdout or None on failure.""" + try: + r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + return r.stdout if r.returncode == 0 else None + except (FileNotFoundError, subprocess.TimeoutExpired, PermissionError): + return None + + +def _detect_gpus() -> List[GPUDevice]: + """Scan all DRM render nodes and classify each GPU.""" + gpus = [] + drm_dir = Path("/sys/class/drm") + + if not drm_dir.exists(): + return gpus + + for card_dir in sorted(drm_dir.glob("card*")): + card_name = card_dir.name # e.g. "card0", "card1" + try: + card_id = int(card_name.replace("card", "")) + except ValueError: + continue + + vendor_path = card_dir / "device" / "vendor" + device_path = card_dir / "device" / "device" + render_node = f"/dev/dri/renderD{128 + card_id}" + + if not vendor_path.exists() or not Path(render_node).exists(): + continue + + vendor_id = vendor_path.read_text().strip() + device_id = device_path.read_text().strip() if device_path.exists() else "0x0000" + vendor_name = VENDOR_MAP.get(vendor_id, "unknown") + + # Get device name from various sources + device_name = _get_gpu_name(vendor_name, card_dir, card_id) + + # Classify discrete vs integrated + is_discrete = _is_discrete_gpu(vendor_name, device_name, device_id) + + # Get VRAM + vram_mb = _get_vram_mb(vendor_name, card_dir) + + # Probe VA-API profiles + vaapi_profiles = _probe_vaapi_profiles(render_node) + + gpu = GPUDevice( + render_node=render_node, + card_id=card_id, + vendor_id=vendor_id, + device_id=device_id, + vendor_name=vendor_name, + device_name=device_name, + is_discrete=is_discrete, + vram_mb=vram_mb, + vaapi_profiles=vaapi_profiles, + ) + gpus.append(gpu) + + return gpus + + +def _get_gpu_name(vendor: str, card_dir: Path, card_id: int) -> str: + """Get GPU name from nvidia-smi, DRI, or DRM.""" + if vendor == "nvidia": + out = _run(["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"]) + if out: + return out.strip().split("\n")[0] + + # Try DRI name + name_path = card_dir / "device" / "product_name" + if name_path.exists(): + return name_path.read_text().strip() + + # Fallback to vendor-based naming + return f"{vendor.upper()} GPU (card{card_id})" + + +def _is_discrete_gpu(vendor: str, name: str, device_id: str) -> bool: + """Classify GPU as discrete or integrated.""" + name_lower = name.lower() + + if vendor == "nvidia": + # NVIDIA GPUs with CUDA are always discrete (for our purposes) + out = _run(["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"]) + return out is not None + + if vendor == "amd": + # Check APU markers first + for marker in AMD_APU_MARKERS: + if marker in name_lower: + return False + # Check discrete markers + for marker in AMD_DISCRETE_MARKERS: + if marker in name_lower: + return True + # If device has VRAM > 1GB, likely discrete + vram = _get_vram_mb(vendor, Path(f"/sys/class/drm/card{device_id}")) + if vram > 1024: + return True + # Default: if we can't tell, assume iGPU (conservative) + return False + + if vendor == "intel": + # Intel is almost always integrated (except Arc) + return "arc" in name_lower + + return False + + +def _get_vram_mb(vendor: str, card_dir: Path) -> int: + """Get VRAM size in MB.""" + if vendor == "nvidia": + out = _run(["nvidia-smi", "--query-gpu=memory.total", "--format=csv,noheader,nounits"]) + if out: + try: + return int(out.strip().split("\n")[0]) + except ValueError: + pass + + # Try DRM VRAM info + vram_path = card_dir / "device" / "mem_info_vram_total" + if vram_path.exists(): + try: + return int(vram_path.read_text().strip()) // (1024 * 1024) + except ValueError: + pass + + return 0 + + +def _probe_vaapi_profiles(render_node: str) -> List[str]: + """Probe VA-API profiles for a render node.""" + out = _run(["vainfo", "--display", "drm", "--device", render_node], timeout=10) + if not out: + return [] + profiles = [] + for line in out.split("\n"): + if "VAProfile" in line and ":" in line: + profiles.append(line.strip().split(":")[0].strip()) + return profiles + + +def _detect_framebuffer() -> Optional[FramebufferDevice]: + """Detect framebuffer device and its capabilities.""" + fb_path = Path("/dev/fb0") + if not fb_path.exists(): + return None + + # Try to get framebuffer info from /sys + fb_sys = Path("/sys/class/graphics/fb0") + width, height, bpp = 1920, 1080, 32 # defaults + + virt_size = fb_sys / "virtual_size" + if virt_size.exists(): + try: + parts = virt_size.read_text().strip().split(",") + width, height = int(parts[0]), int(parts[1]) + except (ValueError, IndexError): + pass + + bits_per_pixel = fb_sys / "bits_per_pixel" + if bits_per_pixel.exists(): + try: + bpp = int(bits_per_pixel.read_text().strip()) + except ValueError: + pass + + return FramebufferDevice( + device_path=str(fb_path), + width=width, + height=height, + bpp=bpp, + capacity_bytes=width * height * (bpp // 8), + ) + + +def _detect_ffmpeg() -> Tuple[bool, List[str], str]: + """Detect FFmpeg and available encoders.""" + out = _run(["ffmpeg", "-encoders"], timeout=10) + if not out: + return False, [], "libx264" + + encoders = [] + for enc in ["h264_nvenc", "hevc_nvenc", "h264_vaapi", "hevc_vaapi", + "h264_amf", "hevc_amf", "libx264", "libx265"]: + if enc in out: + encoders.append(enc) + + preferred = encoders[0] if encoders else "libx264" + return True, encoders, preferred + + +def _get_arch() -> str: + """Detect system architecture.""" + import platform + machine = platform.machine() + if machine == "aarch64": + return "arm64" + return machine + + +# ── Main probe function ────────────────────────────────────────────────────── + +def probe_device(hostname: str = "") -> DeviceCapabilities: + """Probe the local device and return complete capability profile. + + Assigns the highest compute tier the device can handle: + GPU_CUDA > GPU_VAAPI > GPU_APU > CPU_FFMPEG > FRAMEBUFFER > ESP32 > RELAY + """ + if not hostname: + import socket + hostname = socket.gethostname() + + caps = DeviceCapabilities(hostname=hostname, os_arch=_get_arch()) + + # 1. Detect GPUs + caps.gpus = _detect_gpus() + + # 2. Detect FFmpeg + caps.has_ffmpeg, caps.ffmpeg_encoders, caps.preferred_encoder = _detect_ffmpeg() + + # 3. Detect framebuffer + caps.framebuffer = _detect_framebuffer() + + # 4. Get system memory + try: + meminfo = Path("/proc/meminfo").read_text() + for line in meminfo.split("\n"): + if line.startswith("MemTotal:"): + kb = int(line.split()[1]) + caps.total_memory_mb = kb // 1024 + break + except (FileNotFoundError, ValueError): + pass + + # 5. Assign tier and configure + caps.tier, caps.ray_resources, caps.preferred_pixel_format, caps.packing_density = \ + _assign_tier(caps) + + # 6. Set encoder options based on tier + caps.encoder_opts = _get_encoder_opts(caps) + + return caps + + +def _assign_tier(caps: DeviceCapabilities): + """Assign compute tier based on detected capabilities.""" + + # Check for NVIDIA discrete with CUDA + for gpu in caps.gpus: + if gpu.vendor_name == "nvidia" and gpu.is_discrete: + return ( + ComputeTier.GPU_CUDA, + {"GPU": 1, "nvidia.com/gpu": 1}, + "yuv444p", + "yuv444_3x", + ) + + # Check for AMD/Intel discrete with VA-API + for gpu in caps.gpus: + if gpu.is_discrete and gpu.vaapi_profiles: + return ( + ComputeTier.GPU_VAAPI, + {"GPU": 1}, + "yuv444p", + "yuv444_3x", + ) + + # Check for AMD APU/iGPU (shared memory, bandwidth-optimized) + for gpu in caps.gpus: + if not gpu.is_discrete and gpu.vendor_name in ("amd", "intel"): + return ( + ComputeTier.GPU_APU, + {"GPU": 1}, + "yuvj420p", # Full-range YUV420, 50% less bandwidth + "independent_y_6x", + ) + + # FFmpeg available (software encode) + if caps.has_ffmpeg: + return ( + ComputeTier.CPU_FFMPEG, + {"CPU": 1}, + "yuv420p", + "macroblock_1x", + ) + + # Framebuffer only + if caps.framebuffer: + return ( + ComputeTier.FRAMEBUFFER, + {"framebuffer": 1}, + "argb8888", # Direct pixel mapping + "framebuffer_raw", + ) + + # Default: relay (network only) + return (ComputeTier.RELAY, {}, "yuv420p", "macroblock_1x") + + +def _get_encoder_opts(caps: DeviceCapabilities) -> List[str]: + """Get FFmpeg encoder options based on tier.""" + if caps.tier == ComputeTier.GPU_CUDA: + return ["-preset", "lossless", "-tune", "zerolatency", "-color_range", "pc"] + if caps.tier == ComputeTier.GPU_VAAPI: + return ["-qp", "0", "-color_range", "pc"] + if caps.tier == ComputeTier.GPU_APU: + return ["-qp", "0", "-color_range", "pc"] + if caps.tier == ComputeTier.CPU_FFMPEG: + return ["-preset", "ultrafast", "-tune", "zerolatency"] + return [] + + +# ── Cluster probe ──────────────────────────────────────────────────────────── + +def probe_cluster(nodes: List[str]) -> List[DeviceCapabilities]: + """Probe all nodes in the cluster via SSH.""" + results = [] + for node in nodes: + out = _run([ + "ssh", "-o", "ConnectTimeout=5", "-o", "StrictHostKeyChecking=no", + f"root@{node}", + "python3 -c \"import sys; sys.path.insert(0, '/opt/vcn'); " + "from device_capability_probe import probe_device; " + "import json; print(json.dumps(probe_device().__dict__, default=str))\"" + ], timeout=15) + if out: + try: + data = json.loads(out.strip()) + # Reconstruct dataclass + data["tier"] = ComputeTier(data["tier"]) + data["gpus"] = [GPUDevice(**g) for g in data.get("gpus", [])] + if data.get("framebuffer"): + data["framebuffer"] = FramebufferDevice(**data["framebuffer"]) + results.append(DeviceCapabilities(**data)) + except (json.JSONDecodeError, TypeError): + results.append(DeviceCapabilities( + hostname=node, tier=ComputeTier.OFFLINE)) + else: + results.append(DeviceCapabilities( + hostname=node, tier=ComputeTier.OFFLINE)) + return results + + +# ── Ray scheduling helper ──────────────────────────────────────────────────── + +def get_ray_placement_strategy(caps: DeviceCapabilities) -> dict: + """Get Ray scheduling parameters for this device's tier. + + Returns dict suitable for @ray.remote() options. + """ + if caps.tier == ComputeTier.GPU_CUDA: + return {"num_gpus": 1, "resources": {"NVENC": 1}} + if caps.tier == ComputeTier.GPU_VAAPI: + return {"num_gpus": 1, "resources": {"VAAPI": 1}} + if caps.tier == ComputeTier.GPU_APU: + return {"num_gpus": 1, "resources": {"APU": 1}} + if caps.tier == ComputeTier.CPU_FFMPEG: + return {"num_cpus": 1} + if caps.tier == ComputeTier.FRAMEBUFFER: + return {"resources": {"framebuffer": 1}} + return {} + + +def get_framebuffer_resolution(caps: DeviceCapabilities) -> Tuple[int, int]: + """Get the optimal framebuffer resolution for this device. + + For framebuffer-only devices, returns the actual fb size. + For GPU devices, returns the VCN resolution that matches. + """ + if caps.framebuffer: + return (caps.framebuffer.width, caps.framebuffer.height) + return caps.max_resolution + + +# ── CLI ────────────────────────────────────────────────────────────────────── + +def main(): + import argparse + parser = argparse.ArgumentParser(description="Device Capability Probe") + parser.add_argument("--json", action="store_true", help="Output as JSON") + parser.add_argument("--cluster", nargs="+", help="Probe remote nodes via SSH") + args = parser.parse_args() + + if args.cluster: + results = probe_cluster(args.cluster) + else: + results = [probe_device()] + + if args.json: + output = [] + for caps in results: + d = asdict(caps) + d["tier"] = caps.tier.name + output.append(d) + print(json.dumps(output, indent=2, default=str)) + else: + for caps in results: + print(f"\n{'='*60}") + print(f" {caps.hostname} ({caps.os_arch})") + print(f" Tier: {caps.tier.name} (value={caps.tier})") + print(f"{'='*60}") + + if caps.gpus: + for gpu in caps.gpus: + dtype = "dGPU" if gpu.is_discrete else "iGPU/APU" + print(f" GPU: {gpu.device_name} [{dtype}]") + print(f" Render: {gpu.render_node} ({gpu.vendor_id}:{gpu.device_id})") + if gpu.vram_mb: + print(f" VRAM: {gpu.vram_mb} MB") + if gpu.vaapi_profiles: + print(f" VA-API: {len(gpu.vaapi_profiles)} profiles") + else: + print(" GPU: none") + + if caps.framebuffer: + fb = caps.framebuffer + print(f" Framebuffer: {fb.device_path} ({fb.width}x{fb.height} {fb.bpp}bpp)") + print(f" Capacity: {fb.capacity_bytes / (1024*1024):.1f} MB") + + print(f" FFmpeg: {'yes' if caps.has_ffmpeg else 'no'}") + if caps.ffmpeg_encoders: + print(f" Encoders: {', '.join(caps.ffmpeg_encoders)}") + print(f" Preferred: {caps.preferred_encoder} ({caps.preferred_pixel_format})") + print(f" Ray resources: {caps.ray_resources}") + print(f" Memory: {caps.total_memory_mb} MB") + + +if __name__ == "__main__": + main()