# Project export: DigitalTwin

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: UC Berkeley AI Hackathon 2025
- Tagline: Digital Twin: A machine learning-powered platform leveraging cough acoustics for non-invasive, continuous respiratory disease screening.
- Devpost: https://devpost.com/software/digitaltwin-syzrjd
- GitHub: https://github.com/haile-teshome/DigitalTwin
- Demo: https://digital-twin-1-six.vercel.app/
- Team: 3 GitHub contributor(s) — Haile Teshome (9 commits), Shwetha Kandhalu (4 commits), SeonMinKIm (3 commits)

## Devpost submission (written by the team)

### Inspiration

: Digital Twin was born from a simple yet powerful idea: What if a person’s cough could tell the story of their respiratory health? In a world shaped by remote care and digital diagnostics, we envisioned a tool that could democratize access to early respiratory disease detection by making it affordable, user-friendly, and non-invasive. We were inspired by the opportunity to reduce diagnostic delays and empower communities with proactive, accessible healthcare.

### What it does

: Digital Twin is a web-based application that allows users to record their cough and receive a prediction of potential respiratory illness. It leverages a machine learning model trained on cough audio data and integrates Vapi to enable real-time voice interaction. The output provides a preliminary classification to support early screening and awareness.

### How we built it

: After extensive planning and discussion, we divided our workflow into specialized components: Built and trained a machine learning model using labeled cough sound data Integrated the Vapi voice API for voice input capture and backend processing Designed a responsive web application interface

### Challenges we ran into

: Difficulty in finding a publicly available dataset of cough sounds from patients with respiratory diseases, which constrained model development and validation Unfamiliarity with Vapi's API and voice processing tools, which required time to understand and implement effectively Limited frontend development experience, especially in deploying interactive and responsive web interfaces To overcome these we: Participated in workshops and studied documentation Conducted frequent internal check-ins and collaborative debugging Supported one another across roles to close skill gaps

### Accomplishments we're proud of

: Developed an interactive web platform from scratch, despite limited prior frontend experience Integrated a voice-based interface for health screening, enhancing accessibility and user experience

### What we learned

: Acoustic data can serve as powerful digital biomarkers when paired with appropriate ML architectures AI tooling can provide a low cost alternatives for remote settings where resourced are limited while providing many of the services which would require a team to do like customer service and support, web deployment, and resource allocation Collaboration, communication, and flexibility are critical in overcoming steep learning curves

### What's next

: Expand the dataset to include more real world data in order improve the quality of predictions as well as generalize across to more respiratory illnesses Deploy on mobile platforms for increasing accessibility to limited resource settings Expand regulatory compliance framework across other countries Generalize platform to include other longitudinal monitoring of health (ie diabetic retinopathy)

## README (from the GitHub repository)

# 👥 Digital Twin

**Digital Twin** is a machine learning-powered platform leveraging cough acoustics for non-invasive, continuous respiratory disease screening.

<img width="1107" alt="Screenshot 2025-06-22 at 10 38 12 AM" src="https://github.com/user-attachments/assets/e7ba5c2e-6334-4f62-8ec7-0ee3e8dff771" />

# VAPI Audio Analysis Client

This Python application provides a simple interface to interact with the VAPI API for audio analysis.

## Setup

1. Install dependencies:
```bash
pip install -r requirements.txt
```

2. Create a `.env` file in the project root with your API key:
```
VAPI_API_KEY=your_api_key_here
```

## Usage

The application provides a simple interface to analyze audio files. Here's how to use it:

```python
from vapi_client import VAPIClient

# Initialize the client
client = VAPIClient()

# Analyze an audio file
analysis = client.analyze_audio("path/to/your/audio/file.wav")

# Save the results
if analysis:
    client.save_analysis_results(analysis, "audio_analysis.json")
```

## Features

- Analyze audio files using the VAPI API
- Save analysis results to JSON files
- Error handling and file management

![PNG image](https://github.com/user-attachments/assets/e9ecc06e-c567-492d-b0a4-8f2184489974)

## Requirements

- Python 3.7+
- Required packages (see requirements.txt):
  - requests
  - python-dotenv


## Detected evidence (automated analysis)

Indexed codebase: 20 recognized source files, 53 KB.
- HTML (language) — detected in the code
- Python (language) — detected in the code
- PyTorch (technology) — claimed on Devpost, not found in the code
- Vercel (technology) — claimed on Devpost, not found in the code

## Codebase structure (from repository index)

### Files (22 of 22)

```
annotator_agent.py
bureau.py
copd_cnn.pth
dataset.py
feature_extractor.py
frontend/dashboard.html
frontend/index.html
frontend/intro1.html
frontend/intro2.html
frontend/intro3.html
frontend/login.html
frontend/permissions.html
frontend/signup.html
frontend/voiceanalysis.html
frontend/wellness.html
model.py
producer_agent.py
README.md
requirements.txt
streamlit_app.py
train.py
vapi_client.py
```

### Dependencies

- requirements.txt: annotated-types@==0.7.0, anyio@==4.9.0, certifi@==2025.6.15, cffi@==1.17.1, exceptiongroup@==1.3.0, h11@==0.16.0, httpcore@==1.0.9, httpx@==0.28.1, idna@==3.10, numpy@==2.2.6, pycparser@==2.22, pydantic@==2.11.7, pydantic_core@==2.33.2, sniffio@==1.3.1, sounddevice@==0.5.2, typing_extensions@==4.14.0, typing-inspection@==0.4.1, vapi-server-sdk@==1.5.1

### Recent commits (newest first)

- Update README.md
- Update train.py
- Update producer_agent.py
- Update model.py
- Update feature_extractor.py
- Update dataset.py
- Update bureau.py
- Update annotator_agent.py
- Update README.md
- Update README.md
- Merge pull request #1 from shwe-kandhalu/main
- dashboard progress
- progress
- add intros
- Add index file
- Add files via upload

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

### requirements.txt

```
annotated-types==0.7.0
anyio==4.9.0
certifi==2025.6.15
cffi==1.17.1
exceptiongroup==1.3.0
h11==0.16.0
httpcore==1.0.9
httpx==0.28.1
idna==3.10
numpy==2.2.6
pycparser==2.22
pydantic==2.11.7
pydantic_core==2.33.2
sniffio==1.3.1
sounddevice==0.5.2
typing-inspection==0.4.1
typing_extensions==4.14.0
vapi-server-sdk==1.5.1

```

### bureau.py

```python
from uagents import Bureau
from annotator_agent import agent as annotator
from producer_agent import producer

bureau = Bureau(port=8000)
bureau.add(annotator); bureau.add(producer)

if __name__=="__main__":
    bureau.run()

```

### feature_extractor.py

```python
import numpy as np
import torchaudio

mel_transform = torchaudio.transforms.MelSpectrogram(
    sample_rate=16000, n_mels=64)

def extract_features(wav_path: str) -> np.ndarray:
    waveform, sr = torchaudio.load(wav_path)
    if sr != 16000:
        waveform = torchaudio.transforms.Resample(sr, 16000)(waveform)
    mel = mel_transform(waveform).log2().clamp(-10, 10)
    return mel.numpy().squeeze(0)

```

### producer_agent.py

```python
import glob, time, os
from uagents import Agent, Context, Model

class FeatureRequest(Model):
    wav_path: str

class AnnotationResponse(Model):
    wav_path: str
    diagnosis: str
    confidence: float

producer = Agent(name="producer", port=8001, seed="producer_seed")
ANNOTATOR = None  # set after running annotator

@producer.on_event("startup")
async def send_all(ctx: Context):
    for wav in glob.glob(os.path.expanduser("~/Desktop/recordings/*.wav")):
        await ctx.send(ANNOTATOR, FeatureRequest(wav_path=wav))
        time.sleep(0.1)

@producer.on_message(model=AnnotationResponse)
async def receive(ctx: Context, sender: str, msg: AnnotationResponse):
    print(f"{msg.wav_path}: {msg.diagnosis} ({msg.confidence:.2f})")

if __name__=="__main__":
    print("Paste annotator agent address here, then run")
    producer.run()

```

### dataset.py

```python
import os, torch, torchaudio
from torch.utils.data import Dataset

class COPDDataset(Dataset):
    def __init__(self, root_dir):
        self.samples = []
        for label in ["copd", "healthy"]:
            for fname in os.listdir(os.path.join(root_dir, label)):
                if fname.endswith(".wav"):
                    self.samples.append((os.path.join(root_dir, label, fname), 1 if label=="copd" else 0))
        self.mel_transform = torchaudio.transforms.MelSpectrogram(sample_rate=16000, n_mels=64)

    def __len__(self):
        return len(self.samples)

    def __getitem__(self, idx):
        path, label = self.samples[idx]
        waveform, sr = torchaudio.load(path)
        if sr != 16000:
            waveform = torchaudio.transforms.Resample(sr, 16000)(waveform)
        mel = self.mel_transform(waveform).log2().clamp(-10, 10)
        return mel.unsqueeze(0), label

```

### model.py

```python
import torch.nn as nn

class ECABlock(nn.Module):
    def __init__(self, channel, k_size=3):
        super().__init__()
        self.avg_pool = nn.AdaptiveAvgPool2d(1)
        self.conv = nn.Conv1d(1, 1, k_size, padding=(k_size-1)//2, bias=False)
        self.sig = nn.Sigmoid()

    def forward(self, x):
        y = self.avg_pool(x).squeeze(-1).permute(0,2,1)
        y = self.sig(self.conv(y)).permute(0,2,1).unsqueeze(-1)
        return x * y

class COPDNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = nn.Conv2d(1,32,3,1,1); self.eca1 = ECABlock(32)
        self.conv2 = nn.Conv2d(32,64,3,1,1); self.eca2 = ECABlock(64)
        self.conv3 = nn.Conv2d(64,128,3,1,1); self.eca3 = ECABlock(128)
        self.pool = nn.MaxPool2d(2,2)
        self.fc = nn.Sequential(nn.AdaptiveAvgPool2d(1), nn.Flatten(), nn.Linear(128,2))

    def forward(self, x):
        x = self.pool(self.eca1(self.conv1(x)))
        x = self.pool(self.eca2(self.conv2(x)))
        x = self.pool(self.eca3(self.conv3(x)))
        return self.fc(x)

```

### annotator_agent.py

```python
from uagents import Agent, Context, Model
import torch, numpy as np
from feature_extractor import extract_features
from model import COPDNet

class FeatureRequest(Model):
    wav_path: str

class AnnotationResponse(Model):
    wav_path: str
    diagnosis: str
    confidence: float

device = torch.device("cpu")
net = COPDNet().to(device)
net.load_state_dict(torch.load("copd_cnn.pth", map_location=device))
net.eval()

agent = Agent(name="annotator", port=8002, seed="annotator_seed")

@agent.on_message(model=FeatureRequest, replies=AnnotationResponse)
async def annotate(ctx: Context, sender: str, msg: FeatureRequest):
    feats = extract_features(msg.wav_path)
    xb = torch.from_numpy(feats).unsqueeze(0).unsqueeze(0).float().to(device)
    with torch.no_grad():
        out = net(xb)
        conf = torch.softmax(out,1)[0]
        pred = int(conf.argmax())
    await ctx.send(sender, AnnotationResponse(
        wav_path=msg.wav_path,
        diagnosis="copd" if pred==1 else "healthy",
        confidence=float(conf[pred])
    ))

if __name__=="__main__":
    agent.run()

```

### train.py

```python
import torch, torch.nn as nn, torch.optim as optim
from torch.utils.data import DataLoader, random_split
from dataset import COPDDataset
from model import COPDNet
from sklearn.metrics import classification_report

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
ds = COPDDataset("data/")
train_len = int(0.8*len(ds))
train_ds, val_ds = random_split(ds, [train_len, len(ds)-train_len])
train_dl = DataLoader(train_ds,16,shuffle=True)
val_dl = DataLoader(val_ds,16)

model = COPDNet().to(device)
opt, crit = optim.Adam(model.parameters(),1e-3), nn.CrossEntropyLoss()

for epoch in range(20):
    model.train()
    for xb, yb in train_dl:
        xb, yb = xb.to(device), yb.to(device)
        opt.zero_grad()
        loss = crit(model(xb), yb)
        loss.backward(); opt.step()
    print(f"Epoch {epoch+1} done")

model.eval()
preds, trues = [],[]
with torch.no_grad():
    for xb, yb in val_dl:
        xb = xb.to(device)
        out = model(xb)
        preds.extend(out.argmax(1).cpu().numpy())
        trues.extend(yb.numpy())
print(classification_report(trues, preds, target_names=["healthy","copd"]))
torch.save(model.state_dict(), "copd_cnn.pth")

```

### streamlit_app.py

```python
import streamlit as st
from streamlit_webrtc import webrtc_streamer, WebRtcMode
import av
import numpy as np
import soundfile as sf
import tempfile

st.set_page_config(page_title="COPD Voice Recorder", layout="centered")
st.title("🎤 COPD Audio Recorder")
st.markdown("This tool records your voice for COPD model analysis. No camera access is required.")

# Define audio-only media constraints (avoid webcam prompt)
media_stream_constraints = {
    "audio": True,
    "video": False
}

# Custom audio processor
class AudioProcessor:
    def __init__(self):
        self.frames = []

    def recv(self, frame: av.AudioFrame) -> av.AudioFrame:
        audio = frame.to_ndarray()
        self.frames.append(audio)
        return frame

# Start webrtc streamer with audio-only
ctx = webrtc_streamer(
    key="audio-only",
    mode=WebRtcMode.SENDRECV,
    media_stream_constraints=media_stream_constraints,
    audio_processor_factory=AudioProcessor
)

# After recording stops
if ctx.state.playing is False and ctx.audio_processor and ctx.audio_processor.frames:
    st.success("✅ Recording complete!")

    # Combine and save audio
    audio_data = np.concatenate(ctx.audio_processor.frames, axis=1).flatten()

    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        sf.write(f.name, audio_data, 48000)
        st.audio(f.name)
        st.info(f"Saved to: `{f.name}`")

        if st.button("🧠 Run COPD Prediction Model"):
            st.warning("Model integration goes here.")

```

### vapi_client.py

```python
import os
import time
import wave
import threading
import webrtcvad
import sounddevice as sd

# === Configuration ===
RATE = 16000
FRAME_DURATION_MS = 30
NUM_CHANNELS = 1
SILENCE_TIMEOUT = 0.5  # not used for auto-stop, kept for sanity
# Change this to your Desktop path
OUTPUT_DIR = os.path.expanduser("~/Desktop/recordings")
os.makedirs(OUTPUT_DIR, exist_ok=True)

# Initialize VAD
vad = webrtcvad.Vad(2)
FRAME_SIZE = int(RATE * FRAME_DURATION_MS / 1000)

# Shared state
buffer = bytearray()
recording = False
stop_requested = False

def recorder_loop():
    global buffer, recording, stop_requested

    with sd.RawInputStream(samplerate=RATE, blocksize=FRAME_SIZE,
                           dtype='int16', channels=NUM_CHANNELS) as stream:
        print("🎤 Listening... speak to begin recording. Press Enter to stop.")
        while not stop_requested:
            frame, _ = stream.read(FRAME_SIZE)
            is_speech = vad.is_speech(frame, RATE)

            if not recording and is_speech:
                recording = True
                buffer = bytearray()
                buffer.extend(frame)
                print("🟢 Recording started.")

            elif recording:
                buffer.extend(frame)

    # When stop is requested
    if recording and buffer:
        save_wav(buffer)

def save_wav(buffer):
    timestamp = time.strftime("%Y%m%d-%H%M%S")
    filename = f"{timestamp}.wav"
    path = os.path.join(OUTPUT_DIR, filename)
    with wave.open(path, 'wb') as wf:
        wf.setnchannels(NUM_CHANNELS)
        wf.setsampwidth(2)
        wf.setframerate(RATE)
        wf.writeframes(buffer)
    print(f"💾 Saved to Desktop: {path}")

def wait_for_enter():
    global stop_requested
    input()
    stop_requested = True
    print("🛑 Stop requested")

if __name__ == "__main__":
    t_rec = threading.Thread(target=recorder_loop)
    t_key = threading.Thread(target=wait_for_enter)

    t_rec.start()
    t_key.start()

    t_key.join()
    t_rec.join()
    print("✅ Done")

```

[10 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]