# Project export: cafe fra.me

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: TreeHacks 2025
- Tagline: Revolutionizing healthcare by replacing spreadsheets with an AI-native and AR-enhanced approach to deliver personalized, context-rich care.
- Devpost: https://devpost.com/software/cafe-fra-me
- GitHub: https://github.com/jeofo/cafeframe
- Demo: https://cafefra.me/
- Video: https://www.youtube.com/embed/b0TtwTqNfVk?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 1 GitHub contributor(s) — Jefferson Ding (1 commits)

## Devpost submission (written by the team)

### Inspiration

I remember sitting in a family doctor’s waiting room, frustrated as I watched the doctor’s machine reboot—only to discover it had no record of vaccines I received at another clinic. With all these so-called robust ERP systems and endless spreadsheets, how could this happen? The truth is, the entire healthcare management system is broken. There’s no shared documentation, no privacy, no transparency between physicians and patients, and a clunky lookup process that forces you to rely on specialists like nurses or assistants. It’s a system crying out for change. What It Does Cafa Fra.me is a two-part solution designed to revolutionize healthcare management: Software: A conversational AI-driven interface that replaces tedious spreadsheets with dynamic, context-rich interactions. Hardware: Innovative AR glasses that bring critical patient information directly in front of physicians’ eyes, eliminating the need for distracting devices. How We Built It There are several key components: Hardware Integration: We partnered with Brilliant.xyz, whose hardware we’ve been experimenting with. The idea was to remove traditional computers and devices that distract from patient care. Now, every piece of critical information is instantly accessible through AR glasses. The Python SDK made integration a breeze. Robust Backend: We developed a powerful backend featuring semantic search capabilities using a vector store. By leveraging agentic chunking, a vector database, and a reranking algorithm, we built a system that accurately extracts information from physicians’ queries. Web3 Security: We built a smart contract on Polygon that employs both asymmetric and symmetric encryption to securely store medical records on-chain. Only authorized personnel can access or update the records. In addition, a React Native app empowers patients to monitor their doctor visits and upload their health data via the Terra API. Challenges We Ran Into Integration Complexity: Combining cutting-edge hardware with a robust, secure backend presented unique challenges, particularly in ensuring seamless communication between components. Data Privacy and Security: Balancing transparency with stringent privacy requirements required careful design, especially with the introduction of blockchain-based solutions. User Adoption: Transitioning healthcare professionals from traditional systems to an AR-driven interface meant overcoming resistance to change and ensuring ease-of-use. Performance Optimization: Implementing semantic search and vector ranking in real-time demanded rigorous optimization to maintain a smooth user experience. Accomplishments We’re Proud Of Successfully integrating AR glasses to deliver real-time, hands-free access to patient information. Developing a backend with advanced semantic search capabilities that accurately understands and responds to physicians’ queries. Building a secure, decentralized system using smart contracts that ensures only authorized personnel can access sensitive medical data. Creating a user-friendly interface that blends innovation with practicality, making healthcare management more intuitive and patient-focused.

## README (from the GitHub repository)

No README available.

## Detected evidence (automated analysis)

Indexed codebase: 7 recognized source files, 27 KB.
- Python (language) — detected in the code
- Google Gemini (technology) — claimed on Devpost, not found in the code
- OpenAI (technology) — claimed on Devpost, not found in the code
- Solidity (language) — claimed on Devpost, not found in the code
- Swift (language) — claimed on Devpost, not found in the code

## Codebase structure (from repository index)

### Files (10 of 10)

```
.gitignore
index.py
lib/__init__.py
lib/db.py
lib/frame.py
lib/gemini.py
lib/ppxtly.py
lib/tools.py
Pipfile
Pipfile.lock
```

### Dependencies

No dependency index available.

### Recent commits (newest first)

- initial commit

## Key source files (fetched from GitHub, selected and truncated for size)

### index.py

```python
from dotenv import load_dotenv

load_dotenv()  # take environment variables from .env.

import asyncio

from frame_sdk import Frame

from lib.db import fetch_data
from lib.frame import display_text
from lib.gemini import queryFrame


async def index():
    async with Frame() as frame:
        # Show connection and battery status
        battery_level = await frame.get_battery_level()
        print(f"Connected - Battery: {battery_level}%")

        # Get user data and display welcome message
        data = await fetch_data()
        print(type(data))
        await display_text(
            frame, f"Hello, {data['doctor']['name']}!\nTap to ask me anything!"
        )

        while True:
            try:
                await frame.motion.wait_for_tap()
                print("tapped!")
                await queryFrame(frame)
                await asyncio.sleep(1)
            except Exception as e:
                print(f"Error index: {e}")
                await asyncio.sleep(1)

    print("disconnected")


if __name__ == "__main__":
    asyncio.run(index())

```

### lib/ppxtly.py

```python
import os
import requests
from typing import Optional, Dict, Any


def query_perplexity(
    query: str,
) -> str:
    """Query the Perplexity AI API with a given input string.

    Args:
        query (str): The input query string
        max_tokens (int, optional): Maximum tokens in response. Defaults to 123.
        temperature (float, optional): Controls randomness. Defaults to 0.2.
        top_p (float, optional): Controls diversity. Defaults to 0.9.

    Returns:
        Dict[str, Any]: API response containing the generated content and metadata

    Raises:
        ValueError: If API key is not set
        requests.RequestException: If API request fails
    """
    max_tokens: int = 123
    temperature: float = 0.2
    top_p: float = 0.9
    api_key = os.getenv("PERPLEXITY_API_KEY")
    if not api_key:
        raise ValueError("PERPLEXITY_API_KEY environment variable not set")

    url = "https://api.perplexity.ai/chat/completions"

    payload = {
        "model": "sonar",
        "messages": [
            {"role": "system", "content": "Be precise and concise."},
            {"role": "user", "content": query},
        ],
        "max_tokens": max_tokens,
        "temperature": temperature,
        "top_p": top_p,
        "stream": False,
    }

    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}

    try:
        response = requests.post(url, json=payload, headers=headers)
        response.raise_for_status()
        return str(response.json())
    except requests.RequestException as e:
        raise requests.RequestException(f"API request failed: {str(e)}")


if __name__ == "__main__":
    print(query_perplexity("what are some side effects of aspirin?"))

```

### lib/db.py

```python
import asyncio
import json
import urllib.parse
import aiohttp


async def fetch_data():
    async with aiohttp.ClientSession() as session:
        async with session.get("http://compute.r0h.in:3000/read") as response:
            encoded_data = await response.text()
            decoded_data = urllib.parse.unquote(encoded_data)
            return json.loads(decoded_data.strip("'"))


async def update_data(data):
    async with aiohttp.ClientSession() as session:
        encoded_data = urllib.parse.quote(json.dumps(data))
        async with session.get(
            f"http://compute.r0h.in:3000/write?key={encoded_data}"
        ) as response:
            return await response.text()


if __name__ == "__main__":
    data = json.dumps(
        {
            "doctor": {
                "id": "doc123",
                "name": "Dr. Smith",
                "hospital_name": "Central Hospital",
                "department": "Cardiology",
                "specialization": "Interventional Cardiology",
            },
            "patients": {
                "pat123": {
                    "name": "John Doe",
                    "public_key": "abc123",
                    "age": 45,
                    "gender": "M",
                    "contact_info": {
                        "phone": "+1234567890",
                        "email": "john@example.com",
                    },
                    "medical_history": [
                        {
                            "date": "2024-01-15",
                            "description": "Regular checkup",
                            "medications": ["Aspirin 81mg"],
                        }
                    ],
                    "terra_data": "Sample terra data for patient 123",
                }
            },
            "appointments": {
                "apt123": {
                    "patient_id": "pat123",
                    "datetime": "2025-02-18T10:00:00",
                    "purpose": "Follow-up",
                }
            },
        }
    )
    print(asyncio.run(update_data(json.loads(data))))
    print(asyncio.run(fetch_data()))

```

### lib/frame.py

```python
from frame_sdk.display import PaletteColors, Alignment


async def display_text(frame, text, color=PaletteColors.WHITE, delay=0.12):
    """Display text on the Frame, automatically handling long content with scrolling.

    Args:
        frame: Frame instance
        text: Text to display
        color: Text color (default: WHITE)
        delay: Scroll delay in seconds (default: 0.12)
    """
    # Estimate if text will fit on screen (rough estimate)
    # Frame display is 640x400 pixels
    CHARS_PER_LINE = 50  # Approximate characters that fit horizontally
    VISIBLE_LINES = 15  # Approximate lines that fit vertically

    # Count lines after word wrapping
    wrapped_lines = []
    for line in text.split("\n"):
        # Simple word wrap
        words = line.split()
        current_line = []
        current_length = 0

        for word in words:
            if current_length + len(word) + 1 <= CHARS_PER_LINE:
                current_line.append(word)
                current_length += len(word) + 1
            else:
                wrapped_lines.append(" ".join(current_line))
                current_line = [word]
                current_length = len(word)

        if current_line:
            wrapped_lines.append(" ".join(current_line))

    total_lines = len(wrapped_lines)

    # If text fits on screen, display normally
    if total_lines <= VISIBLE_LINES:
        await frame.display.write_text(
            text, max_width=640, align=Alignment.TOP_LEFT, color=color
        )
        await frame.display.show()
    else:
        # For longer text, use scroll_text
        await frame.display.scroll_text(
            text,
            lines_per_frame=5,
            delay=delay,
            color=color,
        )


async def display_centered_text(frame, text, color=PaletteColors.WHITE):
    """Display text centered on the Frame.

    Args:
        frame: Frame instance
        text: Text to display
        color: Text color (default: WHITE)
    """
    await frame.display.write_text(
        text, max_width=640, max_height=400, align=Alignment.MIDDLE_CENTER, color=color
    )
    await frame.display.show()


async def display_scroll(frame, text, color=PaletteColors.WHITE):
    await frame.display.scroll_text(text, color=color)


async def record(frame):
    await frame.microphone.save_audio_file("instructions.wav")

```

### lib/tools.py

```python
import asyncio
from typing import Dict, List, Optional
from datetime import datetime
from lib.db import fetch_data, update_data

data = asyncio.run(fetch_data())


# Doctor-related functions
def get_doctor_info() -> str:
    """Get current doctor's information"""
    print("FUNCTION CALLED: get_doctor_info")
    return str(data["doctor"])


# Patient-related functions
def get_patient_by_id(patient_id: str) -> str:
    """Get patient information by ID"""
    print("FUNCTION CALLED: get_patient_by_id")

    return str(data["patients"].get(patient_id)) or "{}"


def get_patient_medical_history(patient_id: str) -> str:
    """Get patient's medical history"""
    print("FUNCTION CALLED: get_patient_medical_history")

    patient = data["patients"].get(patient_id)
    return str(patient["medical_history"]) if patient else "[]"


def get_patient_terra_data() -> str:
    """Get patient's Terra health data"""
    print("FUNCTION CALLED: get_patient_terra_data")

    return str(data["terra"])


# Appointment-related functions
def get_appointments_for_date(target_date: str) -> str:
    """Get all appointments for a specific date
    Args:
        target_date (str): Date in YYYY-MM-DD format
    Returns:
        str: List of appointments for the given date
    """
    print("FUNCTION CALLED: get_appointments_for_date")

    matching_appointments = []

    for apt_id, apt in data["appointments"].items():
        apt_date = datetime.fromisoformat(apt["datetime"]).strftime("%Y-%m-%d")
        if apt_date == target_date:
            matching_appointments.append({"appointment_id": apt_id, **apt})

    return str(matching_appointments)


def get_patient_appointments(patient_id: str) -> str:
    """Get all appointments for a specific patient"""
    print("FUNCTION CALLED: get_patient_appointments")

    matching_appointments = []

    for apt_id, apt in data["appointments"].items():
        if apt["patient_id"] == patient_id:
            matching_appointments.append({"appointment_id": apt_id, **apt})

    return str(matching_appointments)


# Utility functions
def get_upcoming_appointments() -> str:
    """Get upcoming appointments, ordered by date"""
    print("FUNCTION CALLED: get_upcoming_appointments")

    now = datetime.now()
    upcoming = []

    for apt_id, apt in data["appointments"].items():
        apt_datetime = datetime.fromisoformat(apt["datetime"])
        if apt_datetime > now:
            patient = data["patients"].get(apt["patient_id"])
            upcoming.append(
                {
                    "appointment_id": apt_id,
                    **apt,
                    "patient_name": patient["name"] if patient else "Unknown",
                }
            )

    # Sort by datetime and limit results
    upcoming.sort(key=lambda x: x["datetime"])
    return str(upcoming[:2])


def search_patients(query: str) -> str:
    """Search patients by name, email, or phone"""
    print("FUNCTION CALLED: search_patients")

    query = query.lower()
    matching_patients = []

    for patient_id, patient in data["patients"].items():
        if (
            query in patient["name"].lower()
            or query in patient["contact_info"]["email"].lower()
            or query in patient["contact_info"]["phone"]
        ):
            matching_patients.append({"id": patient_id, **patient})

    return str(matching_patients)


def get_patient_summary(patient_id: str) -> str:
    """Get comprehensive patient summary including history and appointments"""
    print("FUNCTION CALLED: get_patient_summary")

    patient = data["patients"].get(patient_id)
    if not patient:
        return None

    return str(
        {
            "id": patient_id,
            **patient,
            "appointments": get_patient_appointments(patient_id),
        }
    )


async def add_appointment(appointment_data: str) -> str:
    """Add a new appointment to the database. Provide input in format:
    "patient_id;YYYY-MM-DDTHH:MM:SS;purpose"
    """
    print("FUNCTION CALLED: add_appointment")

    appointment_data = dict(
        zip(["patient_id", "datetime", "purpose"], appointment_data.split(";"))
    )
    required_fields = ["patient_id", "datetime", "purpose"]
    for field in required_fields:
        if field not in appointment_data:
            raise ValueError(f"Missing required field: {field}")

    if appointment_data["patient_id"] not in data["patients"]:
        print(appointment_data["patient_id"])
        print(data["patients"])
        raise ValueError(f"Patient {appointment_data['patient_id']} not found")

    try:
        datetime.fromisoformat(appointment_data["datetime"])
    except ValueError:
        raise ValueError("datetime must be in ISO format (YYYY-MM-DDTHH:MM:SS)")

    new_apt_id = f"apt{len(data['appointments']) + 1}"

    data["appointments"][new_apt_id] = {
        "patient_id": appointment_data["patient_id"],
        "datetime": appointment_data["datetime"],
        "purpose": appointment_data["purpose"],
    }

    await update_data(data)

    return str(data["appointments"])


async def add_medical_history(patient_id: str, history_entry: str) -> str:
    """Add a new medical history entry for a patient.
    history_entry must be in format:
    "date;description;medication1,medication2,..."
    """
    print("FUNCTION CALLED: add_medical_history")

    history_entry = dict(
        zip(["date", "description", "medications"], history_entry.split(";"))
    )

    history_entry["medications"] = history_entry.get("medications", "").split(",")
    patient = data["patients"].get(patient_id)
    if not patient:
        raise ValueError(f"Patient {patient_id} not found")

    required_fields = ["date", "description"]
    for field in required_fields:
        if field not in history_entry:
            raise ValueError(f"Missing required field: {field}")

    try:
        datetime.strptime(history_entry["date"], "%Y-%m-%d")
    except ValueError:
        raise ValueError("date must be in YYYY-MM-D
[truncated — 1426 more characters]
```

### lib/gemini.py

```python
import json
import asyncio
import os
from google import genai
from google.genai import types
from lib.frame import display_text, record
from lib.tools import (
    get_doctor_info,
    get_patient_by_id,
    get_patient_medical_history,
    get_patient_terra_data,
    get_appointments_for_date,
    get_upcoming_appointments,
    get_patient_appointments,
    search_patients,
    get_patient_summary,
    add_appointment,
    add_medical_history,
)
from lib.ppxtly import query_perplexity

client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])


def sync_add_appointment(
    patient_id: str, doctor_id: str, date: str, time: str, reason: str
) -> str:
    """Add a new appointment synchronously.

    Args:
        patient_id: ID of the patient
        doctor_id: ID of the doctor
        date: Appointment date (YYYY-MM-DD)
        time: Appointment time (HH:MM)
        reason: Reason for appointment
    Returns:
        str: Confirmation message
    """
    appointment_data = {
        "patient_id": patient_id,
        "doctor_id": doctor_id,
        "date": date,
        "time": time,
        "reason": reason,
    }
    print(f"Adding appointment: {appointment_data}")
    return "Appointment will be added"


def sync_add_medical_history(
    patient_id: str, condition: str, notes: str, date: str
) -> str:
    """Add medical history synchronously.

    Args:
        patient_id: ID of the patient
        condition: Medical condition or diagnosis
        notes: Additional notes about the condition
        date: Date of the condition/diagnosis (YYYY-MM-DD)
    Returns:
        str: Confirmation message
    """
    history_data = {
        "patient_id": patient_id,
        "condition": condition,
        "notes": notes,
        "date": date,
    }
    print(f"Adding medical history: {history_data}")
    return "Medical history will be added"


chat = client.chats.create(
    model="gemini-2.0-flash-001",
    config=types.GenerateContentConfig(
        system_instruction='You are an AI healthcare assistant designed for AR glasses with limited display space called "Frame". Your responses must be extremely concise, clear, and context-aware. You provide quick access to patient data, schedules, and alerts. If you cannot or don\'t know a piece of information, you can use the perplexity tool to search for answers',
        tools=[
            get_doctor_info,
            get_patient_by_id,
            get_patient_medical_history,
            get_patient_terra_data,
            get_appointments_for_date,
            get_upcoming_appointments,
            get_patient_appointments,
            search_patients,
            get_patient_summary,
            sync_add_appointment,
            sync_add_medical_history,
            query_perplexity,
        ],
        automatic_function_calling={"disable": True},
    ),
)


# Map of function names to their handlers
function_map = {
    "get_doctor_info": get_doctor_info,
    "get_patient_by_id": get_patient_by_id,
    "get_patient_medical_history": get_patient_medical_history,
    "get_patient_terra_data": get_patient_terra_data,
    "get_appointments_for_date": get_appointments_for_date,
    "get_upcoming_appointments": get_upcoming_appointments,
    "get_patient_appointments": get_patient_appointments,
    "search_patients": search_patients,
    "get_patient_summary": get_patient_summary,
    "sync_add_appointment": sync_add_appointment,
    "sync_add_medical_history": sync_add_medical_history,
    "query_perplexity": query_perplexity,
}


async def queryFrame(frame):
    if os.path.exists("instructions.wav"):
        os.remove("instructions.wav")
        print("removed instructions.wav")

    print("starting record...")
    await display_text(frame, "Listening...")
    await asyncio.sleep(0.1)
    await record(frame)
    print("recorded instructions.wav")

    if os.path.exists("instructions.wav"):
        file = client.files.upload(file="instructions.wav")
        response = chat.send_message(["audio:", file])
        print("Response received:")

        # Get the first part from the response
        if response and response.candidates and response.candidates[0].content.parts:
            part = response.candidates[0].content.parts[0]

            # Check if we got a function call
            if hasattr(part, "function_call") and part.function_call:
                function_call = part.function_call
                name = function_call.name
                args = function_call.args  # args is already a dict
                print(f"Function call received: {name} with args: {args}")

                if name in function_map:
                    try:
                        if name == "sync_add_appointment":
                            # Handle async appointment addition
                            appointment_data = {
                                "patient_id": args["patient_id"],
                                "doctor_id": args["doctor_id"],
                                "date": args["date"],
                                "time": args["time"],
                                "reason": args["reason"],
                            }
                            await add_appointment(appointment_data)
                            result = "Appointment added successfully"
                        elif name == "sync_add_medical_history":
                            # Handle async medical history addition
                            history_data = {
                                "patient_id": args["patient_id"],
                                "condition": args["condition"],
                                "notes": args["notes"],
                                "date": args["date"],
                            }
                            await add_medical_history(history_data)
                            result = "Medical history added successfully"
                        else:
                            # Handle regular sync functions
                  
[truncated — 6895 more characters]
```