# Project export: brain2bach

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 2026
- Tagline: Listen to your mind
- Devpost: https://devpost.com/software/brain2beet
- GitHub: https://github.com/MarkOfUs/brain2bach
- Video: https://www.youtube.com/embed/ictQIHWAEK4?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent
- Team: 3 GitHub contributor(s) — Markus Hoehn (20 commits), guinterface (3 commits), lcollomb-SU (1 commits)

## Devpost submission (written by the team)

No Devpost description available.

## README (from the GitHub repository)

No README available.

## Detected evidence (automated analysis)

Indexed codebase: 28 recognized source files, 128 KB.
- CSS (language) — detected in the code
- Flask (technology) — detected in the code
- HTML (language) — detected in the code
- JavaScript (language) — detected in the code
- OpenAI (technology) — detected in the code
- Python (language) — detected in the code
- PyTorch (technology) — detected in the code

## Codebase structure (from repository index)

### Files (42 of 42)

```
.gitignore
app.py
eeg_signal.py
eegToEmotion.ipynb
EmotionInference.ipynb
emotions_and_audio_to_suno_pipeline.py
musicInference.py
README.md
requirements.txt
routes.py
runpod_pipeline.py
runpod-emotion-inference/.dockerignore
runpod-emotion-inference/client_call_runsync.py
runpod-emotion-inference/Dockerfile
runpod-emotion-inference/emotion_inference.py
runpod-emotion-inference/handler.py
runpod-emotion-inference/models/.gitkeep
runpod-emotion-inference/models/cnn_faced_best.pt
runpod-emotion-inference/README.md
runpod-emotion-inference/requirements.txt
runpod-music-inference/client_call_runsync.py
runpod-music-inference/Dockerfile
runpod-music-inference/handler.py
runpod-music-inference/models/.gitkeep
runpod-music-inference/models/sub-17_run5_bilstm.pt
runpod-music-inference/models/sub-17_run5_env2mel.pt
runpod-music-inference/music_inference.py
runpod-music-inference/README.md
runpod-music-inference/requirements.txt
serial_eeg.py
serial_recorder.py
signals.py
static/css/main.css
static/css/theme.css
static/js/eeg.js
static/js/recordings.js
suno_pipeline.py
templates/base.html
templates/index.html
templates/recordings.html
templates/shell.html
waves.py
```

### Dependencies

- requirements.txt: eventlet@>=0.33, flask@>=3.0, flask-socketio@>=5.3, numpy@>=1.24, openai@>=1.0, pyserial@>=3.5, python-dotenv@>=1.0, python-socketio@>=5.9, requests@>=2.28, scipy@>=1.10
- runpod-emotion-inference/requirements.txt: h5py@>=3.9, numpy@>=1.24, pandas@>=2.0, requests@>=2.31, runpod@>=1.7.0, scipy@>=1.10, torch@>=2.1
- runpod-music-inference/requirements.txt: h5py@==3.10.0, librosa@==0.10.2.post1, numpy@==1.26.4, requests, runpod, scipy@==1.10.1, soundfile@==0.12.1

### Recent commits (newest first)

- Delete __pycache__ directory
- Update README.md
- fixes
- Commit
- Enhance README and app functionality for EEG to music generation
- Add files via upload
- Add files via upload
- Clean up comments in emotions_and_audio_to_suno_pipeline.py
- new images
- Delete slides/image directory
- Merge branch 'main' of https://github.com/MarkOfUs/brain2bach
- new sildes
- Rename file to emotions_and_audio_to_suno_pipeline.py
- rename
- Add files via upload
- Add waves.py for data recording and processing
- update slides
- Merge branch 'main' of https://github.com/MarkOfUs/brain2bach
- slides
- Update section headers in README.md

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

### requirements.txt

```
python-dotenv>=1.0
flask>=3.0
flask-socketio>=5.3
requests>=2.28
openai>=1.0
numpy>=1.24
scipy>=1.10
pyserial>=3.5
python-socketio>=5.9
eventlet>=0.33

```

### runpod-emotion-inference/requirements.txt

```
runpod>=1.7.0
numpy>=1.24
scipy>=1.10
pandas>=2.0
torch>=2.1
h5py>=3.9
requests>=2.31

```

### runpod-music-inference/requirements.txt

```
runpod
numpy==1.26.4
scipy==1.10.1
h5py==3.10.0
librosa==0.10.2.post1
soundfile==0.12.1
requests

```

### runpod-music-inference/Dockerfile

```
# CPU-friendly base (works on GPU endpoints too, but won't have CUDA-accelerated torch).
FROM python:3.10-slim

WORKDIR /app

# Needed for building some scientific deps, and for libsndfile (soundfile).
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    python3-dev \
    libsndfile1 \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .

# Install CPU-only torch from the PyTorch CPU wheel index (smaller + faster).
# If you want CUDA wheels instead, remove the --index-url line and just pip install torch.
RUN pip install --no-cache-dir --index-url https://download.pytorch.org/whl/cpu torch==2.2.2 \
 && pip install --no-cache-dir -r requirements.txt

COPY music_inference.py handler.py ./
COPY models/ /app/models/

CMD ["python", "-u", "handler.py"]

```

### runpod-emotion-inference/Dockerfile

```
# CPU-friendly base (works on GPU endpoints too, but won't have CUDA-accelerated torch).
FROM python:3.10-slim

WORKDIR /app

# Needed for installing torcheeg from source (and compiling deps like spectrum).
RUN apt-get update && apt-get install -y --no-install-recommends \
    git \
    build-essential \
    python3-dev \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Install torcheeg v1.1.3 but patch its scipy upper-bound (mirrors the notebook approach).
RUN rm -rf torcheeg_src \
 && git clone --depth 1 --branch v1.1.3 https://github.com/torcheeg/torcheeg.git torcheeg_src \
 && sed -i 's/scipy>=1\.7\.3[[:space:]]*,[[:space:]]*<=\s*1\.10\.1/scipy>=1.7.3/' torcheeg_src/setup.py \
 && pip install --no-cache-dir ./torcheeg_src \
 && rm -rf torcheeg_src

COPY emotion_inference.py handler.py ./

# Put your checkpoint here (see README)
# You must provide this file in ./models before building, or comment this out and use a volume/env-var.
COPY models/cnn_faced_best.pt /app/models/cnn_faced_best.pt

CMD ["python", "-u", "handler.py"]

```

### app.py

```python
import os
import time
import json
import base64
import threading
from pathlib import Path

# Load .env before reading env vars (optional)
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    pass

import requests
from flask import Flask, render_template, send_from_directory, request
from flask_socketio import SocketIO

from suno_pipeline import run_suno_from_emotion_probs, EMOTION_LABELS

# ============================================================
# ENVIRONMENT (REQUIRED)
# ============================================================

OPENAI_API_KEY = (os.getenv("OPENAI_API_KEY") or "").strip()
SUNO_API_TOKEN = (os.getenv("SUNO_API_TOKEN") or "").strip()
RUNPOD_API_KEY = (os.getenv("RUNPOD_API_KEY") or "").strip()
MOCK_EEG = os.getenv("MOCK_EEG", "0").lower() in ("1", "true", "yes")
SERIAL_PORT = os.getenv("SERIAL_PORT", "COM3")

if not OPENAI_API_KEY:
    raise RuntimeError("OPENAI_API_KEY not set")
if not SUNO_API_TOKEN:
    raise RuntimeError("SUNO_API_TOKEN not set")
if not RUNPOD_API_KEY:
    raise RuntimeError("RUNPOD_API_KEY not set")

# ============================================================
# PATHS
# ============================================================

BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
RECORDINGS_DIR = DATA_DIR / "recordings"
SNAPSHOT_PATH = RECORDINGS_DIR / "capture_snapshot.mat"

RECORDINGS_DIR.mkdir(parents=True, exist_ok=True)

# ============================================================
# RUNPOD ENDPOINTS (from README)
# ============================================================

EMOTION_RUNPOD_URL = "https://api.runpod.ai/v2/kw804mmqrwyhzz/runsync"

# ============================================================
# FLASK
# ============================================================

app = Flask(
    __name__,
    template_folder="templates",
    static_folder="static"
)

socketio = SocketIO(
    app,
    cors_allowed_origins="*",
    async_mode="threading"
)

# ============================================================
# EEG RECORDER
# ============================================================

def _create_recorder():
    from serial_recorder import MockEEGRecorder, SerialEEGRecorder

    if MOCK_EEG:
        return MockEEGRecorder(
            fs=50,
            snapshot_path=SNAPSHOT_PATH,
            snapshot_interval_s=2.0,
        )

    try:
        import serial
        with serial.Serial(SERIAL_PORT, 115200, timeout=0.1) as _:
            pass
    except Exception as e:
        print(f"[EEG] Serial port {SERIAL_PORT} not available ({e}). Using mock recorder. Set MOCK_EEG=1 to suppress.")
        return MockEEGRecorder(
            fs=50,
            snapshot_path=SNAPSHOT_PATH,
            snapshot_interval_s=2.0,
        )

    return SerialEEGRecorder(
        port=SERIAL_PORT,
        baud=115200,
        fs=50,
        snapshot_path=SNAPSHOT_PATH,
        snapshot_interval_s=2.0,
    )

recorder = _create_recorder()

# ============================================================
# CLIENT STATE
# ============================================================

ACTIVE_CLIENTS = set()
LATEST_EMOTION = {"label": "UNKNOWN", "probs": {}}

# ============================================================
# ROUTES
# ============================================================

@app.route("/")
def root():
    return render_template(
        "index.html",
        mock_eeg=MOCK_EEG,
    )

@app.route("/recordings/<path:filename>")
def download(filename):
    return send_from_directory(RECORDINGS_DIR, filename, as_attachment=True)

# ============================================================
# EEG STREAM
# ============================================================

def eeg_stream_loop(sid):
    try:
        while sid in ACTIVE_CLIENTS:
            sample = recorder.get_latest()
            socketio.emit(
                "eeg_data",
                {
                    "ch1": [float(sample[0])],
                    "ch2": [float(sample[1])],
                    "ch3": [float(sample[2])],
                },
                to=sid
            )
            time.sleep(0.02)
    finally:
        ACTIVE_CLIENTS.discard(sid)

# ============================================================
# RUNPOD EMOTION (mat → base64 → emotion probs)
# ============================================================

def run_emotion_inference(sid=None):
    """Send mat snapshot to Emotion RunPod, return emotion_probs dict for Suno."""
    global LATEST_EMOTION

    if not SNAPSHOT_PATH.exists():
        raise FileNotFoundError(
            "No EEG recording available. Start recording, wait a few seconds, then stop before creating a song."
        )

    with open(SNAPSHOT_PATH, "rb") as f:
        mat_b64 = base64.b64encode(f.read()).decode("utf-8")

    payload = {
        "input": {
            "mat_b64": mat_b64,
            "window_start": 0,
            "window_end": 250,
            "topk": 9,
            "mat_channel_names": ["FP1", "FZ", "FP2"]
        }
    }

    headers = {
        "Authorization": f"Bearer {RUNPOD_API_KEY}",
        "Content-Type": "application/json"
    }
    r = requests.post(EMOTION_RUNPOD_URL, headers=headers, json=payload, timeout=120)

    if r.status_code == 401:
        raise RuntimeError(
            "RunPod API key rejected (401). Check RUNPOD_API_KEY in .env or environment. "
            "Ensure the key is enabled and has Serverless/AI API permissions at runpod.io/settings."
        )
    r.raise_for_status()
    resp = r.json()
    out = resp.get("output", resp)

    classes = out.get("classes", [])
    probs_list = out.get("probs", [])
    topk = out.get("topk", [])

    if not classes or not probs_list or len(classes) != len(probs_list):
        raise RuntimeError(
           
[truncated — 3949 more characters]
```

### routes.py

```python
from flask import Blueprint, render_template

routes = Blueprint("routes", __name__)

@routes.route("/")
def eeg_dashboard():
    return render_template("eeg.html")

```

### signals.py

```python
import numpy as np

class EEGSignalGenerator:
    def __init__(self, fs=250):
        self.fs = fs
        self.t = 0

    def get_samples(self, n_samples=10):
        time = np.arange(self.t, self.t + n_samples) / self.fs
        self.t += n_samples

        ch1 = np.sin(2 * np.pi * 10 * time) + 0.2 * np.random.randn(n_samples)
        ch2 = np.sin(2 * np.pi * 12 * time) + 0.2 * np.random.randn(n_samples)
        ch3 = np.sin(2 * np.pi * 8 * time) + 0.2 * np.random.randn(n_samples)

        return ch1, ch2, ch3

```

### eeg_signal.py

```python
import numpy as np

class EEGGenerator:
    """
    Simple EEG signal simulator with three channels.

    This simulates EEG-like oscillations with noise.
    Sampling frequency is configurable.
    """

    def __init__(self, fs=250):
        self.fs = fs
        self.sample_index = 0

    def generate(self, n_samples=10):
        """
        Generate n_samples for three EEG channels.

        Returns:
            ch1, ch2, ch3 as Python lists (JSON serializable)
        """

        t = (self.sample_index + np.arange(n_samples)) / self.fs
        self.sample_index += n_samples

        # EEG-like rhythms (alpha-ish bands) + noise
        ch1 = 1.0 * np.sin(2 * np.pi * 10 * t) + 0.2 * np.random.randn(n_samples)
        ch2 = 0.8 * np.sin(2 * np.pi * 12 * t) + 0.2 * np.random.randn(n_samples)
        ch3 = 0.6 * np.sin(2 * np.pi * 8 * t)  + 0.2 * np.random.randn(n_samples)

        return ch1.tolist(), ch2.tolist(), ch3.tolist()

```

### serial_eeg.py

```python
import serial
import time
import threading
import numpy as np

class SerialEEGReader:
    def __init__(self, port, baud, fs, vref=3.3, adc_max=65535):
        self.port = port
        self.baud = baud
        self.fs = fs
        self.vref = vref
        self.adc_max = adc_max

        self.latest_samples = np.zeros(3, dtype=np.float32)
        self.lock = threading.Lock()
        self.running = False

    def raw_to_volts(self, raw: int) -> float:
        return raw * self.vref / self.adc_max

    def start(self):
        self.running = True
        thread = threading.Thread(target=self._read_loop, daemon=True)
        thread.start()

    def stop(self):
        self.running = False

    def _read_loop(self):
        with serial.Serial(self.port, self.baud, timeout=1) as ser:
            time.sleep(2)

            next_time = time.time()
            period = 1.0 / self.fs

            while self.running:
                line = ser.readline().decode("utf-8", errors="ignore").strip()
                if not line:
                    continue

                parts = line.split(",")
                if len(parts) != 3:
                    continue

                try:
                    raw0, raw1, raw2 = map(int, parts)
                except ValueError:
                    continue

                sample = np.array([
                    self.raw_to_volts(raw0),
                    self.raw_to_volts(raw1),
                    self.raw_to_volts(raw2)
                ], dtype=np.float32)

                with self.lock:
                    self.latest_samples = sample

                next_time += period
                sleep_time = next_time - time.time()
                if sleep_time > 0:
                    time.sleep(sleep_time)

    def get_latest(self):
        with self.lock:
            return self.latest_samples.copy()

```

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