Troubleshooting & FAQ

# Check API connectivity
curl -H "Authorization: Bearer YOUR_TOKEN" \
  https://api.neurascale.io/v2/health
 
# Check device availability
curl -H "Authorization: Bearer YOUR_TOKEN" \
  https://api.neurascale.io/v2/devices
 
# Check system status
curl https://status.neurascale.io/api/v2/status.json

from neurascale import NeuraScaleClient
import time
 
async def health_check():
    client = NeuraScaleClient(api_key="YOUR_API_KEY")
 
    start_time = time.time()
    status = await client.health.check()
    latency = (time.time() - start_time) * 1000
 
    print(f"API Health: {status}")
    print(f"Response Time: {latency:.1f}ms")
 
    # Check device connectivity
    devices = await client.devices.list()
    print(f"Available Devices: {len(devices)}")
 
    for device in devices:
        print(f"  {device.id}: {device.status}")

{
  "error": {
    "code": "TOKEN_EXPIRED",
    "message": "JWT token has expired"
  }
}

from neurascale import NeuraScaleClient
import asyncio
from datetime import datetime, timedelta
 
class AuthenticatedClient:
    def __init__(self, email, password):
        self.email = email
        self.password = password
        self.token = None
        self.token_expiry = None
        self.client = None
 
    async def ensure_authenticated(self):
        if not self.token or datetime.now() > self.token_expiry:
            await self.refresh_token()
 
    async def refresh_token(self):
        # Get new token
        auth_response = await self.authenticate()
        self.token = auth_response['access_token']
        self.token_expiry = datetime.now() + timedelta(seconds=auth_response['expires_in'])
 
        # Update client
        self.client = NeuraScaleClient(api_key=self.token)
 
    async def api_call(self, method, *args, **kwargs):
        await self.ensure_authenticated()
        return await getattr(self.client, method)(*args, **kwargs)

{
  "error": {
    "code": "RATE_LIMIT_EXCEEDED",
    "message": "Rate limit of 1000 requests/hour exceeded"
  }
}

import asyncio
import random
from functools import wraps
 
def with_retry(max_retries=3, base_delay=1):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except RateLimitError as e:
                    if attempt == max_retries - 1:
                        raise
 
                    # Exponential backoff with jitter
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    await asyncio.sleep(delay)
 
            return None
        return wrapper
    return decorator
 
@with_retry(max_retries=5, base_delay=2)
async def make_api_call():
    return await client.devices.list()

TimeoutError: Request timed out after 30 seconds

MemoryError: Unable to allocate array with shape (1000000, 64)

Warning: Processing latency exceeds target (120ms > 100ms)

# Optimize client configuration
client = NeuraScaleClient(
    api_key="YOUR_KEY",
    performance_mode="ultra_low_latency",
    buffer_size=128,  # Minimal buffering
    batch_size=1      # Process immediately
)
 
# Use gRPC for lowest latency
grpc_client = NeuraScaleGRPCClient(
    endpoint="api.neurascale.io:443",
    compression=False,  # Disable compression
    keepalive_time_ms=30000
)

class PacketLossHandler:
    def __init__(self, tolerance=0.02):  # 2% tolerance
        self.expected_sequence = 0
        self.tolerance = tolerance
        self.loss_count = 0
        self.total_packets = 0
 
    def check_packet(self, packet):
        self.total_packets += 1
 
        if packet.sequence_number != self.expected_sequence:
            # Packet loss detected
            missed = packet.sequence_number - self.expected_sequence
            self.loss_count += missed
 
            loss_rate = self.loss_count / self.total_packets
            if loss_rate > self.tolerance:
                raise PacketLossError(f"Packet loss rate: {loss_rate:.1%}")
 
        self.expected_sequence = packet.sequence_number + 1

from neurascale.processing import ProcessingStage
 
class CustomPreprocessor(ProcessingStage):
    def __init__(self, custom_param=1.0):
        super().__init__()
        self.custom_param = custom_param
 
    async def process(self, data):
        # Your custom preprocessing logic
        processed = self.custom_algorithm(data, self.custom_param)
        return processed
 
    def custom_algorithm(self, data, param):
        # Implement your algorithm
        return data * param
 
# Use in pipeline
pipeline = ProcessingPipeline()
pipeline.add_stage(CustomPreprocessor(custom_param=2.0))

# Create multi-device session
session = await client.sessions.create(
    name="Multi-device Recording",
    device_ids=["openbci_001", "emotiv_001", "muse_001"],
    sync_method="hardware_trigger"  # or "software_sync"
)
 
# Handle synchronized data
async for data_packet in session.stream():
    # data_packet contains data from all devices
    for device_id, device_data in data_packet.devices.items():
        print(f"Data from {device_id}: {len(device_data.channels)} channels")

# OpenBCI calibration
await device.calibrate(
    calibration_type="impedance",
    target_impedance=5000,  # 5kΩ
    timeout=60
)
 
# Emotiv calibration
await device.calibrate(
    calibration_type="contact_quality",
    minimum_quality=0.8,
    timeout=120
)
 
# Custom calibration
calibration_data = await device.record_calibration(
    duration=30,  # 30 seconds
    instructions="Close your eyes and relax"
)
 
await device.apply_calibration(calibration_data)

from neurascale.monitoring import PerformanceMonitor
 
monitor = PerformanceMonitor()
monitor.enable_metrics([
    'latency',
    'packet_loss',
    'memory_usage',
    'cpu_usage'
])
 
# Your application code
await process_neural_data()
 
# Get performance report
report = monitor.generate_report()
print(report.summary())