Sequence Diagrams

# Data Packet Structure
class NeuralDataPacket:
    timestamp: datetime       # UTC timestamp
    device_id: str           # Unique device identifier
    sequence_number: int     # Packet sequence counter
    samples: np.ndarray      # Shape: (n_channels, n_samples)
    sample_rate: int         # Hz (250, 500, 1000)
    channel_mask: int        # Active channels bitmask
    battery_level: float     # 0.0 to 1.0
    signal_quality: List[float]  # Per-channel quality

{
    "topic": "device.data.raw",
    "device_id": "openbci_cyton_001",
    "timestamp": "2024-01-15T10:30:45.123Z",
    "data": {
        "packets": ["..."],
        "batch_size": 1000,
        "duration_ms": 4000
    }
}

{
    "type": "data",
    "device_id": "openbci_cyton_001",
    "channels": [1.23, 4.56, 7.89],
    "timestamp": 1705317045123,
    "quality": [0.98, 0.99, 0.97]
}

{
  "header": {
    "alg": "RS256",
    "typ": "JWT",
    "kid": "key_id_2024"
  },
  "payload": {
    "sub": "user_uuid",
    "email": "user@example.com",
    "role": "researcher",
    "permissions": ["read:devices", "write:sessions"],
    "iat": 1705320000,
    "exp": 1705323600,
    "iss": "neurascale.io",
    "aud": "neurascale-api"
  },
  "signature": "..."
}

# Rate limiting implementation
@rate_limit(
    key=lambda: f"login:{request.remote_addr}",
    rate="5/15m",
    method="sliding_window"
)
async def login(credentials: LoginRequest):
    # Login logic
    pass

from datetime import datetime, timedelta
from typing import Optional, Tuple
import jwt
from argon2 import PasswordHasher
from redis import Redis
 
class AuthService:
    def __init__(self, kms_client, db, redis: Redis):
        self.kms = kms_client
        self.db = db
        self.redis = redis
        self.ph = PasswordHasher()
 
    async def authenticate(
        self,
        email: str,
        password: str
    ) -> Tuple[str, str]:
        # Fetch user
        user = await self.db.fetch_one(
            "SELECT * FROM users WHERE email = ?",
            email
        )
 
        if not user:
            raise AuthenticationError("Invalid credentials")
 
        # Verify password
        try:
            self.ph.verify(user.password_hash, password)
        except:
            raise AuthenticationError("Invalid credentials")
 
        # Generate tokens
        access_token = await self._generate_access_token(user)
        refresh_token = await self._generate_refresh_token(user)
 
        # Cache session
        await self._cache_session(user.id, {
            "user_id": user.id,
            "email": user.email,
            "role": user.role,
            "permissions": user.permissions
        })
 
        return access_token, refresh_token
 
    async def _generate_access_token(self, user) -> str:
        payload = {
            "sub": str(user.id),
            "email": user.email,
            "role": user.role,
            "permissions": user.permissions,
            "iat": datetime.utcnow(),
            "exp": datetime.utcnow() + timedelta(hours=1)
        }
 
        # Sign with KMS
        return await self.kms.sign_jwt(payload)

class HIPAACompliance:
    def __init__(self):
        self.requirements = {
            "access_control": {
                "authentication": "Multi-factor required",
                "authorization": "Role-based (RBAC)",
                "audit_trail": "All access logged"
            },
            "encryption": {
                "at_rest": "AES-256-GCM",
                "in_transit": "TLS 1.3",
                "key_management": "Cloud KMS"
            },
            "data_integrity": {
                "checksums": "SHA-256 for all files",
                "versioning": "Immutable audit log",
                "backup": "Daily encrypted backups"
            },
            "consent_management": {
                "storage": "Encrypted document store",
                "tracking": "Consent status per data use",
                "revocation": "Immediate data isolation"
            }
        }
 
    async def validate_access(self, user, resource, action):
        # Check user authentication
        if not await self.verify_mfa(user):
            raise AuthenticationError("MFA required")
 
        # Check authorization
        if not self.rbac.has_permission(user, resource, action):
            raise AuthorizationError("Access denied")
 
        # Log access attempt
        await self.audit.log({
            "user": user.id,
            "resource": resource,
            "action": action,
            "timestamp": datetime.utcnow(),
            "ip_address": user.ip_address
        })

de_identification:
  - Remove direct identifiers
  - Generalize quasi-identifiers
  - Add statistical noise
  - K-anonymity (k=5)
 
data_minimization:
  - Collect only necessary data
  - Automatic data expiration
  - Purpose limitation
  - Consent-based retention
 
access_patterns:
  clinician:
    - Own patients only
    - Time-limited access
    - Purpose declaration
  researcher:
    - De-identified data only
    - IRB approval required
    - Aggregated results
  admin:
    - Audit access only
    - No patient data view
    - System metrics only

# Example query with permissions
query GetDeviceData($deviceId: ID!) @authenticated {
  device(id: $deviceId) @requirePermission(action: "read") {
    id
    name
    status
    sessions(last: 10) @cost(complexity: 5) {
      id
      startTime
      recordings @cost(complexity: 10) {
        id
        duration
        sampleCount
        quality @computed
      }
    }
  }
}

// DataLoader implementation
class DeviceLoader extends DataLoader<string, Device> {
  async batchLoadFn(ids: string[]): Promise<Device[]> {
    // Batch fetch from database
    const devices = await db.query(
      'SELECT * FROM devices WHERE id = ANY($1)',
      [ids]
    );
 
    // Maintain order and handle missing
    return ids.map(id =>
      devices.find(d => d.id === id) ||
      new Error(`Device ${id} not found`)
    );
  }
}
 
// Query complexity calculation
function calculateComplexity(query: DocumentNode): number {
  let complexity = 0;
 
  visit(query, {
    Field(node) {
      // Base cost
      complexity += 1;
 
      // Additional cost for lists
      if (node.selectionSet) {
        const listMultiplier = getListMultiplier(node);
        complexity += listMultiplier *
          node.selectionSet.selections.length;
      }
    }
  });
 
  return complexity;
}
 
// Caching strategy
const cacheConfig = {
  device: { ttl: 300, tags: ['device'] },
  session: { ttl: 60, tags: ['session'] },
  recording: { ttl: 3600, tags: ['recording'] },
  analysis: { ttl: 86400, tags: ['analysis'] }
};

# Device Service Schema
type Device @key(fields: "id") {
  id: ID!
  name: String!
  type: String!
  status: DeviceStatus!
  firmwareVersion: String
  # Extended by other services
}
 
# Session Service Schema
extend type Device @key(fields: "id") {
  id: ID! @external
  sessions(
    first: Int
    last: Int
    after: String
    before: String
  ): SessionConnection!
}
 
type Session @key(fields: "id") {
  id: ID!
  deviceId: ID!
  startTime: DateTime!
  endTime: DateTime
  status: SessionStatus!
  device: Device!
}
 
# Analysis Service Schema
extend type Session @key(fields: "id") {
  id: ID! @external
  analysis: Analysis
  metrics: SessionMetrics!
}