# Project export: DUI-Vision
This document was generated by HackStack to give an AI agent context about a hackathon project. Sections are labeled with their provenance; content marked as truncated was cut to keep this document small.
## Project metadata
- Hackathon: TreeHacks 2026
- Tagline: AI-powered field sobriety testing on NVIDIA Jetson, ensuring 100% legal privacy and accuracy.
- Devpost: https://devpost.com/software/dui-sobriety-test-ai-agent
- GitHub: https://github.com/ethans0ng/treehack2026
- Video: https://www.youtube.com/embed/Lht-nCzVVn0?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — ethans0ng (14 commits)
## Devpost submission (written by the team)
### Inspiration
I've attended police academies at both my last university, Duke, and at Stanford. Last Wednesday in class, I watched a live Standard Field Sobriety Test (SFST) training (and conducted tests of my own on drunk volunteers). Learning about the process was frustrating -- a single DUI takes at least 4-6 hours of paperwork, keeping officers off the streets from saving lives. The DUI-Vision automates the grunt work of sobriety testing and provides complementary evidence in court.
### What it does
Hardware-integrated Edge AI assistant using body-cam feed to perform real-time Horizontal Gaze Nystagmus (HGN) detection on Jetson Nano. -- HGN is the sobriety test with the highest accuracy rate; it looks at movement of the eye when following an object or at certain angles; inebriated persons will have uncontrollable movement of the eye -- Model detects eyes and pupils automatically and tracks involuntary eye patterns (spasms) -- Automatically uses data to start populating the NHTSA-standard HGN reports (e.g. scores/eye) -- All processing is done entirely on the device
### How we built it
Hardware: NVIDIA Jetson Orin Nano Super + USB Camera Software: Custom pupil-tracking algorithm using OpenCV and NanoOwl Jitter Engine: Filters head movement from actual Nystagmus jerking and differentiates between frequential spasms of eye vs. normal smooth movement Edge Stack: Local FastAPI server and CSS dashboard on Jetson to serve real-time results
### Challenges we ran into
This hackathon was an absolute hardware war for me. As an EE it was humbling to see how much work there is going to just make a connection on "software." I spent hours figuring out SSH/Serial Console and combatting network collisions. I've never used a terminal or SSH'ed into a machine before this, or even used codex to help edit code so it was a steep learning curve. Accomplishments that I'm proud of First hackathon & Competing Solo Not quitting dealing with the difficult hardware and unstable network Not throwing the jetson on the ground after another connection failure Having a working product and local web server running real time What I learned Developer kits are a lot harder than they look. Latency is so important for edge AI products. AI Code tools can help a lot but if you don't have a clear architectural vision or idea it can't do much. Once the project grows big the errors become more and more. Kind of like a cancer.
### What's next
: Field Sobriety Test (FST) Edge Assistant Multi-Modal Agent -- integrating Whisper on a second Jetson to transcribe the suspect interview and auto-fill initial contact and arrest forms. Rest of the SFST -- adding CV models for the "Walk and Turn" and "One Leg Stand" tests, among many. Miniaturization -- making the logic work on a Orin-based body cam rig or recording glasses
## README (from the GitHub repository)
# DUI-Vision: Field Sobriety Test (FST) Edge Assistant
DUI-Vision is a Jetson-based edge system that runs computer vision on live video to assist
Horizontal Gaze Nystagmus (HGN) field testing workflows in near real time.
## What it does
- Captures face/eye video from a local camera
- Tracks pupil motion and estimates HGN indicators
- Computes:
- Lack of smooth pursuit (L/R)
- Nystagmus prior to 45° (L/R)
- Distinct nystagmus at max deviation (L/R)
- Vertical nystagmus estimate
- Detects excessive head movement during tests
- Publishes completed session results to a local API
- Serves a web dashboard for latest result + session history
## Tech stack
- **Edge/device**: NVIDIA Jetson Nano
- **Languages**: Python 3, JavaScript, HTML, CSS
- **CV/ML**: OpenCV, NumPy, PyTorch, NanoOWL (OWL-ViT), Pillow
- **Backend/API**: Python `http.server` + `ThreadingHTTPServer` (REST)
- **Storage**: SQLite + CSV export
## Run it locally
```bash
# Terminal 1: start API + dashboard
python3 api_server.py
# Terminal 2: run HGN edge capture
python3 hgn_tracker3.py
Dashboard: http://localhost:8000
API receives finalized sessions at POST /api/session/finish
## Why this matters for judges
This demonstrates a practical edge-first safety workflow:
- on-device inference (no mandatory cloud dependency),
- lightweight persistence,
- structured result delivery,
- browser dashboard for review.
## Privacy note
This is a field operations prototype; no public live law-enforcement endpoint is exposed in this
release. Data handling is local-first, and sensitive operational data is intentionally not
published publicly.
## Notes
test_eyes.py is a validation utility used during development to test pupil extraction behavior
before full pipeline integration.
## Detected evidence (automated analysis)
Indexed codebase: 9 recognized source files, 103 KB.
- CSS (language) — detected in the code
- HTML (language) — detected in the code
- JavaScript (language) — detected in the code
- Python (language) — detected in the code
- PyTorch (technology) — claimed on Devpost, not found in the code
## Codebase structure (from repository index)
### Files (9 of 9)
```
api_server_new.py
hgn_test_full.py
hgn_test.py
README.md
results_writer.py
static/app.js
static/styles.css
template/index.html
test_eyes.py
```
### Dependencies
No dependency index available.
### Recent commits (newest first)
- Create README.md
- Delete api_server.py
- Delete hgn_tracker2.py
- Delete hgn_tracker3.py
- Delete hgn_tracker1.py
- Delete static/a
- Add files via upload
- Delete template/a
- Add files via upload
- Create a
- Add files via upload
- Create a
- Add files via upload
- Add files via upload
## Key source files (fetched from GitHub, selected and truncated for size)
### static/app.js
```javascript
const dom = {
badge: document.getElementById('live-badge'),
latestNode: document.getElementById('latest-result'),
warningNode: document.getElementById('latest-warning'),
body: document.getElementById('sessions-body'),
limit: document.getElementById('limit'),
pollRate: document.getElementById('poll-rate'),
refreshBtn: document.getElementById('refresh-btn'),
lastUpdated: document.getElementById('last-updated'),
sessionCount: document.getElementById('session-count'),
statusCard: document.querySelector('.status-card')
};
const state = {
latestFingerprint: null,
isFetching: false,
timer: null
};
function setBadge(message, level) {
if (!dom.badge) {
return;
}
dom.badge.classList.remove('live', 'alert', 'warn');
dom.badge.classList.add(level);
dom.badge.innerHTML = `${message}`;
}
function setWarning(message, level) {
if (!dom.warningNode) {
return;
}
dom.warningNode.textContent = message;
dom.warningNode.className = `status-text status-${level}`;
}
function toBinary(value) {
if (value === null || value === undefined || value === '') {
return 'n/a';
}
const raw = String(value).trim();
if (raw === '0' || raw === '1') {
return raw;
}
const numeric = Number(raw);
if (Number.isNaN(numeric)) {
return raw;
}
return numeric ? '1' : '0';
}
function binaryPair(left, right) {
return `${toBinary(left)}/${toBinary(right)}`;
}
function formatDate(value) {
if (!value) {
return 'n/a';
}
const parsed = new Date(value);
if (Number.isNaN(parsed.valueOf())) {
return String(value);
}
return parsed.toLocaleString([], {
month: 'short',
day: '2-digit',
year: 'numeric',
hour: '2-digit',
minute: '2-digit',
second: '2-digit',
hour12: false
});
}
function createCell(value, cssClass = '') {
const cell = document.createElement('td');
if (cssClass.includes('status-chip')) {
const chip = document.createElement('span');
chip.className = cssClass;
chip.textContent = value;
cell.appendChild(chip);
return cell;
}
cell.textContent = value;
if (cssClass) {
cell.className = cssClass;
}
return cell;
}
function renderRows(sessions) {
if (!dom.body) {
return;
}
dom.body.innerHTML = '';
sessions.forEach((s) => {
const headWarningCount = Number(s.head_warning_count || 0);
const warnText = headWarningCount >= 2 ? 'void/retest' : 'ok';
const warnClass = headWarningCount >= 2 ? 'status-chip status-chip--warn' : 'status-chip status-chip--ok';
const row = document.createElement('tr');
if (headWarningCount >= 2) {
row.classList.add('row--warn');
}
row.appendChild(createCell(formatDate(s.created_at)));
row.appendChild(createCell(s.subject_name || 'Unknown'));
row.appendChild(createCell(binaryPair(s.lack_of_smooth_pursuit_left_binary, s.lack_of_smooth_pursuit_right_binary)));
row.appendChild(createCell(binaryPair(s.nystagmus_prior_to_45_left_binary, s.nystagmus_prior_to_45_right_binary)));
row.appendChild(createCell(binaryPair(s.distinct_nystagmus_max_deviation_left_binary, s.distinct_nystagmus_max_deviation_right_binary)));
const vert = Number(s.vertical_nystagmus || 0);
row.appendChild(createCell(Number.isFinite(vert) ? vert.toFixed(1) : 'n/a'));
row.appendChild(createCell(warnText, warnClass));
dom.body.appendChild(row);
});
}
function renderLatest(sessions) {
if (dom.sessionCount) {
dom.sessionCount.textContent = `${sessions.length} sessions loaded`;
}
if (dom.lastUpdated) {
dom.lastUpdated.textContent = `Last update: ${new Date().toLocaleTimeString([], { hour: '2-digit', minute: '2-digit', second: '2-digit' })}`;
}
if (!sessions.length) {
dom.latestNode.textContent = 'No test yet.';
setWarning('Awaiting first sync.', 'neutral');
setBadge('Waiting', 'warn');
dom.body.innerHTML = '';
return;
}
const latest = sessions[0];
const fp = `${latest.session_id || ''}|${latest.created_at || ''}`;
const isNew = state.latestFingerprint && state.latestFingerprint !== fp;
state.latestFingerprint = fp;
const headWarningCount = Number(latest.head_warning_count || 0);
const hasHeadWarning = headWarningCount >= 2;
dom.latestNode.textContent =
`${latest.subject_name || 'Unknown'} · ${formatDate(latest.created_at)} · ` +
`SP ${binaryPair(latest.lack_of_smooth_pursuit_left_binary, latest.lack_of_smooth_pursuit_right_binary)} · ` +
`Prior45 ${binaryPair(latest.nystagmus_prior_to_45_left_binary, latest.nystagmus_prior_to_45_right_binary)} · ` +
`MaxDev ${binaryPair(latest.distinct_nystagmus_max_deviation_left_binary, latest.distinct_nystagmus_max_deviation_right_binary)} · ` +
`Vert ${Number(latest.vertical_nystagmus || 0).toFixed(1)}`;
if (hasHeadWarning) {
setWarning('Head movement warning: HIGH. Result may be void.', 'warn');
setBadge('Head warning', 'alert');
} else {
setWarning('Head movement warning: none.', 'ok');
setBadge('Monitoring', 'live');
}
if (isNew) {
if (dom.statusCard) {
dom.statusCard.classList.remove('pulse');
void dom.statusCard.offsetWidth;
dom.statusCard.classList.add('pulse');
}
}
renderRows(sessions);
}
async function fetchSessions() {
if (state.isFetching) {
return;
}
state.isFetching = true;
if (dom.refreshBtn) {
dom.refreshBtn.disabled = true;
}
setBadge('Syncing', 'warn');
try {
const limit = Number(dom.limit.value || 25);
const res = await fetch(`/api/sessions?limit=${encodeURIComponent(limit)}`, { cache: 'no-store' });
if (!res.ok) {
throw new Error(`HTTP ${res.status}`);
}
const data = await res.json();
const sessions = Array.isArray(data.items) ? data.items : [];
renderLatest(sessions);
} catch (error) {
setBadge('Offline', 'warn');
setWarning(`Unable to contact /api/sessions (${error.message})`, 'neutral');
} finally {
state.isFetching = fa
[truncated — 1311 more characters]
```
### test_eyes.py
```python
import cv2
import numpy as np
import torch
import gc
from PIL import Image
from nanoowl.owl_predictor import OwlPredictor
gc.collect()
torch.cuda.empty_cache()
predictor = OwlPredictor(
"google/owlvit-base-patch32",
image_encoder_engine="data/owl_image_encoder_patch32.engine"
)
cap = cv2.VideoCapture(0, cv2.CAP_V4L2)
def get_pupil_location(eye_img_bgr, is_left_eye):
if eye_img_bgr.size == 0: return None
h_orig, w_orig, _ = eye_img_bgr.shape
# 1. Your Left/Right Shadow Dodge (Still critical)
if is_left_eye:
x_start, x_end = int(w_orig * 0.35), int(w_orig * 0.85)
else:
x_start, x_end = int(w_orig * 0.15), int(w_orig * 0.65)
y_start, y_end = int(h_orig * 0.30), int(h_orig * 0.70)
crop = eye_img_bgr[y_start:y_end, x_start:x_end]
if crop.size == 0: return None
gray = cv2.cvtColor(crop, cv2.COLOR_BGR2GRAY)
# 2. THE GLINT KILLER: This specifically erases the bright screen reflection
# inside your dark pupil, making it a solid dark mass again.
gray = cv2.medianBlur(gray, 7)
# 3. Find the dark blob among the sclera
# We calculate the average brightness of the eye box, and threshold
# anything darker than average to become our pure white tracking blob
mean_val = np.mean(gray)
_, thresh = cv2.threshold(gray, mean_val - 15, 255, cv2.THRESH_BINARY_INV)
# 4. Find the center of that blob
contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
if not contours: return None
contours = sorted(contours, key=cv2.contourArea, reverse=True)
crop_area = crop.shape[0] * crop.shape[1]
for cnt in contours:
area = cv2.contourArea(cnt)
# Ignore tiny camera noise AND ignore giant shadows that fill the box
if area < 10 or area > (crop_area * 0.5):
continue
M = cv2.moments(cnt)
if M["m00"] != 0:
px = int(M["m10"] / M["m00"])
py = int(M["m01"] / M["m00"])
return (px + x_start, py + y_start)
return None
print("Scanning for a frame with BOTH eyes...")
try:
while True:
ret, frame = cap.read()
if not ret: continue
small_frame = cv2.resize(frame, (640, 480))
image_np = cv2.cvtColor(small_frame, cv2.COLOR_BGR2RGB).copy()
image_pil_safe = Image.fromarray(image_np)
output_eyes = predictor.predict(
image=image_pil_safe,
text=["human eye"],
text_encodings=None,
threshold=0.15
)
if len(output_eyes.boxes) >= 2:
print(f"Found {len(output_eyes.boxes)} eyes. Drawing and saving...")
# Sort boxes by X coordinate (left to right across the screen)
# The first box [0] is the left side of the screen, the second [1] is the right
boxes = sorted(output_eyes.boxes.cpu().numpy().astype(int), key=lambda b: b[0])
for i, box in enumerate(boxes[:2]):
x1, y1, x2, y2 = box
h, w, _ = small_frame.shape
x1, y1 = max(0, x1), max(0, y1)
x2, y2 = min(w, x2), min(h, y2)
cv2.rectangle(small_frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
eye_crop = small_frame[y1:y2, x1:x2]
# Determine left/right based on index
is_left_eye = (i == 0)
pupil_rel = get_pupil_location(eye_crop, is_left_eye)
if pupil_rel:
px, py = pupil_rel
gx, gy = x1 + px, y1 + py
cv2.circle(small_frame, (gx, gy), 5, (0, 0, 255), -1)
print(f"Pupil marked at X: {gx}, Y: {gy}")
else:
print("Pupil rejected (blob too big or too small).")
cv2.imwrite("debug.jpg", small_frame)
print("SUCCESS: 'debug.jpg' saved. Stopping script.")
break
del output_eyes
torch.cuda.empty_cache()
except KeyboardInterrupt:
print("Script manually stopped.")
except Exception as e:
print(f"An error occurred: {e}")
finally:
cap.release()
```
### results_writer.py
```python
import csv
import json
import os
import sqlite3
from datetime import date, datetime
from typing import Any, Dict, List, Optional
DB_PATH = os.path.join("data", "results.db")
EXPORT_DIR = "exports"
METRIC_BINARY_THRESHOLD = 40.0
def _ensure_dir(path: str) -> None:
os.makedirs(path, exist_ok=True)
def _now_utc_iso() -> str:
return datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
def _coerce_int(value: Any, default: int = 0) -> int:
try:
return int(value)
except (TypeError, ValueError):
return default
def _coerce_float(value: Any, default: float = 0.0) -> float:
try:
return float(value)
except (TypeError, ValueError):
return default
def init_db(db_path: str = DB_PATH) -> None:
_ensure_dir(os.path.dirname(db_path))
conn = sqlite3.connect(db_path)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS hgn_sessions (
session_id TEXT PRIMARY KEY,
created_at TEXT NOT NULL,
subject_name TEXT DEFAULT '',
stop_time TEXT DEFAULT '',
arrest_time TEXT DEFAULT '',
head_warning_count INTEGER DEFAULT 0,
head_movement_too_much INTEGER DEFAULT 0,
max_head_movement REAL DEFAULT 0.0,
lack_of_smooth_pursuit_left_real REAL DEFAULT 0.0,
lack_of_smooth_pursuit_left_binary INTEGER DEFAULT 0,
lack_of_smooth_pursuit_right_real REAL DEFAULT 0.0,
lack_of_smooth_pursuit_right_binary INTEGER DEFAULT 0,
nystagmus_prior_to_45_left_real REAL DEFAULT 0.0,
nystagmus_prior_to_45_left_binary INTEGER DEFAULT 0,
nystagmus_prior_to_45_right_real REAL DEFAULT 0.0,
nystagmus_prior_to_45_right_binary INTEGER DEFAULT 0,
distinct_nystagmus_max_deviation_left_real REAL DEFAULT 0.0,
distinct_nystagmus_max_deviation_left_binary INTEGER DEFAULT 0,
distinct_nystagmus_max_deviation_right_real REAL DEFAULT 0.0,
distinct_nystagmus_max_deviation_right_binary INTEGER DEFAULT 0,
vertical_nystagmus REAL DEFAULT 0.0,
vertical_nystagmus_binary INTEGER DEFAULT 0,
payload_json TEXT DEFAULT ''
)
"""
)
conn.commit()
conn.close()
def _get_conn(db_path: str = DB_PATH) -> sqlite3.Connection:
_ensure_dir(os.path.dirname(db_path))
return sqlite3.connect(db_path)
def _as_row_dict(cursor: sqlite3.Cursor, row: sqlite3.Row) -> Dict[str, Any]:
if row is None:
return {}
return {col[0]: row[i] for i, col in enumerate(cursor.description)} if row else {}
def _session_row_from_payload(payload: Dict[str, Any]) -> Dict[str, Any]:
metrics = payload.get("metrics", {})
left = (payload.get("binary") or {}).get("left", {})
right = (payload.get("binary") or {}).get("right", {})
left_scores = (payload.get("scores") or {}).get("left", {})
right_scores = (payload.get("scores") or {}).get("right", {})
row = {
"session_id": payload.get("session_id") or _now_utc_iso(),
"created_at": payload.get("created_at") or _now_utc_iso(),
"subject_name": payload.get("subject_name", ""),
"stop_time": payload.get("stop_time", ""),
"arrest_time": payload.get("arrest_time", ""),
"head_warning_count": _coerce_int(payload.get("head_warning_count", 0)),
"head_movement_too_much": _coerce_int(payload.get("head_movement_too_much", 0)),
"max_head_movement": _coerce_float(payload.get("max_head_movement", 0.0)),
}
row["lack_of_smooth_pursuit_left_real"] = _coerce_float(
left_scores.get("lack_of_smooth_pursuit", 0.0)
)
row["lack_of_smooth_pursuit_right_real"] = _coerce_float(
right_scores.get("lack_of_smooth_pursuit", 0.0)
)
row["nystagmus_prior_to_45_left_real"] = _coerce_float(
left_scores.get("nystagmus_prior_to_45", 0.0)
)
row["nystagmus_prior_to_45_right_real"] = _coerce_float(
right_scores.get("nystagmus_prior_to_45", 0.0)
)
row["distinct_nystagmus_max_deviation_left_real"] = _coerce_float(
left_scores.get("distinct_nystagmus_max_deviation", 0.0)
)
row["distinct_nystagmus_max_deviation_right_real"] = _coerce_float(
right_scores.get("distinct_nystagmus_max_deviation", 0.0)
)
row["vertical_nystagmus"] = _coerce_float(metrics.get("vertical_nystagmus", 0.0))
row["lack_of_smooth_pursuit_left_binary"] = _coerce_int(left.get("lack_of_smooth_pursuit", 0))
row["lack_of_smooth_pursuit_right_binary"] = _coerce_int(right.get("lack_of_smooth_pursuit", 0))
row["nystagmus_prior_to_45_left_binary"] = _coerce_int(left.get("nystagmus_prior_to_45", 0))
row["nystagmus_prior_to_45_right_binary"] = _coerce_int(right.get("nystagmus_prior_to_45", 0))
row["distinct_nystagmus_max_deviation_left_binary"] = _coerce_int(
left.get("distinct_nystagmus_max_deviation", 0)
)
row["distinct_nystagmus_max_deviation_right_binary"] = _coerce_int(
right.get("distinct_nystagmus_max_deviation", 0)
)
row["vertical_nystagmus_binary"] = (
1 if _coerce_float(row["vertical_nystagmus"]) >= METRIC_BINARY_THRESHOLD else 0
)
row["payload_json"] = json.dumps(payload, ensure_ascii=False)
return row
def _append_csv(payload_row: Dict[str, Any]) -> None:
_ensure_dir(EXPORT_DIR)
export_date = str(date.today())
csv_path = os.path.join(EXPORT_DIR, f"{export_date}.csv")
fieldnames = [
"session_id",
"created_at",
"subject_name",
"stop_time",
"arrest_time",
"head_warning_count",
"head_movement_too_much",
"max_head_movement",
"lack_of_smooth_pursuit_left_real",
"lack_of_smooth_pursuit_left_binary",
"lack_of_smooth_pursuit_right_real",
"lack_of_smooth_pursuit_right_binary",
"nystagmus_prior_to_45_left_real",
"nystagmus_prior_to_45_left_binary",
[truncated — 3910 more characters]
```
### api_server_new.py
```python
import json
import os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Dict
from urllib.parse import parse_qs, urlparse, unquote
from results_writer import get_session, list_sessions, save_session
HOST = os.getenv("HGN_API_HOST", "0.0.0.0")
PORT = int(os.getenv("HGN_API_PORT", "8000"))
ROOT = Path(__file__).resolve().parent
TEMPLATES_DIR = ROOT / "templates"
STATIC_DIR = ROOT / "static"
SCRIPT_DIR = ROOT / "script"
def _dashboard_html() -> str:
template_path = TEMPLATES_DIR / "index.html"
if template_path.exists():
return template_path.read_text(encoding="utf-8")
return """
HGN Edge Dashboard
Stanford Police · Sobriety Operations
HGN Edge Dashboard
Real-time operational view of edge-collected HGN sessions.