Companion files
Copy these files into the paths shown in Live YOLO depth on the NPU with AI Hub and native QNN.
These files are prototype tutorial code. Review error handling, memory ownership, tensor metadata discovery, and deployment policies before using them in production.
qnn_dlc_runner.cpp
#include <QNN/QnnInterface.h>
#include <QNN/System/QnnSystemContext.h>
#include <QNN/System/QnnSystemInterface.h>
#include <QNN/QnnTypes.h>
#include <dlfcn.h>
#include <algorithm>
#include <chrono>
#include <cstdarg>
#include <cstring>
#include <fstream>
#include <iostream>
#include <numeric>
#include <stdexcept>
#include <string>
#include <vector>
using QnnInterfaceGetProvidersFn = Qnn_ErrorHandle_t (*)(const QnnInterface_t***, uint32_t*);
using QnnSystemInterfaceGetProvidersFn = Qnn_ErrorHandle_t (*)(const QnnSystemInterface_t***, uint32_t*);
#define CHECK_QNN(expr) do { \
Qnn_ErrorHandle_t _e = (expr); \
if (_e != QNN_SUCCESS) { \
std::cerr << "QNN call failed: " #expr " -> 0x" << std::hex << (uint64_t)_e << std::dec << std::endl; \
throw std::runtime_error("QNN error"); \
} \
} while(0)
static std::vector<uint8_t> readFile(const std::string& path) {
std::ifstream f(path, std::ios::binary);
if (!f) throw std::runtime_error("failed to open " + path);
f.seekg(0, std::ios::end);
size_t n = (size_t)f.tellg();
f.seekg(0);
std::vector<uint8_t> b(n);
f.read((char*)b.data(), n);
if (!f) throw std::runtime_error("failed to read " + path);
return b;
}
static void writeFile(const std::string& path, const void* data, size_t n) {
std::ofstream f(path, std::ios::binary);
if (!f) throw std::runtime_error("failed to open output " + path);
f.write((const char*)data, n);
}
static size_t dtypeSize(Qnn_DataType_t dt) {
switch (dt) {
case QNN_DATATYPE_FLOAT_32: return 4;
case QNN_DATATYPE_FLOAT_16: return 2;
case QNN_DATATYPE_UINT_8: return 1;
case QNN_DATATYPE_INT_8: return 1;
case QNN_DATATYPE_UINT_16: return 2;
case QNN_DATATYPE_INT_16: return 2;
case QNN_DATATYPE_UINT_32: return 4;
case QNN_DATATYPE_INT_32: return 4;
default: throw std::runtime_error("unsupported dtype " + std::to_string((int)dt));
}
}
static const Qnn_TensorV2_t& tv2(const Qnn_Tensor_t& t) {
if (t.version != QNN_TENSOR_VERSION_2) throw std::runtime_error("expected tensor v2 metadata");
return t.v2;
}
static uint64_t elemCount(const Qnn_TensorV2_t& t) {
uint64_t n = 1;
for (uint32_t i = 0; i < t.rank; ++i) n *= t.dimensions[i];
return n;
}
static Qnn_Tensor_t makeAppTensor(const Qnn_Tensor_t& meta, void* data, size_t bytes, Qnn_TensorType_t type) {
Qnn_Tensor_t t = meta;
if (t.version == QNN_TENSOR_VERSION_1) {
t.v1.type = type;
t.v1.memType = QNN_TENSORMEMTYPE_RAW;
t.v1.clientBuf.data = data;
t.v1.clientBuf.dataSize = bytes;
} else if (t.version == QNN_TENSOR_VERSION_2) {
t.v2.type = type;
t.v2.memType = QNN_TENSORMEMTYPE_RAW;
t.v2.clientBuf.data = data;
t.v2.clientBuf.dataSize = bytes;
} else {
throw std::runtime_error("unsupported tensor version");
}
return t;
}
static void logCb(const char* fmt, QnnLog_Level_t level, uint64_t timestamp, va_list args) {
(void)fmt; (void)level; (void)timestamp; (void)args;
}
int main(int argc, char** argv) {
std::string dlc = "../aihub_compiled/yolo26n-depth-aihub-qcs8275-qnn-dlc.dlc";
std::string input = "../aihub_dlc_test/input/images.raw";
std::string output = "qnn_app_output.raw";
int loops = 200;
int warmup = 10;
bool serverMode = false;
for (int i = 1; i < argc; ++i) {
std::string a = argv[i];
if (a == "--dlc" && i + 1 < argc) dlc = argv[++i];
else if (a == "--input" && i + 1 < argc) input = argv[++i];
else if (a == "--output" && i + 1 < argc) output = argv[++i];
else if (a == "--loops" && i + 1 < argc) loops = std::stoi(argv[++i]);
else if (a == "--warmup" && i + 1 < argc) warmup = std::stoi(argv[++i]);
else if (a == "--server") serverMode = true;
else { std::cerr << "usage: " << argv[0] << " [--dlc file.dlc] [--input images.raw] [--output out.raw] [--loops N] [--warmup N] [--server]\n"; return 2; }
}
void* htpLib = dlopen("libQnnHtp.so", RTLD_NOW | RTLD_LOCAL);
if (!htpLib) { std::cerr << "dlopen libQnnHtp.so failed: " << dlerror() << "\n"; return 1; }
void* systemLib = dlopen("libQnnSystem.so", RTLD_NOW | RTLD_LOCAL);
if (!systemLib) { std::cerr << "dlopen libQnnSystem.so failed: " << dlerror() << "\n"; return 1; }
auto getProviders = (QnnInterfaceGetProvidersFn)dlsym(htpLib, "QnnInterface_getProviders");
if (!getProviders) { std::cerr << "dlsym QnnInterface_getProviders failed\n"; return 1; }
const QnnInterface_t** providers = nullptr;
uint32_t numProviders = 0;
CHECK_QNN(getProviders(&providers, &numProviders));
if (numProviders == 0) throw std::runtime_error("no QNN providers");
const auto& qnn = providers[0]->QNN_INTERFACE_VER_NAME;
auto getSystemProviders = (QnnSystemInterfaceGetProvidersFn)dlsym(systemLib, "QnnSystemInterface_getProviders");
if (!getSystemProviders) { std::cerr << "dlsym QnnSystemInterface_getProviders failed\n"; return 1; }
const QnnSystemInterface_t** systemProviders = nullptr;
uint32_t numSystemProviders = 0;
CHECK_QNN(getSystemProviders(&systemProviders, &numSystemProviders));
if (numSystemProviders == 0) throw std::runtime_error("no QNN System providers");
const auto& sys = systemProviders[0]->QNN_SYSTEM_INTERFACE_VER_NAME;
Qnn_LogHandle_t logger = nullptr;
if (qnn.logCreate) qnn.logCreate(logCb, QNN_LOG_LEVEL_WARN, &logger);
Qnn_BackendHandle_t backend = nullptr;
CHECK_QNN(qnn.backendCreate(logger, nullptr, &backend));
Qnn_DeviceHandle_t device = nullptr;
if (qnn.deviceCreate) CHECK_QNN(qnn.deviceCreate(logger, nullptr, &device));
auto bin = readFile(dlc);
// Extract embedded HTP context binary from the DLC. qnn-net-run does this internally for --dlc_path.
// QnnContext_createFromBinary() expects the context record, not necessarily the whole DLC file.
QnnSystemDlc_Handle_t dlcHandle = nullptr;
QnnSystemDlc_RecordHandle_t* recordHandles = nullptr;
uint32_t numRecordHandles = 0;
QnnSystemDlc_RecordHandle_t contextRecord = nullptr;
const uint8_t* contextData = bin.data();
uint64_t contextSize = bin.size();
if (sys.systemDlcCreateFromFile && sys.systemDlcGetRecordsByType && sys.systemDlcReadRecordDataMemoryMapped) {
Qnn_ErrorHandle_t de = sys.systemDlcCreateFromFile(logger, dlc.c_str(), &dlcHandle);
if (de == QNN_SUCCESS && dlcHandle) {
de = sys.systemDlcGetRecordsByType(dlcHandle,
QNN_SYSTEM_DLC_RECORD_PREFIX_HTP_CACHE_RECORD,
1,
&recordHandles,
&numRecordHandles);
if (de == QNN_SUCCESS && numRecordHandles > 0 && recordHandles && recordHandles[0]) {
contextRecord = recordHandles[0];
de = sys.systemDlcReadRecordDataMemoryMapped(contextRecord, &contextData, &contextSize);
if (de == QNN_SUCCESS) {
std::cerr << "extracted HTP context record from DLC: " << contextSize << " bytes\n";
} else {
std::cerr << "failed reading HTP context record; falling back to whole DLC\n";
contextData = bin.data();
contextSize = bin.size();
}
} else {
std::cerr << "no HTP context record found in DLC; falling back to whole DLC\n";
}
} else {
std::cerr << "systemDlcCreateFromFile failed; falling back to whole DLC\n";
}
}
QnnSystemContext_Handle_t sysCtx = nullptr;
const QnnSystemContext_BinaryInfo_t* info = nullptr;
const char* graphName = nullptr;
uint32_t numInputs = 0, numOutputs = 0;
Qnn_Tensor_t* metaInputs = nullptr;
Qnn_Tensor_t* metaOutputs = nullptr;
Qnn_Tensor_t fallbackInput = QNN_TENSOR_INIT;
Qnn_Tensor_t fallbackOutput = QNN_TENSOR_INIT;
uint32_t inDims[4] = {1, 320, 320, 3};
uint32_t outDims[4] = {1, 1, 320, 320};
bool haveMetadata = false;
if (sys.systemContextCreate && sys.systemContextGetMetaData) {
if (sys.systemContextCreate(&sysCtx) == QNN_SUCCESS &&
sys.systemContextGetMetaData(sysCtx, contextData, (Qnn_ContextBinarySize_t)contextSize, &info) == QNN_SUCCESS) {
uint32_t numGraphs = 0;
QnnSystemContext_GraphInfo_t* graphs = nullptr;
if (info->version == QNN_SYSTEM_CONTEXT_BINARY_INFO_VERSION_1) {
numGraphs = info->contextBinaryInfoV1.numGraphs;
graphs = info->contextBinaryInfoV1.graphs;
} else if (info->version == QNN_SYSTEM_CONTEXT_BINARY_INFO_VERSION_2) {
numGraphs = info->contextBinaryInfoV2.numGraphs;
graphs = info->contextBinaryInfoV2.graphs;
} else if (info->version == QNN_SYSTEM_CONTEXT_BINARY_INFO_VERSION_3) {
numGraphs = info->contextBinaryInfoV3.numGraphs;
graphs = info->contextBinaryInfoV3.graphs;
}
if (numGraphs >= 1) {
auto& gmeta = graphs[0];
if (gmeta.version == QNN_SYSTEM_CONTEXT_GRAPH_INFO_VERSION_1) {
graphName = gmeta.graphInfoV1.graphName; numInputs = gmeta.graphInfoV1.numGraphInputs; metaInputs = gmeta.graphInfoV1.graphInputs; numOutputs = gmeta.graphInfoV1.numGraphOutputs; metaOutputs = gmeta.graphInfoV1.graphOutputs;
} else if (gmeta.version == QNN_SYSTEM_CONTEXT_GRAPH_INFO_VERSION_2) {
graphName = gmeta.graphInfoV2.graphName; numInputs = gmeta.graphInfoV2.numGraphInputs; metaInputs = gmeta.graphInfoV2.graphInputs; numOutputs = gmeta.graphInfoV2.numGraphOutputs; metaOutputs = gmeta.graphInfoV2.graphOutputs;
} else if (gmeta.version == QNN_SYSTEM_CONTEXT_GRAPH_INFO_VERSION_3) {
graphName = gmeta.graphInfoV3.graphName; numInputs = gmeta.graphInfoV3.numGraphInputs; metaInputs = gmeta.graphInfoV3.graphInputs; numOutputs = gmeta.graphInfoV3.numGraphOutputs; metaOutputs = gmeta.graphInfoV3.graphOutputs;
}
haveMetadata = (graphName && numInputs == 1 && numOutputs == 1 && metaInputs && metaOutputs);
}
}
}
if (!haveMetadata) {
std::cerr << "QNN system metadata unavailable for DLC; using known graph/tensor metadata fallback\n";
graphName = "graph_ymndtmzg";
numInputs = 1;
numOutputs = 1;
fallbackInput.version = QNN_TENSOR_VERSION_2;
fallbackInput.v2.id = 1;
fallbackInput.v2.name = "images";
fallbackInput.v2.type = QNN_TENSOR_TYPE_APP_WRITE;
fallbackInput.v2.dataFormat = QNN_TENSOR_DATA_FORMAT_FLAT_BUFFER;
fallbackInput.v2.dataType = QNN_DATATYPE_FLOAT_32;
fallbackInput.v2.rank = 4;
fallbackInput.v2.dimensions = inDims;
fallbackInput.v2.memType = QNN_TENSORMEMTYPE_RAW;
fallbackOutput.version = QNN_TENSOR_VERSION_2;
fallbackOutput.v2.id = 842;
fallbackOutput.v2.name = "output_0";
fallbackOutput.v2.type = QNN_TENSOR_TYPE_APP_READ;
fallbackOutput.v2.dataFormat = QNN_TENSOR_DATA_FORMAT_FLAT_BUFFER;
fallbackOutput.v2.dataType = QNN_DATATYPE_FLOAT_32;
fallbackOutput.v2.rank = 4;
fallbackOutput.v2.dimensions = outDims;
fallbackOutput.v2.memType = QNN_TENSORMEMTYPE_RAW;
metaInputs = &fallbackInput;
metaOutputs = &fallbackOutput;
}
std::cerr << "graph: " << graphName << " inputs=" << numInputs << " outputs=" << numOutputs << "\n";
if (numInputs != 1 || numOutputs != 1) throw std::runtime_error("runner currently expects 1 input/1 output");
const auto& inMeta = tv2(metaInputs[0]);
const auto& outMeta = tv2(metaOutputs[0]);
size_t inBytes = elemCount(inMeta) * dtypeSize(inMeta.dataType);
size_t outBytes = elemCount(outMeta) * dtypeSize(outMeta.dataType);
std::cerr << "input: " << inMeta.name << " bytes=" << inBytes << " dtype=0x" << std::hex << inMeta.dataType << std::dec << "\n";
std::cerr << "output: " << outMeta.name << " bytes=" << outBytes << " dtype=0x" << std::hex << outMeta.dataType << std::dec << "\n";
auto inputBuf = readFile(input);
if (inputBuf.size() != inBytes) throw std::runtime_error("input size mismatch: got " + std::to_string(inputBuf.size()) + " expected " + std::to_string(inBytes));
std::vector<uint8_t> outputBuf(outBytes);
Qnn_ContextHandle_t context = nullptr;
CHECK_QNN(qnn.contextCreateFromBinary(backend, device, nullptr, contextData, (Qnn_ContextBinarySize_t)contextSize, &context, nullptr));
Qnn_GraphHandle_t graph = nullptr;
CHECK_QNN(qnn.graphRetrieve(context, graphName, &graph));
Qnn_Tensor_t inTensor = makeAppTensor(metaInputs[0], inputBuf.data(), inputBuf.size(), QNN_TENSOR_TYPE_APP_WRITE);
Qnn_Tensor_t outTensor = makeAppTensor(metaOutputs[0], outputBuf.data(), outputBuf.size(), QNN_TENSOR_TYPE_APP_READ);
for (int i = 0; i < warmup; ++i) {
CHECK_QNN(qnn.graphExecute(graph, &inTensor, 1, &outTensor, 1, nullptr, nullptr));
}
if (serverMode) {
// Persistent mode. Read commands from stdin:
// RUN <input.raw> <output.raw>
// Reply on stdout:
// OK <ms> <output_bytes>
// or:
// ERR <message>
std::cout << "READY " << inBytes << " " << outBytes << std::endl;
std::string cmd, inPath, outPath;
while (std::cin >> cmd) {
if (cmd == "QUIT" || cmd == "EXIT") break;
if (cmd != "RUN") {
std::cout << "ERR expected RUN" << std::endl;
continue;
}
if (!(std::cin >> inPath >> outPath)) {
std::cout << "ERR missing paths" << std::endl;
break;
}
try {
auto frameBuf = readFile(inPath);
if (frameBuf.size() != inBytes) {
std::cout << "ERR input_size got=" << frameBuf.size() << " expected=" << inBytes << std::endl;
continue;
}
std::memcpy(inputBuf.data(), frameBuf.data(), inBytes);
auto t0 = std::chrono::steady_clock::now();
CHECK_QNN(qnn.graphExecute(graph, &inTensor, 1, &outTensor, 1, nullptr, nullptr));
auto t1 = std::chrono::steady_clock::now();
double oneMs = std::chrono::duration<double, std::milli>(t1 - t0).count();
writeFile(outPath, outputBuf.data(), outputBuf.size());
std::cout << "OK " << oneMs << " " << outputBuf.size() << std::endl;
} catch (const std::exception& e) {
std::cout << "ERR " << e.what() << std::endl;
}
}
} else {
std::vector<double> ms;
ms.reserve(loops);
for (int i = 0; i < loops; ++i) {
auto t0 = std::chrono::steady_clock::now();
CHECK_QNN(qnn.graphExecute(graph, &inTensor, 1, &outTensor, 1, nullptr, nullptr));
auto t1 = std::chrono::steady_clock::now();
ms.push_back(std::chrono::duration<double, std::milli>(t1 - t0).count());
}
writeFile(output, outputBuf.data(), outputBuf.size());
double sum = std::accumulate(ms.begin(), ms.end(), 0.0);
double avg = sum / ms.size();
std::sort(ms.begin(), ms.end());
double p50 = ms[ms.size()/2];
double p90 = ms[(size_t)(ms.size()*0.90)];
double p99 = ms[std::min(ms.size()-1, (size_t)(ms.size()*0.99))];
std::cout << "loops=" << loops << " warmup=" << warmup << "\n";
std::cout << "avg_ms=" << avg << " p50_ms=" << p50 << " p90_ms=" << p90 << " p99_ms=" << p99 << " fps=" << (1000.0/avg) << "\n";
std::cout << "wrote=" << output << " bytes=" << outBytes << "\n";
}
if (qnn.contextFree) qnn.contextFree(context, nullptr);
if (qnn.deviceFree && device) qnn.deviceFree(device);
if (qnn.backendFree) qnn.backendFree(backend);
if (qnn.logFree && logger) qnn.logFree(logger);
if (sysCtx && sys.systemContextFree) sys.systemContextFree(sysCtx);
if (contextRecord && sys.systemDlcFreeRecord) sys.systemDlcFreeRecord(contextRecord);
if (dlcHandle && sys.systemDlcFree) sys.systemDlcFree(dlcHandle);
dlclose(systemLib);
dlclose(htpLib);
return 0;
}
Makefile
CXX ?= g++
CXXFLAGS := -std=c++17 -O3 -Wall -Wextra -I/usr/include -I/usr/include/QNN
LDFLAGS := -ldl
all: qnn_dlc_runner
qnn_dlc_runner: qnn_dlc_runner.cpp
$(CXX) $(CXXFLAGS) $< -o $@ $(LDFLAGS)
clean:
rm -f qnn_dlc_runner
live_aihub_qnn_native.py
#!/usr/bin/env python3
"""Live USB camera depth viewer using the persistent AI Hub QNN/NPU native app."""
import argparse
import subprocess
import sys
import tempfile
import time
from pathlib import Path
import cv2
import numpy as np
from ultralytics.utils.plotting import colorize_depth
ROOT = Path(__file__).resolve().parent
APP = ROOT / "qnn_app" / "qnn_dlc_runner"
CONTEXT = ROOT / "qnn_context" / "yolo26n-depth-aihub-qcs8275-context.bin.bin"
SIZE = 320
def cam_id(value):
try:
return int(value)
except ValueError:
return value
def letterbox_bgr(frame, size=SIZE):
h, w = frame.shape[:2]
scale = min(size / h, size / w)
nh, nw = int(round(h * scale)), int(round(w * scale))
resized = cv2.resize(frame, (nw, nh), interpolation=cv2.INTER_LINEAR)
canvas = np.full((size, size, 3), 114, dtype=np.uint8)
top = (size - nh) // 2
left = (size - nw) // 2
canvas[top : top + nh, left : left + nw] = resized
rgb = cv2.cvtColor(canvas, cv2.COLOR_BGR2RGB)
x = (rgb.astype(np.float32) / 255.0)[None, ...]
meta = (top, left, nh, nw, h, w)
return x, meta
def unletterbox_depth(depth_320, meta):
top, left, nh, nw, h, w = meta
crop = depth_320[top : top + nh, left : left + nw]
return cv2.resize(crop, (w, h), interpolation=cv2.INTER_LINEAR)
def read_ready(proc, timeout=20.0):
deadline = time.time() + timeout
lines = []
while time.time() < deadline:
line = proc.stdout.readline()
if not line:
if proc.poll() is not None:
raise RuntimeError("native QNN app exited early:\n" + "".join(lines))
time.sleep(0.01)
continue
lines.append(line)
if line.startswith("READY "):
print("Native QNN app ready:", line.strip())
return line.strip()
# Keep setup messages visible; useful if something fails.
print("[qnn]", line.rstrip())
raise TimeoutError("Timed out waiting for native QNN app READY. Last output:\n" + "".join(lines[-30:]))
def main():
ap = argparse.ArgumentParser(description="Live AI Hub QNN/NPU native depth camera")
ap.add_argument("--camera", default=0, type=cam_id, help="Camera index/path, e.g. 0 or /dev/video0")
ap.add_argument("--width", default=640, type=int)
ap.add_argument("--height", default=480, type=int)
ap.add_argument("--display-width", default=1280, type=int, help="Display width; 0 keeps native stacked width")
ap.add_argument("--display-height", default=0, type=int, help="Display height; 0 preserves aspect")
ap.add_argument("--font-scale", default=1.0, type=float)
ap.add_argument("--metric", action="store_true", help="Use metric color mode")
ap.add_argument("--vmax", default=20.0, type=float, help="Max meters for --metric mode")
ap.add_argument("--save-dir", default="aihub_qnn_native_captures")
ap.add_argument("--app", default=str(APP))
ap.add_argument("--context", default=str(CONTEXT))
args = ap.parse_args()
app = Path(args.app)
context = Path(args.context)
if not app.exists():
raise SystemExit(f"Missing native app: {app}\nBuild it with: cd {ROOT/qnn_app} && make")
if not context.exists():
raise SystemExit(f"Missing QNN context: {context}")
cv2.setNumThreads(1)
tmp = tempfile.TemporaryDirectory(prefix="qnn_live_")
tmpdir = Path(tmp.name)
in_raw = tmpdir / "input.raw"
out_raw = tmpdir / "output.raw"
# The native runner validates the initial --input path even in --server mode,
# so create a correctly-sized dummy NHWC float32 input before launching it.
np.zeros((1, SIZE, SIZE, 3), dtype=np.float32).tofile(in_raw)
cmd = [str(app), "--dlc", str(context), "--input", str(in_raw), "--output", str(out_raw), "--warmup", "10", "--loops", "1", "--server"]
print("Starting native QNN app:", " ".join(cmd))
# stderr is merged so we can show setup logs and still wait for READY.
proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
try:
read_ready(proc)
cap = cv2.VideoCapture(args.camera, cv2.CAP_V4L2)
if not cap.isOpened():
raise SystemExit(f"Could not open camera {args.camera!r}. Try --camera /dev/video0 or check: ls /dev/video*")
cap.set(cv2.CAP_PROP_FRAME_WIDTH, args.width)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, args.height)
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
save_dir = Path(args.save_dir)
save_dir.mkdir(exist_ok=True)
print("Live AI Hub QNN native depth started")
print("Controls: q/Esc quit, s save current RGB/depth/overlay")
last_depth = None
last_color = None
last_overlay = None
last_frame = None
frame_idx = 0
fps_smooth = None
while True:
ok, frame = cap.read()
if not ok:
print("Camera read failed")
break
last_frame = frame.copy()
x, meta = letterbox_bgr(frame, SIZE)
x.tofile(in_raw)
t0 = time.perf_counter()
proc.stdin.write(f"RUN {in_raw} {out_raw}\n")
proc.stdin.flush()
line = proc.stdout.readline().strip()
while line and not (line.startswith("OK ") or line.startswith("ERR ")):
# Forward any unexpected QNN log lines.
print("[qnn]", line)
line = proc.stdout.readline().strip()
if line.startswith("ERR "):
raise RuntimeError(line)
if not line.startswith("OK "):
raise RuntimeError("native QNN app produced no response")
qnn_ms = float(line.split()[1])
total_ms = (time.perf_counter() - t0) * 1000.0
inst_fps = 1000.0 / total_ms if total_ms > 0 else 0.0
fps_smooth = inst_fps if fps_smooth is None else (0.9 * fps_smooth + 0.1 * inst_fps)
d320 = np.fromfile(out_raw, dtype=np.float32).reshape(1, 1, SIZE, SIZE)[0, 0]
depth = unletterbox_depth(d320, meta).astype(np.float32)
last_depth = depth
if args.metric:
color = colorize_depth(depth, vmin=0.0, vmax=args.vmax, cmap="inferno", mode="metric")
else:
color = colorize_depth(depth, cmap="spectral", mode="disparity")
overlay = cv2.addWeighted(frame, 0.45, color, 0.55, 0)
last_color = color
last_overlay = overlay
view = np.hstack([frame, color, overlay])
valid = depth[np.isfinite(depth) & (depth > 0)]
if valid.size:
txt = f"QNN {qnn_ms:.1f} ms | total {total_ms:.1f} ms | {fps_smooth:.1f} FPS | depth {valid.min():.1f}/{valid.mean():.1f}/{valid.max():.1f} m | q quit | s save"
else:
txt = f"QNN {qnn_ms:.1f} ms | total {total_ms:.1f} ms | {fps_smooth:.1f} FPS | q quit | s save"
fs = args.font_scale
thick_bg = max(3, int(round(4 * fs)))
thick_fg = max(1, int(round(1.5 * fs)))
y_top = int(38 * fs)
y_bottom = view.shape[0] - int(18 * fs)
for text, pos in [(txt, (15, y_top)), ("RGB | DEPTH | OVERLAY", (15, y_bottom))]:
cv2.putText(view, text, pos, cv2.FONT_HERSHEY_SIMPLEX, fs, (0, 0, 0), thick_bg, cv2.LINE_AA)
cv2.putText(view, text, pos, cv2.FONT_HERSHEY_SIMPLEX, fs, (255, 255, 255), thick_fg, cv2.LINE_AA)
show = view
if args.display_width or args.display_height:
h, w = view.shape[:2]
if args.display_width and args.display_height:
dw, dh = args.display_width, args.display_height
elif args.display_width:
dw = args.display_width
dh = max(1, int(h * dw / w))
else:
dh = args.display_height
dw = max(1, int(w * dh / h))
show = cv2.resize(view, (dw, dh), interpolation=cv2.INTER_AREA if dw < w else cv2.INTER_LINEAR)
cv2.namedWindow("AI Hub QNN Native Live Depth", cv2.WINDOW_NORMAL)
cv2.imshow("AI Hub QNN Native Live Depth", show)
key = cv2.waitKey(1) & 0xFF
if key in (ord("q"), 27):
break
if key == ord("s") and last_depth is not None:
stamp = time.strftime("%Y%m%d-%H%M%S")
cv2.imwrite(str(save_dir / f"{stamp}_rgb.jpg"), last_frame)
cv2.imwrite(str(save_dir / f"{stamp}_depth_color.png"), last_color)
cv2.imwrite(str(save_dir / f"{stamp}_overlay.png"), last_overlay)
np.save(save_dir / f"{stamp}_depth.npy", last_depth)
print("Saved", save_dir.resolve(), stamp)
frame_idx += 1
cap.release()
cv2.destroyAllWindows()
finally:
try:
if proc.stdin:
proc.stdin.write("QUIT\n")
proc.stdin.flush()
except Exception:
pass
try:
proc.terminate()
proc.wait(timeout=2)
except Exception:
proc.kill()
tmp.cleanup()
if __name__ == "__main__":
main()
live_aihub_qnn_native.sh
#!/usr/bin/env bash
set -euo pipefail
cd "$(dirname "$0")"
source ~/yolo-depth-venv/bin/activate
python ./live_aihub_qnn_native.py "$@"