find_best_genre.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [],
   "source": [
    "import argparse\n",
    "import time\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import optparse\n",
    "from dotenv import load_dotenv\n",
    "import random\n",
    "import soundfile as sf\n",
    "import sounddevice as sd\n",
    "\n",
    "from brainflow.board_shim import BoardShim, BrainFlowInputParams, BoardIds, BrainFlowPresets\n",
    "from brainflow import ml_model\n",
    "from brainflow.ml_model import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_random_clip_file(genre: str) -> str:\n",
    "    i = random.randint(0, 9)\n",
    "    return os.path.join(\"genre_clip_data\", genre, f'{genre}.0000{i}.wav')\n",
    "\n",
    "def play_random_clip_file(genre: str) -> None:\n",
    "    data, fs = sf.read(get_random_clip_file(genre), dtype='float32')\n",
    "    sd.play(data, fs)\n",
    "\n",
    "def get_preds_for_genre(genre: str, t_steps_per_pred: int = 50, recording_time: float = 5.0):\n",
    "    # prepare device\n",
    "    load_dotenv()\n",
    "    BoardShim.enable_dev_board_logger()\n",
    "\n",
    "    board_id = 0\n",
    "    serial_port = os.getenv(\"SERIAL_PORT\")\n",
    "\n",
    "    params = BrainFlowInputParams()\n",
    "    params.serial_port = serial_port\n",
    "    board = BoardShim(board_id, params)\n",
    "    board.prepare_session()\n",
    "\n",
    "    play_random_clip_file(genre)\n",
    "    board.start_stream()\n",
    "    time.sleep(recording_time)\n",
    "    data = board.get_board_data()\n",
    "    board.stop_stream()\n",
    "    board.release_session()\n",
    "\n",
    "    model = MLModel(BrainFlowModelParams(1, 0))\n",
    "    num_preds = data.shape[1] // t_steps_per_pred\n",
    "    preds = []\n",
    "    for i in range(num_preds):\n",
    "        model.prepare()\n",
    "        prediction = model.predict(data[:, i * t_steps_per_pred : (i+1) * t_steps_per_pred])\n",
    "        model.release()\n",
    "        preds.append(prediction)\n",
    "    \n",
    "    return preds"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.0"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "np.mean(get_preds_for_genre(\"classical\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "250"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# prepare device\n",
    "load_dotenv()\n",
    "BoardShim.enable_dev_board_logger()\n",
    "\n",
    "board_id = 0\n",
    "serial_port = os.getenv(\"SERIAL_PORT\")\n",
    "\n",
    "params = BrainFlowInputParams()\n",
    "params.serial_port = serial_port\n",
    "board = BoardShim(board_id, params)\n",
    "board.prepare_session()\n",
    "board.get_sampling_rate(board_id)   # per second, presumably"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "board.start_stream ()\n",
    "time.sleep(2)\n",
    "# data = board.get_current_board_data (256) # get latest 256 packages or less, doesnt remove them from internal buffer\n",
    "data = board.get_board_data()  # get all data and remove it from internal buffer\n",
    "board.stop_stream()\n",
    "board.release_session()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = MLModel(BrainFlowModelParams(0, 0))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0 [1.]\n",
      "1 [1.]\n",
      "2 [1.]\n",
      "3 [1.]\n",
      "4 [1.]\n",
      "5 [1.]\n",
      "6 [1.]\n",
      "7 [1.]\n"
     ]
    }
   ],
   "source": [
    "t_steps_per_pred = 50\n",
    "num_preds = data.shape[1] // t_steps_per_pred\n",
    "for i in range(num_preds):\n",
    "    model.prepare()\n",
    "    prediction = model.predict(data[:, i * t_steps_per_pred : (i+1) * t_steps_per_pred])\n",
    "    model.release()\n",
    "    print(i, prediction)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}