PYTHON SENSOR ARRAY

Real-Time Data Collection • Batch Validation • Anomaly Detection

🖥️ Core Architecture

Twelve IoT nodes distributed across Cold Storage Bay 3. Each node samples pH, temperature, humidity, and vibration at 10Hz. Data streams converge at central validator running on Raspberry Pi cluster (Raspberry Pi 4 Model B × 4, bonded Ethernet).

12
Active Nodes
10
Samples/sec/node
480
Data Points/sec
<2
ms Latency
Server rack with blinking status lights in industrial facility

💾 Batch Validator Script

This is the script I wrote last Tuesday. Runs at 04:00 local time every morning. Checks overnight batch stability. If any parameter drifts beyond tolerance, it wakes me up.

batch_validator.py
import numpy as np
from datetime import datetime
from dataclasses import dataclass

@dataclass
class BatchParameters:
    ph_target: float = 6.2
    ph_tolerance: float = 0.1
    temp_target: float = 4.0
    temp_tolerance: float = 0.3
    humidity_target: float = 97.0
    humidity_tolerance: float = 2.0

class SensorArrayValidator:
    def __init__(self, node_count: int = 12):
        self.nodes = [Node(i) for i in range(node_count)]
        self.batch_manifest = None
        
    def validate_batch(self, batch_id: str) -> bool:
        # Aggregate readings from all nodes
        readings = self._collect_all_readings()
        params = BatchParameters()
        
        # Check pH stability across array
        ph_variance = np.var([r.ph for r in readings])
        if ph_variance > (params.ph_tolerance ** 2):
            self._trigger_alarm("PH variance exceeds tolerance", batch_id)
            return False
            
        # Validate temperature envelope
        temp_min, temp_max = min(r.temp for r in readings), max(r.temp for r in readings)
        if (temp_max - temp_min) > (2 * params.temp_tolerance):
            self._trigger_alarm("Thermal gradient breach", batch_id)
            return False
            
        # Commit to manifest
        self.batch_manifest = {
            'batch_id': batch_id,
            'timestamp': datetime.now().isoformat(),
            'nodes_validated': len(self.nodes),
            'status': 'GREEN_LIGHT'
        }
        return True
    
    def _collect_all_readings(self):
        # Serial handshake with each node
        return [node.sample() for node in self.nodes]
    
    def _trigger_alarm(self, message: str, batch_id: str):
        # Wake-up protocol: SMS + siren + dashboard flash
        logger.critical(f"[{batch_id}] {message}")
        sms_gateway.send(+17155551234, f"ALARM: {message}")

🌐 Live Dashboard Preview

The web interface renders as a living organism. Each node pulses in time with its sampling rate. Green = stable. Amber = approaching boundary. Crimson = immediate intervention required.

I built this with WebGL because matplotlib felt too slow. Because when you're watching bacterial replication curves, latency is the enemy.