eeg_ascii.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Converts .eeg file to pandas dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import numpy as np\n",
    "import struct\n",
    "\n",
    "def eeg_to_ascii(file_name, chanlist='all', triallist='all', typerange='all', accepttype='all', rtrange='all', responsetype='all', data_format='auto'):\n",
    "    \"\"\"This function reads the data from a binary EEG file, extracts and scales the data, and returns it in ASCII format in volts.\n",
    "\n",
    "    Parameters:\n",
    "\n",
    "    - **file_name (str)**: The name of the binary EEG file you want to read.\n",
    "    - **chanlist (str or list)**: List of channels to read from the file. Default is 'all', which reads all channels in the file.\n",
    "    - **triallist (str or list)**: List of trials to read from the file. Default is 'all', which reads all trials in the file.\n",
    "    - **typerange (str, list or tuple)**: Range of type codes to read from the file. Default is 'all', which reads all type codes in the file.\n",
    "    - **accepttype (str, list or tuple)**: Range of accept codes to read from the file. Default is 'all', which reads all accept codes in the file.\n",
    "    - **rtrange (str, list or tuple)**: Range of reaction times to read from the file. Default is 'all', which reads all reaction times in the file.\n",
    "    - **responsetype (str, list, or tuple)**: Range of response codes to read from the file. Default is 'all', which reads all response codes in the file.\n",
    "    - **data_format (str)**: The format in which to return the data. Default is 'auto', which returns the data in its native format.\n",
    "\n",
    "    Returns:\n",
    "\n",
    "    - **data (numpy.ndarray)**: A 3D array containing the EEG data (in volts) for each trial, channel, and time point.\n",
    "    - **chan_names (list of str)**: A list of channel names.\n",
    "    - **rate (int)**: The sample rate (in Hz) of the EEG data.\n",
    "    - **xmin (float)**: The minimum time (in seconds) of the data.\n",
    "    - **swep_headers (list of tuples)**: A list of tuples containing the sweep header information for each trial. Each tuple contains the following information: (accept, ttype, correct, rt, response, sweep_num).\"\"\"\n",
    "    \n",
    "\n",
    "    #Check if file ends with .eeg\n",
    "    if not file_name.endswith('.eeg'):\n",
    "        raise ValueError(\"File must be a binary EEG file (.eeg).\")\n",
    "\n",
    "    if not os.path.isfile(file_name):\n",
    "        raise ValueError(f\"File {file_name} not found.\")\n",
    "    \n",
    "    with open(file_name, 'rb') as f:\n",
    "        try:\n",
    "            # Read general part of the ERP header and set variables\n",
    "            f.read(20) # skip revision number\n",
    "            f.read(342) # skip the first 362 bytes\n",
    "\n",
    "            nsweeps = struct.unpack('<H', f.read(2))[0]  # number of sweeps\n",
    "            f.read(4)  # skip 4 bytes\n",
    "            pnts = struct.unpack('<H', f.read(2))[0]  # number of points per waveform\n",
    "            chan = struct.unpack('<H', f.read(2))[0]  # number of channels\n",
    "            f.read(4)  # skip 4 bytes\n",
    "            rate = struct.unpack('<H', f.read(2))[0]  # sample rate (Hz)\n",
    "            f.read(127)  # skip 127 bytes\n",
    "            xmin = struct.unpack('<f', f.read(4))[0]  # in s\n",
    "            xmax = struct.unpack('<f', f.read(4))[0]  # in s\n",
    "            f.read(387)  # skip 387 bytes\n",
    "\n",
    "            # Read electrode configuration\n",
    "            chan_names = []\n",
    "            baselines = []\n",
    "            sensitivities = []\n",
    "            calibs = []\n",
    "            factors = []\n",
    "            for elec in range(chan):\n",
    "                chan_name = f.read(10).decode('ascii').strip('\\x00')\n",
    "                chan_names.append(chan_name)\n",
    "                f.read(37)  # skip 37 bytes\n",
    "                baseline = struct.unpack('<H', f.read(2))[0]\n",
    "                baselines.append(baseline)\n",
    "                f.read(10)  # skip 10 bytes\n",
    "                sensitivity = struct.unpack('<f', f.read(4))[0]\n",
    "                sensitivities.append(sensitivity)\n",
    "                f.read(8)  # skip 8 bytes\n",
    "                calib = struct.unpack('<f', f.read(4))[0]\n",
    "                calibs.append(calib)\n",
    "                factor = calib * sensitivity / 204.8\n",
    "                factors.append(factor)\n",
    "\n",
    "        except struct.error:\n",
    "            raise ValueError(\"Error reading binary file. File may be corrupted or not in the expected format.\")\n",
    "        except Exception as e:\n",
    "            raise ValueError(f\"Error reading file: {e}\")\n",
    "\n",
    "    # Read and process epoch datapoints data\n",
    "    data = np.empty((nsweeps, len(chan_names), pnts), dtype=float)\n",
    "    sweep_headers = []\n",
    "\n",
    "    # Constants for the sweep header size in bytes and data point size in bytes\n",
    "    SWEEP_HEAD_SIZE = 13\n",
    "    DATA_POINT_SIZE = 4\n",
    "\n",
    "    with open(file_name, 'rb') as f:\n",
    "        # Ensure the file pointer is at the beginning of the EEG data\n",
    "        f.seek((900 + chan * 75))\n",
    "\n",
    "        for sweep in range(nsweeps):\n",
    "            # Read the sweep header\n",
    "            try:\n",
    "                # f.read(SWEEP_HEAD_SIZE)\n",
    "                accept = struct.unpack('<c', f.read(1))[0]\n",
    "                ttype = struct.unpack('<h', f.read(2))[0]\n",
    "                correct = struct.unpack('<h', f.read(2))[0]\n",
    "                rt = struct.unpack('<f', f.read(4))[0]\n",
    "                response = struct.unpack('<h', f.read(2))[0]\n",
    "                # reserved  struct.unpack('<h', f.read(2))[0]\n",
    "                f.read(2) # skip 2 bytes\n",
    "                sweep_headers.append((accept, ttype, correct, rt, response, sweep))\n",
    "            except struct.error:\n",
    "                raise ValueError(\"Error reading sweep header. File may be corrupted or not in the expected format.\")\n",
    "            except Exception as e:\n",
    "                raise ValueError(f\"Error reading sweep header: {e}\")\n",
    "\n",
    "            for point in range(pnts):\n",
    "                for channel in range(chan):\n",
    "                    try:\n",
    "                        # Read the data point as a 4-byte integer\n",
    "                        value = struct.unpack('<l', f.read(DATA_POINT_SIZE))[0]\n",
    "\n",
    "                        # Scale the data point to microvolts and store it in the data array\n",
    "                        data[sweep, channel, point] = value * factors[channel]\n",
    "                    except struct.error:\n",
    "                        raise ValueError(\"Error reading data points. File may be corrupted or not in the expected format.\")\n",
    "                    except Exception as e:\n",
    "                        raise ValueError(f\"Error reading data points: {e}\")\n",
    "\n",
    "    #Convert data from microvolts to volts\n",
    "    data = data * 1e-6\n",
    "    # Return relevant data in ASCII format\n",
    "    return data, chan_names, rate, xmin, sweep_headers"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example Usage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "file_name = \"/home/woess/current assignment/read_neuro_scratch/16017_rest_ec.eeg\"\n",
    "\n",
    "\n",
    "eeg_data, channels, sample_rate, xmix = eeg_to_ascii(file_name)\n",
    "print(\"Channels:\", channels)\n",
    "print(\"EEG data:\", eeg_data)\n",
    "print(\"Sample rate:\", sample_rate)\n",
    "print(\"Xmin:\", xmix)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Place eeg_data into a pandas dataframe\n",
    "import pandas as pd\n",
    "import matplotlib.pyplot as plt\n",
    "\n",
    "df = pd.DataFrame(eeg_data[0].transpose(), columns=channels)\n",
    "#Display decimals (change '10' to the number of decimals you want to display)\n",
    "pd.set_option('display.float_format', lambda x: '%.50f' % x)\n",
    "df\n",
    "\n",
    "# #plot the data\n",
    "# plt.plot(df.index.to_list(),df[\"O2\"].to_list())\n",
    "#plt.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "mne",
   "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.10.10"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}