The AI Newscaster: Zero-Dependency News with Arduino Uno Q

# =============================================================================
# main.py  MPU Controller for AI Newscaster
# =============================================================================
# Runs on the Qualcomm QRB2210 MPU (Linux) side of the Arduino UNO Q.
# Handles: news fetching (RSS), article formatting, face detection, 
# and Bridge RPC communication with the MCU.
#
# NOTE: Pressing SELECT will trigger a face presence scan via the 
# VideoObjectDetection brick. Once a face is detected, it will prompt 
# the user to select a category, then fetch news for that category.
#
# Bridge RPC Protocol:
#   Provides to MCU:
#     get_display_data()  Returns pipe-delimited display data string
#   Receives from MCU:
#     on_scan_pressed, on_category_select:<idx>, on_story_nav:<dir>, on_back
# =============================================================================

import os
import json
import time
import threading
import urllib.request
import xml.etree.ElementTree as ET

from arduino.app_utils import Bridge, App

try:
    from arduino.app_bricks.video_objectdetection import VideoObjectDetection
except ImportError:
    print("[WARN] VideoObjectDetection brick not available.")
    VideoObjectDetection = None
# =============================================================================
# Configuration
# =============================================================================

# File paths (on MPU filesystem)
BASE_DIR    = os.path.dirname(os.path.abspath(__file__))
CONFIG_PATH = os.path.join(BASE_DIR, "config.json")
# Google News RSS categories (matching the MCU's category labels)
CATEGORIES = [
    "business", "entertainment", "general",
    "health", "science", "sports", "technology"
]

# =============================================================================
# Global State
# =============================================================================

# System state  determines what the MCU displays
current_state = "idle"

# Display data  pipe-delimited string sent to MCU
display_data = "state:idle"

# Configuration
config = {}

# Currently selected category
selected_category_idx = 0

# Briefing data
briefing_stories = []  # List of {headline, summary, category}
current_story_idx = 0

# Thread lock for state changes
state_lock = threading.Lock()

# Camera index
CAMERA_INDEX = 0

# Video Object Detector
video_detector = None
is_scanning = False
face_detected_event = threading.Event()
if VideoObjectDetection:
    try:
        video_detector = VideoObjectDetection(confidence=0.4, debounce_sec=1.5)
    except Exception as e:
        print(f"[WARN] Failed to initialize VideoObjectDetection: {e}")

# =============================================================================
# Configuration & Persistence
# =============================================================================

def load_config():
    """Load API keys and settings from config.json."""
    global config, CAMERA_INDEX
    if not os.path.exists(CONFIG_PATH):
        print(f"[WARN] Config file not found at {CONFIG_PATH}")
        print("Creating template config.json...")
        template = {
            "camera_index": 0,
            "news_country": "US"
        }
        with open(CONFIG_PATH, 'w') as f:
            json.dump(template, f, indent=2)
        print(f"[INFO] Template config.json created at {CONFIG_PATH}")
        config = template
        return True  # Continue with defaults

    with open(CONFIG_PATH, 'r') as f:
        config = json.load(f)

    CAMERA_INDEX = config.get("camera_index", 0)
    return True



# =============================================================================
# State Management
# =============================================================================

def set_state(new_state, data_fields=None):
    """
    Update the system state and display data.
    Thread-safe. The MCU polls get_display_data() to receive this.
    """
    global current_state, display_data

    with state_lock:
        current_state = new_state
        parts = [f"state:{new_state}"]
        if data_fields:
            for key, value in data_fields.items():
                parts.append(f"{key}:{value}")
        display_data = "|".join(parts)
        print(f"[STATE]  {new_state}")

# =============================================================================
# News Fetching
# =============================================================================

def fetch_news(topics):
    """
    Fetch top headlines using Google News RSS for the given topic categories.
    """
    country = config.get("news_country", "US")
    all_articles = []

    rss_topic_map = {
        "business": "BUSINESS",
        "entertainment": "ENTERTAINMENT",
        "general": "WORLD",
        "health": "HEALTH",
        "science": "SCIENCE",
        "sports": "SPORTS",
        "technology": "TECHNOLOGY"
    }

    for topic in topics:
        try:
            gnews_topic = rss_topic_map.get(topic.lower(), "WORLD")
            url = f"https://news.google.com/rss/headlines/section/topic/{gnews_topic}?hl=en-{country}&gl={country}&ceid={country}:en"

            req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
            with urllib.request.urlopen(req) as response:
                xml_data = response.read()

            root = ET.fromstring(xml_data)
            channel = root.find("channel")

            if channel is not None:
                items = channel.findall("item")[:5]
                for item in items:
                    title_elem = item.find("title")
                    desc_elem = item.find("description")
                    source_elem = item.find("source")

                    title = title_elem.text if title_elem is not None and title_elem.text else "Untitled"
                    desc = desc_elem.text if desc_elem is not None and desc_elem.text else ""
                    source = source_elem.text if source_elem is not None and source_elem.text else ""

                    all_articles.append({
                        "title": title,
                        "description": desc,
                        "source": source,
                        "category": topic
                    })
                print(f"[NEWS] Fetched {len(items)} articles for '{topic}'")
            else:
                print(f"[NEWS] RSS error for '{topic}' (no channel)")

        except Exception as e:
            print(f"[NEWS] Error fetching '{topic}': {e}")

    return all_articles

# =============================================================================
# News Formatting
# =============================================================================

def summarize_news(articles):
    """
    Return raw news stories (no AI summary) but utilize the full 250-char space.
    """
    stories = []
    if not articles:
        return stories

    print(f"[NEWS] Extracting {len(articles)} articles...")

    import re
    import html
    for i, article in enumerate(articles):
        desc = article.get("description", "No summary available.")
        desc = html.unescape(re.sub(r'<[^>]+>', '', desc))
        
        stories.append({
            "headline": (article.get("title", "Untitled"))[:80],
            "summary": desc[:250],
            "category": article.get("category", "").title()
        })
        print(f"  -> Formatted article {i+1}/{len(articles)}")

    return stories

# =============================================================================
# Scan & Category Selection
# =============================================================================

def on_scan_pressed():
    """User pressed SELECT in IDLE  trigger scan, then go to category selection."""
    # Prevent multiple threads from starting
    if current_state != "idle":
        return True

    def scan_thread():
        global briefing_stories, current_story_idx, is_scanning

        set_state("scanning")

        if video_detector:
            print("[SCAN] Waiting for face...")
            is_scanning = True
            face_detected_event.clear()
            
            # Wait up to 5 seconds for the camera to detect something
            found = face_detected_event.wait(timeout=5.0)
            is_scanning = False
            
            if not found:
                print("[SCAN] No face detected. Returning to idle.")
                set_state("error", {"message": "No face detected"})
                time.sleep(2)
                set_state("idle")
                return
            else:
                print("[SCAN] Face detected! Proceeding to fetch...")
        else:
            print("[SCAN] Video detector not available, skipping scan.")
            time.sleep(1)

        set_state("pick_category")

    thread = threading.Thread(target=scan_thread, daemon=True)
    thread.start()
    return True


def on_category_select(args):
    """User selected a category  fetch news for it."""
    global selected_category_idx
    try:
        selected_category_idx = int(args)
    except ValueError:
        selected_category_idx = 0
        
    if 0 <= selected_category_idx < len(CATEGORIES):
        selected_category = CATEGORIES[selected_category_idx]
    else:
        selected_category = "general"
        
    def fetch_thread():
        global briefing_stories, current_story_idx
        set_state("fetching")
        
        articles = fetch_news([selected_category])
        if not articles:
            set_state("error", {"message": "No news found. Check WiFi."})
            return
            
        stories = summarize_news(articles)
        if stories:
            briefing_stories = stories
            current_story_idx = 0
            set_state("briefing", format_story_data(0))
        else:
            set_state("error", {"message": "News service unavailable."})
            
    thread = threading.Thread(target=fetch_thread, daemon=True)
    thread.start()
    return True


def on_story_nav(args):
    """User navigated stories (up/down)."""
    global current_story_idx

    if args == "up" and current_story_idx > 0:
        current_story_idx -= 1
    elif args == "down" and current_story_idx < len(briefing_stories) - 1:
        current_story_idx += 1

    set_state("briefing", format_story_data(current_story_idx))
    return True


def on_back():
    """User pressed BACK  return to idle from most states."""
    set_state("idle")
    return True




# =============================================================================
# Display Data Formatting
# =============================================================================

def format_story_data(story_idx):
    """Format a story for the MCU display data protocol."""
    if not briefing_stories or story_idx >= len(briefing_stories):
        return {
            "story_idx": "0",
            "story_count": "0",
            "headline": "No stories available",
            "category": ""
        }

    story = briefing_stories[story_idx]
    
    def clean_text(text, max_len):
        text = text.replace("\n", " ").replace("\r", " ").replace("|", "-")
        return text[:max_len]

    return {
        "story_idx": str(story_idx),
        "story_count": str(len(briefing_stories)),
        "headline": clean_text(story["headline"], 80),
        "category": clean_text(story.get("category", ""), 15)
    }




# =============================================================================
# Bridge RPC Providers
# =============================================================================

def get_display_data():
    """Returns the current display data string. Called by MCU polling."""
    return display_data

def get_summary_data():
    """Returns JUST the summary for the current story to bypass RPC limits."""
    if current_state != "briefing" or not briefing_stories:
        return ""
    if current_story_idx >= len(briefing_stories):
        return ""
    
    story = briefing_stories[current_story_idx]
    text = story.get("summary", "")
    text = text.replace("\n", " ").replace("\r", " ").replace("|", "-")
    return text[:250]

# =============================================================================
# Bridge Registration
# =============================================================================

def register_bridge_handlers():
    """Register all Bridge RPC handlers."""

    # Simple handlers (no arguments)
    Bridge.provide("on_scan_pressed",           on_scan_pressed)
    Bridge.provide("on_back",                   on_back)

    # Data provider
    Bridge.provide("get_display_data",          get_display_data)
    Bridge.provide("get_summary_data",          get_summary_data)

    # Parameterized handlers
    for i in range(len(CATEGORIES)):
        name = f"on_category_select:{i}"
        Bridge.provide(name, lambda idx=i: on_category_select(str(idx)))

    Bridge.provide("on_story_nav:up",   lambda: on_story_nav("up"))
    Bridge.provide("on_story_nav:down", lambda: on_story_nav("down"))

# =============================================================================
# Main Entry Point
# =============================================================================

def main():
    print("=" * 60)
    print("  AI NEWSCASTER  Starting up...")
    print("=" * 60)

    # Load configuration
    load_config()



    # Bind Vision Callbacks
    if video_detector:
        def on_all_detections(detections: dict):
            if not detections:
                return
            print(f"[VISION] Detections: {detections}")
            
            global is_scanning
            if is_scanning:
                face_detected_event.set()

        try:
            video_detector.on_detect_all(on_all_detections)
            print("[INFO] Video Object Detection bound to AI-Newscaster-FaceDetect model")
        except Exception as e:
            print(f"[WARN] Failed to bind vision callbacks: {e}")

    # Register Bridge handlers
    register_bridge_handlers()
    print("[INFO] Bridge RPC handlers registered")

    # Set initial state
    set_state("idle")

    print("[INFO] AI Newscaster ready!")
    print("=" * 60)

    # Start the App Lab event loop
    App.run()


if __name__ == "__main__":
    main()

#  AI Newscaster

An intelligent, face-detecting news briefing device built on the Arduino UNO Q (4GB).

Press the button  the webcam scans for a person  select a category  your news briefing appears on the LCD.

---

## Table of Contents

1. [Overview](#overview)
2. [Bill of Materials](#bill-of-materials)
3. [Phase 1: Setup](#phase-1-setup)
4. [Phase 2: Hardware Wiring](#phase-2-hardware-wiring)
5. [Phase 3: Edge Impulse Model Training](#phase-3-edge-impulse-model-training)
6. [Phase 4: App Lab Project Setup](#phase-4-app-lab-project-setup)
7. [Phase 5: Code Deployment](#phase-5-code-deployment)
8. [Phase 6: Usage Guide](#phase-6-usage-guide)
9. [License](#license)

---

## Overview

The AI Newscaster runs on the Arduino UNO Q's dual-processor architecture:

| Side | Processor | Role |
|------|-----------|------|
| **MPU** (Linux) | Qualcomm QRB2210 | Video object detection (Person), RSS fetching |
| **MCU** (Arduino) | STM32U585 | LCD rendering, button input, screen state machine |

The two processors communicate via **Bridge RPC** (Remote Procedure Calls).

### Key Features
- **On-device person detection**  uses the native VideoObjectDetection brick to wake up.
- **Zero-Dependency Architecture**  uses standard Python libraries for lightning-fast, crash-free execution.
- **No phone or computer needed**  the entire UX happens on the device.

---

## Bill of Materials

### Core Components

| # | Component | Purpose |
|---|-----------|---------|
| 1 | Arduino UNO Q (4GB) | Main board (QRB2210 MPU + STM32U585 MCU) |
| 2 | KooingTech HBV-W202012HD USB UVC mini webcam | Face capture |
| 3 | Waveshare 2" LCD Module (240320, ST7789V) | Display output |
| 4 | 14 Tactile Button Module | User input (UP/DOWN/SELECT/BACK) |
| 5 | Arduino 8-in-1 USB-C Hub | Peripheral connectivity |
| 6 | 5V/3A 10000mAh Power Bank | Portable power |

---

## Phase 1: Setup

This project uses a 100% free, private, and local approach. It uses RSS feeds for lightning-fast news retrieval, and native computer vision to detect when a user is present. You do **not** need any API keys or Google Cloud projects!

### 1.1 RSS News Feeds

News is fetched directly from Google News RSS feeds using Python's built-in `urllib` and `xml` libraries. It requires no authentication, has no strict rate limits, and requires zero external Python packages to install.

### 1.2 Arduino App Lab & WiFi Setup

1. Download and install **Arduino App Lab** (version 0.7 or newer).
2. Connect your **Arduino UNO Q** to your computer using a USB-C cable.
3. Open Arduino App Lab and ensure your board is detected in the top menu.
4. **Connect the board to WiFi**:
   - Open the **Terminal** tab at the bottom of the App Lab window (this connects you directly to the board's Linux MPU).
   - Type the following command and press Enter to open the visual network manager:
     ```nmtui```
   - A visual menu will appear in your terminal. Use your keyboard's **Arrow Keys** to select **"Activate a connection"** and press Enter.
   - Select your WiFi network from the list, press Enter, type your password, and hit Enter again.
   - Once connected, press **ESC** a few times to exit the menu.

---

## Phase 2: Hardware Wiring

### Pin Assignment Table

#### ST7789V LCD  MCU (SPI)

| LCD Pin | UNO Q Pin | Function |
|---------|-----------|----------|
| VCC | **3.3V** | Power (**NOT 5V!**) |
| GND | GND | Ground |
| DIN (MOSI) | D11 | SPI Data |
| CLK (SCK) | D13 | SPI Clock |
| CS | D10 | Chip Select |
| DC | D9 | Data/Command |
| RST | D8 | Reset |
| BL | D7 | Backlight |

#### 4-Button Module  MCU

| Button | UNO Q Pin | Function |
|--------|-----------|----------|
| Button 1 | D2 | UP |
| Button 2 | D3 | DOWN |
| Button 3 | D4 | SELECT |
| Button 4 | D5 | BACK |
| GND | GND | Common ground |

#### USB Webcam  MPU (via hub)

| Connection | Path |
|------------|------|
| Webcam USB |  USB-C Hub  UNO Q USB-C port |

### Assembly Steps

1. **Place the UNO Q** on your work surface, USB-C port facing you.
2. **Wire the LCD:** Connect VCC3.3V, GNDGND, then the 6 signal wires (DIND11, CLKD13, CSD10, DCD9, RSTD8, BLD7). Use short wires for SPI signals.
3. **Wire the buttons:** Connect 4 signal pins to D2-D5, and the module's GND to board GND. The sketch uses `INPUT_PULLUP`, so buttons should pull pins LOW when pressed.
4. **Connect the USB hub:** Plug into UNO Q's USB-C port. Plug webcam into a USB-A port on the hub.
5. **Power:** Connect your PC to the hub's PD passthrough or directly to the UNO Q.

---

## Phase 3: Edge Impulse Model Training

Train a single-class **face detection** model. This model only needs to detect *that a face exists*  identification is handled by the device's native flow.

### 3.1 Create Project
1. Log in to [Edge Impulse Studio](https://studio.edgeimpulse.com)
2. Create a new project: **AI-Newscaster-FaceDetect**

### 3.2 Upload Data
1. Collect 200-500 face images (varied lighting, angles, skin tones)
   - Recommended: use a public dataset like [WIDER FACE](http://shuoyang1213.me/WIDERFACE/)
2. **Data acquisition**  **Add data**  **Upload data**
3. Select **Automatically split between training and testing**
4. Confirm this is an **Object Detection** project

### 3.3 Label Data
1. **Data acquisition**  **AI Labeling**
2. Select **Bounding box labeling with OWL-ViT**
3. Set prompt: `"A human face (face, 0.2)"`
4. Click **Label Preview Data**, verify accuracy
5. Click **Label Dataset**
6. Review and manually fix any bad labels

### 3.4 Design Impulse
1. **Impulse design**  **Create Impulse**
2. Set target device: **Arduino UNO Q**
3. Image input: **320320**, resize mode: **Fit shortest axis**
4. Processing block: **Image**
5. Learning block: **Object Detection (Images)**
6. **Save Impulse**

### 3.5 Train
1. **Impulse design**  **Image**: color depth **RGB**, click **Save parameters**, then **Generate features**
2. **Impulse design**  **Object detection**: 
   - Under **Neural network architecture**, click **Choose a different model** and select **MobileNetV2 SSD FPN-Lite (320x320 only)**.
   - Number of training cycles: **50**
   - Learning rate: **0.1**
3. Click **Save & train**
4. Target: **mAP@50 > 0.50** (unoptimized float32)

### 3.6 Deploy
1. **Model testing**  **Classify all** (verify on test data)
2. **Deployment**  search **Arduino UNO Q**  **Unoptimized (float32)**  **Build**
3. Note your project ID from the URL (you'll need it for `app.yaml`)

---

## Phase 4: App Lab Project Setup

### 4.1 Create Project
1. Open Arduino App Lab
2. Create a new project: **AI Newscaster**

### 4.2 Add Video Object Detection Brick
1. Go to the **Bricks** panel
2. Add **Video Object Detection** (`video_object_detection`)
3. Under **AI Models**, link your Edge Impulse account
4. Select the **AI-Newscaster-FaceDetect** model
5. Verify `app.yaml` has the correct `model_id` (edit manually via terminal if needed)

### 4.3 Install Arduino Libraries
In App Lab's library manager, install:
- **Adafruit GFX Library**
- **Adafruit ST7735 and ST7789 Library**

---

## Phase 5: Code Deployment

### 5.1 Configure app.yaml
Check `app.yaml` and confirm it has your actual Edge Impulse project ID.

### 5.2 Deploy
1. Click **Run** in App Lab
2. App Lab will:
   - Compile the C++ sketch for the STM32
   - Deploy the video_object_detection Brick container
   - Start the Python environment

---

## Phase 6: Usage Guide

### Daily Use
1. Power on the device (plug in the power bank).
2. Wait for the LCD to show **"Hello! Press SELECT to get your news..."**
3. Press **SELECT** to trigger the camera.
4. Look into the camera! Once it detects a face, you will be prompted to select a news category.
5. Use **UP/DOWN** to highlight a category and press **SELECT** to confirm.
6. The device fetches your news. Once your briefing appears, use **UP/DOWN** to navigate stories.
7. Press **BACK** to return to the idle screen.

---

## License

This project is open source. Built for the Arduino Project Hub.