# Project export: Semidirectional YOLO-Based Aquatic Unit (SYBAU) 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 vision system that detects and tracks targets to direct a 100 PSI smart water stream — enabling the future of interactive productivity tools. - Devpost: https://devpost.com/software/semi-directional-yolo-based-aquatic-unit-sybau - GitHub: https://github.com/SuperBigMac/treehacks-2026 - Video: https://www.youtube.com/embed/7NZX4gsD-Yo?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 2 GitHub contributor(s) — ehtun (21 commits), Ryota (10 commits) ## Devpost submission (written by the team) ### Inspiration Imagine jokingly trolling your friend with a face-tracking water gun — that playful idea sparked our exploration into embodied AI, systems that not only perceive the world but physically interact with it in meaningful ways. The project began as a fun concept brainstormed over dinner: a face-tracking system capable of directing a high-pressure water stream. As development progressed, the idea evolved beyond novelty into a broader investigation of human-centered applications, where we believe in a strong use-case for productivity/education, allowing everyone to "lock-in" for their meetings with a fresh blast of water. How We Built It SYBAU combines real-time computer vision, robotic arm manipulation, and high-pressure fluid control into a modular platform. Vision and AI We used YOLOv8-medium for real-time object detection and tracking, running locally on an NVIDIA RTX 4070 GPU to enable low-latency edge inference. A 360° fisheye camera provides wide environmental awareness, allowing the system to detect targets anywhere within its field of view. The detection pipeline includes: Image acquisition from the fisheye camera Object detection using YOLOv8 Target selection and tracking Conversion of bounding box coordinates into servo motion commands A real-time feedback control loop that continuously adjusts the robotic arm to keep the target (e.g., a face) centered in the frame Servo positioning maps image-space coordinates into angular control: $$ \theta_x = f_x(u), \quad \theta_y = f_y(v) $$ where (u,v) are the detected object coordinates in image space. Mechanical and Control Systems The physical actuation system consists of: Dynamixel servos (daisy-chained) for smooth pan/tilt targeting OpenRB-150 microcontroller running low-level Arduino firmware for deterministic control A 12V diaphragm water pump (RV/marine-grade) A 12V relay to safely switch pump power Interchangeable nozzle tips for different water stream profiles Custom 3D-printed mount The architecture separates perception and control: High-level edge AI and computer vision are implemented in Python on the GPU Low-level firmware on the microcontroller handles real-time servo positioning and relay switching This separation allows fast visual inference while maintaining stable, responsive hardware control through a closed-loop feedback system. Challenges We Faced Selecting and sourcing the right materials and components to balance durability, safety, and performance Calibrating the fisheye camera and compensating for lens distortion in the vision pipeline Interfacing high-level AI software with low-level embedded firmware Designing reliable wiring and power systems for stable hardware operation Ensuring mechanical alignment so all moving parts operate smoothly as an integrated system Working with wide-angle optics and translating distorted image coordinates into accurate servo control Designing and CAD-ing custom 3D-printed parts robust enough to withstand motion, water pressure, and repeated use What We Learned How to design and integrate a complete moving robotic system combining AI, electronics, and mechanical engineering Practical lessons in hardware reliability, real-time control, and system integration How to build and tune a fully functional AI-controlled water delivery system Future Directions While currently focused on face tracking, SYBAU can generalize to many detection tasks in the future. Potential applications include: Fire detection and suppression assistance Agricultural or home pest mitigation Interactive entertainment and amusement use cases, and haptic feedback for movies etc. ## README (from the GitHub repository) ## Treehacks 2026 Project (Ryota Sato, Ethan Htun) Creating a product that helps keep you and your meetings on track ## Detected evidence (automated analysis) Indexed codebase: 14 recognized source files, 34 KB. - Python (language) — detected in the code - PyTorch (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (21 of 21) ``` __init__.py .gitignore brain.py hardware/api.py hardware/final_firmware/final_firmware.ino hardware/mock_api.py hardware/README.md hardware/robotic_arm/robotic_arm.ino hardware/scan_dynamixel/scan_dynamixel.ino hardware/water_test/water_test.ino main.py README.md requirements.txt vision/__init__.py vision/camera.py vision/face_detection_short_range.tflite vision/fisheye_utils.py vision/inference.py vision/pipeline.py vision/README.md vision/runner.py ``` ### Dependencies - requirements.txt: opencv-python, pyserial, ultralytics@>=8.0.0 ### Recent commits (newest first) - Merge branch 'main' of https://github.com/SuperBigMac/treehacks-2026 - calibrate aiming - update title + desc - changes - Merge pull request #3 from SuperBigMac/cv++ - add vision - Merge branch 'main' of https://github.com/SuperBigMac/treehacks-2026 - add center crop and rotation fix - Merge branch 'main' of https://github.com/SuperBigMac/treehacks-2026 - changed range to -90 to 90 - minor cleanup - add auto timeout - updates - Merge pull request #2 from SuperBigMac/facial-feedback - add brain and fisheye calibration v0 - changed firmware - added firmware code - add heartbeat and loop logic, as well as mock hardware api - add disconnect timeout - revert to the og ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` opencv-python pyserial ultralytics>=8.0.0 ``` ### main.py ```python import time from hardware.api import HardwareAPI, HardwareDisconnectedError from hardware.mock_api import MockHardwareAPI from vision.runner import FacePipelineRunner from brain import Brain PORT = "COM10" # PORT = "/dev/tty.usbmodem101" BAUD = 9600 message = "2" target_x = 0.52 target_y = 0.45 if __name__ == "__main__": start_time = time.time() pipeline = FacePipelineRunner( show_window=True, camera_index=1, rotate_180=True, center_crop_fraction=1, ) pipeline.start() print("Pipeline running. Check state with pipeline.get_state()") # display dummy point pipeline.update_state(pause_detection=False, target_x=target_x, target_y=target_y) try: hardware_api = HardwareAPI(port=PORT, baudrate=BAUD) # hardware_api = MockHardwareAPI(port=PORT, baudrate=BAUD) except Exception as e: print(f"Error initializing hardware API: {e}") exit(1) # initialize the brain (gain_deg: scale on fisheye angle, 1.0=use as-is; bias: mechanical offset) # Lower Kp and gain_deg to reduce oscillation so crosshair can settle on face brain = Brain(hardware_api, target_x, target_y, gain_deg=0.7, Kp=0.4) try: timestamp_ms = 0 print_interval = 0.3 last_print = time.time() last_heartbeat = time.time() heartbeat_interval = 0.3 # Arduino timeout 500ms; 300ms keeps alive without flooding while pipeline.is_alive(): state = pipeline.get_state() timestamp_ms = state["timestamp_ms"] if state.get("is_running"): faces = state.get("faces", []) fw, fh = state.get("frame_width"), state.get("frame_height") if len(faces) > 0 and fw is not None and fh is not None: try: brain.run( faces, frame_width=fw, frame_height=fh, center_crop_fraction=state.get("center_crop_fraction") ) except HardwareDisconnectedError as e: print(f"Hardware: {e}") if time.time() - last_print >= print_interval: n = state["num_faces"] print(f"State: {n} face(s), ts={timestamp_ms} ms") for i, (x1, y1, x2, y2) in enumerate(faces): print(f" face[{i}] box=({x1}, {y1}, {x2}, {y2})") if len(faces) > 0: print(f"arm_x: {brain.arm_x}, arm_y: {brain.arm_y}") last_print = time.time() if time.time() - last_heartbeat >= heartbeat_interval: try: hardware_api.send_heartbeat() last_heartbeat = time.time() except HardwareDisconnectedError as e: print(f"Hardware: {e}") time.sleep(0.1) except KeyboardInterrupt: brain.hardware_api.send_message("x 0", rate_limit=False) brain.hardware_api.send_message("y 0", rate_limit=False) brain.hardware_api.send_message("0", rate_limit=False) print("Keyboard Interrupt. Shutting down.") finally: # Always shut down the pipeline before exiting so the Manager stays alive # until the child exits. Otherwise the subprocess hits EOFError when it # touches the shared state after main (and the Manager) are gone. pipeline.request_quit() pipeline.join(timeout=2.0) hardware_api.close() ``` ### __init__.py ```python # make this a parent package ``` ### brain.py ```python """ Orchestrates hardware and vision: move camera so face centroid aligns with target. - Offset = derivative (we integrate to get position); plus P term so we don't miss the target. - Command = integral + Kp * offset + bias (P+I). """ import math from typing import List, Tuple # Consider "on target" when crosshair is within this many pixels of face centroid (so small oscillations still fire) ON_TARGET_RADIUS_PX = 40 from hardware.api import HardwareAPI from vision.fisheye_utils import circle_radius_px_from_frame, offset_to_angle class Brain: # Servo bounds (degrees); firmware uses x in [-45, 45], y in [-20, 30] ARM_X_BOUNDS = (-45, 45) ARM_Y_BOUNDS = (-20, 30) def __init__( self, hardware_api: HardwareAPI, target_x: float, target_y: float, gain_deg: float = 1.0, max_step_deg: float = 2.0, Kp: float = 0.4, dead_zone_deg: float = 1.5, arm_x_bias_deg: float = 0.0, arm_y_bias_deg: float = 0.0, ): self.hardware_api = hardware_api self.target_x = float(target_x) self.target_y = float(target_y) self.gain_deg = gain_deg self.max_step_deg = max_step_deg self.Kp = Kp # proportional term so we don't overshoot / miss target self.dead_zone_deg = dead_zone_deg # no correction when |pan_deg| and |tilt_deg| both below this self.arm_x_bias_deg = arm_x_bias_deg self.arm_y_bias_deg = arm_y_bias_deg self.is_shooting = False self.arm_x = 0.0 self.arm_y = 0.0 def _largest_face( self, detections: List[Tuple[int, int, int, int]], ) -> Tuple[int, int, int, int]: """Face bbox (x1, y1, x2, y2) with largest area.""" def area(box: Tuple[int, int, int, int]) -> int: x1, y1, x2, y2 = box return (x2 - x1) * (y2 - y1) return max(detections, key=area) def run( self, detections: List[Tuple[int, int, int, int]], frame_width: int | None = None, frame_height: int | None = None, center_crop_fraction: float | None = None, ) -> None: if len(detections) == 0: self.arm_x = 0.0 self.arm_y = 0.0 return w = frame_width if frame_width is not None else 640 h = frame_height if frame_height is not None else 480 box = self._largest_face(detections) x1, y1, x2, y2 = box centroid_x = ((x1 + x2) / 2) / w centroid_y = ((y1 + y2) / 2) / h # Work in frame pixel space (from video feed / state) centroid_x_px = centroid_x * w centroid_y_px = centroid_y * h target_x_px = self.target_x * w if self.target_x <= 1.0 else self.target_x target_y_px = self.target_y * h if self.target_y <= 1.0 else self.target_y # Fisheye circle radius from current frame size (viewport = cut-off circle) circle_r = circle_radius_px_from_frame(w, h) pan_deg = offset_to_angle( centroid_x_px - target_x_px, radius_px=circle_r, crop_fraction=center_crop_fraction, ) tilt_deg = offset_to_angle( centroid_y_px - target_y_px, radius_px=circle_r, crop_fraction=center_crop_fraction, ) # Dead zone: when error is small, don't integrate or apply P so the arm can settle in_dead_zone = abs(pan_deg) < self.dead_zone_deg and abs(tilt_deg) < self.dead_zone_deg # Offset = derivative; integrate (accumulate) to get position. Positive y = tilt up so flip tilt. if not in_dead_zone: step_x = max(-self.max_step_deg, min(self.max_step_deg, self.gain_deg * pan_deg)) step_y = max(-self.max_step_deg, min(self.max_step_deg, -self.gain_deg * tilt_deg)) self.arm_x += step_x self.arm_y += step_y lo_x, hi_x = self.ARM_X_BOUNDS lo_y, hi_y = self.ARM_Y_BOUNDS self.arm_x = max(lo_x, min(hi_x, self.arm_x)) self.arm_y = max(lo_y, min(hi_y, self.arm_y)) # P + I: integral (arm_x/y) + proportional term so we pull toward target and don't miss p_x = 0.0 if in_dead_zone else self.Kp * pan_deg p_y = 0.0 if in_dead_zone else self.Kp * tilt_deg x_cmd = self.arm_x + p_x + self.arm_x_bias_deg y_cmd = self.arm_y - p_y + self.arm_y_bias_deg x_deg = int(round(max(lo_x, min(hi_x, x_cmd)))) y_deg = int(round(max(lo_y, min(hi_y, y_cmd)))) self.hardware_api.send_message(f"x {x_deg}", rate_limit=False) self.hardware_api.send_message(f"y {y_deg}", rate_limit=False) # Shoot when target point is inside face box or within ON_TARGET_RADIUS_PX of face centroid tx_frame = self.target_x * w if self.target_x <= 1.0 else self.target_x ty_frame = self.target_y * h if self.target_y <= 1.0 else self.target_y inside_box = x1 <= tx_frame <= x2 and y1 <= ty_frame <= y2 dist_to_centroid = math.sqrt((tx_frame - centroid_x_px) ** 2 + (ty_frame - centroid_y_px) ** 2) on_target = inside_box or dist_to_centroid <= ON_TARGET_RADIUS_PX self.is_shooting = on_target self.hardware_api.send_message("1" if on_target else "0", rate_limit=False) ``` ### vision/__init__.py ```python # turns this into a package ``` ### hardware/mock_api.py ```python """ Mock hardware API: same interface as api.HardwareAPI but no serial. Prints when messages are sent and when rate limiting kicks in. """ import time from hardware.api import DEFAULT_RATE_LIMIT_MS, HardwareDisconnectedError class MockHardwareAPI: """ Drop-in mock for HardwareAPI. No serial port; prints send/rate-limit activity. """ def __init__(self, port: str = "mock", baudrate: int = 9600, rate_limit_ms: int = DEFAULT_RATE_LIMIT_MS): self.port = port self.baudrate = baudrate self._rate_limit_ms = rate_limit_ms self._connected = True self.last_message = time.time() def send_message(self, message: str, verbose: bool = False, rate_limit: bool = True) -> None: """Send a message (mock: just print). Prints when rate limited.""" if rate_limit and (time.time() - self.last_message) < (self._rate_limit_ms / 1000): print("[mock hardware] Rate limited.") return print(f"[mock hardware] Sent: {message!r}") if rate_limit: self.last_message = time.time() def send_heartbeat(self) -> None: """Send heartbeat (mock: just print).""" print("[mock hardware] Heartbeat sent.") def close(self) -> None: """No-op for mock.""" self._connected = False print("[mock hardware] Close (no-op).") ``` ### vision/runner.py ```python """ For running vision pipeline in multiprocessing """ from multiprocessing import Manager, Process from vision.pipeline import run_pipeline_in_process def _default_state() -> dict: return { "is_running": False, "timestamp_ms": 0, "num_faces": 0, "faces": [], "quit_requested": False, "center_crop_fraction": None, } class FacePipelineRunner: """Runs FaceCameraPipeline in a subprocess. State is shared via Manager().dict().""" def __init__( self, show_window: bool = True, camera_index: int = 0, frame_fps: int = 30, window_name: str = "Video Feed", rotate_180: bool = False, center_crop_fraction: float | None = None, ): self._manager = Manager() self._state = self._manager.dict() self._state.update(_default_state()) self._process = Process( target=run_pipeline_in_process, args=(self._state,), kwargs={ "show_window": show_window, "camera_index": camera_index, "frame_fps": frame_fps, "window_name": window_name, "rotate_180": rotate_180, "center_crop_fraction": center_crop_fraction, }, ) def start(self) -> None: self._process.start() def get_state(self) -> dict: """Snapshot of pipeline state (is_running, timestamp_ms, num_faces, faces).""" return dict(self._state) def update_state(self, **kwargs: object) -> None: """Write state from main process; the vision subprocess can read these keys.""" for k, v in kwargs.items(): self._state[k] = v def request_quit(self) -> None: self._state["quit_requested"] = True def is_alive(self) -> bool: return self._process.is_alive() def join(self, timeout: float | None = None) -> None: self._process.join(timeout=timeout) ``` ### vision/inference.py ```python """ Face detection inference using YOLOv8m (Ultralytics). """ import os import urllib.request import cv2 from ultralytics import YOLO # Face-trained YOLOv8m weights (Bingsu/adetailer on Hugging Face) MODEL_FILENAME = "face_yolov8m.pt" MODEL_URL = ( "https://huggingface.co/Bingsu/adetailer/resolve/main/face_yolov8m.pt" ) def _default_model_path() -> str: return os.path.join(os.path.dirname(__file__), MODEL_FILENAME) def _ensure_model(path: str) -> None: if os.path.exists(path): return print("Downloading YOLOv8m face detection model...") urllib.request.urlretrieve(MODEL_URL, path) class FaceDetectorInference: """ Runs face detection on video frames using YOLOv8m (face-trained). Use as a context manager so the detector is closed properly. """ def __init__( self, model_path: str | None = None, min_detection_confidence: float = 0.5, **kwargs, ): self._model_path = model_path or _default_model_path() _ensure_model(self._model_path) self._model = YOLO(self._model_path) self._conf = min_detection_confidence self._kwargs = kwargs def detect( self, frame_bgr: cv2.typing.MatLike, timestamp_ms: int, ) -> list[tuple[int, int, int, int]]: """ Run face detection on a single frame. Args: frame_bgr: BGR image (e.g. from cv2.VideoCapture). timestamp_ms: Frame timestamp in milliseconds (unused; for API compatibility). Returns: List of bounding boxes as (x1, y1, x2, y2) in image coordinates. """ results = self._model.predict( frame_bgr, conf=self._conf, verbose=False, **self._kwargs, ) boxes = [] for r in results: if r.boxes is None: continue for box in r.boxes: xyxy = box.xyxy[0] x1 = int(xyxy[0].item()) y1 = int(xyxy[1].item()) x2 = int(xyxy[2].item()) y2 = int(xyxy[3].item()) boxes.append((x1, y1, x2, y2)) return boxes def close(self) -> None: # Ultralytics YOLO doesn't require explicit close; no-op for API compatibility pass def __enter__(self) -> "FaceDetectorInference": return self def __exit__(self, exc_type, exc_val, exc_tb) -> None: self.close() ``` ### vision/camera.py ```python """ Camera capture API. No inference or display logic. """ import time from typing import Tuple import cv2 import numpy as np DEFAULT_READ_RETRIES = 5 DEFAULT_RETRY_DELAY_SEC = 0.05 DEFAULT_RECONNECT_DELAY_SEC = 0.5 class Camera: """ Camera capture with retries and reconnection. Use as context manager. Optional width/height sets resolution (e.g. 3840×1920 for Picam360). """ def __init__( self, camera_index: int = 0, width: int | None = None, height: int | None = None, read_retries: int = DEFAULT_READ_RETRIES, retry_delay_sec: float = DEFAULT_RETRY_DELAY_SEC, reconnect_delay_sec: float = DEFAULT_RECONNECT_DELAY_SEC, ): self._index = camera_index self._width = width self._height = height self._read_retries = read_retries self._retry_delay_sec = retry_delay_sec self._reconnect_delay_sec = reconnect_delay_sec self._cap = self._open() if not self._cap.isOpened(): raise RuntimeError( f"Could not open camera (index={camera_index}). " "In use, no permission, or wrong index?" ) def _open(self) -> cv2.VideoCapture: cap = cv2.VideoCapture(self._index) if cap.isOpened(): cap.set(cv2.CAP_PROP_BUFFERSIZE, 1) if self._width is not None: cap.set(cv2.CAP_PROP_FRAME_WIDTH, self._width) if self._height is not None: cap.set(cv2.CAP_PROP_FRAME_HEIGHT, self._height) return cap def read(self) -> Tuple[bool, np.ndarray | None]: """ Read one frame. Retries on failure, then tries one reconnect. Returns: (True, frame) on success, (False, None) if still no frame after retries and reconnect. """ ret, frame = self._cap.read() if ret: return True, frame for _ in range(self._read_retries): time.sleep(self._retry_delay_sec) ret, frame = self._cap.read() if ret: return True, frame # Reconnect once self._cap.release() time.sleep(self._reconnect_delay_sec) self._cap = self._open() if not self._cap.isOpened(): return False, None ret, frame = self._cap.read() return ret, frame if ret else None def close(self) -> None: if self._cap is not None: self._cap.release() self._cap = None def __enter__(self) -> "Camera": return self def __exit__(self, exc_type, exc_val, exc_tb) -> None: self.close() ``` ### vision/fisheye_utils.py ```python """ Simple angle geometry for converting pixel positions to angles (pan/tilt). Viewport = rectangle cutting off a circular fisheye image (equidistant: angle ∝ r). Circle radius is derived from the actual frame size (video feed / state). """ import math from typing import Tuple # Angle (deg) from center to edge of the circle (~180° total → 90° to edge) ANGLE_AT_EDGE_DEG = 90.0 MAX_THETA_DEG = ANGLE_AT_EDGE_DEG # Circle covers center 2/3 of frame width → radius = frame_width * this fraction CIRCLE_RADIUS_FRACTION_OF_WIDTH = 1.0 / 3.0 # Legacy / pixel_to_angle default when no frame size given WIDTH = 3840 HEIGHT = 1920 CENTER_X = WIDTH / 2 CENTER_Y = HEIGHT / 2 RADIUS_PX = WIDTH / 3 RADIUS_X = RADIUS_PX RADIUS_Y = RADIUS_PX def circle_radius_px_from_frame(frame_width: int, frame_height: int) -> float: """Fisheye circle radius in frame pixels (viewport = cut-off circle).""" return frame_width * CIRCLE_RADIUS_FRACTION_OF_WIDTH def pixel_to_angle( x: float, y: float, center_x: float = CENTER_X, center_y: float = CENTER_Y, radius_px: float = RADIUS_PX, max_theta_deg: float = MAX_THETA_DEG, crop_fraction: float | None = None, ) -> Tuple[float, float]: """ Convert pixel (x, y) to angles. Returns (theta_deg, phi_deg): theta 0° at center, phi 0°–360°. When center_crop_fraction is used, the visible FOV is smaller; pass crop_fraction (e.g. 0.6) so angle scale matches the cropped image. """ f = crop_fraction if crop_fraction is not None else 1.0 effective_max_theta = max_theta_deg * f dx = x - center_x dy = y - center_y r_pixel = math.sqrt(dx * dx + dy * dy) theta = (r_pixel / radius_px) * effective_max_theta if radius_px > 0 else 0.0 phi_rad = math.atan2(dy, dx) phi = math.degrees(phi_rad) if phi < 0: phi += 360.0 return theta, phi def face_box_to_angle( box: Tuple[float, float, float, float], center_x: float = CENTER_X, center_y: float = CENTER_Y, radius_px: float = RADIUS_PX, max_theta_deg: float = MAX_THETA_DEG, crop_fraction: float | None = None, ) -> Tuple[float, float]: """Convert face bbox (x1, y1, x2, y2) to (theta_deg, phi_deg) using box center.""" x1, y1, x2, y2 = box cx = (x1 + x2) / 2 cy = (y1 + y2) / 2 return pixel_to_angle( cx, cy, center_x, center_y, radius_px, max_theta_deg, crop_fraction ) def offset_to_angle( delta_pixels: float, radius_px: float = RADIUS_PX, max_theta_deg: float = MAX_THETA_DEG, crop_fraction: float | None = None, ) -> float: """Convert radial pixel offset to angle in degrees. Pass crop_fraction when using a center-cropped image so scale matches FOV.""" f = crop_fraction if crop_fraction is not None else 1.0 effective_max_theta = max_theta_deg * f return (delta_pixels / radius_px) * effective_max_theta if radius_px > 0 else 0.0 ``` [2 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]