From e11da2ca233e7fd4bda081d5121e49bebd2faf74 Mon Sep 17 00:00:00 2001 From: Kulkarni Rutuparna Ratnakar <151852360+K-Rutuparna1087@users.noreply.github.com> Date: Fri, 13 Feb 2026 16:39:26 +0530 Subject: [PATCH] Build NeuroLog Cortex EEG GUI application --- README.md | 87 +++++++++++++- neuro_log/__init__.py | 9 ++ neuro_log/api.py | 245 ++++++++++++++++++++++++++++++++++++++ neuro_log/app.py | 29 +++++ neuro_log/gui.py | 271 ++++++++++++++++++++++++++++++++++++++++++ neuro_log/recorder.py | 122 +++++++++++++++++++ neuro_log/utils.py | 76 ++++++++++++ requirements.txt | 5 + 8 files changed, 843 insertions(+), 1 deletion(-) create mode 100644 neuro_log/__init__.py create mode 100644 neuro_log/api.py create mode 100644 neuro_log/app.py create mode 100644 neuro_log/gui.py create mode 100644 neuro_log/recorder.py create mode 100644 neuro_log/utils.py create mode 100644 requirements.txt diff --git a/README.md b/README.md index 7bbe5e3..75d63c6 100644 --- a/README.md +++ b/README.md @@ -1 +1,86 @@ -# NeuroLog \ No newline at end of file +# NeuroLog + +NeuroLog is a Python desktop application for **real-time EEG acquisition** with the **Emotiv Cortex API**. It provides a PyQt6 GUI to connect to Cortex, view a live EEG waveform, start/stop recordings, add manual markers, and export recordings in common research formats. + +## Features + +- Secure WebSocket connection to Cortex (`wss://localhost:6868`) +- Authentication with `clientId` and `clientSecret` +- Headset discovery for Emotiv EPOC X+ +- Session creation and EEG stream subscription +- Real-time EEG plotting in GUI +- Start/Stop recording controls +- Metadata capture (subject, experiment, notes) +- Manual event markers during recording +- Export to: + - **CSV** (`timestamp` + channel columns) + - **NumPy** (`.npy` array) + - **FIF** (`.fif` via MNE, includes annotations) + +## Installation + +1. Ensure Python **3.10+** is installed. +2. (Recommended) create and activate a virtual environment. +3. Install dependencies: + +```bash +pip install -r requirements.txt +``` + +## Configure Cortex credentials + +Set your Emotiv Cortex credentials as environment variables before launching the app: + +```bash +export CORTEX_CLIENT_ID="your_client_id" +export CORTEX_CLIENT_SECRET="your_client_secret" +``` + +> On Windows PowerShell use `$env:CORTEX_CLIENT_ID="..."` and `$env:CORTEX_CLIENT_SECRET="..."`. + +## Running the application + +From the repository root: + +```bash +python -m neuro_log.app +``` + +## Connecting Emotiv EPOC X+ + +1. Turn on your EPOC X+ headset. +2. Open Emotiv Launcher and ensure Cortex service is running. +3. Make sure the headset is connected and available to Cortex. +4. In NeuroLog, click **Connect to Cortex**. +5. Once connected, click **Start Recording**. + +## Recording workflow + +1. Fill in session metadata fields (subject / experiment / notes). +2. Click **Start Recording**. +3. Use the annotation field and **Add Marker** to insert event markers. +4. Click **Stop + Save** to export files. + +Files are written to the `recordings/` folder with timestamped names. + +## Exported file formats + +For each session, NeuroLog saves: + +- `*.csv` — row-wise samples with first column `timestamp` +- `*.npy` — NumPy array of EEG samples (`n_samples x n_channels`) +- `*.fif` — MNE Raw file with EEG channels and marker annotations +- `*.json` — metadata and marker summary + +## Project structure + +```text +neuro_log/ + app.py # application entrypoint + api.py # Cortex API WebSocket client + gui.py # PyQt6 GUI and live plotting + recorder.py # recording buffers and export + utils.py # shared helpers and dataclasses +requirements.txt +README.md +``` diff --git a/neuro_log/__init__.py b/neuro_log/__init__.py new file mode 100644 index 0000000..d5376c3 --- /dev/null +++ b/neuro_log/__init__.py @@ -0,0 +1,9 @@ +"""NeuroLog package for real-time EEG acquisition with Emotiv Cortex.""" + +__all__ = [ + "api", + "app", + "gui", + "recorder", + "utils", +] diff --git a/neuro_log/api.py b/neuro_log/api.py new file mode 100644 index 0000000..feab5e9 --- /dev/null +++ b/neuro_log/api.py @@ -0,0 +1,245 @@ +"""Cortex API WebSocket client for Emotiv EEG streaming.""" + +from __future__ import annotations + +import json +import logging +import ssl +import threading +import time +from dataclasses import dataclass +from typing import Any, Callable + +from websocket import WebSocketApp + +LOGGER = logging.getLogger(__name__) + + +@dataclass(slots=True) +class HeadsetInfo: + """Basic headset information discovered from Cortex.""" + + headset_id: str + status: str + battery_percent: float | None + + +class CortexClient: + """Threaded JSON-RPC client for the Emotiv Cortex WebSocket API. + + This client manages: + - secure websocket connection to Cortex + - authentication and token management + - headset discovery and session creation + - stream subscription and callbacks for EEG packets + """ + + def __init__( + self, + client_id: str, + client_secret: str, + url: str = "wss://localhost:6868", + ) -> None: + self.client_id = client_id + self.client_secret = client_secret + self.url = url + + self._ws: WebSocketApp | None = None + self._thread: threading.Thread | None = None + self._connected = threading.Event() + self._stop_requested = threading.Event() + + self._msg_id = 1 + self._pending: dict[int, dict[str, Any]] = {} + self._pending_event = threading.Event() + self._pending_lock = threading.Lock() + + self.auth_token: str | None = None + self.session_id: str | None = None + + self.on_eeg: Callable[[dict[str, Any]], None] | None = None + self.on_status: Callable[[str], None] | None = None + self.on_battery: Callable[[float], None] | None = None + + @property + def connected(self) -> bool: + """Return whether websocket is currently connected.""" + return self._connected.is_set() + + def connect(self, timeout_s: float = 10.0) -> None: + """Connect to Cortex websocket endpoint and wait for readiness.""" + if self._thread and self._thread.is_alive(): + return + + self._stop_requested.clear() + self._ws = WebSocketApp( + self.url, + on_open=self._on_open, + on_message=self._on_message, + on_error=self._on_error, + on_close=self._on_close, + ) + + def _run() -> None: + assert self._ws is not None + # Cortex usually uses a self-signed certificate on localhost. + self._ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) + + self._thread = threading.Thread(target=_run, name="cortex-ws", daemon=True) + self._thread.start() + + if not self._connected.wait(timeout=timeout_s): + raise TimeoutError("Timed out waiting for Cortex websocket connection.") + + def close(self) -> None: + """Close websocket and stop background thread.""" + self._stop_requested.set() + if self._ws: + self._ws.close() + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=2) + self._connected.clear() + + def authorize(self) -> str: + """Request user authorization and return Cortex auth token.""" + result = self._rpc( + "authorize", + { + "clientId": self.client_id, + "clientSecret": self.client_secret, + "debit": 1, + }, + ) + token = result["cortexToken"] + self.auth_token = token + return token + + def query_headsets(self) -> list[HeadsetInfo]: + """Query available headsets and parse their metadata.""" + result = self._rpc("queryHeadsets", {}) + headsets: list[HeadsetInfo] = [] + for item in result: + battery = item.get("battery") + headsets.append( + HeadsetInfo( + headset_id=item.get("id", ""), + status=item.get("status", "unknown"), + battery_percent=float(battery) if battery is not None else None, + ) + ) + return headsets + + def create_session(self, headset_id: str) -> str: + """Create an active Cortex session for streaming.""" + if not self.auth_token: + raise RuntimeError("Cannot create session before authorize().") + + result = self._rpc( + "createSession", + { + "cortexToken": self.auth_token, + "headset": headset_id, + "status": "active", + }, + ) + self.session_id = result["id"] + return self.session_id + + def subscribe_eeg(self) -> dict[str, Any]: + """Subscribe current session to EEG stream.""" + if not self.auth_token or not self.session_id: + raise RuntimeError("Cannot subscribe without auth token and session.") + + return self._rpc( + "subscribe", + { + "cortexToken": self.auth_token, + "session": self.session_id, + "streams": ["eeg", "dev"], + }, + ) + + def update_session_status(self, status: str) -> dict[str, Any]: + """Update session status (active, close).""" + if not self.auth_token or not self.session_id: + raise RuntimeError("Session not initialized.") + + return self._rpc( + "updateSession", + { + "cortexToken": self.auth_token, + "session": self.session_id, + "status": status, + }, + ) + + def _on_open(self, _ws: WebSocketApp) -> None: + LOGGER.info("Connected to Cortex websocket.") + self._connected.set() + if self.on_status: + self.on_status("Connected") + + def _on_error(self, _ws: WebSocketApp, error: Any) -> None: + LOGGER.error("Websocket error: %s", error) + if self.on_status: + self.on_status(f"Error: {error}") + + def _on_close(self, _ws: WebSocketApp, _status_code: int, _msg: str) -> None: + LOGGER.info("Cortex websocket closed.") + self._connected.clear() + if self.on_status: + self.on_status("Disconnected") + + def _on_message(self, _ws: WebSocketApp, message: str) -> None: + payload = json.loads(message) + + # RPC response path + if "id" in payload: + with self._pending_lock: + self._pending[payload["id"]] = payload + self._pending_event.set() + return + + # Data stream path + if "eeg" in payload and self.on_eeg: + self.on_eeg(payload) + + if "dev" in payload: + dev = payload["dev"] + if isinstance(dev, list) and len(dev) > 2 and self.on_battery: + battery = dev[2] + try: + self.on_battery(float(battery)) + except (TypeError, ValueError): + pass + + def _rpc(self, method: str, params: dict[str, Any], timeout_s: float = 10.0) -> Any: + if not self._ws or not self.connected: + raise RuntimeError("WebSocket is not connected.") + + msg_id = self._msg_id + self._msg_id += 1 + + request = { + "jsonrpc": "2.0", + "id": msg_id, + "method": method, + "params": params, + } + self._ws.send(json.dumps(request)) + + start = time.monotonic() + while time.monotonic() - start < timeout_s: + with self._pending_lock: + response = self._pending.pop(msg_id, None) + if not self._pending: + self._pending_event.clear() + + if response is not None: + if "error" in response: + raise RuntimeError(f"Cortex error for {method}: {response['error']}") + return response.get("result") + + self._pending_event.wait(timeout=0.05) + + raise TimeoutError(f"Timed out waiting for Cortex RPC response: {method}") diff --git a/neuro_log/app.py b/neuro_log/app.py new file mode 100644 index 0000000..9e3baac --- /dev/null +++ b/neuro_log/app.py @@ -0,0 +1,29 @@ +"""Application entrypoint for NeuroLog.""" + +from __future__ import annotations + +import sys + +from PyQt6.QtWidgets import QApplication + +from neuro_log.api import CortexClient +from neuro_log.gui import NeuroLogWindow +from neuro_log.recorder import EEGRecorder +from neuro_log.utils import configure_logging, load_credentials + + +def main() -> int: + """Start the NeuroLog GUI application.""" + configure_logging() + creds = load_credentials() + + app = QApplication(sys.argv) + api_client = CortexClient(client_id=creds.client_id, client_secret=creds.client_secret) + recorder = EEGRecorder() + window = NeuroLogWindow(api_client=api_client, recorder=recorder) + 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..d078791 --- /dev/null +++ b/neuro_log/gui.py @@ -0,0 +1,271 @@ +"""PyQt6 GUI for real-time EEG visualization and recording controls.""" + +from __future__ import annotations + +import queue +import time +from collections import deque +from pathlib import Path + +import numpy as np +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, + QStatusBar, + QTextEdit, + QVBoxLayout, + QWidget, +) + +from neuro_log.api import CortexClient +from neuro_log.recorder import EEGRecorder +from neuro_log.utils import SessionMetadata, ensure_output_dir, now_utc_iso + + +class NeuroLogWindow(QMainWindow): + """Main NeuroLog window integrating stream management, plotting, and export.""" + + def __init__(self, api_client: CortexClient, recorder: EEGRecorder) -> None: + super().__init__() + self.api_client = api_client + self.recorder = recorder + + self.setWindowTitle("NeuroLog - Emotiv EEG Recorder") + self.resize(1100, 720) + + self._sample_queue: queue.Queue[dict] = queue.Queue() + self._plot_buffer: deque[list[float]] = deque(maxlen=500) + self._timestamps: deque[float] = deque(maxlen=500) + self._channel_labels: list[str] = [] + self._sample_rate_hz: float = 128.0 + + self._init_widgets() + self._bind_callbacks() + + self.refresh_timer = QTimer(self) + self.refresh_timer.timeout.connect(self._process_incoming_samples) + self.refresh_timer.start(40) + + def _init_widgets(self) -> None: + central = QWidget() + layout = QGridLayout(central) + + self.status_connection = QLabel("Disconnected") + self.status_sample_rate = QLabel("Sample rate: -- Hz") + self.status_battery = QLabel("Battery: --%") + + status_box = QGroupBox("Device Status") + status_layout = QFormLayout(status_box) + status_layout.addRow("Connection", self.status_connection) + status_layout.addRow("Sampling", self.status_sample_rate) + status_layout.addRow("Battery", self.status_battery) + + self.subject_input = QLineEdit() + self.experiment_input = QLineEdit() + self.notes_input = QTextEdit() + self.notes_input.setPlaceholderText("Session notes...") + + metadata_box = QGroupBox("Session Metadata") + metadata_layout = QFormLayout(metadata_box) + metadata_layout.addRow("Subject", self.subject_input) + metadata_layout.addRow("Experiment", self.experiment_input) + metadata_layout.addRow("Notes", self.notes_input) + + self.marker_input = QLineEdit() + self.marker_input.setPlaceholderText("e.g., stimulus_onset") + self.marker_button = QPushButton("Add Marker") + + marker_box = QGroupBox("Manual Annotation") + marker_layout = QHBoxLayout(marker_box) + marker_layout.addWidget(self.marker_input) + marker_layout.addWidget(self.marker_button) + + self.connect_button = QPushButton("Connect to Cortex") + self.start_button = QPushButton("Start Recording") + self.stop_button = QPushButton("Stop + Save") + self.stop_button.setEnabled(False) + + controls_box = QGroupBox("Controls") + controls_layout = QVBoxLayout(controls_box) + controls_layout.addWidget(self.connect_button) + controls_layout.addWidget(self.start_button) + controls_layout.addWidget(self.stop_button) + controls_layout.addStretch(1) + + self.figure = Figure(figsize=(8, 5), tight_layout=True) + self.canvas = FigureCanvas(self.figure) + self.axes = self.figure.add_subplot(111) + self.axes.set_title("Live EEG") + self.axes.set_xlabel("Samples") + self.axes.set_ylabel("Amplitude (uV)") + + layout.addWidget(status_box, 0, 0) + layout.addWidget(metadata_box, 1, 0) + layout.addWidget(marker_box, 2, 0) + layout.addWidget(controls_box, 3, 0) + layout.addWidget(self.canvas, 0, 1, 4, 1) + layout.setColumnStretch(1, 1) + + self.setCentralWidget(central) + self.setStatusBar(QStatusBar()) + + def _bind_callbacks(self) -> None: + self.connect_button.clicked.connect(self.connect_cortex) + self.start_button.clicked.connect(self.start_recording) + self.stop_button.clicked.connect(self.stop_and_save) + self.marker_button.clicked.connect(self.add_marker) + + self.api_client.on_eeg = self._on_eeg + self.api_client.on_status = self._on_status + self.api_client.on_battery = self._on_battery + + def connect_cortex(self) -> None: + """Connect, authorize, query headset, and subscribe to EEG stream.""" + try: + self.statusBar().showMessage("Connecting to Cortex...") + self.api_client.connect() + self.api_client.authorize() + + headsets = self.api_client.query_headsets() + if not headsets: + raise RuntimeError("No headset detected. Please pair your EPOC X+.") + + headset = headsets[0] + self.api_client.create_session(headset.headset_id) + sub_result = self.api_client.subscribe_eeg() + + eeg_stream = next((s for s in sub_result.get("success", []) if s.get("streamName") == "eeg"), None) + if eeg_stream: + self._channel_labels = eeg_stream.get("cols", [])[2:] + + self.status_connection.setText(f"Connected ({headset.headset_id})") + if headset.battery_percent is not None: + self.status_battery.setText(f"Battery: {headset.battery_percent:.0f}%") + self.statusBar().showMessage("Connected and subscribed to EEG stream.", 5000) + except Exception as exc: # UI-safe message relay + QMessageBox.critical(self, "Connection Error", str(exc)) + self.statusBar().showMessage("Connection failed.", 5000) + + def start_recording(self) -> None: + """Start recording incoming EEG packets.""" + try: + if not self.api_client.connected: + raise RuntimeError("Connect to Cortex before recording.") + labels = self._channel_labels or [f"CH{i+1}" for i in range(14)] + self.recorder.start(labels, sample_rate_hz=self._sample_rate_hz) + self.start_button.setEnabled(False) + self.stop_button.setEnabled(True) + self.statusBar().showMessage("Recording started.") + except Exception as exc: + QMessageBox.warning(self, "Start Recording", str(exc)) + + def stop_and_save(self) -> None: + """Stop recording and export files to recordings/ directory.""" + try: + self.recorder.stop() + metadata = SessionMetadata( + subject=self.subject_input.text().strip() or "unknown", + experiment_name=self.experiment_input.text().strip() or "untitled", + notes=self.notes_input.toPlainText().strip(), + ) + output_dir = ensure_output_dir("recordings") + stem = f"{metadata.subject}_{metadata.experiment_name}_{int(time.time())}" + paths = self.recorder.export_all(Path(output_dir / stem), metadata) + + self.start_button.setEnabled(True) + self.stop_button.setEnabled(False) + self.statusBar().showMessage( + f"Saved CSV/NPY/FIF: {paths['csv'].name}, {paths['npy'].name}, {paths['fif'].name}", + 8000, + ) + except Exception as exc: + QMessageBox.warning(self, "Save Error", str(exc)) + + def add_marker(self) -> None: + """Add a manual annotation marker at current wall-clock timestamp.""" + label = self.marker_input.text().strip() + if not label: + return + self.recorder.add_marker(timestamp=time.time(), label=label) + self.marker_input.clear() + self.statusBar().showMessage(f"Marker added: {label}", 3000) + + def _on_eeg(self, payload: dict) -> None: + """Receive EEG packet callback from websocket thread.""" + self._sample_queue.put(payload) + + def _on_status(self, text: str) -> None: + self.status_connection.setText(text) + + def _on_battery(self, value: float) -> None: + self.status_battery.setText(f"Battery: {value:.0f}%") + + def _process_incoming_samples(self) -> None: + """Drain queue, update recorder and refresh plot at UI timer cadence.""" + updated = False + while True: + try: + payload = self._sample_queue.get_nowait() + except queue.Empty: + break + + eeg = payload.get("eeg") + if not isinstance(eeg, list) or len(eeg) < 4: + continue + + # Cortex eeg payload commonly: [COUNTER, INTERPOLATED, CH1..CHn, MARKERSYNC] + timestamp = time.time() + values = [float(v) for v in eeg[2:-1]] if len(eeg) > 3 else [float(v) for v in eeg[2:]] + self._timestamps.append(timestamp) + self._plot_buffer.append(values) + self.recorder.add_sample(timestamp, values) + updated = True + + if self._channel_labels and not self.recorder.channel_labels: + self.recorder.channel_labels = self._channel_labels + + sid = payload.get("sid") + if sid: + self.status_connection.setText(f"Session: {sid}") + + if updated: + self._redraw_plot() + + def _redraw_plot(self) -> None: + """Draw first channel as a lightweight live waveform.""" + if not self._plot_buffer: + return + + data = np.asarray(self._plot_buffer) + first_channel = data[:, 0] + + self.axes.cla() + self.axes.plot(first_channel, linewidth=1.0, color="#1f77b4") + self.axes.set_title(f"Live EEG - {self._channel_labels[0] if self._channel_labels else 'Channel 1'}") + self.axes.set_xlabel("Samples") + self.axes.set_ylabel("Amplitude (uV)") + self.canvas.draw_idle() + + def closeEvent(self, event) -> None: # noqa: N802 (Qt naming) + """Shutdown Cortex session and websocket when GUI exits.""" + try: + if self.api_client.session_id and self.api_client.auth_token: + self.api_client.update_session_status("close") + self.api_client.close() + finally: + super().closeEvent(event) + + @property + def session_info(self) -> str: + """Expose short session string used by tests/debugging.""" + return now_utc_iso() diff --git a/neuro_log/recorder.py b/neuro_log/recorder.py new file mode 100644 index 0000000..a4edb1a --- /dev/null +++ b/neuro_log/recorder.py @@ -0,0 +1,122 @@ +"""Recording buffer and export utilities for NeuroLog EEG sessions.""" + +from __future__ import annotations + +import csv +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import mne +import numpy as np + +from neuro_log.utils import SessionMetadata, dump_json + + +@dataclass(slots=True) +class EventMarker: + """A manual event marker created during recording.""" + + timestamp: float + label: str + + +class EEGRecorder: + """Collects EEG samples and exports them in multiple research-friendly formats.""" + + def __init__(self) -> None: + self._recording = False + self._timestamps: list[float] = [] + self._samples: list[list[float]] = [] + self._markers: list[EventMarker] = [] + self.channel_labels: list[str] = [] + self.sample_rate_hz: float | None = None + + @property + def is_recording(self) -> bool: + """Return current recording state.""" + return self._recording + + def start(self, channel_labels: list[str], sample_rate_hz: float | None = None) -> None: + """Begin a new recording and reset in-memory buffers.""" + self._recording = True + self.channel_labels = channel_labels + self.sample_rate_hz = sample_rate_hz + self._timestamps.clear() + self._samples.clear() + self._markers.clear() + + def stop(self) -> None: + """Stop recording without clearing captured data.""" + self._recording = False + + def add_sample(self, timestamp: float, values: list[float]) -> None: + """Store one EEG sample if recording is active.""" + if not self._recording: + return + self._timestamps.append(timestamp) + self._samples.append(values) + + def add_marker(self, timestamp: float, label: str) -> None: + """Attach a manual event marker to the active recording.""" + if self._recording and label.strip(): + self._markers.append(EventMarker(timestamp=timestamp, label=label.strip())) + + def export_all(self, output_prefix: Path, metadata: SessionMetadata) -> dict[str, Path]: + """Save recording data to CSV, NPY, FIF, plus metadata JSON.""" + if not self._samples: + raise RuntimeError("No EEG samples recorded; cannot export empty dataset.") + + paths = { + "csv": output_prefix.with_suffix(".csv"), + "npy": output_prefix.with_suffix(".npy"), + "fif": output_prefix.with_suffix(".fif"), + "meta": output_prefix.with_suffix(".json"), + } + + self._export_csv(paths["csv"]) + self._export_npy(paths["npy"]) + self._export_fif(paths["fif"], metadata) + self._export_metadata(paths["meta"], metadata) + return paths + + def _export_csv(self, path: Path) -> None: + with path.open("w", newline="", encoding="utf-8") as fh: + writer = csv.writer(fh) + writer.writerow(["timestamp", *self.channel_labels]) + for ts, values in zip(self._timestamps, self._samples, strict=True): + writer.writerow([ts, *values]) + + def _export_npy(self, path: Path) -> None: + data = np.asarray(self._samples, dtype=np.float32) + np.save(path, data) + + def _export_fif(self, path: Path, metadata: SessionMetadata) -> None: + sample_rate = self.sample_rate_hz or 128.0 + channels = self.channel_labels or [f"EEG{i+1}" for i in range(len(self._samples[0]))] + + data = np.asarray(self._samples, dtype=np.float64).T + info = mne.create_info(ch_names=channels, sfreq=sample_rate, ch_types="eeg") + raw = mne.io.RawArray(data, info, verbose=False) + + for marker in self._markers: + onset = marker.timestamp - self._timestamps[0] + raw.annotations.append(onset=onset, duration=0.0, description=marker.label) + + raw.info["description"] = ( + f"subject={metadata.subject}; experiment={metadata.experiment_name}; " + f"notes={metadata.notes}" + ) + raw.save(path, overwrite=True, verbose=False) + + def _export_metadata(self, path: Path, metadata: SessionMetadata) -> None: + payload: dict[str, Any] = { + "subject": metadata.subject, + "experiment_name": metadata.experiment_name, + "notes": metadata.notes, + "sample_rate_hz": self.sample_rate_hz, + "channels": self.channel_labels, + "samples": len(self._samples), + "markers": [marker.__dict__ for marker in self._markers], + } + dump_json(path, payload) diff --git a/neuro_log/utils.py b/neuro_log/utils.py new file mode 100644 index 0000000..06ba307 --- /dev/null +++ b/neuro_log/utils.py @@ -0,0 +1,76 @@ +"""Utility helpers for NeuroLog.""" + +from __future__ import annotations + +import json +import logging +import os +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +LOGGER = logging.getLogger("neuro_log") + + +@dataclass(slots=True) +class CortexCredentials: + """Container for Emotiv Cortex API credentials.""" + + client_id: str + client_secret: str + + +@dataclass(slots=True) +class SessionMetadata: + """User-provided metadata associated with a recording session.""" + + subject: str + experiment_name: str + notes: str + + +def configure_logging(level: int = logging.INFO) -> None: + """Configure module-level logging format and level.""" + logging.basicConfig( + level=level, + format="%(asctime)s | %(levelname)s | %(name)s | %(message)s", + ) + + +def now_utc_iso() -> str: + """Return the current UTC timestamp in ISO-8601 format.""" + return datetime.now(tz=timezone.utc).isoformat() + + +def load_credentials() -> CortexCredentials: + """Load Cortex credentials from environment variables. + + Raises: + RuntimeError: If required variables are missing. + """ + + client_id = os.getenv("CORTEX_CLIENT_ID", "").strip() + client_secret = os.getenv("CORTEX_CLIENT_SECRET", "").strip() + + if not client_id or not client_secret: + raise RuntimeError( + "Missing Cortex credentials. Set CORTEX_CLIENT_ID and " + "CORTEX_CLIENT_SECRET environment variables." + ) + + return CortexCredentials(client_id=client_id, client_secret=client_secret) + + +def ensure_output_dir(base_dir: str | Path = "recordings") -> Path: + """Create and return the output directory for recordings.""" + + path = Path(base_dir) + path.mkdir(parents=True, exist_ok=True) + return path + + +def dump_json(path: Path, data: dict[str, Any]) -> None: + """Write dictionary data as pretty JSON to a file.""" + + path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..8583a9f --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +PyQt6>=6.7 +websocket-client>=1.8 +numpy>=1.26 +mne>=1.7 +matplotlib>=3.8