diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c18dd8d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/README.md b/README.md index 7bbe5e3..cb16150 100644 --- a/README.md +++ b/README.md @@ -1 +1,143 @@ -# NeuroLog \ No newline at end of file +# NeuroLog + +NeuroLog is a Python 3.10+ desktop application for real-time EEG acquisition from an Emotiv EPOC X+ headset via the Emotiv Cortex WebSocket API (`wss://localhost:6868`). + +It provides: +- Secure Cortex connection + authentication (`clientId` / `clientSecret`) +- Headset discovery and active session creation +- EEG stream subscription and live plotting in a PyQt6 GUI +- Recording controls with manual markers +- Session metadata capture (subject, experiment, notes) +- Export formats: CSV, NumPy `.npy`, and MNE `.fif` + +--- + +## Project Structure + +```text +NeuroLog/ +├── neuro_log/ +│ ├── __init__.py +│ ├── app.py # Entry point +│ ├── api.py # Cortex API client +│ ├── gui.py # PyQt6 GUI and live plotting +│ ├── recorder.py # Recording buffer and file exporters +│ └── utils.py # Helpers and environment config +├── requirements.txt +└── README.md +``` + +--- + +## Installation + +1. **Create and activate a virtual environment**: + +```bash +python -m venv .venv +source .venv/bin/activate # Linux/macOS +# .venv\Scripts\activate # Windows +``` + +2. **Install dependencies**: + +```bash +pip install -r requirements.txt +``` + +3. **Ensure Emotiv Cortex service is running** on your machine. + +--- + +## Configure Cortex Credentials + +NeuroLog reads credentials from environment variables: + +- `EMOTIV_CLIENT_ID` +- `EMOTIV_CLIENT_SECRET` +- optional: `EMOTIV_SSL_VERIFY` (`1` to enforce TLS cert verification, default `0`) + +Example: + +```bash +export EMOTIV_CLIENT_ID="your_client_id" +export EMOTIV_CLIENT_SECRET="your_client_secret" +export EMOTIV_SSL_VERIFY="0" +``` + +> Obtain client credentials from your Emotiv developer account and Cortex app configuration. + +--- + +## Connect the Emotiv Headset + +1. Power on your **Emotiv EPOC X+** headset. +2. Plug in the Emotiv USB receiver/dongle. +3. Open Emotiv Launcher / Cortex service and verify the device is paired. +4. Start NeuroLog and click **Connect**. +5. NeuroLog will: + - request access + - authorize + - query available headsets + - connect to the first available headset + - create an active session + - subscribe to EEG stream + +--- + +## Run the Application + +```bash +python -m neuro_log.app +``` + +## Quick Start (Avoid the "Not Ready" popup) + +1. Launch app: `python -m neuro_log.app` +2. Click **Connect** and wait until status shows connected/subscribed. +3. Confirm sampling rate is populated. +4. Click **Start Recording**. + +`Start Recording` is intentionally disabled until EEG subscription is ready. + +--- + +GUI features: +- **Live plot** of recent EEG activity +- **Start Recording** / **Stop Recording** buttons +- **Manual marker** input (`label`, optional `value`) +- **Session metadata** input (subject, experiment, notes) +- Status labels for: + - connection status + - sampling rate + - battery level + +--- + +## Recording and Export Formats + +On stop recording, NeuroLog writes files to `recordings/`: + +1. **CSV** (`*.csv`) + - Columns: `timestamp` + EEG channel labels + - Useful for quick inspection and spreadsheet workflows + +2. **NumPy array** (`*.npy`) + - Shape: `(n_samples, n_channels)` + - Efficient for scientific Python pipelines + +3. **MNE FIF** (`*.fif`) + - Created with `mne.io.RawArray` + - Contains EEG channels, sampling rate, and marker annotations + - Ready for advanced EEG processing in MNE + +4. **Marker CSV** (`*_markers.csv`) + - Manual marker timestamps + labels/values + +--- + +## Notes + +- The Cortex stream payload layout can vary by SDK version. This project includes practical defaults for EEG and battery updates, but you may adapt parsing for your specific firmware/API version. +- If certificate verification fails on localhost, use `EMOTIV_SSL_VERIFY=0`. +- For production/lab deployments, enable TLS verification whenever possible. diff --git a/neuro_log/__init__.py b/neuro_log/__init__.py new file mode 100644 index 0000000..376ca78 --- /dev/null +++ b/neuro_log/__init__.py @@ -0,0 +1,9 @@ +"""NeuroLog package for real-time EEG acquisition and recording.""" + +__all__ = [ + "api", + "app", + "gui", + "recorder", + "utils", +] diff --git a/neuro_log/api.py b/neuro_log/api.py new file mode 100644 index 0000000..a44e1ac --- /dev/null +++ b/neuro_log/api.py @@ -0,0 +1,322 @@ +"""Cortex API client for Emotiv EEG streaming over secure WebSocket. + +This module wraps the JSON-RPC workflow required by the Emotiv Cortex API: +1. Connect to `wss://localhost:6868` +2. Request access and authenticate +3. Query headset information +4. Create an active session +5. Subscribe to EEG and motion/system streams + +The client runs the WebSocket in a background thread and emits parsed updates +through callback hooks suitable for GUI integration. +""" + +from __future__ import annotations + +import json +import logging +import queue +import ssl +import threading +import time +from dataclasses import dataclass +from typing import Any, Callable + +import websocket + +LOGGER = logging.getLogger(__name__) + + +@dataclass(slots=True) +class CortexConfig: + """Configuration required to authenticate against Cortex.""" + + client_id: str + client_secret: str + debit: int = 1 + license: str = "" + host: str = "localhost" + port: int = 6868 + ssl_verify: bool = False + + @property + def url(self) -> str: + """Return the secure websocket endpoint URL.""" + return f"wss://{self.host}:{self.port}" + + +class CortexApiClient: + """Threaded client for Cortex JSON-RPC and EEG stream subscription.""" + + def __init__(self, config: CortexConfig) -> None: + self.config = config + self.ws: websocket.WebSocketApp | None = None + self.ws_thread: threading.Thread | None = None + + self._running = threading.Event() + self._opened = threading.Event() + self._authorized = threading.Event() + + self._request_id = 0 + self._pending: dict[int, queue.Queue[dict[str, Any]]] = {} + self._lock = threading.Lock() + + self.cortex_token: str | None = None + self.session_id: str | None = None + self.headset_id: str | None = None + + self.channel_labels: list[str] = [] + self.sampling_rate_hz: float | None = None + self.battery_percent: int | None = None + + self.on_status: Callable[[str], None] | None = None + self.on_eeg: Callable[[dict[str, Any]], None] | None = None + self.on_battery: Callable[[int], None] | None = None + + def connect(self, timeout_s: float = 10.0) -> None: + """Connect and complete the authentication/session setup flow.""" + self._set_status("Connecting to Cortex...") + self._running.set() + + sslopt: dict[str, Any] + if self.config.ssl_verify: + sslopt = {} + else: + sslopt = {"cert_reqs": ssl.CERT_NONE, "check_hostname": False} + + self.ws = websocket.WebSocketApp( + self.config.url, + on_open=self._on_open, + on_message=self._on_message, + on_error=self._on_error, + on_close=self._on_close, + ) + + self.ws_thread = threading.Thread( + target=self.ws.run_forever, + kwargs={"sslopt": sslopt, "ping_interval": 20, "ping_timeout": 10}, + daemon=True, + name="CortexWebSocketThread", + ) + self.ws_thread.start() + + if not self._opened.wait(timeout_s): + raise TimeoutError("Timed out while connecting to Cortex websocket.") + + self._setup_authorized_session(timeout_s=timeout_s) + self._set_status("Connected and streaming-ready") + + def disconnect(self) -> None: + """Close session and stop websocket thread gracefully.""" + self._running.clear() + if self.ws: + try: + if self.session_id and self.cortex_token: + self._rpc( + "updateSession", + { + "cortexToken": self.cortex_token, + "session": self.session_id, + "status": "close", + }, + timeout_s=3, + ) + except Exception: + LOGGER.exception("Failed to close Cortex session cleanly") + self.ws.close() + + if self.ws_thread and self.ws_thread.is_alive(): + self.ws_thread.join(timeout=2) + + self._set_status("Disconnected") + + def subscribe_eeg(self) -> dict[str, Any]: + """Subscribe to EEG and battery/system streams for updates.""" + if not self.cortex_token or not self.session_id: + raise RuntimeError("Client must be connected/authenticated before subscribing.") + + response = self._rpc( + "subscribe", + { + "cortexToken": self.cortex_token, + "session": self.session_id, + "streams": ["eeg", "dev"], + }, + ) + + for stream_info in response.get("success", []): + stream_name = stream_info.get("streamName") + if stream_name == "eeg": + cols = stream_info.get("cols", []) + self.channel_labels = [ + col for col in cols if col not in {"COUNTER", "INTERPOLATED", "MARKER_HARDWARE", "MARKERS"} + ] + self.sampling_rate_hz = stream_info.get("sampleRate") + + if not self.channel_labels: + # Fallback for SDK variants that omit EEG labels in subscribe metadata. + self.channel_labels = [ + "AF3", + "F7", + "F3", + "FC5", + "T7", + "P7", + "O1", + "O2", + "P8", + "T8", + "FC6", + "F4", + "F8", + "AF4", + ] + + if not self.sampling_rate_hz: + self.sampling_rate_hz = 128.0 + + self._set_status("Subscribed to EEG stream") + return response + + def create_marker(self, label: str, value: str = "", port: str = "python") -> dict[str, Any]: + """Inject a manual marker into the current session.""" + if not self.cortex_token or not self.session_id: + raise RuntimeError("Cannot create marker before session is active.") + + return self._rpc( + "injectMarker", + { + "cortexToken": self.cortex_token, + "session": self.session_id, + "label": label, + "value": value, + "port": port, + "time": int(time.time() * 1000), + }, + ) + + def _setup_authorized_session(self, timeout_s: float) -> None: + self._rpc( + "requestAccess", + {"clientId": self.config.client_id, "clientSecret": self.config.client_secret}, + timeout_s=timeout_s, + ) + + auth = self._rpc( + "authorize", + { + "clientId": self.config.client_id, + "clientSecret": self.config.client_secret, + "debit": self.config.debit, + "license": self.config.license, + }, + timeout_s=timeout_s, + ) + self.cortex_token = auth.get("cortexToken") + if not self.cortex_token: + raise RuntimeError(f"Authorization failed: {auth}") + + query = self._rpc("queryHeadsets", {}, timeout_s=timeout_s) + headsets = query if isinstance(query, list) else query.get("headsets", []) + if not headsets: + raise RuntimeError("No Emotiv headset found. Check dongle/headset pairing.") + + headset = headsets[0] + self.headset_id = headset.get("id") + if not self.headset_id: + raise RuntimeError(f"Headset payload missing id: {headset}") + + self._rpc( + "controlDevice", + {"command": "connect", "headset": self.headset_id}, + timeout_s=timeout_s, + ) + + session = self._rpc( + "createSession", + { + "cortexToken": self.cortex_token, + "headset": self.headset_id, + "status": "active", + }, + timeout_s=timeout_s, + ) + self.session_id = session.get("id") + if not self.session_id: + raise RuntimeError(f"Could not create Cortex session: {session}") + + self._authorized.set() + + def _rpc(self, method: str, params: dict[str, Any], timeout_s: float = 8.0) -> dict[str, Any]: + if not self.ws: + raise RuntimeError("WebSocket is not initialized") + + request_id = self._next_id() + msg = {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params} + + response_queue: queue.Queue[dict[str, Any]] = queue.Queue(maxsize=1) + self._pending[request_id] = response_queue + + self.ws.send(json.dumps(msg)) + + try: + raw = response_queue.get(timeout=timeout_s) + except queue.Empty as exc: + self._pending.pop(request_id, None) + raise TimeoutError(f"RPC timeout for method '{method}'") from exc + + error = raw.get("error") + if error: + raise RuntimeError(f"Cortex RPC error for {method}: {error}") + + return raw.get("result", {}) + + def _next_id(self) -> int: + with self._lock: + self._request_id += 1 + return self._request_id + + def _on_open(self, _: websocket.WebSocketApp) -> None: + self._set_status("WebSocket open") + self._opened.set() + + def _on_message(self, _: websocket.WebSocketApp, message: str) -> None: + try: + payload = json.loads(message) + except json.JSONDecodeError: + LOGGER.warning("Received non-JSON payload: %s", message) + return + + if "id" in payload: + request_id = payload["id"] + response_queue = self._pending.pop(request_id, None) + if response_queue: + response_queue.put(payload) + return + + if "eeg" in payload: + if self.on_eeg: + self.on_eeg(payload) + return + + if "dev" in payload: + dev = payload.get("dev", []) + # Device payload varies by SDK version; battery often appears as int in index 2/3. + battery_candidates = [v for v in dev if isinstance(v, int)] + if battery_candidates: + self.battery_percent = max(0, min(100, battery_candidates[-1])) + if self.on_battery: + self.on_battery(self.battery_percent) + return + + def _on_error(self, _: websocket.WebSocketApp, error: Any) -> None: + LOGGER.error("Cortex websocket error: %s", error) + self._set_status(f"Connection error: {error}") + + def _on_close(self, _: websocket.WebSocketApp, close_status_code: int, close_msg: str) -> None: + self._set_status(f"WebSocket closed ({close_status_code}): {close_msg}") + + def _set_status(self, status: str) -> None: + LOGGER.info("Cortex status: %s", status) + if self.on_status: + self.on_status(status) diff --git a/neuro_log/app.py b/neuro_log/app.py new file mode 100644 index 0000000..fec6809 --- /dev/null +++ b/neuro_log/app.py @@ -0,0 +1,33 @@ +"""Application entry point for NeuroLog.""" + +from __future__ import annotations + +import sys + +from PyQt6.QtWidgets import QApplication, QMessageBox + +from neuro_log.api import CortexApiClient +from neuro_log.gui import NeuroLogMainWindow +from neuro_log.utils import build_cortex_config_from_env, configure_logging + + +def main() -> int: + """Launch the NeuroLog GUI application.""" + configure_logging() + + app = QApplication(sys.argv) + + try: + config = build_cortex_config_from_env() + except Exception as exc: + QMessageBox.critical(None, "Configuration Error", str(exc)) + return 1 + + client = CortexApiClient(config) + window = NeuroLogMainWindow(client) + window.show() + return app.exec() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/neuro_log/gui.py b/neuro_log/gui.py new file mode 100644 index 0000000..0ec2e11 --- /dev/null +++ b/neuro_log/gui.py @@ -0,0 +1,281 @@ +"""PyQt6 GUI for real-time NeuroLog EEG acquisition and recording.""" + +from __future__ import annotations + +import time +from collections import deque +from pathlib import Path + +from matplotlib.backends.backend_qtagg import FigureCanvasQTAgg as FigureCanvas +from matplotlib.figure import Figure +from PyQt6.QtCore import QTimer +from PyQt6.QtWidgets import ( + QFormLayout, + QGridLayout, + QGroupBox, + QHBoxLayout, + QLabel, + QLineEdit, + QMainWindow, + QMessageBox, + QPushButton, + QTextEdit, + QVBoxLayout, + QWidget, +) + +from neuro_log.api import CortexApiClient +from neuro_log.recorder import EEGRecorder, SessionMetadata + + +class NeuroLogMainWindow(QMainWindow): + """Main application window for EEG streaming and recording control.""" + + def __init__(self, api_client: CortexApiClient) -> None: + super().__init__() + self.api_client = api_client + self.recorder = EEGRecorder(output_dir=Path("recordings")) + + self.setWindowTitle("NeuroLog - Emotiv EEG Recorder") + self.resize(1200, 800) + + self.channel_labels: list[str] = [] + self.plot_window_s = 8 + self.plot_buffers: dict[str, deque[float]] = {} + self.plot_time: deque[float] = deque(maxlen=1024) + self.connected_and_subscribed = False + + self._build_ui() + self._configure_callbacks() + + self.redraw_timer = QTimer(self) + self.redraw_timer.timeout.connect(self._refresh_plot) + self.redraw_timer.start(120) + + def _build_ui(self) -> None: + root = QWidget() + root_layout = QVBoxLayout(root) + + status_group = QGroupBox("Status") + status_layout = QGridLayout(status_group) + self.connection_status = QLabel("Disconnected") + self.sampling_rate_label = QLabel("-") + self.battery_label = QLabel("-") + self.hint_label = QLabel("Click Connect first, then Start Recording.") + status_layout.addWidget(QLabel("Connection:"), 0, 0) + status_layout.addWidget(self.connection_status, 0, 1) + status_layout.addWidget(QLabel("Sampling rate:"), 0, 2) + status_layout.addWidget(self.sampling_rate_label, 0, 3) + status_layout.addWidget(QLabel("Battery:"), 0, 4) + status_layout.addWidget(self.battery_label, 0, 5) + status_layout.addWidget(self.hint_label, 1, 0, 1, 6) + + control_group = QGroupBox("Recording Controls") + control_layout = QHBoxLayout(control_group) + self.connect_button = QPushButton("Connect") + self.start_button = QPushButton("Start Recording") + self.stop_button = QPushButton("Stop Recording") + self.start_button.setEnabled(False) + self.stop_button.setEnabled(False) + + self.marker_label_input = QLineEdit() + self.marker_label_input.setPlaceholderText("marker label") + self.marker_value_input = QLineEdit() + self.marker_value_input.setPlaceholderText("optional value") + self.add_marker_button = QPushButton("Add Marker") + self.add_marker_button.setEnabled(False) + + control_layout.addWidget(self.connect_button) + control_layout.addWidget(self.start_button) + control_layout.addWidget(self.stop_button) + control_layout.addWidget(self.marker_label_input) + control_layout.addWidget(self.marker_value_input) + control_layout.addWidget(self.add_marker_button) + + metadata_group = QGroupBox("Session Metadata") + metadata_layout = QFormLayout(metadata_group) + self.subject_input = QLineEdit() + self.experiment_input = QLineEdit() + self.notes_input = QTextEdit() + self.notes_input.setMaximumHeight(100) + + metadata_layout.addRow("Subject:", self.subject_input) + metadata_layout.addRow("Experiment:", self.experiment_input) + metadata_layout.addRow("Notes:", self.notes_input) + + self.figure = Figure(figsize=(10, 5)) + self.canvas = FigureCanvas(self.figure) + self.axes = self.figure.add_subplot(111) + self.axes.set_title("Live EEG") + self.axes.set_xlabel("Time (s)") + self.axes.set_ylabel("Amplitude (uV)") + + root_layout.addWidget(status_group) + root_layout.addWidget(control_group) + root_layout.addWidget(metadata_group) + root_layout.addWidget(self.canvas, stretch=1) + + self.setCentralWidget(root) + + def _configure_callbacks(self) -> None: + self.api_client.on_status = self.connection_status.setText + self.api_client.on_eeg = self._on_eeg_message + self.api_client.on_battery = lambda val: self.battery_label.setText(f"{val}%") + + self.connect_button.clicked.connect(self._connect_client) + self.start_button.clicked.connect(self._start_recording) + self.stop_button.clicked.connect(self._stop_recording) + self.add_marker_button.clicked.connect(self._add_marker) + + def _connect_client(self) -> None: + self.connect_button.setEnabled(False) + self.connection_status.setText("Connecting...") + + try: + self.api_client.connect() + self.api_client.subscribe_eeg() + except Exception as exc: + self.connect_button.setEnabled(True) + self.connected_and_subscribed = False + self.start_button.setEnabled(False) + self.add_marker_button.setEnabled(False) + QMessageBox.critical(self, "Connection Error", str(exc)) + return + + self.channel_labels = self.api_client.channel_labels + if not self.channel_labels: + self.connect_button.setEnabled(True) + self.connected_and_subscribed = False + self.start_button.setEnabled(False) + self.add_marker_button.setEnabled(False) + QMessageBox.critical( + self, + "Subscription Error", + "Connected to Cortex but no EEG channels were returned. " + "Check that the headset is connected and streaming EEG.", + ) + return + + sample_rate = self.api_client.sampling_rate_hz or 128.0 + self.sampling_rate_label.setText(f"{sample_rate} Hz") + self._init_plot_buffers(self.channel_labels) + + self.connected_and_subscribed = True + self.start_button.setEnabled(True) + self.add_marker_button.setEnabled(True) + self.connect_button.setText("Connected") + self.hint_label.setText("Connected. You can now start recording.") + + def _start_recording(self) -> None: + if not self.connected_and_subscribed: + QMessageBox.warning( + self, + "Not Ready", + "You must connect first. Click Connect and wait for EEG subscription to complete.", + ) + return + + metadata = SessionMetadata( + subject=self.subject_input.text().strip() or "unknown", + experiment_name=self.experiment_input.text().strip() or "baseline", + notes=self.notes_input.toPlainText().strip(), + ) + + self.recorder.start( + metadata=metadata, + channel_labels=self.channel_labels, + sampling_rate_hz=self.api_client.sampling_rate_hz or 128.0, + ) + + self.start_button.setEnabled(False) + self.stop_button.setEnabled(True) + self.hint_label.setText("Recording in progress...") + + def _stop_recording(self) -> None: + try: + saved_paths = self.recorder.stop() + except Exception as exc: + QMessageBox.warning(self, "Recording", str(exc)) + self.start_button.setEnabled(True) + self.stop_button.setEnabled(False) + self.hint_label.setText("Connected. You can start recording when ready.") + return + + self.start_button.setEnabled(True) + self.stop_button.setEnabled(False) + self.hint_label.setText("Recording stopped and exported.") + + file_lines = "\n".join(f"{fmt.upper()}: {path}" for fmt, path in saved_paths.items()) + QMessageBox.information(self, "Saved", f"Recording exported:\n{file_lines}") + + def _add_marker(self) -> None: + label = self.marker_label_input.text().strip() + value = self.marker_value_input.text().strip() + if not label: + QMessageBox.warning(self, "Marker", "Please provide a marker label.") + return + + now = time.time() + self.recorder.add_marker(now, label, value) + + try: + if self.api_client.session_id: + self.api_client.create_marker(label=label, value=value) + except Exception: + # Marker is still stored locally even if cortex injection fails. + pass + + self.marker_label_input.clear() + self.marker_value_input.clear() + + def _on_eeg_message(self, payload: dict) -> None: + if not self.channel_labels: + return + + try: + timestamp, values = EEGRecorder.extract_eeg_values(payload, len(self.channel_labels)) + except Exception: + return + + self.recorder.add_sample(timestamp, values) + + self.plot_time.append(timestamp) + for idx, label in enumerate(self.channel_labels): + self.plot_buffers[label].append(values[idx]) + + def _init_plot_buffers(self, channels: list[str]) -> None: + self.plot_time = deque(maxlen=2048) + self.plot_buffers = {label: deque(maxlen=2048) for label in channels} + + def _refresh_plot(self) -> None: + if not self.plot_time or not self.plot_buffers: + return + + t0 = self.plot_time[-1] + min_t = t0 - self.plot_window_s + xs = [t - t0 for t in self.plot_time if t >= min_t] + if not xs: + return + + self.axes.clear() + offset_step = 70.0 + + for idx, label in enumerate(self.channel_labels[:8]): + ys_full = list(self.plot_buffers[label]) + ys = ys_full[-len(xs) :] + shifted = [y + idx * offset_step for y in ys] + self.axes.plot(xs, shifted, linewidth=0.9, label=label) + + self.axes.set_title("Live EEG (last 8s, first 8 channels)") + self.axes.set_xlabel("Time relative to now (s)") + self.axes.set_ylabel("Amplitude + offset") + self.axes.legend(loc="upper left", ncols=4, fontsize=8) + self.axes.grid(alpha=0.2) + self.canvas.draw_idle() + + def closeEvent(self, event) -> None: # noqa: N802 + try: + self.api_client.disconnect() + except Exception: + pass + super().closeEvent(event) diff --git a/neuro_log/recorder.py b/neuro_log/recorder.py new file mode 100644 index 0000000..c34a526 --- /dev/null +++ b/neuro_log/recorder.py @@ -0,0 +1,152 @@ +"""Recording pipeline and file exporters for NeuroLog EEG sessions.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Any + +import mne +import numpy as np +import pandas as pd + +from neuro_log.utils import sanitize_filename + + +@dataclass(slots=True) +class SessionMetadata: + """User-provided metadata associated with one recording session.""" + + subject: str + experiment_name: str + notes: str = "" + + +@dataclass(slots=True) +class Marker: + """Manual marker inserted during recording.""" + + timestamp: float + label: str + value: str = "" + + +@dataclass +class EEGRecorder: + """Stateful EEG recorder that accumulates samples and persists to disk.""" + + output_dir: Path = Path("recordings") + recording: bool = False + metadata: SessionMetadata | None = None + channel_labels: list[str] = field(default_factory=list) + sampling_rate_hz: float = 128.0 + + timestamps: list[float] = field(default_factory=list) + samples: list[list[float]] = field(default_factory=list) + markers: list[Marker] = field(default_factory=list) + + def start(self, metadata: SessionMetadata, channel_labels: list[str], sampling_rate_hz: float) -> None: + """Start a fresh recording buffer with metadata and stream layout.""" + self.output_dir.mkdir(parents=True, exist_ok=True) + self.recording = True + self.metadata = metadata + self.channel_labels = channel_labels + self.sampling_rate_hz = sampling_rate_hz + + self.timestamps.clear() + self.samples.clear() + self.markers.clear() + + def stop(self) -> dict[str, Path]: + """Stop recording and save all outputs. + + Returns a mapping of file format to output path. + """ + self.recording = False + if not self.samples: + raise RuntimeError("No samples available to save.") + + base_name = self._build_base_name() + csv_path = self.output_dir / f"{base_name}.csv" + npy_path = self.output_dir / f"{base_name}.npy" + fif_path = self.output_dir / f"{base_name}.fif" + marker_path = self.output_dir / f"{base_name}_markers.csv" + + data_array = np.asarray(self.samples, dtype=np.float64) + self._save_csv(csv_path, data_array) + np.save(npy_path, data_array) + self._save_fif(fif_path, data_array) + self._save_markers(marker_path) + + return {"csv": csv_path, "npy": npy_path, "fif": fif_path, "markers": marker_path} + + def add_sample(self, timestamp_s: float, eeg_values: list[float]) -> None: + """Store one EEG sample frame when recording is enabled.""" + if not self.recording: + return + + if len(eeg_values) != len(self.channel_labels): + raise ValueError( + f"EEG sample length ({len(eeg_values)}) does not match channel count ({len(self.channel_labels)})." + ) + + self.timestamps.append(timestamp_s) + self.samples.append(eeg_values) + + def add_marker(self, timestamp_s: float, label: str, value: str = "") -> None: + """Store a manual marker.""" + self.markers.append(Marker(timestamp=timestamp_s, label=label, value=value)) + + def _save_csv(self, path: Path, data_array: np.ndarray) -> None: + columns = ["timestamp"] + self.channel_labels + df = pd.DataFrame(np.column_stack([self.timestamps, data_array]), columns=columns) + + if self.metadata: + df.attrs["subject"] = self.metadata.subject + df.attrs["experiment_name"] = self.metadata.experiment_name + df.attrs["notes"] = self.metadata.notes + + df.to_csv(path, index=False) + + def _save_fif(self, path: Path, data_array: np.ndarray) -> None: + # MNE expects shape (n_channels, n_times) + data = data_array.T + info = mne.create_info(ch_names=self.channel_labels, sfreq=self.sampling_rate_hz, ch_types="eeg") + raw = mne.io.RawArray(data, info, verbose="ERROR") + + for marker in self.markers: + onset = max(0.0, marker.timestamp - self.timestamps[0]) + raw.annotations.append(onset=onset, duration=0.0, description=marker.label) + + if self.metadata: + raw.info["subject_info"] = {"his_id": self.metadata.subject} + raw.info["description"] = f"experiment={self.metadata.experiment_name}; notes={self.metadata.notes}" + + raw.save(path, overwrite=True, verbose="ERROR") + + def _save_markers(self, path: Path) -> None: + marker_df = pd.DataFrame( + [ + {"timestamp": marker.timestamp, "label": marker.label, "value": marker.value} + for marker in self.markers + ] + ) + marker_df.to_csv(path, index=False) + + def _build_base_name(self) -> str: + now = datetime.now().strftime("%Y%m%d_%H%M%S") + subject = self.metadata.subject if self.metadata else "unknown" + experiment = self.metadata.experiment_name if self.metadata else "session" + return sanitize_filename(f"{subject}_{experiment}_{now}") + + @staticmethod + def extract_eeg_values(eeg_payload: dict[str, Any], channel_count: int) -> tuple[float, list[float]]: + """Parse Cortex EEG payload into timestamp and channel vector.""" + eeg = eeg_payload.get("eeg", []) + if len(eeg) < channel_count: + raise ValueError(f"Unexpected EEG payload length: {len(eeg)}") + + timestamp = float(eeg_payload.get("time", 0.0)) / 1000.0 + values = [float(v) for v in eeg[:channel_count]] + return timestamp, values diff --git a/neuro_log/utils.py b/neuro_log/utils.py new file mode 100644 index 0000000..cf52620 --- /dev/null +++ b/neuro_log/utils.py @@ -0,0 +1,56 @@ +"""General utility helpers for the NeuroLog application.""" + +from __future__ import annotations + +import logging +import os +import re +from dataclasses import dataclass + +from neuro_log.api import CortexConfig + + +def configure_logging() -> None: + """Initialize process-wide logging configuration.""" + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", + ) + + +@dataclass(slots=True) +class Credentials: + """Container for Cortex client credentials.""" + + client_id: str + client_secret: str + + +def load_credentials_from_env() -> Credentials: + """Load Cortex credentials from environment variables. + + Required environment variables: + - EMOTIV_CLIENT_ID + - EMOTIV_CLIENT_SECRET + """ + client_id = os.getenv("EMOTIV_CLIENT_ID", "").strip() + client_secret = os.getenv("EMOTIV_CLIENT_SECRET", "").strip() + if not client_id or not client_secret: + raise RuntimeError( + "Missing credentials. Set EMOTIV_CLIENT_ID and EMOTIV_CLIENT_SECRET environment variables." + ) + return Credentials(client_id=client_id, client_secret=client_secret) + + +def build_cortex_config_from_env() -> CortexConfig: + """Construct Cortex configuration from environment variables.""" + creds = load_credentials_from_env() + ssl_verify = os.getenv("EMOTIV_SSL_VERIFY", "0") == "1" + return CortexConfig(client_id=creds.client_id, client_secret=creds.client_secret, ssl_verify=ssl_verify) + + +def sanitize_filename(value: str) -> str: + """Convert arbitrary text into a filesystem-safe filename chunk.""" + value = value.strip().replace(" ", "_") + value = re.sub(r"[^a-zA-Z0-9_.-]", "", value) + return value or "recording" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..7faf504 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +PyQt6>=6.6 +websocket-client>=1.8 +matplotlib>=3.8 +numpy>=1.26 +pandas>=2.1 +mne>=1.6