MobiCam (GitHub Repository) and Pyrants address two core engineering challenges:
MyCam Server built with Kotlin & CameraX) with a Python AI client using OpenCV, MobileNet-SSD deep neural networks, and a continuous socket-flushing buffer (FastStreamReader).By combining low-level socket stream control with on-device deep learning object detection, MobiCam eliminates stream delay while supporting remote global monitoring over 4G/5G via Tailscale VPN.
Here is the exact data flow for MobiCam's zero-latency video processing and Pyrants' interactive code execution engine.
+-----------------------------------------------------------------------------------+
| MOBILE CAMERA NODE (ANDROID / IOS) |
| [ Android Phone Camera ] |
| | (CameraX Frame Ingest @ 720p/480p 30 FPS) |
| v |
| [ MyCam Server App ] (MjpegStreamServer.kt on HTTP:8080) |
+----------|------------------------------------------------------------------------+
| (MJPEG Stream over Wi-Fi / USB / Tailscale 4G/5G Mesh)
v
+-----------------------------------------------------------------------------------+
| DESKTOP AI CLIENT (PYTHON / OPENCV) |
| [ FastStreamReader Thread ] |
| |---> Flushes Socket Buffer Continuously (0-Lag Frame Pull) |
| v |
| [ AI Engine: MobileNet-SSD Caffe DNN & HOG Detector ] |
| |---> Classifies Objects: Person, Pets, Vehicles |
| |---> Calculates Bounding Box Overlays |
| v |
| [ Event Triggers & Actions ] |
| +---> Auto 5-Second Video Clips (.mp4) -> recordings/ |
| +---> Auto High-Res Person Snapshots (.jpg) -> snapshots/ |
| +---> Audible Security Audio Alarm Trigger |
| v |
| [ Main GUI Viewport & CLI Window ] (main.py / cli_cam.py) |
+-----------------------------------------------------------------------------------++-----------------------------------------------------------------------------------+
| NEXT.JS 16 DASHBOARD (FRONTEND) |
| [ Interactive Student IDE Editor ] |
| | (User Python Code Payload) |
| v |
| [ WebSockets Gateway Client ] |
+----------|------------------------------------------------------------------------+
| (JSON Payload over WSS)
v
+-----------------------------------------------------------------------------------+
| PYRANTS EVALUATION ENGINE (BACKEND) |
| [ Python AST Security Validator ] ---> Blocks Restricted AST Nodes |
| | (Clean Code Tree) |
| v |
| [ Sandboxed Worker Scope ] ---> Redirects Stdout / Stderr Buffers |
| |---> Enforces 3s Execution Timeout Guard |
| v |
| [ Live Telemetry Stream ] ---> Pushes Terminal Outputs to Frontend |
+-----------------------------------------------------------------------------------+FastStreamReader)Standard OpenCV cv2.VideoCapture or HTTP requests buffer incoming network frames in an internal queue. Over Wi-Fi or cellular networks, this buffer creates an escalating 2-5 second video delay.
MobiCam solves this with a custom threading class called FastStreamReader:
ai_engine.py)MobileNetSSD_deploy.caffemodel) combined with OpenCV's cv2.dnn.readNetFromCaffe.Person Only, Pets (dogs, cats), and Vehicles (cars, buses, motorbikes).VideoWriter to capture a 5-second .mp4 clip to recordings/..jpg images to snapshots/ and events/, and triggers audible sound alerts for perimeter security.MyCam Server)MjpegStreamServer.kt) on port 8080.FastStreamReader in Python)import cv2
import threading
import urllib.request
import numpy as np
class FastStreamReader:
"""Continuously flushes socket frame buffers to achieve 0-latency streaming."""
def __init__(self, url: str):
self.url = url
self.stream = urllib.request.urlopen(url)
self.bytes = b''
self.frame = None
self.stopped = False
self.lock = threading.Lock()
# Start background frame reading thread
self.thread = threading.Thread(target=self._update, daemon=True)
self.thread.start()
def _update(self):
while not self.stopped:
try:
self.bytes += self.stream.read(4096)
a = self.bytes.find(b'ÿØ') # JPEG start marker
b = self.bytes.find(b'ÿÙ') # JPEG end marker
if a != -1 and b != -1:
jpg = self.bytes[a:b+2]
self.bytes = self.bytes[b+2:] # Flush consumed buffer
img = cv2.imdecode(np.frombuffer(jpg, dtype=np.uint8), cv2.IMREAD_COLOR)
with self.lock:
self.frame = img
except Exception as e:
print(f"[FastStreamReader Error]: {e}")
break
def read(self):
with self.lock:
return self.frame is not None, self.frame
def stop(self):
self.stopped = Trueai_engine.py)import cv2
import numpy as np
import time
CLASSES = ["background", "aeroplane", "bicycle", "bird", "boat",
"bottle", "bus", "car", "cat", "chair", "cow", "diningtable",
"dog", "horse", "motorbike", "person", "pottedplant", "sheep",
"sofa", "train", "tvmonitor"]
class MobiCamAIEngine:
def __init__(self, prototxt_path: str, model_path: str, confidence_thresh=0.5):
self.net = cv2.dnn.readNetFromCaffe(prototxt_path, model_path)
self.confidence_thresh = confidence_thresh
def detect_objects(self, frame: np.ndarray):
(h, w) = frame.shape[:2]
# Prepare 300x300 blob for MobileNet-SSD DNN
blob = cv2.dnn.blobFromImage(cv2.resize(frame, (300, 300)), 0.007843, (300, 300), 127.5)
self.net.setInput(blob)
detections = self.net.forward()
results = []
for i in range(detections.shape[2]):
confidence = detections[0, 0, i, 2]
if confidence > self.confidence_thresh:
idx = int(detections[0, 0, i, 1])
label = CLASSES[idx]
box = detections[0, 0, i, 3:7] * np.array([w, h, w, h])
(startX, startY, endX, endY) = box.astype("int")
results.append({
"label": label,
"confidence": float(confidence),
"box": (startX, startY, endX, endY)
})
# Draw visual bounding box and label overlay
cv2.rectangle(frame, (startX, startY), (endX, endY), (0, 255, 0), 2)
text = f"{label}: {confidence * 100:.1f}%"
cv2.putText(frame, text, (startX, startY - 10),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)
return results, frameimport ast
import sys
import io
FORBIDDEN_NODES = {ast.Import, ast.ImportFrom, ast.Global, ast.Nonlocal}
def validate_python_ast(code_str: str) -> bool:
"""Validates student Python submission for restricted AST calls."""
try:
tree = ast.parse(code_str)
for node in ast.walk(tree):
if type(node) in FORBIDDEN_NODES:
return False
if isinstance(node, ast.Call) and getattr(node.func, 'id', '') in {'exec', 'eval', 'open', '__import__'}:
return False
return True
except SyntaxError:
return False
def run_pyrants_sandbox(code_str: str) -> dict:
if not validate_python_ast(code_str):
return {"status": "error", "message": "Restricted call detected by Pyrants AST validator."}
buffer = io.StringIO()
sys.stdout = buffer
sys.stderr = buffer
safe_scope = {"__builtins__": {"print": print, "range": range, "len": len, "int": int, "str": str}}
try:
exec(code_str, safe_scope)
return {"status": "success", "output": buffer.getvalue()}
except Exception as err:
return {"status": "runtime_error", "output": str(err)}
finally:
sys.stdout = sys.__stdout__
sys.stderr = sys.__stderr__MyCam Server (MyCamServer.apk in releases/), built with Kotlin, Android CameraX, and Jetpack Compose.