# Project export: Netra AI 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: Cal Hacks 11.0 - Tagline: Meet Netra, the next generation AI-powered video stream security app that transforms the way you monitor your space. Netra allows for instant querying of events and gives real time alerts of events. - Devpost: https://devpost.com/software/netra-ai - GitHub: https://github.com/devangsharmadj/netra - Video: https://www.youtube.com/embed/wxnFaQ2-1sA?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 1 GitHub contributor(s) — Devang Sharma (1 commits) ## Devpost submission (written by the team) ### Inspiration Vibha (one of our teammates) was in India last summer visiting her grandparents. Someone stole something in front of her grandma's house, so she was asked through look through hours of footage to determine what exactly happened. While brainstorming, the team decided that AI could be used to address this issue and that's how Netra AI was born. ### What it does Netra AI has two main functionalities. The first one is that the user can query for events. For example, the user can ask if a person came to their house and if so, what time. The second functionality is that users can get alerts on events with detailed descriptions. ### How we built it For the frontend and backend, we used the Reflex framework. To analyze the video files we used Gemini. For the speech to text feature in the chatbot, we used DeepGram. To detect the motion and record the 5 second clip, we used OpenCV. We used ChromaDB to store the alert message produced from the processing. ### Challenges we ran into One challenge that we had was learning the Reflex framework. We wanted to display the alert messages on the website, but we were not able to do so. We also had issues with the motion detection being too sensitive, but we were able to fix it by changing the sensitivity. ### Accomplishments we're proud of One thing we are proud of is being able to successfully implement the speech to text feature. We felt that this feature was important because it would be useful for people on the go to easily communicate with the system. Additionally, we wanted to make our system more accessible. Vibha's grandma struggles to type on the keyboard, so in this case a speech feature would be useful for people like her. ### What we learned Through this experience, we learned many technologies from the sponsors like Gemini, ChromaDB, Reflex, and DeepGram which we used in our project. We also learned how to work better as a team. We were able to make progress by delegating tasks. When we felt stuck, we swapped computers with each other to get a fresh set of eyes. This proved to be very valuable as we were able to resolve many issues this way. ### What's next 1) Real Time Text Alerts with Detailed Descriptions of Important Events Text user that, for example “Your daughter has arrived home” or “There is a fire outside your home.” 2) More Training of the AI Model Train the AI model on video data from different countries and people. 3) Improving Security and Privacy Implement Multi-Factor Authentication and Encryption ## README (from the GitHub repository) Reflex x LLamaIndex ## UI For Llama Deploy Follow the tutorial here: https://github.com/run-llama/llama_deploy/tree/main/examples/python_fullstack ## Detected evidence (automated analysis) Indexed codebase: 16 recognized source files, 41 KB. - OpenAI (technology) — detected in the code - Python (language) — detected in the code - CSS (language) — claimed on Devpost, not found in the code - Google Gemini (technology) — claimed on Devpost, not found in the code - HTML (language) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (19 of 19) ``` .gitignore dockerfile frontend/__init__.py frontend/components/badge.py frontend/components/hint.py frontend/components/reset.py frontend/components/settings.py frontend/frontend.py frontend/speechreflex.py frontend/speechtotext.py frontend/state.py frontend/style.py frontend/views/chat.py frontend/views/templates.py frontend/views/video.py frontend/webcam.py README.md requirements.txt rxconfig.py ``` ### Dependencies - requirements.txt: chromadb@==0.5.15, google-generativeai@==0.8.3, greenlet@==3.1.1, grpcio-status@==1.67.0, httptools@==0.6.4, llama-deploy@==0.1.3, openai@==1.52.0, opencv-python@==4.10.0.84, PyAutoGUI@==0.9.53, pygame@==2.5.2, reflex-audio-capture@==0.0.4, reflex-img-comparison-slider@==0.0.1, reflex-webcam@==0.0.6, replicate@==0.26.0, uvloop@==0.21.0 ### Recent commits (newest first) - first commit - Merge pull request #2 from reflex-dev/alek/up - Upgate - Update state.py - Update README.md - Update README.md - Merge pull request #1 from carlosabadia/add-template - update - add template - Create README.md ## Key source files (fetched from GitHub, selected and truncated for size) ### requirements.txt ``` chromadb==0.5.15 google-generativeai==0.8.3 greenlet==3.1.1 grpcio-status==1.67.0 httptools==0.6.4 llama-deploy==0.1.3 openai==1.52.0 opencv-python==4.10.0.84 PyAutoGUI==0.9.53 pygame==2.5.2 reflex-audio-capture==0.0.4 reflex-img-comparison-slider==0.0.1 reflex-webcam==0.0.6 replicate==0.26.0 uvloop==0.21.0 ``` ### dockerfile ``` # This Dockerfile is used to deploy a simple single-container Reflex app instance. FROM python:3.10-slim # Copy local context to `/app` inside container (see .dockerignore) WORKDIR /app COPY . . # Install app requirements and reflex in the container # Deploy templates and prepare app # Download all npm dependencies and compile frontend RUN apt-get clean && apt-get update \ && apt-get --no-install-recommends install zip unzip curl -y \ && pip install -r requirements.txt \ && reflex export --frontend-only --no-zip # Needed until Reflex properly passes SIGTERM on backend. STOPSIGNAL SIGKILL # Always apply migrations before starting the backend. CMD [ -d alembic ] && reflex db migrate; reflex run --env prod ``` ### rxconfig.py ```python import reflex as rx from frontend.style import create_colors_dict config = rx.Config( app_name="frontend", api_url="http://localhost:9000", backend_port=9000, deployment_name="deployment", tailwind={ "darkMode": "class", "theme": { "colors": { **create_colors_dict(), }, }, }, ) ``` ### frontend/speechtotext.py ```python import requests from datetime import datetime # Set your Deepgram API key API_KEY = '65a890e9867c3dda76519eb728d08d69547761c6' URL = 'https://api.deepgram.com/v1/listen?model=nova-2&smart_format=true' # Prepare the headers headers = { 'Authorization': f'Token {API_KEY}' } # Get the audio file from the user # audio_file_path = input("Enter the path to the audio file you want to upload: ") # Open the audio file in binary mode with open(f'{datetime.now()}', 'rb') as audio_file: # Send the POST request with the audio file response = requests.post(URL, headers=headers, data=audio_file) # Check if the request was successful if response.status_code == 200: # Parse the JSON response transcript_data = response.json() # Extract the transcription text transcript = transcript_data['results']['channels'][0]['alternatives'][0]['transcript'] # Print the transcribed sentence print("Transcription result:") print(transcript) else: # If there's an error, print the error message print("Error:", response.status_code) print("Message:", response.text) ``` ### frontend/style.py ```python # style.py from reflex.constants.colors import ColorType STYLESHEETS = [ "https://fonts.googleapis.com/css2?family=Instrument+Sans:ital,wght@0,400;0,500;0,600;0,700;1,400;1,500;1,600;1,700&family=Poppins:ital,wght@0,400;0,500;0,600;0,700;1,400;1,500;1,600;1,700&family=Inter:wght@400;500;600;700&family=Roboto:ital,wght@0,400;0,500;0,700;1,400;1,500;1,700&family=Open+Sans:ital,wght@0,400;0,600;0,700;1,400;1,600;1,700&family=Lato:ital,wght@0,400;0,700;1,400;1,700&display=swap" ] # Default Radix Colors def create_colors_dict() -> dict: colors_dict = {} for color in ColorType.__args__: if color not in ["black", "white"]: colors_dict[color] = { shade: f"var(--{color}-{shade})" for shade in range(1, 13) } # Append the alpha colors colors_dict[f"{color}A"] = { shade: f"var(--{color}-a{shade})" for shade in range(1, 13) } # Add accent palette colors_dict["accent"] = {shade: f"var(--accent-{shade})" for shade in range(1, 13)} colors_dict["accentA"] = { shade: f"var(--accent-a{shade})" for shade in range(1, 13) } return colors_dict ``` ### frontend/frontend.py ```python import reflex as rx from frontend import style from frontend.state import SettingsState from frontend.components.settings import settings_icon from frontend.components.reset import reset from frontend.views.templates import templates from frontend.views.chat import chat, action_bar from frontend.views.video import video from frontend.speechreflex import audio # from frontend.state import GeminiStarter def index() -> rx.Component: return rx.theme( rx.el.style( f""" :root {{ --font-family: "{SettingsState.font_family}", sans-serif; }} """ ), # Top bar with the reset and settings buttons rx.box( reset(), settings_icon(), class_name="top-4 right-4 absolute flex flex-row items-center gap-3.5", ), # Main content rx.vstack( video(), # Video rx.box( templates() ), rx.box( rx.scroll_area( chat(), action_bar(), ) ), # # Chat history # # Action bar # Prompt examples # class_name="relative flex flex-col justify-between gap-20 mx-auto px-6 pt-16 lg:pt-6 pb-6 max-w-4xl h-screen", align="center" ), accent_color=SettingsState.color, ) app = rx.App() app = rx.App(stylesheets=style.STYLESHEETS, style={"font_family": "var(--font-family)"}) app.add_page( index, title="Netra", description="Your AI security assistant" ) ``` ### frontend/speechreflex.py ```python from urllib.request import urlopen import reflex as rx import requests from reflex_audio_capture import AudioRecorderPolyfill, get_codec, strip_codec_part API_KEY = '65a890e9867c3dda76519eb728d08d69547761c6' URL = 'https://api.deepgram.com/v1/listen?model=nova-2&smart_format=true' # Prepare the headers headers = { 'Authorization': f'Token {API_KEY}' } # from openai import AsyncOpenAI # client = AsyncOpenAI() REF = "myaudio" class Audio(rx.State): """The app state.""" has_error: bool = False processing: bool = False transcript: list[str] = [] timeslice: int = 0 device_id: str = "" use_mp3: bool = True async def on_data_available(self, chunk: str): mime_type, _, codec = get_codec(chunk).partition(";") audio_type = mime_type.partition("/")[2] if audio_type == "mpeg": audio_type = "mp3" print(len(chunk), mime_type, codec, audio_type) with urlopen(strip_codec_part(chunk)) as audio_data: # print(type(audio_data)) with open("output_audio.mp3", "wb") as audio_file: audio_file.write(audio_data.read()) with open(f'output_audio.mp3', 'rb') as audio_file: # Send the POST request with the audio file response = requests.post(URL, headers=headers, data=audio_file) # Check if the request was successful if response.status_code == 200: # Parse the JSON response transcript_data = response.json() # Extract the transcription text transcript = transcript_data['results']['channels'][0]['alternatives'][0]['transcript'] # Print the transcribed sentence print("Transcription result:") print(transcript) else: # If there's an error, print the error message print("Error:", response.status_code) print("Message:", response.text) # try: # self.processing = True # yield # transcription = await client.audio.transcriptions.create( # model="whisper-1", # file=("temp." + audio_type, audio_data.read(), mime_type), # ) # except Exception as e: # self.has_error = True # yield capture.stop() # raise # finally: # self.processing = False self.transcript.append(transcript) def set_timeslice(self, value): self.timeslice = value[0] def set_device_id(self, value): self.device_id = value yield capture.stop() def on_error(self, err): print(err) def on_load(self): # We can start the recording immediately when the page loads return capture.start() capture = AudioRecorderPolyfill.create( id=REF, on_data_available=Audio.on_data_available, on_error=Audio.on_error, timeslice=Audio.timeslice, device_id=Audio.device_id, use_mp3=Audio.use_mp3, ) def input_device_select(): return rx.select.root( rx.select.trigger(placeholder="Select Input Device"), rx.select.content( rx.foreach( capture.media_devices, lambda device: rx.cond( device.deviceId & device.kind == "audioinput", rx.select.item(device.label, value=device.deviceId), ), ), ), on_change=Audio.set_device_id, ) def audio() -> rx.Component: return rx.container( rx.vstack( capture, rx.cond( capture.is_recording, rx.button("Stop Recording", on_click=capture.stop()), rx.button( rx.icon(tag="mic"), on_click=capture.start(), ), ), style={"width": "100%", "> *": {"width": "100%"}}, ), size="1", margin_y="2em", ) # Add state and page to the app. app = rx.App() app.add_page(audio) ``` ### frontend/state.py ```python import asyncio import json import os import uuid import time import httpx import reflex as rx from openai import AsyncOpenAI import google.generativeai as genai import json # def gemini_starter(): GEMINI_API_KEY = 'AIzaSyBjWprVJ6UMCQXHE4GnO7OapYn1r_0ejak' genai.configure(api_key=GEMINI_API_KEY) def upload_to_gemini(path, mime_type=None): """Uploads the given file to Gemini.""" file = genai.upload_file(path, mime_type=mime_type) print(f"Uploaded file '{file.display_name}' as: {file.uri}") return file def wait_for_files_active(files): """Waits for the given files to be active.""" print("Waiting for file processing...") for name in (file.name for file in files): file = genai.get_file(name) while file.state.name == "PROCESSING": print(".", end="", flush=True) time.sleep(10) file = genai.get_file(name) if file.state.name != "ACTIVE": raise Exception(f"File {file.name} failed to process") print("...all files ready") print() # Create the model generation_config = { "temperature": 1, "top_p": 0.95, "top_k": 40, "max_output_tokens": 8192, "response_mime_type": "application/json", } model = genai.GenerativeModel( model_name="gemini-1.5-pro-002", generation_config=generation_config, ) # Upload your video file files = [ upload_to_gemini(r"./assets/calhacks.mp4", mime_type="video/mp4"), ] # Wait for the files to be processed wait_for_files_active(files) chat_session = model.start_chat( history=[ { "role": "user", "parts": [ files[0], ], }, ] ) class SettingsState(rx.State): # The accent color for the app color: str = "violet" # The font family for the app font_family: str = "Poppins" class State(rx.State): # The current question being asked. question: str # Whether the app is processing a question. processing: bool = False # Keep track of the chat history as a list of (question, answer) tuples. chat_history: list[tuple[str, str]] = [] user_id: str = str(uuid.uuid4()) async def answer(self): # Set the processing state to True. self.processing = True yield # convert chat history to a list of dictionaries chat_history_dicts = [] for chat_history_tuple in self.chat_history: chat_history_dicts.append( {"role": "user", "content": chat_history_tuple[0]} ) chat_history_dicts.append( {"role": "assistant", "content": chat_history_tuple[1]} ) self.chat_history.append((self.question, "")) # Clear the question input. question = self.question self.question = "" # Yield here to clear the frontend input before continuing. yield # This is where I am calling the gemini api prompt = f"{question}. Give the response with the key being 'details'. Please give the response in a paragraph format. Only provide it in a detailed response. Provide a detailed and descriptive explanation in natural language." # Send the modified prompt to the model response = chat_session.send_message(prompt) # client = httpx.AsyncClient() # # call the agentic workflow # input_payload = { # "chat_history_dicts": chat_history_dicts, # "user_input": question, # } # deployment_name = os.environ.get("DEPLOYMENT_NAME", "MyDeployment") # apiserver_url = os.environ.get("APISERVER_URL", "http://localhost:4501") # response = await client.post( # f"{apiserver_url}/deployments/{deployment_name}/tasks/create", # json={"input": json.dumps(input_payload)}, # timeout=60, # ) response_text = response.text response_data = json.loads(response_text) details = response_data.get('details') answer = details for i in range(len(answer)): # Pause to show the streaming effect. await asyncio.sleep(0.01) # Add one letter at a time to the output. self.chat_history[-1] = ( self.chat_history[-1][0], answer[: i + 1], ) yield # Add to the answer as the chatbot responds. answer = "" yield # async for item in session: # if hasattr(item.choices[0].delta, "content"): # if item.choices[0].delta.content is None: # break # answer += item.choices[0].delta.content # self.chat_history[-1] = (self.chat_history[-1][0], answer) # yield # Ensure the final answer is added to chat history if answer: self.chat_history[-1] = (self.chat_history[-1][0], answer) yield # Set the processing state to False. self.processing = False async def handle_key_down(self, key: str): if key == "Enter": async for t in self.answer(): yield t def clear_chat(self): # Reset the chat history and processing state self.chat_history = [] self.processing = False ``` ### frontend/webcam.py ```python # """Take screenshots and video recordings from webcam.""" # import time # from pathlib import Path # from urllib.request import urlopen # from PIL import Image # import reflex as rx # import reflex_webcam as webcam # # Identifies a particular webcam component in the DOM # WEBCAM_REF = "webcam" # VIDEO_FILE_NAME = "video.webm" # # The path containing the app # APP_PATH = Path(__file__) # APP_MODULE_DIR = APP_PATH.parent # SOURCE_CODE = [ # APP_MODULE_DIR.parent.parent / "custom_components/reflex_webcam/webcam.py", # APP_PATH, # APP_MODULE_DIR.parent / "requirements.txt", # ] # # Mark Upload as used so StaticFiles can get mounted on /_upload # rx.upload() # class State(rx.State): # last_screenshot: Image.Image | None = None # last_screenshot_timestamp: str = "" # loading: bool = False # recording: bool = False # def handle_screenshot(self, img_data_uri: str): # """Webcam screenshot upload handler. # Args: # img_data_uri: The data uri of the screenshot (from upload_screenshot). # """ # if self.loading: # return # self.last_screenshot_timestamp = time.strftime("%H:%M:%S") # with urlopen(img_data_uri) as img: # self.last_screenshot = Image.open(img) # self.last_screenshot.load() # # convert to webp during serialization for smaller size # self.last_screenshot.format = "WEBP" # type: ignore # def _video_path(self) -> Path: # return Path(rx.get_upload_dir()) / VIDEO_FILE_NAME # @rx.var(cache=True) # def video_exists(self) -> bool: # if not self.recording: # return self._video_path().exists() # return False # def on_start_recording(self): # self.recording = True # print("Started recording") # with self._video_path().open("wb") as f: # f.write(b"") # def _strip_codec_part(self, chunk: str) -> str: # parts = chunk.split(";") # for part in parts: # if "codecs=" in part: # parts.remove(part) # break # return ";".join(parts) # def handle_video_chunk(self, chunk: str): # print("Got video chunk", len(chunk)) # with self._video_path().open("ab") as f: # with urlopen(self._strip_codec_part(chunk)) as vid: # f.write(vid.read()) # def on_stop_recording(self): # print(f"Stopped recording: {self._video_path()}") # self.recording = False # def start_recording(self, ref: str): # """Start recording a video.""" # return webcam.start_recording( # ref, # on_data_available=State.handle_video_chunk, # on_start=State.on_start_recording, # on_stop=State.on_stop_recording, # timeslice=1000, # ) # def last_screenshot_widget() -> rx.Component: # """Widget for displaying the last screenshot and timestamp.""" # return rx.box( # rx.cond( # State.last_screenshot, # rx.fragment( # rx.image(src=State.last_screenshot), # rx.text(State.last_screenshot_timestamp), # ), # rx.center( # rx.text("Click image to capture.", size="4"), # ), # ), # height="270px", # ) # def webcam_upload_component(ref: str) -> rx.Component: # """Component for displaying webcam preview and uploading screenshots. # Args: # ref: The ref of the webcam component. # Returns: # A reflex component. # """ # return rx.vstack( # webcam.webcam( # id=ref, # on_click=webcam.upload_screenshot( # ref=ref, # handler=State.handle_screenshot, # type: ignore # ), # audio=True, # ), # rx.cond( # ~State.recording, # rx.button( # "🟢 Start Recording", # on_click=State.start_recording(ref), # color_scheme="green", # size="4", # ), # rx.button( # "🟤 Stop Recording", # on_click=webcam.stop_recording(ref), # color_scheme="tomato", # size="4", # ), # ), # rx.cond( # State.video_exists, # rx.link( # "Download Last Video", href=rx.get_upload_url(VIDEO_FILE_NAME), size="4" # ), # ), # last_screenshot_widget(), # width="320px", # align="center", # ) # def index() -> rx.Component: # return rx.fragment( # rx.color_mode.button(position="top-right"), # rx.center( # webcam_upload_component(WEBCAM_REF), # padding_top="3em", # ), # *[ # rx.vstack( # rx.heading(f"Source Code: {p.name}"), # rx.code_block( # p.read_text(), # language="python", # width="90%", # overflow_x="auto", # ), # margin_top="5em", # padding_x="1em", # width="100vw", # align="center", # ) # for p in SOURCE_CODE # ], # ) # app = rx.App() # app.add_page(index) """Reflex custom component Webcam.""" from __future__ import annotations from typing import Any, List import reflex as rx from reflex.vars import Var class Webcam(rx.Component): """Wrapper for react-webcam component.""" # The React library to wrap. library = "react-webcam" # The React component tag. tag = "Webcam" # If the tag is the default export from the module, you can set is_default = True. # This is normally used when components don't have curly braces around them when importing. is [truncated — 4343 more characters] ``` ### frontend/views/video.py ```python import reflex as rx def video() -> rx.Component: return rx.box( rx.video( url="/calhacks.mp4", controls=True, #position="relative", # width="50%", # height="100%", border="none", ), margin_bottom="2rem", ) ``` [6 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]