# Project export: Mind+ 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: Your brain, but optimized. A new way to interact with agents - Devpost: https://devpost.com/software/neurofocus-myvzab - GitHub: https://github.com/bigbrainw/Treehacks26-mac - Demo: https://github.com/bigbrainw/treehacks26-jetson - Video: https://www.youtube.com/embed/UHJ3r3hC67o?enablejsapi=1&hl=en_US&rel=0&start=&version=3&wmode=transparent - Team: 1 GitHub contributor(s) — Journalwere (3 commits) ## Devpost submission (written by the team) ### Inspiration Today’s AI is powerful, but it’s reactive. Even the most advanced agents still require us to actively type prompts. We must tell the system what we want, how we feel, and when we need help. But humans don’t operate through prompts. Our cognitive state changes continuously. Mind+ was inspired by a question: What if AI could understand your brain state in real time and respond before you ask? ### What it does Mind+ monitors your brain activity in real time using an Emotiv EEG headset to understand whether you are truly engaged or mentally drifting. When studying or reading difficult material, we often think we’re focusing. But in reality, our minds start wandering. NeuroFocus can detect: When you are in deep focus When your attention starts to drift When cognitive load becomes too high When the material may be too abstract or overwhelming Instead of waiting for you to realize you’re distracted, the system can: Gently alert you when your focus drops Suggest a break before burnout Offer clarification when cognitive strain increases Activate an AI agent to assist automatically And this isn’t limited to studying. Because it integrates with your computer environment, NeuroFocus works across: Coding Reading research papers Gaming Writing Any digital task Your brain becomes an input device. Instead of manually prompting AI, your mental state becomes the signal. ### How we built it We used an Emotiv EEG headset to collect real-time mental state data and streamed it to a Jetson Nano for processing. The Jetson runs our pipeline inside Docker, and we exposed the service remotely using ngrok so the system can work from anywhere with low setup friction. In parallel, we built a lightweight computer-context layer that actively monitors the window/app the user is currently working in, so the system understands what task the user is doing without needing manual input. Finally, we connected these two signals, brain state + on-screen context, into a Claude-based agent, which can decide when to proactively engage (e.g., when focus drops, when distraction rises, or when the task becomes cognitively demanding) instead of relying on the user to type prompts. ### Challenges we ran into A core challenge was making the agent safe + useful with the exact signals we get from the Emotiv Cortex API. Cortex gives us high-level performance metrics like attention, engagement, excitement, stress, and relaxation (plus EEG power bands/motion), but mapping those labels to “what the user needs right now” is not straightforward. For example, a drop in focus could mean mind-wandering, fatigue, or simply switching tasks; a rise in stress might mean productive struggle or frustration; and high engagement doesn’t always mean the user is learning effectively. This ambiguity made it hard to design reliable triggers and avoid false positives. Privacy was another big issue because our system combines brain-state metrics with computer context (the active window/app). Even if we only track window titles, it can still reveal sensitive behavior, so we had to minimize what data is logged and carefully control what is passed to the agent. Finally, we tested multiple agent setups. We originally tried running the agent locally using Ollama for maximum privacy, but the on-device models weren’t consistently strong enough to give helpful, context-aware support from signals like attention/stress/relaxation in real time. That’s why we moved to a Claude-based agent, so interventions (explanations, reframes, break suggestions, “ask a guiding question”) are actually high-quality instead of generic. ### Accomplishments we're proud of We’re proud that the system worked better than we expected in a real usage setting: we successfully streamed Emotiv Cortex mental-state metrics (like attention, engagement, stress, and relaxation) into our Jetson Nano processing pipeline and used them to trigger meaningful agent behaviors in real time. We also demonstrated a new interaction pattern for AI agents. Instead of constantly typing prompts, the agent can respond based on your cognitive state, plus what you’re doing on your computer. Finally, we built an end-to-end demo where we used brain commands to drive a complete pipeline that can order food, proving this isn’t just a visualization project, it’s a full neuroadaptive agent workflow. ### What we learned We learned that building neuroadaptive systems is less about raw EEG data and more about interpretation and restraint. Even though the Emotiv Cortex API gives clean high-level metrics like attention, engagement, stress, and relaxation, those labels don’t directly translate into user intent. Brain-state signals are probabilistic, personal, and context-dependent, so designing reliable triggers requires calibration and thoughtful thresholds, not blind automation. We also learned that proactive AI must be carefully timed. Just because focus drops doesn’t mean the agent should interrupt immediately. The hardest problem wasn’t signal processing; it was designing interventions that feel supportive instead of intrusive. Finally, we learned that model capability matters. Running everything locally sounds ideal for privacy, but real-time cognitive assistance requires strong reasoning. Choosing the right balance between privacy, performance, and usability is critical when building brain-integrated AI systems. ### What's next Our next step is turning Mind+ from a hackathon prototype into a usable, everyday product. While the Emotiv headset is powerful, it’s too bulky and impractical for daily life. We’re now working on a lightweight, low-channel EEG design that can capture core mental-state signals related to attention and cognitive drift. Even if it’s not as powerful as a full multi-channel system, our goal is to make it “good enough” for real-world use, portable, comfortable, and affordable. We’re also focusing on improving personalization, refining state-detection models, and making the neuroadaptive agent feel natural and seamless. Long term, we want Mind+ to become a practical brain–computer interface layer that integrates directly into everyday workflows, not just a demo, but a real cognitive companion. ## README (from the GitHub repository) # Emotiv Cortex API Python Examples This folder contains the official Emotiv Cortex API Python examples from [Emotiv/cortex-example](https://github.com/Emotiv/cortex-example). ## Requirements - Python 2.7+ or Python 3.4+ - Install dependencies: `pip install -r requirements.txt` ## Getting Started 1. **Download EMOTIV Launcher** from [emotiv.com](https://www.emotiv.com/products/emotiv-launcher) 2. **Register your Cortex App** to get Client ID and Client Secret: https://emotiv.gitbook.io/cortex-api#create-a-cortex-app 3. Update `your_app_client_id` and `your_app_client_secret` in each script ## Files | File | Description | |------|-------------| | `cortex.py` | Core Cortex API wrapper (WebSocket, JSON-RPC, event handling) | | `sub_data.py` | Subscribe to EEG, motion, performance metrics, band power | | `record.py` | Record and export data to CSV/EDF | | `marker.py` | Inject markers during recording | | `mental_command_train.py` | Train mental command actions (push, pull, etc.) | | `facial_expression_train.py` | Train facial expression actions | | `live_advance.py` | Live mental command data + sensitivity control | | `query_records.py` | Query, download, and export records | ## Focus Agent (Main App) **Real activity + time-on-page + EEG → Jetson.** When you stay on difficult content too long, triggers agent for help. Feedback in overlay. ```bash python app.py # Real activity + real Emotiv EEG python app.py --mock # Mock EEG (no headset, for testing) python app.py --long 45 # 45 sec on page before trigger python app.py --no-feedback # No overlay window ``` - **Activity:** Real app/window/URL via ActivityMonitor (lecture, reading, coding, etc.) - **Time:** SessionTracker fires at `warn` (5s) and `long` (10s) – stuck trigger at 10s - **EEG:** Real Emotiv headset (default); `--mock` for testing without headset - **Mental command:** Requires trained profile; set `EMOTIV_PROFILE` in .env to match your Emotiv BCI profile name - **On long threshold:** POSTs to Jetson `/eeg` with context + duration + mental_state, shows feedback - **Streams:** activity (with `duration_seconds`), eeg, mental_state over WebSocket ## Jetson Collector (WebSocket) Legacy/standalone collector: stream EEG + activity to Jetson. ```bash python collector.py --url wss://YOUR_NGROK_URL --show-feedback python collector_mock.py --url wss://YOUR_NGROK_URL --show-feedback ``` - **Sends:** `eeg`, `mental_state`, `activity` over WebSocket - **Receives:** Agent feedback (`{"type": "feedback", "feedback": "..."}`) in overlay - **Processor:** Jetson runs processor (HTTP + WebSocket) ## Data Structures See [DATA_STRUCTURES.md](DATA_STRUCTURES.md) for the exact JSON payloads sent to ngrok: WebSocket (`activity`, `eeg`, `mental_state`, `reading_help`) and HTTP POST `/eeg`. ## Documentation Full API docs: https://emotiv.gitbook.io/cortex-api/ ## Detected evidence (automated analysis) Indexed codebase: 34 recognized source files, 217 KB. - Python (language) — detected in the code - Docker (technology) — claimed on Devpost, not found in the code ## Codebase structure (from repository index) ### Files (39 of 39) ``` .env.example .gitignore activity/__init__.py activity/linux.py activity/macos.py AGENT_REQUEST.json agent_request.py app.py certificates/rootCA.pem check_emotiv_creds.py collector_mock.py collector.py config.py cortex.py data_schema.py DATA_STRUCTURES.md eeg.py facial_expression_train.py feedback_window.py live_advance.py live_reading_test.py marker.py mental_command_train.py mental_state_parser.py MENTAL_STATE.md query_records.py README.md record.py requirements.txt send_sample_to_jetson.py send_to_jetson.py sub_data.py test_activity.py test_feedback_from_jetson.py test_feedback_window.py test_met.py test_on_mac.py test_pdf_stuck.py time_tracker.py ``` ### Dependencies - requirements.txt: python-dispatch, python-dotenv, requests, websocket-client ### Recent commits (newest first) - Refactor Emotiv client credentials handling and enhance feedback display. Updated config to support alternative environment variable names for client ID and secret. Improved error handling in EmotivCortexClient to provide clearer messages for invalid credentials. Replaced feedback label with a scrollable text area in the feedback window for better user experience. - Enhance Cortex API integration and documentation. Updated Cortex class to handle profiles loaded by other applications without unloading. Improved README with detailed usage instructions for the Focus Agent and Jetson Collector. Refactored send_sample_to_jetson and send_to_jetson scripts to dynamically construct the Jetson URL. Added reading section detection in macOS activity monitoring. Enhanced activity context to include reading section information. Implemented macOS-specific functions to retrieve browser URLs and selected text for better user context awareness. - first commit ## Key source files (fetched from GitHub, selected and truncated for size) ### MENTAL_STATE.md ```markdown # Mental State from Emotiv Cortex API How we parse the `met` (performance metrics) stream. **The array order is defined by `cols` from the subscribe response** – we use cols when available; otherwise fallback indices for EPOC/Insight/Flex. ## Emotiv met Stream Format (EPOC / Insight / Flex) ``` ['eng.isActive','eng','exc.isActive','exc','lex','str.isActive','str','rel.isActive','rel','int.isActive','int','attention.isActive','attention'] ``` | Index | Label | Meaning | |-------|---------------|---------| | 0–1 | eng.isActive, eng | Engagement – immersion in activity (0–1) | | 2–4 | exc.isActive, exc, lex | Excitement, long-term excitement | | 5–6 | str.isActive, str | Stress – emotional tension when completing a task (0–1) | | 7–8 | rel.isActive, rel | Relaxation – calm focus after intense work | | 9–10 | int.isActive, int | Interest – attraction or aversion to stimuli | | 11–12 | attention.isActive, attention | Attention – sustained focus on a single task | All values are 0–1. `null` when signal quality is too poor. ## When to Call User "Confused" or "Stuck" | State | Conditions | |-------------|------------| | **Stuck** | Low engagement (<0.35) + high stress (>0.55) + low attention (<0.4) | | **Confused**| Low engagement (<0.4) + high stress (>0.5), or stress >0.55 | | **Distracted** | Low attention (<0.35) + low engagement (<0.45) | | **Focused** | Engagement ≥0.5, attention ≥0.4, stress <0.6 | ## Usage - `mental_state_parser.parse_met_to_mental_state(metrics)` – parses raw `met` dict into `MentalStateSnapshot` - `mental_state_parser.derive_mental_state_label(ms)` – returns `"confused"`, `"stuck"`, `"distracted"`, or `"focused"` The app uses the derived label in `reading_help` payloads and for agent feedback. ``` ### DATA_STRUCTURES.md ```markdown # Data Structures Sent to Ngrok / Jetson Backend Two transport modes: **WebSocket** (streaming) and **HTTP POST** (on stuck trigger). **Canonical agent request:** See `agent_request.py` and `AGENT_REQUEST.json` for the exact JSON format. --- ## 1. WebSocket (e.g. `wss://YOUR_NGROK.ngrok-free.app`) JSON messages sent every ~2 seconds (configurable via `SEND_INTERVAL`). Each message has a `type` field. ### 1a. `activity` ```json { "type": "activity", "timestamp": 1739612345.678, "activity": { "app_name": "Chrome", "window_title": "CS224N Lecture 5 — Backpropagation | Stanford", "context_type": "website", "context_id": "Chrome::cs224n.stanford.edu", "reading_section": null, "duration_seconds": 12.5 } } ``` | Field | Type | Description | |-------|------|-------------| | `app_name` | string | Browser/app (Chrome, Safari, VS Code, etc.) | | `window_title` | string | Tab/window title | | `context_type` | string | `website` \| `file` \| `terminal` \| `browser` \| `app` | | `context_id` | string | Stable id for context (e.g. `App::domain`) | | `reading_section` | string? | Selected text, URL section, or "Page N of M" (PDF) | | `page_number` | int? | Current page (PDFs; optional) | | `file_path` | string? | Full path to file (optional) | | `duration_seconds` | float? | Time spent in this context | --- ### 1b. `eeg` ```json { "type": "eeg", "timestamp": 1739612345.678, "eeg": { "metrics": { "met": [true, 0.65, true, 0.42, 0.38, true, 0.55, true, 0.72, true, 0.48, true, 0.58], "time": 1739612345.678 } }, "activity": { "app_name": "Chrome", "window_title": "Lecture Notes...", "context_type": "website", "context_id": "Chrome::example.edu", "reading_section": null, "duration_seconds": 12.5 } } ``` `metrics.met` is the Emotiv performance metrics array (engagement, stress, etc.). --- ### 1c. `mental_state` ```json { "type": "mental_state", "timestamp": 1739612345.678, "mental_state": { "engagement": 0.55, "stress": 0.35, "relaxation": 0.45, "focus": 0.52, "excitement": null, "interest": null, "metrics": { "met": [true, 0.65, true, 0.42, 0.38], "time": 1739612345.678 } } } ``` | Field | Type | Description | |-------|------|-------------| | `engagement` | float? | 0–1 | | `stress` | float? | 0–1 | | `relaxation` | float? | 0–1 | | `focus` | float? | 0–1 | | `excitement` | float? | 0–1 | | `interest` | float? | 0–1 | | `metrics` | dict | Raw `met` from Emotiv | --- ### 1d. `reading_help` (on stuck trigger, ~10s on same page) ```json { "type": "reading_help", "timestamp": 1739612345.678, "activity": { "app_name": "Chrome", "window_title": "Complex Paper — arXiv", "context_type": "website", "context_id": "Chrome::arxiv.org", "reading_section": null, "duration_seconds": 10.2 }, "mental_state": { "engagement": 0.32, "stress": 0.58, "relaxation": 0.42, "focus": 0.35 [truncated — 1321 more characters] ``` ### requirements.txt ``` websocket-client python-dispatch python-dotenv requests ``` ### app.py ```python #!/usr/bin/env python3 """ Focus Agent — main application. Real activity monitoring + time-on-page tracking + EEG (real or mock) → Jetson. When you stay on difficult content too long, triggers agent for help. Feedback in overlay. Usage: python app.py # Real activity + mock EEG python app.py --eeg # Real Emotiv headset (requires .env) python app.py --long 45 # 45 sec on page before trigger python app.py --no-feedback # No overlay window """ import argparse import json import os import signal import sys import threading import time from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) try: import websocket except ImportError: websocket = None import config from activity import ActivityMonitor from agent_request import build_agent_request, build_post_eeg_body, build_reading_help_ws_message from data_schema import ( ActivitySnapshot, CollectorPayload, EEGMetricsSnapshot, MentalStateSnapshot, ) from feedback_window import FeedbackWindow from mental_state_parser import parse_met_to_mental_state from time_tracker import SessionTracker, SessionEvent, SessionEventType # Overlay window identifiers — when focused, use last real context for session/help OVERLAY_APP_TITLES = ("Agent Feedback", "Python | Agent Feedback") OVERLAY_APP_NAME = "Python" def _is_overlay(ctx) -> bool: """True if ctx is the feedback overlay (don't use for session tracking).""" if not ctx: return False title = (getattr(ctx, "window_title", None) or "").strip() app = (getattr(ctx, "app_name", None) or "").strip() if "Agent Feedback" in title: return True if app == OVERLAY_APP_NAME and any(t in title for t in OVERLAY_APP_TITLES): return True return False def _ctx_to_snapshot(ctx, duration_seconds: float | None = None) -> ActivitySnapshot | None: if not ctx: return None return ActivitySnapshot( app_name=getattr(ctx, "app_name", "") or "", window_title=getattr(ctx, "window_title", "") or "", context_type=getattr(ctx, "context_type", "app") or "app", context_id=getattr(ctx, "context_id", "") or "", reading_section=getattr(ctx, "reading_section", None), duration_seconds=duration_seconds, ) # --- Shared state --- class AppState: def __init__(self): self.last_mental_state: MentalStateSnapshot | None = None self.lock = threading.Lock() def set_mental_state(self, ms: MentalStateSnapshot): with self.lock: self.last_mental_state = ms def get_mental_state(self) -> MentalStateSnapshot | None: with self.lock: return self.last_mental_state def run_app( jetson_ws_url: str, jetson_http_base: str, use_mock_eeg: bool = True, show_feedback: bool = True, warn_sec: float = 120, long_sec: float = 180, follow_up_interval_sec: float = 300, poll_interval: float = 0.3, ) -> None: if not websocket: print("Error: pip install websocket-client") sys.exit(1) state = AppState() running = True ws_ref: dict = {"ws": None} activity = ActivityMonitor(poll_interval=poll_interval) session_tracker = SessionTracker( warn_threshold_sec=min(warn_sec, max(1, long_sec - 30)), long_threshold_sec=long_sec, follow_up_interval_sec=follow_up_interval_sec, ) # Overlay exclusion: when overlay is focused, use last real context for session/help last_real_context = [None] # list to allow mutation in closure def stop(_=None, __=None): nonlocal running running = False signal.signal(signal.SIGINT, stop) signal.signal(signal.SIGTERM, stop) # Feedback window feedback_cb = None if show_feedback: win = FeedbackWindow(width=360, height=160, use_poll=False) feedback_cb = win.update_feedback win.root.protocol("WM_DELETE_WINDOW", lambda: (stop(), win.root.destroy())) win.update_feedback("Monitoring... Stay on a difficult page to trigger help.") def send_payload(payload: CollectorPayload): if ws_ref["ws"] and ws_ref["ws"].sock and ws_ref["ws"].sock.connected: try: d = payload.to_dict() ws_ref["ws"].send(json.dumps(d)) act = payload.activity ms = payload.mental_state parts = [] if act and (act.app_name or act.context_type): title = (act.window_title or "")[:35] parts.append(f"{act.app_name or '?'} | {title}{'...' if len(act.window_title or '') > 35 else ''} | {act.context_type or '?'}") if ms: ms_parts = [] if ms.engagement is not None: ms_parts.append(f"eng={ms.engagement:.2f}") if ms.stress is not None: ms_parts.append(f"stress={ms.stress:.2f}") if ms.focus is not None: ms_parts.append(f"focus={ms.focus:.2f}") if ms.relaxation is not None: ms_parts.append(f"relax={ms.relaxation:.2f}") parts.append(f"mental_state=[{', '.join(ms_parts) or 'metrics'}]") print(f" [WS] Sent to backend: {payload.type}" + (" | " + " | ".join(parts) if parts else "")) except Exception as e: print(" Send error:", e) def _make_activity_snapshot(duration_seconds: float | None = None) -> ActivitySnapshot | None: ctx = activity.get_current_activity() if not ctx: return None return _ctx_to_snapshot(ctx, duration_seconds) # Session events → help request (POST /eeg + WebSocket reading_help) def on_session_event(event: SessionEvent): if event.event_type not in (SessionEventType.LONG_THRESHOLD, SessionEventType.FOLLOW_UP [truncated — 9882 more characters] ``` ### test_feedback_window.py ```python """Test the feedback window in isolation (no Emotiv, no Jetson).""" import threading import time from feedback_window import FeedbackWindow def main(): f = FeedbackWindow() def inject(): time.sleep(2) f.root.after(0, lambda: f.update_feedback("Consider taking a short break – stress is elevated.")) time.sleep(4) f.root.after(0, lambda: f.update_feedback("You've been focused for a while. Stretch or look away for 20 seconds.")) threading.Thread(target=inject, daemon=True).start() print("Feedback window open. Sample messages will appear in 2s and 6s.") f.run() if __name__ == "__main__": main() ``` ### test_activity.py ```python """Test activity monitoring only - no Emotiv, no Jetson.""" import time from activity import ActivityMonitor def main(): monitor = ActivityMonitor(poll_interval=2.0) def on_change(ctx, prev): print(f" [CHANGE] {prev.display_name if prev else '—'} -> {ctx.display_name}") monitor.on_context_change(on_change) print("Monitoring active app/window (Ctrl+C to stop)...\n") while True: ctx = monitor.get_current_activity() if ctx: print(f"[{time.strftime('%H:%M:%S')}] {ctx.display_name}") print(f" type={ctx.context_type} id={ctx.context_id}") if ctx.reading_section: r = ctx.reading_section print(f" reading_section: {r[:80]}{'...' if len(r) > 80 else ''}") else: print(f"[{time.strftime('%H:%M:%S')}] (no context)") time.sleep(2) if __name__ == "__main__": main() ``` ### check_emotiv_creds.py ```python #!/usr/bin/env python3 """Verify Emotiv credentials load and provide -32021 troubleshooting steps.""" import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import config cid = config.EMOTIV_CLIENT_ID secret = config.EMOTIV_CLIENT_SECRET print("Emotiv credentials check") print("-" * 40) print(f" EMOTIV_CLIENT_ID: {'loaded (' + str(len(cid)) + ' chars)' if cid else 'MISSING'}") print(f" EMOTIV_CLIENT_SECRET: {'loaded (' + str(len(secret)) + ' chars)' if secret else 'MISSING'}") if not cid or not secret: print("\n Add to .env: EMOTIV_CLIENT_ID=... and EMOTIV_CLIENT_SECRET=...") sys.exit(1) print("\n If you still get -32021 Invalid Client Credentials:") print(" 1. EMOTIV Launcher: log in, Settings -> Authorized Apps -> approve your app") print(" 2. Emotiv Developer: emotiv.com/developer -> verify Client ID/Secret match .env") print(" 3. Unpublished app: only the creator's EmotivID can use it") print("\n To run without headset: python app.py (uses mock EEG)") ``` ### test_met.py ```python """ Test script: subscribe to performance metrics (met) only. met stream: engagement, excitement, stress, relaxation, interest, attention On free tier: ~0.1 Hz (1 sample every 10 seconds) """ from cortex import Cortex class SubMet: def __init__(self, app_client_id, app_client_secret, **kwargs): self.c = Cortex(app_client_id, app_client_secret, debug_mode=False, **kwargs) self.c.bind(create_session_done=self.on_create_session_done) self.c.bind(new_data_labels=self.on_new_data_labels) self.c.bind(new_met_data=self.on_new_met_data) self.c.bind(inform_error=self.on_inform_error) def start(self, headset_id=''): if headset_id: self.c.set_wanted_headset(headset_id) self.c.open() def on_create_session_done(self, *args, **kwargs): print("Session ready, subscribing to 'met'...") self.c.sub_request(['met']) def on_new_data_labels(self, *args, **kwargs): data = kwargs.get('data') print(f"met labels: {data.get('labels', [])}") def on_new_met_data(self, *args, **kwargs): data = kwargs.get('data') print("pm data:", data) def on_inform_error(self, *args, **kwargs): print("Error:", kwargs.get('error_data')) def main(): import os from pathlib import Path from dotenv import load_dotenv load_dotenv(Path(__file__).parent / '.env') client_id = os.environ.get('client_id') or os.environ.get('CORTEX_CLIENT_ID') client_secret = os.environ.get('client_secret') or os.environ.get('CORTEX_CLIENT_SECRET') if not client_id or not client_secret: raise SystemExit('Missing credentials. Set client_id and client_secret in .env') s = SubMet(client_id, client_secret) s.start() if __name__ == '__main__': main() ``` ### send_sample_to_jetson.py ```python """ Send sample EEG + activity data to Jetson (no Emotiv headset required). Use this to verify the Mac -> Jetson pipeline works. """ import os import time from pathlib import Path from dotenv import load_dotenv from activity import ActivityMonitor import requests # Sample data matching real Emotiv format SAMPLE_MET = {"met": [True, 0.65, True, 0.42, 0.38, True, 0.55, True, 0.72, True, 0.48, True, 0.58], "time": time.time()} SAMPLE_POW = {"pow": [2.1, 1.0, 0.25, 0.38, 0.23] * 14, "time": time.time()} # 70 values for EPOC X SAMPLE_MOT = {"mot": [11, 0, 0.64, -0.52, -0.46, -0.33, 0.94, -0.37, 0.04, -40.1, -4.1, -12.4], "time": time.time()} SAMPLE_DEV = {"signal": 1.0, "dev": [4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 4, 100], "batteryPercent": 72, "time": time.time()} def send_sample(jetson_url: str, count: int = 5): """Send count sample payloads to Jetson.""" activity = ActivityMonitor(poll_interval=1.0) for i in range(count): ctx = activity.get_current_activity() context = ctx.to_dict() if ctx else {} body = { "timestamp": time.time(), "streams": { "met": SAMPLE_MET, "pow": SAMPLE_POW, "mot": SAMPLE_MOT, "dev": SAMPLE_DEV, }, "context": context, } try: r = requests.post( jetson_url, json=body, headers={ "Content-Type": "application/json", "ngrok-skip-browser-warning": "1", }, timeout=10, ) print(f"[{i+1}/{count}] POST {jetson_url} -> {r.status_code}") if r.status_code != 200: print(f" {r.text[:150]}") except requests.RequestException as e: print(f"[{i+1}/{count}] POST failed: {e}") if i < count - 1: time.sleep(2) def main(): load_dotenv(Path(__file__).parent / ".env") base = os.environ.get("JETSON_URL", "https://8061-68-65-164-46.ngrok-free.app").rstrip("/") jetson_url = f"{base}/eeg" print(f"Sending 5 sample payloads to {jetson_url}\n") send_sample(jetson_url, count=5) print("\nDone. Check Jetson logs to confirm receipt.") if __name__ == "__main__": main() ``` ### test_feedback_from_jetson.py ```python """ Test: POST sample data to /eeg, show feedback from the response in window. Feedback comes from the POST response body (e.g. {"feedback": "message"}). """ import os import threading import time from pathlib import Path from dotenv import load_dotenv from activity import ActivityMonitor from feedback_window import FeedbackWindow import requests SAMPLE_STREAMS = { "met": {"met": [True, 0.65, True, 0.42, 0.38, True, 0.55, True, 0.72, True, 0.48, True, 0.58], "time": time.time()}, "pow": {"pow": [2.1, 1.0, 0.25, 0.38, 0.23] * 14, "time": time.time()}, "mot": {"mot": [11, 0, 0.64, -0.52, -0.46, -0.33, 0.94, -0.37, 0.04, -40.1, -4.1, -12.4], "time": time.time()}, "dev": {"signal": 1.0, "dev": [4] * 14 + [100], "batteryPercent": 72, "time": time.time()}, } SEND_INTERVAL = 2.0 def _send_and_show_feedback(eeg_url: str, window: FeedbackWindow): """POST to /eeg, parse response for feedback, update window.""" activity = ActivityMonitor(poll_interval=1.0) count = 0 while True: count += 1 ctx = activity.get_current_activity() context = ctx.to_dict() if ctx else {} body = {"timestamp": time.time(), "streams": SAMPLE_STREAMS, "context": context} try: r = requests.post( eeg_url, json=body, headers={"Content-Type": "application/json", "ngrok-skip-browser-warning": "1"}, timeout=10, ) print(f"[{count}] POST {eeg_url} -> {r.status_code}") if r.status_code == 200 and r.text: try: data = r.json() feedback = data.get("feedback") or data.get("message") if feedback: window.root.after(0, lambda t=feedback: window.update_feedback(t)) print(f" feedback: {feedback[:60]}...") except ValueError: pass except requests.RequestException as e: print(f"[{count}] POST failed: {e}") time.sleep(SEND_INTERVAL) def main(): load_dotenv(Path(__file__).parent / ".env") base = os.environ.get("JETSON_URL", "https://8061-68-65-164-46.ngrok-free.app").rstrip("/") eeg_url = f"{base}/eeg" print(f"POST {eeg_url} (feedback in response)") print("Close window to exit.\n") w = FeedbackWindow() # No poll_url – feedback from POST response t = threading.Thread(target=_send_and_show_feedback, args=(eeg_url, w), daemon=True) t.start() w.run() if __name__ == "__main__": main() ``` [24 more indexed source files omitted to keep this export small. The full file list is in the Codebase structure section above.]