Security Camera Example

import asyncio
import logging
from typing import Any, Dict

import numpy as np
from dotenv import load_dotenv
from poster_generator import generate_and_post_poster
from security_camera_processor import (
    PackageDetectedEvent,
    PackageDisappearedEvent,
    PersonDetectedEvent,
    PersonDisappearedEvent,
    SecurityCameraProcessor,
)
from vision_agents.core import Agent, Runner, User
from vision_agents.core.agents import AgentLauncher
from vision_agents.plugins import deepgram, elevenlabs, gemini, getstream

load_dotenv()

logger = logging.getLogger(__name__)

PACKAGE_THEFT_DELAY_SECONDS = 3.0
_pending_theft_tasks: Dict[str, asyncio.Task] = {}
_package_history: Dict[str, Dict[str, Any]] = {}


async def create_agent(**kwargs) -> Agent:
    llm = gemini.LLM("gemini-2.5-flash-lite")

    security_processor = SecurityCameraProcessor(
        fps=5,
        time_window=1800,  # 30 minutes
        thumbnail_size=80,
        detection_interval=2.0,
        bbox_update_interval=0.3,
        model_path="weights_custom.pt",
        package_conf_threshold=0.7,
        max_tracked_packages=1,
    )

    agent = Agent(
        edge=getstream.Edge(),
        agent_user=User(name="Security AI", id="agent"),
        instructions="Read @instructions.md",
        processors=[security_processor],
        llm=llm,
        tts=elevenlabs.TTS(),
        stt=deepgram.STT(eager_turn_detection=True),
    )

    # Merge processor events with agent events
    agent.events.merge(security_processor.events)

    # Register LLM functions
    @llm.register_function(
        description="Get the number of unique visitors detected in the last 30 minutes."
    )
    async def get_visitor_count() -> Dict[str, Any]:
        count = security_processor.get_visitor_count()
        state = security_processor.state()
        return {
            "unique_visitors": count,
            "total_detections": state["total_face_detections"],
            "time_window": f"{state['time_window_minutes']} minutes",
        }

    @llm.register_function(
        description="Register the current person's face with a name so they can be recognized in the future."
    )
    async def remember_my_face(name: str) -> Dict[str, Any]:
        return security_processor.register_current_face_as(name)

    # Subscribe to events
    @agent.events.subscribe
    async def on_person_detected(event: PersonDetectedEvent):
        if event.is_new:
            agent.logger.info(f"🚨 NEW PERSON ALERT: {event.face_id} detected!")
        else:
            agent.logger.info(f"👤 Returning visitor: {event.face_id}")
            await agent.say(f"Welcome back, {event.face_id}!")

    @agent.events.subscribe
    async def on_package_disappeared(event: PackageDisappearedEvent):
        picker_display = event.picker_name or (
            event.picker_face_id[:8] if event.picker_face_id else "unknown"
        )
        
        async def delayed_theft_check():
            await asyncio.sleep(PACKAGE_THEFT_DELAY_SECONDS)
            del _pending_theft_tasks[event.package_id]
            
            if event.package_id in _package_history:
                _package_history[event.package_id]["picked_up_by"] = picker_display
            
            if event.picker_face_id:
                face_image = security_processor.get_face_image(event.picker_face_id)
                if face_image is not None:
                    await handle_package_theft(
                        agent, face_image, picker_display, security_processor
                    )
        
        _pending_theft_tasks[event.package_id] = asyncio.create_task(
            delayed_theft_check()
        )

    return agent


async def handle_package_theft(
    agent: Agent,
    face_image: np.ndarray,
    suspect_name: str,
    processor: SecurityCameraProcessor,
) -> None:
    await agent.say(
        f"Alert! Package stolen by {suspect_name}! Generating wanted poster."
    )

    poster_bytes, tweet_url = await generate_and_post_poster(
        face_image,
        suspect_name,
        post_to_x_enabled=True,
        tweet_caption=f'🚨 WANTED: {suspect_name} caught "stealing" a package!',
    )

    if poster_bytes:
        processor.share_image(poster_bytes, duration=8.0)
        await agent.say("Here's the wanted poster for the package thief!")
        
        if tweet_url:
            await agent.say("Wanted poster also posted to X!")


async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
    call = await agent.create_call(call_type, call_id)
    async with agent.join(call):
        await agent.finish()


if __name__ == "__main__":
    Runner(AgentLauncher(create_agent=create_agent, join_call=join_call)).cli()

PACKAGE_THEFT_DELAY_SECONDS = 3.0

@agent.events.subscribe
async def on_package_disappeared(event: PackageDisappearedEvent):
    async def delayed_theft_check():
        await asyncio.sleep(PACKAGE_THEFT_DELAY_SECONDS)
        # Confirm package is gone and trigger workflow
        ...
    
    _pending_theft_tasks[event.package_id] = asyncio.create_task(
        delayed_theft_check()
    )

@llm.register_function(
    description="Get the number of unique visitors detected in the last 30 minutes."
)
async def get_visitor_count() -> Dict[str, Any]:
    count = security_processor.get_visitor_count()
    return {"unique_visitors": count, ...}

@llm.register_function(
    description="Register the current person's face with a name."
)
async def remember_my_face(name: str) -> Dict[str, Any]:
    return security_processor.register_current_face_as(name)

# Also available:
# - get_visitor_details()
# - get_package_count()
# - get_package_details()
# - get_activity_log()
# - get_known_faces()

security_processor = SecurityCameraProcessor(
    fps=5,  # Frames per second to process
    time_window=1800,  # Time window in seconds (30 min)
    thumbnail_size=80,  # Thumbnail size in pixels
    detection_interval=2.0,  # Seconds between full face detection
    bbox_update_interval=0.3,  # Seconds between bbox updates
    model_path="weights_custom.pt",  # YOLO model path
    package_conf_threshold=0.7,  # Package detection confidence
    max_tracked_packages=1,  # Single-package mode
    face_match_tolerance=0.6,  # Face matching tolerance (lower = stricter)
)

@agent.events.subscribe
async def on_person_detected(event: PersonDetectedEvent):
    # event.face_id, event.name, event.is_new, event.detection_count
    ...

@agent.events.subscribe
async def on_person_disappeared(event: PersonDisappearedEvent):
    # event.face_id, event.name
    ...

@agent.events.subscribe
async def on_package_detected(event: PackageDetectedEvent):
    # event.package_id, event.is_new, event.confidence
    ...

@agent.events.subscribe
async def on_package_disappeared(event: PackageDisappearedEvent):
    # event.package_id, event.picker_face_id, event.picker_name
    ...