Read a frame, inspect a result, make a decision, and display it yourself.
Computer vision that keeps your code visible.
CVGO simplifies repetitive OpenCV and MediaPipe setup while keeping
the familiar while True flow easy to read, edit, and customize.
Simple code, without hiding the process.
Start with useful defaults. Keep control of frames, detections, conditions, output, and cleanup when your project grows.
Face, hand, pose, holistic, gesture, object detection, and segmentation.
FPS, timers, alarms, Arduino serial, Telegram photos, and editable thresholds.
Install CVGO.
Use Python 3.10, 3.11, or 3.12 in a dedicated virtual environment.
python -m pip install cvgo
py -3.11 -m venv .venv-cvgo
.venv-cvgo\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install cvgo
python3.11 -m venv .venv-cvgo
source .venv-cvgo/bin/activate
python -m pip install --upgrade pip
python -m pip install cvgo
python -m cvgo check
python -m cvgo check --camera 4
Install only opencv-contrib-python for CVGO. Avoid another
OpenCV package variant in the same environment. CVGO is a universal
Python wheel; on AArch64, pip must still find compatible OpenCV and
MediaPipe wheels for the target operating system.
Defaults that stay customizable.
Constructors work with no arguments for beginners. Optional parameters remain available when a camera or project needs different behavior.
Main defaults
- Camera()
- Camera 0, OpenCV CAP_ANY
- FaceDetector()
- Fast engine, maximum one face
- FaceLandmarks()
- Maximum one face
- HandTracker()
- Maximum two hands
- PoseTracker()
- One main pose, complexity 1
- GestureRecognizer()
- Video mode, two hands
- ObjectDetector()
- Video mode, 10 objects
- Serial()
- Automatic port, 9600 baud
- Telegram()
- Environment config, 30 s cooldown
- Timer()
- One-second duration
- Smoother()
- Alpha 0.45
Python naming style
- Classes use
PascalCase:PoseTracker. - Methods use
snake_case:put_text(). - Constants use
UPPER_CASE:BIT_DROWSY. - Use
fps.read()for the short, readable FPS API.
MediaPipe task modes
ObjectDetector and GestureRecognizer support three modes. Video remains the compatible default; live keeps camera loops responsive.
- mode="image"
- Independent still images
- mode="video"
- Synchronous sequential frames
- mode="live"
- Latest asynchronous camera result
- result_ready
- First result has completed
Consistent bounding boxes
Face, hand, pose, and object boxes share one predictable API. Pose boxes ignore low-visibility landmarks by default.
- box.xyxy
- Left, top, right, bottom
- box.center
- Center pixel coordinates
- box.area
- Box area in pixels
- box.draw()
- Rectangle and optional label
Advanced parameters, without losing simplicity.
Keep the basic examples unchanged. Add only the named parameters needed by a camera, model, output, or calibrated project.
Every omitted parameter keeps its documented default. Parameters after an asterisk are keyword-only, so their names stay visible.
camera = go.Camera()
camera = go.Camera(4)
camera = go.Camera(4, width=1280, height=720, fps=30)
- Confidence values use
0.0to1.0. - OpenCV colors use BGR, not RGB.
Nonekeeps the system or library default.static=Trueis for unrelated still images.mode="live"keeps task-based camera loops responsive.
Camera and drawing Choose a source, request capture settings, and customize the GUI.
go.Camera(source=0, *, width=None, height=None, fps=None, backend=None)
| Parameter | Default | What it changes |
|---|---|---|
| source | 0 | Camera index, video path, or stream URL. |
| width | None | Requested capture width in pixels. |
| height | None | Requested capture height in pixels. |
| fps | None | Requested camera frame rate. |
| backend | None | OpenCV backend; None uses CAP_ANY. |
Width, height, and FPS are requests; a camera driver may choose
the nearest supported value. Read camera.size after
opening, or use camera.capture for raw OpenCV settings.
Display and text
| API / parameter | Default | What it changes |
|---|---|---|
| show.title | "CVGO" | Window title. |
| show.delay | 1 | Keyboard polling delay in milliseconds. |
| show.quit_key | "q" | One character that closes the loop. |
| close.windows | True | Destroy OpenCV GUI windows; use False for a headless check. |
| put_text.position | (20, 35) | Text origin in pixels. |
| put_text.color | (0, 255, 0) | Text color in BGR. |
| put_text.scale | 0.7 | OpenCV font scale. |
| put_text.thickness | 2 | Text stroke width. |
| put_text.background | False | Add a black text background. |
| box.draw.color | (0, 255, 0) | Shared bounding-box color. |
| box.draw.thickness | 2 | Shared bounding-box thickness. |
| box.draw.label | None | Optional shared bounding-box label. |
Diagnostics
go.system_info()
go.check_camera(source=0, *, backend=None)
Use these from a custom support tool, or run
python -m cvgo check --camera 4. The camera check
reads one frame, reports its dimensions, and closes without
opening a GUI window.
camera = go.Camera(1, width=1280, height=720, fps=30)
go.put_text(
frame,
"Security active",
color=(0, 255, 255),
background=True,
)
camera.show(frame, title="Security Camera", quit_key="x")
Face detection and landmarks Control face count, confidence, iris refinement, boxes, and drawing.
go.FaceDetector(*, max_faces=1, padding=10, model=0, detection_confidence=0.5, engine="auto", refine=False, tracking_confidence=0.5)
go.FaceLandmarks(*, max_faces=1, refine=False, detection_confidence=0.5, tracking_confidence=0.5)
| Parameter | Default | What it changes |
|---|---|---|
| max_faces | 1 | Maximum faces returned per frame. |
| padding | 10 | Extra pixels around a detector box. |
| model | 0 | Fast model: 0 near-range or 1 full-range. |
| engine | "auto" | Select auto, fast, or Face Mesh-compatible mesh. |
| refine | False | Mesh only: refine eyes and lips and add iris landmarks. |
| detection_confidence | 0.5 | Minimum initial face detection confidence. |
| tracking_confidence | 0.5 | Mesh only: minimum landmark tracking confidence. |
The default auto engine uses lightweight MediaPipe Face
Detection. It keeps the compatible mesh engine when
refine=True or a custom
tracking_confidence is supplied. Every fast
FaceBox also exposes confidence.
raw_result remains available in both modes;
detector.faces contains landmarks only in mesh mode.
Face result methods
| API / parameter | Default | What it changes |
|---|---|---|
| face.box.padding | 10 | Extra pixels around landmark bounds. |
| face.draw.style | "contours" | contours, tesselation, iris, or all. |
| face.draw.color | (0, 255, 0) | Landmark and connection color. |
| face.draw.thickness | 1 | Connection thickness. |
| face.draw.radius | 1 | Landmark radius. |
| FaceBox.draw.color | (0, 255, 0) | Box and label color. |
| FaceBox.draw.thickness | 2 | Box and label thickness. |
| FaceBox.draw.label | "Face" | Box label; None hides it. |
detector = go.FaceDetector(
max_faces=2,
detection_confidence=0.7,
model=0,
)
landmarker = go.FaceLandmarks(
max_faces=2,
refine=True,
detection_confidence=0.7,
)
faces = landmarker.detect(frame)
for face in faces:
face.draw(frame, style="tesselation", color=(255, 180, 0))
face.box(padding=20).draw(frame, label="Tracked face")
Hand tracking Balance speed, accuracy, hand count, handedness, and drawing.
go.HandTracker(*, max_hands=2, model_complexity=1, detection_confidence=0.5, tracking_confidence=0.5, static=False, mirrored=False)
| Parameter | Default | What it changes |
|---|---|---|
| max_hands | 2 | Maximum hands returned per frame. |
| model_complexity | 1 | 0 is lighter; 1 is more accurate. |
| detection_confidence | 0.5 | Minimum hand detection confidence. |
| tracking_confidence | 0.5 | Minimum landmark tracking confidence. |
| static | False | Use True for independent still images. |
| mirrored | False | Use True if the input was already flipped horizontally. |
Hand result methods
| API / parameter | Default | What it changes |
|---|---|---|
| hand.box.padding | 10 | Extra pixels around the hand. |
| hand.draw.color | (0, 255, 0) | Connection color. |
| hand.draw.point_color | (255, 0, 255) | Landmark color. |
| hand.draw.thickness | 2 | Connection thickness. |
| hand.draw.radius | 2 | Landmark radius. |
| HandBox.draw.label | "Hand" | Box label; None hides it. |
tracker = go.HandTracker(
max_hands=1,
model_complexity=0,
detection_confidence=0.7,
)
hands = tracker.detect(frame)
for hand in hands:
hand.draw(frame, color=(255, 200, 0), point_color=(0, 0, 255))
Pose tracking Select Lite, Full, or Heavy and tune visibility-based boxes.
go.PoseTracker(*, model_complexity=1, detection_confidence=0.5, tracking_confidence=0.5, smooth=True, segmentation=False, static=False)
| Parameter | Default | What it changes |
|---|---|---|
| model_complexity | 1 | 0 Lite, 1 Full, or 2 Heavy. |
| detection_confidence | 0.5 | Minimum pose detection confidence. |
| tracking_confidence | 0.5 | Minimum landmark tracking confidence. |
| smooth | True | Smooth landmarks and an optional segmentation mask. |
| segmentation | False | Also produce pose.mask. |
| static | False | Use True for independent still images. |
Pose result methods
| API / parameter | Default | What it changes |
|---|---|---|
| pose.visible.confidence | 0.5 | Required landmark visibility. |
| pose.box.padding | 20 | Extra pixels around the visible body. |
| pose.box.min_visibility | 0.5 | Ignore weaker landmarks when building the box. |
| pose.draw.color | (0, 255, 0) | Skeleton connection color. |
| pose.draw.point_color | (255, 0, 255) | Landmark color. |
| pose.draw.thickness | 2 | Connection thickness. |
| pose.draw.radius | 2 | Landmark radius. |
| PoseBox.draw.label | "Person" | Box label; None hides it. |
Use model_complexity=0 for an STB or low-power board.
tracker = go.PoseTracker(
model_complexity=0,
detection_confidence=0.6,
segmentation=True,
)
pose = tracker.detect(frame)
if pose:
pose.box(padding=30, min_visibility=0.6).draw(
frame,
label="Person",
)
Holistic tracking Configure the combined face, pose, hands, and segmentation pipeline.
go.HolisticTracker(*, model_complexity=1, detection_confidence=0.5, tracking_confidence=0.5, smooth=True, refine_face=False, segmentation=False, static=False)
| Parameter | Default | What it changes |
|---|---|---|
| model_complexity | 1 | Pose model complexity: 0, 1, or 2. |
| detection_confidence | 0.5 | Minimum initial detection confidence. |
| tracking_confidence | 0.5 | Minimum landmark tracking confidence. |
| smooth | True | Smooth landmarks and an optional mask. |
| refine_face | False | Refine landmarks around the eyes and lips. |
| segmentation | False | Also produce result.mask. |
| static | False | Use True for independent still images. |
result.draw(frame, face=True, pose=True, hands=True)
can show or hide each landmark group independently. All three
Boolean parameters default to True.
tracker = go.HolisticTracker(
model_complexity=0,
refine_face=True,
)
result = tracker.detect(frame)
if result:
result.draw(frame, face=False, pose=True, hands=True)
Object detection Filter labels, tune confidence, select a task mode, or load a model.
go.ObjectDetector(model_path=None, *, confidence=0.5, max_objects=10, allow=None, deny=None, locale="en", mode="video", stream=None, download=True)
| Parameter | Default | What it changes |
|---|---|---|
| model_path | None | Compatible custom .tflite model path. |
| confidence | 0.5 | Minimum object score. |
| max_objects | 10 | Maximum results per frame. |
| allow | None | Return only these labels, such as ["person"]. |
| deny | None | Exclude these labels. |
| locale | "en" | Preferred display-name locale in model metadata. |
| mode | "video" | image, video, or asynchronous live. |
| stream | None | Legacy option; new code should use mode. |
| download | True | Download the default model when not cached. |
allow and deny cannot be combined. In
live mode, detect() returns the latest completed
result; result_ready identifies the first completion.
Detected object drawing
| Parameter | Default | What it changes |
|---|---|---|
| color | (0, 255, 0) | Box and label color. |
| thickness | 2 | Box and label thickness. |
| show_score | True | Include confidence in the label. |
detector = go.ObjectDetector(
confidence=0.65,
max_objects=3,
allow=["person", "car"],
)
objects = detector.detect(frame)
for item in objects:
item.draw(frame, color=(0, 200, 255), show_score=False)
Gesture recognition and task models Tune gesture stages, use asynchronous results, and prepare offline models.
go.GestureRecognizer(model_path=None, *, max_hands=2, gesture_confidence=0.5, detection_confidence=0.5, presence_confidence=0.5, tracking_confidence=0.5, mirrored=False, mode="video", stream=None, download=True)
| Parameter | Default | What it changes |
|---|---|---|
| model_path | None | Compatible custom .task model path. |
| max_hands | 2 | Maximum hands recognized per frame. |
| gesture_confidence | 0.5 | Minimum score for a recognized gesture. |
| detection_confidence | 0.5 | Minimum hand detection confidence. |
| presence_confidence | 0.5 | Minimum hand presence confidence. |
| tracking_confidence | 0.5 | Minimum landmark tracking confidence. |
| mirrored | False | Use True if input was already flipped. |
| mode | "video" | image, video, or asynchronous live. |
| stream | None | Legacy option; new code should use mode. |
| download | True | Download the default model when not cached. |
Gesture result methods
| API / parameter | Default | What it changes |
|---|---|---|
| gesture.box.padding | 10 | Extra pixels around the gesture hand. |
| gesture.draw.color | (0, 255, 0) | Connections and box color. |
| gesture.draw.point_color | (255, 0, 255) | Hand landmark color. |
model = go.download_model(
"gesture_recognizer",
directory="models",
timeout=180,
)
recognizer = go.GestureRecognizer(
model,
max_hands=1,
gesture_confidence=0.7,
)
Model download parameters
| Parameter | Default | What it changes |
|---|---|---|
| name | Required | object_detection or gesture_recognizer. |
| directory | None | Custom download folder. |
| force | False | Download again when a valid model exists. |
| timeout | 120.0 | Download timeout in seconds. |
Set CVGO_MODEL_DIR to change the shared model cache.
CVGO verifies each pinned model's header and SHA-256 checksum and
downloads a damaged or incomplete cache file again.
Segmentation and timing Choose a segmentation model and tune masks, timers, smoothing, and FPS.
| API / parameter | Default | What it changes |
|---|---|---|
| SelfieSegmenter.model | 1 | 0 general; 1 landscape/webcam. |
| foreground.threshold | 0.5 | Minimum mask value treated as foreground. |
| apply.background | (0, 0, 0) | BGR color or image matching the frame size. |
| apply.threshold | 0.5 | Foreground cutoff. |
| blur.amount | 35 | Blur kernel; an even value is raised to the next odd value. |
| blur.threshold | 0.5 | Foreground cutoff. |
| Timer.seconds | 1.0 | Time a condition must remain true. |
| Smoother.alpha | 0.45 | Lower is smoother; higher reacts faster. |
| FPS.update_every | 1.0 | Seconds between displayed FPS updates. |
segmenter = go.SelfieSegmenter(model=0)
timer = go.Timer(1.5)
smoother = go.Smoother(alpha=0.3)
fps = go.FPS(update_every=0.5)
result = segmenter.segment(frame)
frame = result.blur(frame, amount=51, threshold=0.6)
Serial, Telegram, and alarm output Configure device connections, notification cooldowns, and sound.
go.Serial(port=None, *, baud=9600, timeout=1.0, reconnect_after=5.0, settle_time=2.0, newline=False, connect=True)
| Serial parameter | Default | What it changes |
|---|---|---|
| port | None | Auto-detect, or use a path such as COM5 or /dev/ttyUSB0. |
| baud | 9600 | Baud rate; it must match the board. |
| timeout | 1.0 | Read timeout in seconds. |
| reconnect_after | 5.0 | Minimum delay between reconnect attempts. |
| settle_time | 2.0 | Wait after a board resets on connect; use 0 when unnecessary. |
| newline | False | Append a newline to outgoing values. |
| connect | True | Connect during construction. |
send() waits for its result.
send_async() uses one ordered worker so a serial
reconnect or write does not hold the camera loop. Both return a
boolean result; the asynchronous form wraps it in a Future.
go.Telegram(token=None, chat_id=None, *, cooldown=30.0, timeout=15.0, silent=False, protect=False)
| Telegram parameter | Default | What it changes |
|---|---|---|
| token | Environment | Bot token or CVGO_TELEGRAM_TOKEN. |
| chat_id | Environment | Target ID or CVGO_TELEGRAM_CHAT_ID. |
| cooldown | 30.0 | Seconds between successful sends using the same key. |
| timeout | 15.0 | HTTP request timeout in seconds. |
| silent | False | Send without notification sound. |
| protect | False | Ask Telegram to protect message content. |
send_message() parameters
| Parameter | Default | What it changes |
|---|---|---|
| text | Required | Message text, from 1 to 4096 characters. |
| key | "message" | Independent cooldown name. |
| force | False | Bypass cooldown intentionally. |
| silent | None | Use or override the constructor setting. |
| protect | None | Use or override the constructor setting. |
| parse_mode | None | Formatting mode such as HTML. |
send_photo() parameters
| Parameter | Default | What it changes |
|---|---|---|
| photo | Required | OpenCV frame, image bytes, or image path. |
| caption | "" | Caption up to 1024 characters. |
| key | "photo" | Independent cooldown name. |
| force | False | Bypass cooldown intentionally. |
| filename | None | Optional uploaded filename. |
| quality | 85 | JPEG quality from 1 to 100 for OpenCV frames. |
| silent | None | Use or override the constructor setting. |
| protect | None | Use or override the constructor setting. |
| parse_mode | None | Caption formatting mode. |
send_message_async() and
send_photo_async() accept the same parameters and use
one background queue. Camera frames are copied before queueing.
Call telegram.close() after the loop; its
wait=True default finishes queued sends.
go.MqttClient(host="localhost", port=1883, *, client_id="", username=None, password=None, keepalive=60, reconnect_after=5.0, tls=False, connect=False)
go.WebSocketClient(url, *, timeout=10.0, reconnect_after=5.0, connect=False)
MQTT publishes and subscriptions use JSON for structured values;
WebSocket send() and receive() do the same.
Both clients support connect(),
reconnect(), close(), context managers,
and asynchronous sending. Install these optional integrations with
pip install "cvgo[robotics]".
import cvgo as go
with go.MqttClient(host="localhost", client_id="cvgo-camera") as mqtt:
mqtt.publish("robot/camera/state", {"person_detected": True})
import cvgo as go
with go.WebSocketClient("ws://localhost:8080") as websocket:
websocket.send({"type": "person_detected", "confidence": 0.94})
message = websocket.receive()
go.Alarm(*, frequency=1500, duration=180, repeat=3, cooldown=0.8)
| Alarm parameter | Default | What it changes |
|---|---|---|
| frequency | 1500 | Beep frequency in Hz on Windows. |
| duration | 180 | Length of each beep in milliseconds. |
| repeat | 3 | Number of beeps per trigger. |
| cooldown | 0.8 | Minimum seconds between alarm starts. |
arduino = go.Serial(
"/dev/ttyUSB0",
baud=115200,
newline=True,
)
telegram = go.Telegram(cooldown=60, silent=True)
alarm = go.Alarm(frequency=1800, repeat=2, cooldown=1.0)
arduino.send_async("1")
telegram.send_message_async("CVGO active")
Advanced Driver Monitor Calibrate eye, head, missing-face, serial, and event behavior.
Driver Monitor groups related thresholds into small configuration objects, so each part can be calibrated without a crowded constructor.
| EyeConfig parameter | Default | What it changes |
|---|---|---|
| closed_threshold | 0.20 | Enter closed-eye state below this EAR. |
| open_threshold | 0.24 | Leave closed-eye state above this EAR. |
| alert_after | 1.5 | Closed-eye seconds before drowsiness is active. |
| smoothing | 0.45 | EAR smoother alpha. |
| HeadConfig parameter | Default | What it changes |
|---|---|---|
| yaw_normal | 0.50 | Calibrated straight-ahead yaw ratio. |
| turn_threshold | 0.12 | Enter looking-away state beyond this offset. |
| turn_release | 0.07 | Leave looking-away state below this offset. |
| turn_alert_after | 0.7 | Looking-away seconds before an alert. |
| pitch_normal | 0.50 | Calibrated upright pitch ratio. |
| down_threshold | 0.055 | Enter head-down state beyond this offset. |
| down_release | 0.030 | Leave head-down state below this offset. |
| down_alert_after | 0.7 | Head-down seconds before an alert. |
| Other parameter | Default | What it changes |
|---|---|---|
| FaceConfig.missing_alert_after | 2.0 | Missing-face seconds before an alert. |
| DriverMonitor.camera | 0 | Camera source or configured Camera. |
| DriverMonitor.serial | False | True for auto serial or a Serial object. |
| DriverMonitor.sound | False | Enable its built-in alarm. |
| serial_repeat_after | 0.5 | Seconds between repeated mask transmissions. |
eyes = go.EyeConfig(
closed_threshold=0.22,
open_threshold=0.26,
alert_after=1.2,
)
head = go.HeadConfig(turn_threshold=0.10, down_threshold=0.05)
face = go.FaceConfig(missing_alert_after=3.0)
monitor = go.DriverMonitor(
camera=go.Camera(4, width=640, height=480),
serial=go.Serial("/dev/ttyUSB0", baud=115200),
sound=True,
eyes=eyes,
head=head,
face=face,
)
monitor.serial_repeat_after = 1.0
Events: drowsy, looking_away,
looking_left, looking_right,
head_down, face_missing, and
normal. The result keeps measurements, durations,
landmarks, alert flags, FPS, mask, and
mask_hex available for custom logic.
Driver Monitor display and shortcut mode
| Method parameter | Default | What it changes |
|---|---|---|
| show.title | "CVGO Driver Monitor" | GUI window title. |
| show.draw_landmarks | True | Draw face landmarks before showing. |
| show.landmark_style | "contours" | Face drawing style. |
| show.quit_key | "q" | GUI quit key. |
| run.show | False | Enable a GUI window in shortcut mode. |
| run.draw_landmarks | False | Draw landmarks in shortcut mode. |
| run.print_status | True | Print status twice per second. |
| run.quit_key | "q" | GUI quit key when show=True. |
Raw access remains available
| Access | Raw or editable value | Use |
|---|---|---|
| camera.capture | cv2.VideoCapture | Additional OpenCV camera properties. |
| tracker.raw_result | MediaPipe result | Features not wrapped by CVGO. |
| face.raw / hand.raw / pose.raw | MediaPipe landmarks | Direct MediaPipe interoperability. |
| item.raw / gesture.raw | Task result or category | Model-specific metadata. |
| face.points / hand.points / pose.points | CVGO points | Readable custom calculations. |
34 complete, copy-ready examples.
Every topic includes a complete Standard / GUI program and a complete CLI / Terminal program, including imports, loops, and cleanup.
CLI examples are ordinary Python scripts that print live status. Stop them with Ctrl+C.
01 Camera and GUIOpen a camera with OpenCV CAP_ANY. Press q to quit the window or Ctrl+C in terminal mode.
"""Example 1: open and display the camera."""
import cvgo as go
camera = go.Camera()
while True:
frame = camera.read()
if frame is None:
break
if not camera.show(frame):
break
camera.close()
"""CLI example 1: read camera information without a preview window."""
import cvgo as go
camera = go.Camera()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
height, width = frame.shape[:2]
info = f"Camera: ON | Size: {width}x{height} | FPS: {fps.read():.1f}"
print(f"\r{info:<70}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
02 Face DetectionDetect faces, draw the boxes, or print the live face count.
"""Example 2: detect faces."""
import cvgo as go
camera = go.Camera()
detector = go.FaceDetector()
while True:
frame = camera.read()
if frame is None:
break
faces = detector.detect(frame)
for face in faces:
face.draw(frame)
if not camera.show(frame):
break
camera.close()
detector.close()
"""CLI example 2: print face detection status."""
import cvgo as go
camera = go.Camera()
detector = go.FaceDetector()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
faces = detector.detect(frame)
status = "DETECTED" if faces else "NOT DETECTED"
info = (
f"Face: {status} | Count: {len(faces)} | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<70}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
detector.close()
03 Face LandmarksRead and draw detailed face landmarks for custom measurements.
"""Example 3: display face landmarks."""
import cvgo as go
camera = go.Camera()
landmarker = go.FaceLandmarks()
while True:
frame = camera.read()
if frame is None:
break
faces = landmarker.detect(frame)
for face in faces:
face.draw(frame)
if not camera.show(frame):
break
camera.close()
landmarker.close()
"""CLI example 3: print face and landmark counts."""
import cvgo as go
camera = go.Camera()
landmarker = go.FaceLandmarks()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
faces = landmarker.detect(frame)
points = sum(len(face) for face in faces)
info = (
f"Faces: {len(faces)} | Landmarks: {points} | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<70}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
landmarker.close()
04 Face MetricsCalculate EAR, yaw, and pitch while keeping thresholds editable.
"""Example 4: read EAR, yaw, and pitch."""
import cvgo as go
camera = go.Camera()
landmarker = go.FaceLandmarks()
while True:
frame = camera.read()
if frame is None:
break
faces = landmarker.detect(frame)
if faces:
face = faces[0]
ear = go.eye_ratio(face)
yaw = go.yaw_ratio(face)
pitch = go.pitch_ratio(face)
go.put_text(frame, f"EAR: {ear:.3f}")
go.put_text(frame, f"Yaw: {yaw:.3f}", (20, 70))
go.put_text(frame, f"Pitch: {pitch:.3f}", (20, 105))
face.draw(frame)
if not camera.show(frame):
break
camera.close()
landmarker.close()
"""CLI example 4: print EAR, yaw, and pitch values."""
import cvgo as go
camera = go.Camera()
landmarker = go.FaceLandmarks()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
faces = landmarker.detect(frame)
if faces:
face = faces[0]
ear = f"{go.eye_ratio(face):.3f}"
yaw = f"{go.yaw_ratio(face):.3f}"
pitch = f"{go.pitch_ratio(face):.3f}"
else:
ear = yaw = pitch = "---"
info = (
f"EAR: {ear} | Yaw: {yaw} | Pitch: {pitch} | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<80}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
landmarker.close()
05 Serial ArduinoConnect automatically and send values to an Arduino.
"""Example 5: send data to Arduino."""
import cvgo as go
arduino = go.Serial()
if arduino.connected:
arduino.send("1")
arduino.close()
"""CLI example 5: send terminal input to Arduino."""
import cvgo as go
arduino = go.Serial()
try:
if not arduino.connected:
print("Arduino: NOT CONNECTED")
else:
print(f"Arduino: CONNECTED | Port: {arduino.port}")
print("Type a value and press Enter. Type q to quit.")
while True:
value = input("Send > ").strip()
if value.lower() == "q":
break
if value:
status = "SENT" if arduino.send(value) else "FAILED"
print(f"{status}: {value}")
except KeyboardInterrupt:
print()
finally:
arduino.close()
06 Face to ArduinoSend face-presence status without repeating unchanged serial data.
"""Example 6: send face detection status to Arduino."""
import cvgo as go
camera = go.Camera()
detector = go.FaceDetector()
arduino = go.Serial()
last_status = None
while True:
frame = camera.read()
if frame is None:
break
faces = detector.detect(frame)
status = 1 if faces else 0
if status != last_status:
if arduino.send(status):
last_status = status
for face in faces:
face.draw(frame)
if not camera.show(frame):
break
camera.close()
detector.close()
arduino.close()
"""CLI example 6: send face presence to Arduino."""
import cvgo as go
camera = go.Camera()
detector = go.FaceDetector()
arduino = go.Serial()
fps = go.FPS()
last_status = None
try:
while True:
frame = camera.read()
if frame is None:
break
faces = detector.detect(frame)
status = 1 if faces else 0
if status != last_status and arduino.send(status):
last_status = status
face_text = "DETECTED" if status else "NOT DETECTED"
serial_text = "CONNECTED" if arduino.connected else "DISCONNECTED"
info = (
f"Face: {face_text} | Arduino: {serial_text} | "
f"Sent: {last_status} | FPS: {fps.read():.1f}"
)
print(f"\r{info:<100}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
detector.close()
arduino.close()
07 Drowsiness DetectionCombine EAR, smoothing, a timer, and an alarm.
"""Example 7: detect drowsiness based on EAR and duration."""
import cvgo as go
EAR_THRESHOLD = 0.20
camera = go.Camera()
landmarker = go.FaceLandmarks()
eye_timer = go.Timer(1.5)
ear_smoother = go.Smoother()
alarm = go.Alarm()
while True:
frame = camera.read()
if frame is None:
break
faces = landmarker.detect(frame)
drowsy = False
if faces:
face = faces[0]
ear = ear_smoother.update(go.eye_ratio(face))
eyes_closed = ear < EAR_THRESHOLD
drowsy = eye_timer.check(eyes_closed)
status = "DROWSY" if drowsy else "NORMAL"
color = (0, 0, 255) if drowsy else (0, 255, 0)
go.put_text(frame, f"Status: {status}", color=color)
go.put_text(frame, f"EAR: {ear:.3f}", (20, 70))
face.draw(frame, color=color)
else:
eye_timer.reset()
ear_smoother.reset()
alarm.trigger(drowsy)
if not camera.show(frame):
break
camera.close()
landmarker.close()
"""CLI example 7: detect drowsiness and print the result."""
import cvgo as go
EAR_THRESHOLD = 0.30
camera = go.Camera()
landmarker = go.FaceLandmarks()
eye_timer = go.Timer(0.5)
ear_smoother = go.Smoother()
alarm = go.Alarm()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
faces = landmarker.detect(frame)
drowsy = False
status = "NO FACE"
ear_text = "---"
if faces:
ear = ear_smoother.update(go.eye_ratio(faces[0]))
drowsy = eye_timer.check(ear < EAR_THRESHOLD)
status = "DROWSY" if drowsy else "NORMAL"
ear_text = f"{ear:.3f}"
else:
eye_timer.reset()
ear_smoother.reset()
alarm.trigger(drowsy)
info = (
f"Status: {status} | EAR: {ear_text} | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<70}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
landmarker.close()
08 Driver MonitorComplete modular monitoring with eyes, head direction, serial output, and FPS.
"""Example 8: final driver monitor project that is still easy to study."""
import cvgo as go
# Detection thresholds
EAR_THRESHOLD = 0.20
YAW_NORMAL = 0.50
YAW_LIMIT = 0.12
PITCH_NORMAL = 0.50
PITCH_LIMIT = 0.055
# Components
camera = go.Camera()
landmarker = go.FaceLandmarks()
arduino = go.Serial()
alarm = go.Alarm()
fps_counter = go.FPS()
# Condition timers
eye_timer = go.Timer(1.5)
turn_timer = go.Timer(0.7)
down_timer = go.Timer(0.7)
missing_timer = go.Timer(2.0)
# Eye value smoother
ear_smoother = go.Smoother()
# Last serial status
last_mask = None
while True:
frame = camera.read()
if frame is None:
break
faces = landmarker.detect(frame)
fps = fps_counter.read()
ear = None
yaw = None
pitch = None
drowsy = False
looking_away = False
head_down = False
face_missing = False
if faces:
face = faces[0]
ear = ear_smoother.update(go.eye_ratio(face))
yaw = go.yaw_ratio(face)
pitch = go.pitch_ratio(face)
eyes_closed = ear < EAR_THRESHOLD
turn_condition = abs(yaw - YAW_NORMAL) > YAW_LIMIT
down_condition = pitch - PITCH_NORMAL > PITCH_LIMIT
drowsy = eye_timer.check(eyes_closed)
looking_away = turn_timer.check(turn_condition)
head_down = down_timer.check(down_condition)
missing_timer.reset()
face.draw(frame)
else:
eye_timer.reset()
turn_timer.reset()
down_timer.reset()
ear_smoother.reset()
face_missing = missing_timer.check(True)
mask = 0
if drowsy:
mask |= go.BIT_DROWSY
if looking_away:
mask |= go.BIT_LOOKING_AWAY
if head_down:
mask |= go.BIT_HEAD_DOWN
if face_missing:
mask |= go.BIT_FACE_MISSING
if mask != last_mask:
if arduino.send(f"{mask:X}"):
last_mask = mask
alerts = []
if drowsy:
alerts.append("DROWSY")
if looking_away:
alerts.append("LOOKING_AWAY")
if head_down:
alerts.append("HEAD_DOWN")
if face_missing:
alerts.append("FACE_MISSING")
status = " | ".join(alerts) if alerts else "NORMAL"
color = (0, 0, 255) if alerts else (0, 255, 0)
ear_text = "-" if ear is None else f"{ear:.3f}"
yaw_text = "-" if yaw is None else f"{yaw:.3f}"
pitch_text = "-" if pitch is None else f"{pitch:.3f}"
go.put_text(
frame,
f"Status: {status}",
(20, 35),
color=color,
background=True,
)
go.put_text(frame, f"EAR: {ear_text}", (20, 70))
go.put_text(frame, f"Yaw: {yaw_text}", (20, 105))
go.put_text(frame, f"Pitch: {pitch_text}", (20, 140))
go.put_text(frame, f"FPS: {fps:.1f} | Mask: {mask:X}", (20, 175))
alarm.trigger(mask != 0)
if not camera.show(frame, title="CVGO Driver Monitor"):
break
camera.close()
landmarker.close()
arduino.close()
"""CLI example 8: complete modular driver monitoring."""
import cvgo as go
EAR_THRESHOLD = 0.20
YAW_NORMAL = 0.50
YAW_LIMIT = 0.12
PITCH_NORMAL = 0.50
PITCH_LIMIT = 0.055
camera = go.Camera()
landmarker = go.FaceLandmarks()
arduino = go.Serial()
alarm = go.Alarm()
fps = go.FPS()
eye_timer = go.Timer(1.5)
turn_timer = go.Timer(0.7)
down_timer = go.Timer(0.7)
missing_timer = go.Timer(2.0)
ear_smoother = go.Smoother()
last_mask = None
try:
while True:
frame = camera.read()
if frame is None:
break
faces = landmarker.detect(frame)
ear = yaw = pitch = None
drowsy = looking_away = head_down = face_missing = False
if faces:
face = faces[0]
ear = ear_smoother.update(go.eye_ratio(face))
yaw = go.yaw_ratio(face)
pitch = go.pitch_ratio(face)
drowsy = eye_timer.check(ear < EAR_THRESHOLD)
looking_away = turn_timer.check(
abs(yaw - YAW_NORMAL) > YAW_LIMIT
)
head_down = down_timer.check(
pitch - PITCH_NORMAL > PITCH_LIMIT
)
missing_timer.reset()
else:
eye_timer.reset()
turn_timer.reset()
down_timer.reset()
ear_smoother.reset()
face_missing = missing_timer.check(True)
mask = 0
if drowsy:
mask |= go.BIT_DROWSY
if looking_away:
mask |= go.BIT_LOOKING_AWAY
if head_down:
mask |= go.BIT_HEAD_DOWN
if face_missing:
mask |= go.BIT_FACE_MISSING
if mask != last_mask and arduino.send(f"{mask:X}"):
last_mask = mask
alerts = []
if drowsy:
alerts.append("DROWSY")
if looking_away:
alerts.append("LOOKING AWAY")
if head_down:
alerts.append("HEAD DOWN")
if face_missing:
alerts.append("FACE MISSING")
status = ", ".join(alerts) if alerts else "NORMAL"
ear_text = "---" if ear is None else f"{ear:.3f}"
yaw_text = "---" if yaw is None else f"{yaw:.3f}"
pitch_text = "---" if pitch is None else f"{pitch:.3f}"
alarm.trigger(mask != 0)
info = (
f"Status: {status} | EAR: {ear_text} | Yaw: {yaw_text} | "
f"Pitch: {pitch_text} | FPS: {fps.read():.1f} | Mask: {mask:X}"
)
print(f"\r{info:<130}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
landmarker.close()
arduino.close()
09 Hand TrackingTrack 21 landmarks, handedness, and confidence for each hand.
"""Example 9: hand tracking, hand labels, and FPS."""
import cvgo as go
camera = go.Camera()
tracker = go.HandTracker()
fps = go.FPS()
while True:
frame = camera.read()
if frame is None:
break
hands = tracker.detect(frame)
for hand in hands:
hand.draw(frame)
label = f"{hand.handedness}: {hand.confidence:.2f}"
hand.box().draw(frame, label=label)
go.put_text(frame, f"Hands: {len(hands)}")
go.put_text(frame, f"FPS: {fps.read():.1f}", (20, 70))
if not camera.show(frame, title="CVGO Hand Tracking"):
break
camera.close()
tracker.close()
"""CLI example 9: print hand tracking results."""
import cvgo as go
camera = go.Camera()
tracker = go.HandTracker()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
hands = tracker.detect(frame)
labels = [
f"{hand.handedness} ({hand.confidence:.2f})"
for hand in hands
]
details = ", ".join(labels) if labels else "NONE"
info = (
f"Hands: {len(hands)} | Detail: {details} | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<100}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
tracker.close()
10 Pose TrackingDetect one main body pose with 33 landmarks.
"""Example 10: body pose tracking."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker()
fps = go.FPS()
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
person_detected = pose is not None
if pose:
pose.draw(frame)
status = "POSE DETECTED" if person_detected else "NO POSE"
color = (0, 255, 0) if person_detected else (0, 0, 255)
go.put_text(frame, status, color=color)
go.put_text(frame, f"FPS: {fps.read():.1f}", (20, 70))
if not camera.show(frame, title="CVGO Pose Tracking"):
break
camera.close()
tracker.close()
"""CLI example 10: print body pose status."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
status = "DETECTED" if pose else "NOT DETECTED"
points = len(pose) if pose else 0
info = (
f"Pose: {status} | Landmarks: {points} | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<80}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
tracker.close()
11 Pose Security Use Pose Lite as a lightweight person box without a skeleton.
"""Example 11: lightweight person security with a pose bounding box."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker(model_complexity=0)
presence_timer = go.Timer(0.5)
alarm = go.Alarm()
fps = go.FPS()
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
person_detected = pose is not None
alert = presence_timer.check(person_detected)
status = "ALERT" if alert else "SAFE"
color = (0, 0, 255) if alert else (0, 255, 0)
if pose:
pose.box(
padding=30,
).draw(
frame,
color=color,
label="Person",
)
go.put_text(frame, f"Status: {status}", color=color)
go.put_text(frame, f"FPS: {fps.read():.1f}", (20, 70))
alarm.trigger(alert)
if not camera.show(frame, title="CVGO Person Security Lite"):
break
camera.close()
tracker.close()
"""CLI example 11: run lightweight person security with Pose Lite."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker(model_complexity=0)
presence_timer = go.Timer(0.5)
alarm = go.Alarm()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
alert = presence_timer.check(pose is not None)
status = "ALERT" if alert else "SAFE"
info = f"Security: {status} | FPS: {fps.read():.1f}"
alarm.trigger(alert)
print(f"\r{info:<60}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
tracker.close()
12 Object DetectionDetect common objects and inspect labels, scores, and boxes.
"""Example 12: general object detection."""
import cvgo as go
camera = go.Camera()
detector = go.ObjectDetector()
fps = go.FPS()
while True:
frame = camera.read()
if frame is None:
break
objects = detector.detect(frame)
for item in objects:
item.draw(frame)
go.put_text(frame, f"Objects: {len(objects)}")
go.put_text(frame, f"Loop FPS: {fps.read():.1f}", (20, 70))
if not camera.show(frame, title="CVGO Object Detection"):
break
camera.close()
detector.close()
"""CLI example 12: print detected object labels and scores."""
import cvgo as go
camera = go.Camera()
detector = go.ObjectDetector()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
objects = detector.detect(frame)
labels = [
f"{item.label} ({item.score:.2f})"
for item in objects[:3]
]
details = ", ".join(labels) if labels else "NONE"
info = (
f"Objects: {len(objects)} | Top: {details} | "
f"Loop FPS: {fps.read():.1f}"
)
print(f"\r{info:<120}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
detector.close()
13 Person SecurityFilter object detection to people for a security monitor.
"""Example 13: detect multiple people for simple security."""
import cvgo as go
camera = go.Camera()
detector = go.ObjectDetector(allow=["person"])
presence_timer = go.Timer(0.5)
alarm = go.Alarm()
while True:
frame = camera.read()
if frame is None:
break
people = detector.detect(frame)
alert = presence_timer.check(bool(people))
for person in people:
person.draw(frame, color=(0, 0, 255))
status = "ALERT" if alert else "SAFE"
color = (0, 0, 255) if alert else (0, 255, 0)
go.put_text(frame, f"Status: {status}", color=color)
go.put_text(frame, f"Count: {len(people)}", (20, 70))
alarm.trigger(alert)
if not camera.show(frame, title="CVGO Person Security"):
break
camera.close()
detector.close()
"""CLI example 13: detect people for a terminal security monitor."""
import cvgo as go
camera = go.Camera()
detector = go.ObjectDetector(allow=["person"])
presence_timer = go.Timer(0.5)
alarm = go.Alarm()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
people = detector.detect(frame)
alert = presence_timer.check(bool(people))
status = "ALERT" if alert else "SAFE"
info = (
f"Security: {status} | People: {len(people)} | "
f"Loop FPS: {fps.read():.1f}"
)
alarm.trigger(alert)
print(f"\r{info:<70}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
detector.close()
14 Gesture RecognitionRecognize supported hand gestures and their confidence.
"""Example 14: hand gesture recognition."""
import cvgo as go
camera = go.Camera()
recognizer = go.GestureRecognizer()
while True:
frame = camera.read()
if frame is None:
break
gestures = recognizer.detect(frame)
for gesture in gestures:
gesture.draw(frame)
if not camera.show(frame, title="CVGO Gesture Recognition"):
break
camera.close()
recognizer.close()
"""CLI example 14: print recognized hand gestures."""
import cvgo as go
camera = go.Camera()
recognizer = go.GestureRecognizer()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
gestures = recognizer.detect(frame)
labels = [
f"{gesture.label} ({gesture.score:.2f})"
for gesture in gestures
if gesture.recognized
]
details = ", ".join(labels) if labels else "NONE"
info = f"Gestures: {details} | Loop FPS: {fps.read():.1f}"
print(f"\r{info:<100}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
recognizer.close()
15 Holistic TrackingTrack face, pose, and both hands through one result.
"""Example 15: face, pose, and hands in one pipeline."""
import cvgo as go
camera = go.Camera()
tracker = go.HolisticTracker()
while True:
frame = camera.read()
if frame is None:
break
result = tracker.detect(frame)
result.draw(frame)
if not camera.show(frame, title="CVGO Holistic Tracking"):
break
camera.close()
tracker.close()
"""CLI example 15: print face, pose, and hand status."""
import cvgo as go
camera = go.Camera()
tracker = go.HolisticTracker()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
result = tracker.detect(frame)
face = "YES" if result.face else "NO"
pose = "YES" if result.pose else "NO"
info = (
f"Face: {face} | Pose: {pose} | Hands: {len(result.hands)} | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<80}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
tracker.close()
16 Selfie SegmentationSeparate a person from the background or measure coverage.
"""Example 16: blur the webcam background."""
import cvgo as go
camera = go.Camera()
segmenter = go.SelfieSegmenter()
while True:
frame = camera.read()
if frame is None:
break
result = segmenter.segment(frame)
frame = result.blur(frame)
if not camera.show(frame, title="CVGO Selfie Segmentation"):
break
camera.close()
segmenter.close()
"""CLI example 16: print foreground coverage from segmentation."""
import cvgo as go
camera = go.Camera()
segmenter = go.SelfieSegmenter()
fps = go.FPS()
try:
while True:
frame = camera.read()
if frame is None:
break
result = segmenter.segment(frame)
coverage = result.foreground().mean() * 100
info = (
f"Person coverage: {coverage:.1f}% | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<70}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
segmenter.close()
17 Telegram SecuritySend a camera photo to Telegram when a person is detected.
"""Example 17: send a Telegram photo when a pose is detected."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker(model_complexity=0)
telegram = go.Telegram()
presence_timer = go.Timer(0.5)
notified = False
pending = None
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
alert = presence_timer.check(pose is not None)
if pending is not None and pending.done():
if not pending.result():
print(f"Telegram: {telegram.last_error}")
pending = None
if pose:
pose.box(padding=30).draw(frame, color=(0, 0, 255), label="Person")
status = "PERSON DETECTED" if alert else "SAFE"
color = (0, 0, 255) if alert else (0, 255, 0)
go.put_text(frame, f"Status: {status}", color=color)
if alert and not notified and pending is None:
pending = telegram.send_photo_async(frame, "Warning: person detected.", key="security")
notified = alert
if not camera.show(frame, title="CVGO Telegram Security"):
break
camera.close()
tracker.close()
telegram.close()
"""CLI example 17: send a Telegram photo when a pose is detected."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker(model_complexity=0)
telegram = go.Telegram()
presence_timer = go.Timer(0.5)
fps = go.FPS()
notified = False
telegram_status = "WAITING"
pending = None
try:
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
alert = presence_timer.check(pose is not None)
if pending is not None and pending.done():
sent = pending.result()
telegram_status = "SENT" if sent else "FAILED"
pending = None
if alert and not notified and pending is None:
pending = telegram.send_photo_async(
frame,
"Warning: person detected.",
key="security",
)
telegram_status = "QUEUED"
notified = True
elif not alert:
notified = False
telegram_status = "WAITING"
status = "PERSON DETECTED" if alert else "SAFE"
info = (
f"Security: {status} | Pose: {'YES' if pose else 'NO'} | "
f"Telegram: {telegram_status} | "
f"Loop FPS: {fps.read():.1f}"
)
print(f"\r{info:<110}", end="", flush=True)
if telegram_status == "FAILED":
print(f"\nTelegram: {telegram.last_error}")
telegram_status = "ERROR SHOWN"
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
tracker.close()
telegram.close()
18 Telegram Person SecuritySend a camera photo to Telegram when one or more people are detected.
"""Example 18: send a Telegram photo when people are detected."""
import cvgo as go
camera = go.Camera()
detector = go.ObjectDetector(allow=["person"])
telegram = go.Telegram()
presence_timer = go.Timer(0.5)
notified = False
pending = None
while True:
frame = camera.read()
if frame is None:
break
people = detector.detect(frame)
alert = presence_timer.check(bool(people))
if pending is not None and pending.done():
if not pending.result():
print(f"Telegram: {telegram.last_error}")
pending = None
for person in people:
person.draw(frame, color=(0, 0, 255))
status = "PEOPLE DETECTED" if alert else "SAFE"
color = (0, 0, 255) if alert else (0, 255, 0)
go.put_text(frame, f"Status: {status}", color=color)
go.put_text(frame, f"Count: {len(people)}", (20, 70))
if alert and not notified and pending is None:
pending = telegram.send_photo_async(
frame,
f"Warning: {len(people)} person(s) detected.",
key="person-security",
)
notified = alert
if not camera.show(frame, title="CVGO Telegram Person Security"):
break
camera.close()
detector.close()
telegram.close()
"""CLI example 18: send a Telegram photo when people are detected."""
import cvgo as go
camera = go.Camera()
detector = go.ObjectDetector(allow=["person"])
telegram = go.Telegram()
presence_timer = go.Timer(0.5)
fps = go.FPS()
notified = False
telegram_status = "WAITING"
pending = None
try:
while True:
frame = camera.read()
if frame is None:
break
people = detector.detect(frame)
alert = presence_timer.check(bool(people))
if pending is not None and pending.done():
sent = pending.result()
telegram_status = "SENT" if sent else "FAILED"
pending = None
if alert and not notified and pending is None:
pending = telegram.send_photo_async(
frame,
f"Warning: {len(people)} person(s) detected.",
key="person-security",
)
telegram_status = "QUEUED"
notified = True
elif not alert:
notified = False
telegram_status = "WAITING"
status = "PEOPLE DETECTED" if alert else "SAFE"
info = (
f"Security: {status} | People: {len(people)} | "
f"Telegram: {telegram_status} | "
f"Loop FPS: {fps.read():.1f}"
)
print(f"\r{info:<110}", end="", flush=True)
if telegram_status == "FAILED":
print(f"\nTelegram: {telegram.last_error}")
telegram_status = "ERROR SHOWN"
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
detector.close()
telegram.close()
19 MQTT Robot StatusPublish pose status as JSON to an MQTT robot topic.
"""Example 19: publish pose status to an MQTT robot topic."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker(model_complexity=0)
mqtt = go.MqttClient(host="localhost", client_id="cvgo-camera", connect=True)
last_detected = None
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
detected = pose is not None
if detected != last_detected:
mqtt.publish(
"robot/camera/pose",
{"person_detected": detected},
)
last_detected = detected
if pose:
pose.box(padding=30).draw(frame, label="Person")
go.put_text(frame, f"Person detected: {detected}")
if not camera.show(frame, title="CVGO MQTT Robot"):
break
camera.close()
tracker.close()
mqtt.close()
"""CLI example 19: publish pose status to an MQTT robot topic."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker(model_complexity=0)
mqtt = go.MqttClient(host="localhost", client_id="cvgo-camera", connect=True)
fps = go.FPS()
last_detected = None
try:
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
detected = pose is not None
if detected != last_detected:
mqtt.publish(
"robot/camera/pose",
{"person_detected": detected},
)
last_detected = detected
info = (
f"MQTT: {'DETECTED' if detected else 'CLEAR'} | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<70}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
tracker.close()
mqtt.close()
20 WebSocket Robot StatusSend pose status as JSON to a WebSocket robot service.
"""Example 20: send pose status to a WebSocket robot service."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker(model_complexity=0)
websocket = go.WebSocketClient("ws://localhost:8080", connect=True)
last_detected = None
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
detected = pose is not None
if detected != last_detected:
websocket.send({"person_detected": detected})
last_detected = detected
if pose:
pose.box(padding=30).draw(frame, label="Person")
go.put_text(frame, f"Person detected: {detected}")
if not camera.show(frame, title="CVGO WebSocket Robot"):
break
camera.close()
tracker.close()
websocket.close()
"""CLI example 20: send pose status to a WebSocket robot service."""
import cvgo as go
camera = go.Camera()
tracker = go.PoseTracker(model_complexity=0)
websocket = go.WebSocketClient("ws://localhost:8080", connect=True)
fps = go.FPS()
last_detected = None
try:
while True:
frame = camera.read()
if frame is None:
break
pose = tracker.detect(frame)
detected = pose is not None
if detected != last_detected:
websocket.send({"person_detected": detected})
last_detected = detected
info = (
f"WebSocket: {'DETECTED' if detected else 'CLEAR'} | "
f"FPS: {fps.read():.1f}"
)
print(f"\r{info:<70}", end="", flush=True)
except KeyboardInterrupt:
pass
finally:
print()
camera.close()
tracker.close()
websocket.close()
Built to be studied and changed.
CVGO is released under the MIT License. The visible camera loops and decisions can grow into learning projects, security tools, or a complete driver-monitoring final project.