Coverage for integrations / providers / neuro_providers.py: 100.0%

1"""Neuro / biometric provider registry — one source of truth for every 

2brainwave, EEG, EMG, HRV, and wearable device HARTOS can talk to. 

3 

4Kept distinct from providers/registry.py (compute APIs: OpenAI, Replicate) 

5because the axes are different. A Muse headset doesn't have 

6`context_length` or `pricing_per_1k_tokens`; it has form factor, 

7transport, signal types, sample rate, channel count. Forcing hardware 

8into the compute-API dataclass would have made a parallel path. 

9 

10This registry + neuro_adapter.py + neuro_discovery.py together are the 

11surface Nunba uses so developers can plug an agent into biometric 

12hardware without re-implementing discovery per-vendor. 

13""" 

14 

15from __future__ import annotations 

16 

17from dataclasses import dataclass 

18from enum import Enum 

19from typing import Dict, List, Optional, Tuple 

20 

21 

22# ─── Enums ──────────────────────────────────────────────────────────── 

23 

24class FormFactor(str, Enum): 

25 HEADSET_EEG = 'headset_eeg' 

26 EARPIECE_EEG = 'earpiece_eeg' 

27 WRISTBAND_PPG = 'wristband_ppg' 

28 RING_PPG = 'ring_ppg' 

29 CAP_EEG = 'cap_eeg' 

30 IMPLANT_ECOG = 'implant_ecog' 

31 GLASSES_EOG = 'glasses_eog' 

32 CHEST_HRV = 'chest_hrv' 

33 UNKNOWN = 'unknown' 

34 

35 

36class Transport(str, Enum): 

37 BLE = 'ble' 

38 USB_HID = 'usb_hid' 

39 USB_SERIAL = 'usb_serial' 

40 WEBSOCKET = 'websocket' 

41 REST = 'rest' 

42 LSL = 'lsl' # Lab Streaming Layer — de-facto neuro transport standard 

43 UNKNOWN = 'unknown' 

44 

45 

46class SignalType(str, Enum): 

47 EEG = 'eeg' 

48 EMG = 'emg' 

49 ECG = 'ecg' 

50 PPG = 'ppg' # photoplethysmography -> heart rate / HRV 

51 EOG = 'eog' # eye movement 

52 ACC = 'acc' # accelerometer 

53 GYRO = 'gyro' 

54 HRV = 'hrv' 

55 SLEEP = 'sleep' 

56 TEMPERATURE = 'temperature' 

57 BRAINWAVE_FINGERPRINT = 'brainwave_fingerprint' 

58 

59 

60class ProviderStatus(str, Enum): 

61 PUBLIC_SDK = 'public_sdk' # docs + pypi/npm package available 

62 CONTACT_REQUIRED = 'contact_required' # e.g. Yneuro — email them 

63 LICENSE_GATED = 'license_gated' # SDK exists but needs paid license 

64 INVITE_ONLY = 'invite_only' # closed beta 

65 DISCONTINUED = 'discontinued' 

66 RESEARCH_ONLY = 'research_only' 

67 

68 

69# ─── Dataclass ──────────────────────────────────────────────────────── 

70 

71@dataclass(frozen=True) 

72class NeuroProvider: 

73 """Static specification of a neural / biometric provider.""" 

74 id: str # slug: 'muse', 'yneuro', 'neurosity' 

75 name: str # display name 

76 form_factor: FormFactor 

77 transport: Tuple[Transport, ...] # primary first — fallbacks after 

78 signals: Tuple[SignalType, ...] 

79 status: ProviderStatus 

80 sample_rate_hz: int = 0 # 0 = unknown or variable 

81 channels: int = 0 # EEG channel count, where applicable 

82 sdk_python: str = '' # pip package name 

83 sdk_node: str = '' # npm package name 

84 docs_url: str = '' 

85 website: str = '' 

86 contact_email: str = '' # populated when status=CONTACT_REQUIRED 

87 ble_service_uuid: str = '' # BLE service UUID, for discovery matching 

88 notes: str = '' 

89 

90 

91# ─── Seeded registry ────────────────────────────────────────────────── 

92 

93NEURO_REGISTRY: Dict[str, NeuroProvider] = { 

94 # ── Yneuro — contact-gated brainwave authentication ────────────── 

95 # Intentionally first so it's visible as the headline provider the 

96 # registry was built around. No public SDK as of 2026-04; email 

97 # them for access. The adapter surface is already wired so the 

98 # day they reply, implementing the client is a ~30 LOC drop-in. 

99 'yneuro': NeuroProvider( 

100 id='yneuro', 

101 name='Yneuro (Neuro ID)', 

102 form_factor=FormFactor.UNKNOWN, 

103 transport=(Transport.UNKNOWN,), 

104 signals=(SignalType.BRAINWAVE_FINGERPRINT,), 

105 status=ProviderStatus.CONTACT_REQUIRED, 

106 website='https://www.yneuro.com/', 

107 contact_email='hello@yneuro.com', 

108 notes=( 

109 'Brainwave-signature authentication. No public SDK or ' 

110 'API documented as of 2026-04. Email for developer ' 

111 'access; the ContactGatedAdapter raises a clear ' 

112 'PermissionError pointing there until credentials land.' 

113 ), 

114 ), 

115 

116 # ── Muse (Choose Muse) — BLE headset EEG, public ───────────────── 

117 'muse': NeuroProvider( 

118 id='muse', 

119 name='Muse (2 / S)', 

120 form_factor=FormFactor.HEADSET_EEG, 

121 transport=(Transport.BLE, Transport.LSL), 

122 signals=(SignalType.EEG, SignalType.PPG, SignalType.ACC), 

123 status=ProviderStatus.PUBLIC_SDK, 

124 sample_rate_hz=256, 

125 channels=4, 

126 sdk_python='muselsl', 

127 docs_url='https://github.com/alexandrebarachant/muse-lsl', 

128 website='https://choosemuse.com/', 

129 ble_service_uuid='0000fe8d-0000-1000-8000-00805f9b34fb', 

130 ), 

131 

132 # ── Neurosity Crown — REST + WebSocket ────────────────────────── 

133 'neurosity': NeuroProvider( 

134 id='neurosity', 

135 name='Neurosity Crown', 

136 form_factor=FormFactor.HEADSET_EEG, 

137 transport=(Transport.WEBSOCKET, Transport.REST), 

138 signals=(SignalType.EEG,), 

139 status=ProviderStatus.PUBLIC_SDK, 

140 sample_rate_hz=256, 

141 channels=8, 

142 sdk_python='neurosity', 

143 sdk_node='@neurosity/sdk', 

144 docs_url='https://docs.neurosity.co/', 

145 website='https://neurosity.co/', 

146 ), 

147 

148 # ── OpenBCI — open hardware (Cyton / Ganglion) via BrainFlow ──── 

149 'openbci': NeuroProvider( 

150 id='openbci', 

151 name='OpenBCI (Cyton / Ganglion)', 

152 form_factor=FormFactor.CAP_EEG, 

153 transport=(Transport.USB_SERIAL, Transport.BLE, Transport.LSL), 

154 signals=(SignalType.EEG, SignalType.EMG, SignalType.ECG), 

155 status=ProviderStatus.PUBLIC_SDK, 

156 sample_rate_hz=250, 

157 channels=8, 

158 sdk_python='brainflow', 

159 docs_url='https://brainflow.readthedocs.io/', 

160 website='https://openbci.com/', 

161 ), 

162 

163 # ── Emotiv (EPOC / Insight) — Cortex WebSocket API ────────────── 

164 'emotiv': NeuroProvider( 

165 id='emotiv', 

166 name='Emotiv (EPOC / Insight)', 

167 form_factor=FormFactor.HEADSET_EEG, 

168 transport=(Transport.WEBSOCKET,), 

169 signals=(SignalType.EEG,), 

170 status=ProviderStatus.LICENSE_GATED, 

171 sample_rate_hz=256, 

172 channels=14, 

173 sdk_python='cortex-v2-example', 

174 docs_url='https://emotiv.gitbook.io/cortex-api/', 

175 website='https://www.emotiv.com/', 

176 notes='Requires an EMOTIV license for raw EEG stream access.', 

177 ), 

178 

179 # ── NextSense — earpiece EEG, invite-only beta ────────────────── 

180 'nextsense': NeuroProvider( 

181 id='nextsense', 

182 name='NextSense', 

183 form_factor=FormFactor.EARPIECE_EEG, 

184 transport=(Transport.BLE,), 

185 signals=(SignalType.EEG,), 

186 status=ProviderStatus.INVITE_ONLY, 

187 website='https://nextsense.io/', 

188 ), 

189 

190 # ── Whoop — wristband, HRV + sleep via REST OAuth ─────────────── 

191 'whoop': NeuroProvider( 

192 id='whoop', 

193 name='Whoop', 

194 form_factor=FormFactor.WRISTBAND_PPG, 

195 transport=(Transport.REST,), 

196 signals=(SignalType.HRV, SignalType.SLEEP, SignalType.PPG), 

197 status=ProviderStatus.PUBLIC_SDK, 

198 docs_url='https://developer.whoop.com/', 

199 website='https://www.whoop.com/', 

200 ), 

201 

202 # ── Oura — ring, HRV + sleep + temperature via REST OAuth ─────── 

203 'oura': NeuroProvider( 

204 id='oura', 

205 name='Oura', 

206 form_factor=FormFactor.RING_PPG, 

207 transport=(Transport.REST,), 

208 signals=( 

209 SignalType.HRV, SignalType.SLEEP, 

210 SignalType.PPG, SignalType.TEMPERATURE, 

211 ), 

212 status=ProviderStatus.PUBLIC_SDK, 

213 docs_url='https://cloud.ouraring.com/v2/docs', 

214 website='https://ouraring.com/', 

215 ), 

216} 

217 

218 

219# ─── Public API ─────────────────────────────────────────────────────── 

220 

221def get_provider(provider_id: str) -> Optional[NeuroProvider]: 

222 return NEURO_REGISTRY.get(provider_id) 

223 

224 

225def list_providers( 

226 form_factor: Optional[FormFactor] = None, 

227 status: Optional[ProviderStatus] = None, 

228 signal: Optional[SignalType] = None, 

229 transport: Optional[Transport] = None, 

230) -> List[NeuroProvider]: 

231 """Filtered view of the registry — any combination of filters.""" 

232 out = list(NEURO_REGISTRY.values()) 

233 if form_factor is not None: 

234 out = [p for p in out if p.form_factor == form_factor] 

235 if status is not None: 

236 out = [p for p in out if p.status == status] 

237 if signal is not None: 

238 out = [p for p in out if signal in p.signals] 

239 if transport is not None: 

240 out = [p for p in out if transport in p.transport] 

241 return out