传感器数据采集 #

┌─────────────────────────────────────────────────────────┐
│                    传感器分类体系                        │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  按输出信号分类：                                        │
│  ├── 模拟传感器：输出连续模拟信号                        │
│  │   └── 需要ADC转换                                    │
│  ├── 数字传感器：输出数字信号                            │
│  │   └── I2C/SPI/UART接口                               │
│  └── 开关传感器：输出高低电平                            │
│      └── 直接GPIO读取                                   │
│                                                         │
│  按测量对象分类：                                        │
│  ├── 环境传感器：温度、湿度、气压、光照                  │
│  ├── 运动传感器：加速度、陀螺仪、磁力计                  │
│  ├── 气体传感器：CO2、PM2.5、有害气体                    │
│  └── 接近传感器：红外、超声波、激光测距                  │
│                                                         │
└─────────────────────────────────────────────────────────┘

package com.example.sensor;

import com.pi4j.Pi4J;
import com.pi4j.io.gpio.digital.*;
import java.util.concurrent.TimeUnit;

public class DHTSensor {

    private static final int MAX_TIMINGS = 85;
    
    private final DigitalOutput output;
    private final DigitalInput input;
    private final int pin;
    
    public static class DHTData {
        public final double temperature;
        public final double humidity;
        public final boolean valid;
        
        public DHTData(double temperature, double humidity, boolean valid) {
            this.temperature = temperature;
            this.humidity = humidity;
            this.valid = valid;
        }
    }
    
    public DHTSensor(var pi4j, int pin) {
        this.pin = pin;
        
        this.output = pi4j.create(DigitalOutput.newConfigBuilder(pi4j)
                .id("dht-output")
                .name("DHT Output")
                .address(pin)
                .shutdown(DigitalState.HIGH)
                .initial(DigitalState.HIGH)
                .provider("pigpio-digital-output"));
        
        this.input = pi4j.create(DigitalInput.newConfigBuilder(pi4j)
                .id("dht-input")
                .name("DHT Input")
                .address(pin)
                .pull(PullResistance.OFF)
                .provider("pigpio-digital-input"));
    }
    
    public DHTData read() {
        byte[] data = new byte[5];
        
        output.low();
        try { TimeUnit.MILLISECONDS.sleep(18); } catch (InterruptedException e) {}
        output.high();
        try { TimeUnit.MICROSECONDS.sleep(40); } catch (InterruptedException e) {}
        
        int lastState = DigitalState.HIGH.getValue();
        int j = 0;
        
        for (int i = 0; i < MAX_TIMINGS; i++) {
            int count = 0;
            while (input.state().getValue() == lastState) {
                count++;
                try { TimeUnit.MICROSECONDS.sleep(1); } catch (InterruptedException e) {}
                if (count == 255) break;
            }
            
            lastState = input.state().getValue();
            
            if (count == 255) break;
            
            if (i >= 4 && i % 2 == 0) {
                data[j / 8] <<= 1;
                if (count > 16) {
                    data[j / 8] |= 1;
                }
                j++;
            }
        }
        
        if (j >= 40 && checkChecksum(data)) {
            double humidity = ((data[0] << 8) | data[1]) / 10.0;
            double temperature = ((data[2] << 8) | data[3]) / 10.0;
            return new DHTData(temperature, humidity, true);
        }
        
        return new DHTData(0, 0, false);
    }
    
    private boolean checkChecksum(byte[] data) {
        return data[4] == (byte) ((data[0] + data[1] + data[2] + data[3]) & 0xFF);
    }
    
    public static void main(String[] args) throws InterruptedException {
        var pi4j = Pi4J.newAutoContext();
        
        DHTSensor dht = new DHTSensor(pi4j, 17);
        
        System.out.println("DHT温湿度传感器测试");
        
        while (true) {
            DHTData data = dht.read();
            
            if (data.valid) {
                System.out.printf("温度: %.1f°C, 湿度: %.1f%%%n", 
                    data.temperature, data.humidity);
            } else {
                System.out.println("读取失败");
            }
            
            Thread.sleep(2000);
        }
    }
}

package com.example.sensor;

import com.pi4j.Pi4J;
import com.pi4j.io.i2c.I2C;
import com.pi4j.io.i2c.I2CConfig;

public class SHT30Sensor {

    private static final int ADDRESS = 0x44;
    
    private static final byte[] CMD_MEASURE_HIGH = {(byte) 0x2C, 0x06};
    private static final byte[] CMD_MEASURE_MEDIUM = {(byte) 0x2C, 0x0D};
    private static final byte[] CMD_MEASURE_LOW = {(byte) 0x2C, 0x10};
    
    private static final byte[] CMD_HEATER_ON = {(byte) 0x30, 0x6D};
    private static final byte[] CMD_HEATER_OFF = {(byte) 0x30, 0x66};
    
    private final I2C device;
    
    public static class SHT30Data {
        public final double temperature;
        public final double humidity;
        
        public SHT30Data(double temperature, double humidity) {
            this.temperature = temperature;
            this.humidity = humidity;
        }
    }
    
    public SHT30Sensor(var pi4j, int bus) {
        this.device = pi4j.create(I2C.newConfigBuilder(pi4j)
                .id("sht30")
                .name("SHT30 Temperature Humidity Sensor")
                .bus(bus)
                .device(ADDRESS)
                .build());
    }
    
    public SHT30Data read() {
        device.write(CMD_MEASURE_HIGH);
        
        try { Thread.sleep(15); } catch (InterruptedException e) {}
        
        byte[] data = new byte[6];
        device.read(data, 0, 6);
        
        int rawTemp = ((data[0] & 0xFF) << 8) | (data[1] & 0xFF);
        int rawHum = ((data[3] & 0xFF) << 8) | (data[4] & 0xFF);
        
        double temperature = -45 + 175 * rawTemp / 65535.0;
        double humidity = 100 * rawHum / 65535.0;
        
        return new SHT30Data(temperature, humidity);
    }
    
    public void setHeater(boolean on) {
        device.write(on ? CMD_HEATER_ON : CMD_HEATER_OFF);
    }
    
    public void softReset() {
        byte[] cmd = {(byte) 0x30, (byte) 0xA2};
        device.write(cmd);
        try { Thread.sleep(1); } catch (InterruptedException e) {}
    }
    
    public static void main(String[] args) throws InterruptedException {
        var pi4j = Pi4J.newAutoContext();
        
        SHT30Sensor sht30 = new SHT30Sensor(pi4j, 1);
        
        System.out.println("SHT30温湿度传感器测试");
        
        while (true) {
            SHT30Data data = sht30.read();
            System.out.printf("温度: %.2f°C, 湿度: %.2f%%%n", 
                data.temperature, data.humidity);
            Thread.sleep(1000);
        }
    }
}

package com.example.sensor;

import com.pi4j.Pi4J;
import com.pi4j.io.i2c.I2C;
import com.pi4j.io.i2c.I2CConfig;

public class BH1750Sensor {

    private static final int ADDRESS_LOW = 0x23;
    private static final int ADDRESS_HIGH = 0x5C;
    
    private static final byte POWER_DOWN = 0x00;
    private static final byte POWER_ON = 0x01;
    private static final byte RESET = 0x07;
    
    private static final byte CONTINUOUS_HIGH_RES = 0x10;
    private static final byte CONTINUOUS_HIGH_RES_2 = 0x11;
    private static final byte CONTINUOUS_LOW_RES = 0x13;
    private static final byte ONE_TIME_HIGH_RES = 0x20;
    private static final byte ONE_TIME_HIGH_RES_2 = 0x21;
    private static final byte ONE_TIME_LOW_RES = 0x23;
    
    private final I2C device;
    private byte mode = CONTINUOUS_HIGH_RES;
    
    public BH1750Sensor(var pi4j, int bus, boolean addressHigh) {
        int address = addressHigh ? ADDRESS_HIGH : ADDRESS_LOW;
        
        this.device = pi4j.create(I2C.newConfigBuilder(pi4j)
                .id("bh1750")
                .name("BH1750 Light Sensor")
                .bus(bus)
                .device(address)
                .build());
        
        initialize();
    }
    
    private void initialize() {
        device.write(POWER_ON);
        device.write(RESET);
        device.write(mode);
    }
    
    public void setMode(byte mode) {
        this.mode = mode;
        device.write(mode);
    }
    
    public double readLux() {
        byte[] data = new byte[2];
        device.read(data, 0, 2);
        
        int raw = ((data[0] & 0xFF) << 8) | (data[1] & 0xFF);
        
        return raw / 1.2;
    }
    
    public void powerDown() {
        device.write(POWER_DOWN);
    }
    
    public void powerOn() {
        device.write(POWER_ON);
    }
    
    public static void main(String[] args) throws InterruptedException {
        var pi4j = Pi4J.newAutoContext();
        
        BH1750Sensor bh1750 = new BH1750Sensor(pi4j, 1, false);
        
        System.out.println("BH1750光照传感器测试");
        
        while (true) {
            double lux = bh1750.readLux();
            System.out.printf("光照强度: %.1f lux%n", lux);
            Thread.sleep(500);
        }
    }
}

package com.example.sensor;

import com.pi4j.Pi4J;
import com.pi4j.io.i2c.I2C;
import com.pi4j.io.i2c.I2CConfig;

public class TSL2561Sensor {

    private static final int ADDRESS_LOW = 0x29;
    private static final int ADDRESS_FLOAT = 0x39;
    private static final int ADDRESS_HIGH = 0x49;
    
    private static final int REG_CONTROL = 0x80;
    private static final int REG_TIMING = 0x81;
    private static final int REG_DATA0LOW = 0x8C;
    private static final int REG_DATA1LOW = 0x8E;
    
    private static final byte POWER_ON = 0x03;
    private static final byte POWER_OFF = 0x00;
    
    private static final int INTEGRATION_13MS = 0x00;
    private static final int INTEGRATION_101MS = 0x01;
    private static final int INTEGRATION_402MS = 0x02;
    
    private static final int GAIN_1X = 0x00;
    private static final int GAIN_16X = 0x10;
    
    private final I2C device;
    private int integrationTime = INTEGRATION_402MS;
    private int gain = GAIN_1X;
    
    public TSL2561Sensor(var pi4j, int bus, int address) {
        this.device = pi4j.create(I2C.newConfigBuilder(pi4j)
                .id("tsl2561")
                .name("TSL2561 Light Sensor")
                .bus(bus)
                .device(address)
                .build());
        
        initialize();
    }
    
    private void initialize() {
        device.writeRegisterByte(REG_CONTROL, POWER_ON);
        device.writeRegisterByte(REG_TIMING, gain | integrationTime);
    }
    
    public void setTiming(int integration, int gainValue) {
        this.integrationTime = integration;
        this.gain = gainValue;
        device.writeRegisterByte(REG_TIMING, gain | integrationTime);
    }
    
    public int[] readRaw() {
        int ch0Low = device.readRegisterByte(REG_DATA0LOW) & 0xFF;
        int ch0High = device.readRegisterByte(REG_DATA0LOW + 1) & 0xFF;
        int ch0 = (ch0High << 8) | ch0Low;
        
        int ch1Low = device.readRegisterByte(REG_DATA1LOW) & 0xFF;
        int ch1High = device.readRegisterByte(REG_DATA1LOW + 1) & 0xFF;
        int ch1 = (ch1High << 8) | ch1Low;
        
        return new int[]{ch0, ch1};
    }
    
    public double readLux() {
        int[] raw = readRaw();
        int ch0 = raw[0];
        int ch1 = raw[1];
        
        double ratio = ch1 / (double) ch0;
        double lux;
        
        if (ratio <= 0.5) {
            lux = 0.0304 * ch0 - 0.062 * ch0 * Math.pow(ratio, 1.4);
        } else if (ratio <= 0.61) {
            lux = 0.0224 * ch0 - 0.031 * ch1;
        } else if (ratio <= 0.80) {
            lux = 0.0128 * ch0 - 0.0153 * ch1;
        } else if (ratio <= 1.3) {
            lux = 0.00146 * ch0 - 0.00112 * ch1;
        } else {
            lux = 0;
        }
        
        if (integrationTime == INTEGRATION_13MS) {
            lux *= 322.0 / 11;
        } else if (integrationTime == INTEGRATION_101MS) {
            lux *= 322.0 / 81;
        } else {
            lux *= 322.0 / 322;
        }
        
        if (gain == GAIN_16X) {
            lux /= 16;
        }
        
        return lux;
    }
    
    public void powerDown() {
        device.writeRegisterByte(REG_CONTROL, POWER_OFF);
    }
    
    public static void main(String[] args) throws InterruptedException {
        var pi4j = Pi4J.newAutoContext();
        
        TSL2561Sensor tsl = new TSL2561Sensor(pi4j, 1, ADDRESS_FLOAT);
        
        System.out.println("TSL2561光照传感器测试");
        
        while (true) {
            double lux = tsl.readLux();
            int[] raw = tsl.readRaw();
            System.out.printf("CH0: %d, CH1: %d, 光照: %.1f lux%n", 
                raw[0], raw[1], lux);
            Thread.sleep(500);
        }
    }
}

package com.example.sensor;

import com.pi4j.Pi4J;
import com.pi4j.io.i2c.I2C;
import com.pi4j.io.i2c.I2CConfig;

public class MPU6050Sensor {

    private static final int ADDRESS = 0x68;
    
    private static final int REG_SELF_TEST_X = 0x0D;
    private static final int REG_SELF_TEST_Y = 0x0E;
    private static final int REG_SELF_TEST_Z = 0x0F;
    private static final int REG_SELF_TEST_A = 0x10;
    private static final int REG_SMPLRT_DIV = 0x19;
    private static final int REG_CONFIG = 0x1A;
    private static final int REG_GYRO_CONFIG = 0x1B;
    private static final int REG_ACCEL_CONFIG = 0x1C;
    private static final int REG_FIFO_EN = 0x23;
    private static final int REG_I2C_MST_CTRL = 0x24;
    private static final int REG_I2C_SLV0_ADDR = 0x25;
    private static final int REG_I2C_SLV0_REG = 0x26;
    private static final int REG_I2C_SLV0_CTRL = 0x27;
    private static final int REG_INT_ENABLE = 0x38;
    private static final int REG_INT_STATUS = 0x3A;
    private static final int REG_ACCEL_XOUT_H = 0x3B;
    private static final int REG_TEMP_OUT_H = 0x41;
    private static final int REG_GYRO_XOUT_H = 0x43;
    private static final int REG_USER_CTRL = 0x6A;
    private static final int REG_PWR_MGMT_1 = 0x6B;
    private static final int REG_PWR_MGMT_2 = 0x6C;
    private static final int REG_FIFO_COUNT_H = 0x72;
    private static final int REG_FIFO_R_W = 0x74;
    private static final int REG_WHO_AM_I = 0x75;
    
    private final I2C device;
    
    private double accelScale = 16384.0;
    private double gyroScale = 131.0;
    
    public static class MPU6050Data {
        public final double accelX, accelY, accelZ;
        public final double gyroX, gyroY, gyroZ;
        public final double temperature;
        
        public MPU6050Data(double ax, double ay, double az,
                          double gx, double gy, double gz, double temp) {
            this.accelX = ax;
            this.accelY = ay;
            this.accelZ = az;
            this.gyroX = gx;
            this.gyroY = gy;
            this.gyroZ = gz;
            this.temperature = temp;
        }
    }
    
    public MPU6050Sensor(var pi4j, int bus) {
        this.device = pi4j.create(I2C.newConfigBuilder(pi4j)
                .id("mpu6050")
                .name("MPU6050 IMU Sensor")
                .bus(bus)
                .device(ADDRESS)
                .build());
        
        initialize();
    }
    
    private void initialize() {
        device.writeRegisterByte(REG_PWR_MGMT_1, (byte) 0x00);
        
        try { Thread.sleep(100); } catch (InterruptedException e) {}
        
        device.writeRegisterByte(REG_SMPLRT_DIV, (byte) 0x07);
        device.writeRegisterByte(REG_CONFIG, (byte) 0x00);
        device.writeRegisterByte(REG_GYRO_CONFIG, (byte) 0x00);
        device.writeRegisterByte(REG_ACCEL_CONFIG, (byte) 0x00);
    }
    
    public int whoAmI() {
        return device.readRegisterByte(REG_WHO_AM_I) & 0xFF;
    }
    
    public void setAccelRange(int range) {
        device.writeRegisterByte(REG_ACCEL_CONFIG, (byte) (range << 3));
        switch (range) {
            case 0: accelScale = 16384.0; break;
            case 1: accelScale = 8192.0; break;
            case 2: accelScale = 4096.0; break;
            case 3: accelScale = 2048.0; break;
        }
    }
    
    public void setGyroRange(int range) {
        device.writeRegisterByte(REG_GYRO_CONFIG, (byte) (range << 3));
        switch (range) {
            case 0: gyroScale = 131.0; break;
            case 1: gyroScale = 65.5; break;
            case 2: gyroScale = 32.8; break;
            case 3: gyroScale = 16.4; break;
        }
    }
    
    public MPU6050Data readAll() {
        byte[] data = new byte[14];
        device.readRegisterBuffer(REG_ACCEL_XOUT_H, data, 0, 14);
        
        int accelX = (short) (((data[0] & 0xFF) << 8) | (data[1] & 0xFF));
        int accelY = (short) (((data[2] & 0xFF) << 8) | (data[3] & 0xFF));
        int accelZ = (short) (((data[4] & 0xFF) << 8) | (data[5] & 0xFF));
        
        int temp = (short) (((data[6] & 0xFF) << 8) | (data[7] & 0xFF));
        
        int gyroX = (short) (((data[8] & 0xFF) << 8) | (data[9] & 0xFF));
        int gyroY = (short) (((data[10] & 0xFF) << 8) | (data[11] & 0xFF));
        int gyroZ = (short) (((data[12] & 0xFF) << 8) | (data[13] & 0xFF));
        
        double temperature = temp / 340.0 + 36.53;
        
        return new MPU6050Data(
            accelX / accelScale,
            accelY / accelScale,
            accelZ / accelScale,
            gyroX / gyroScale,
            gyroY / gyroScale,
            gyroZ / gyroScale,
            temperature
        );
    }
    
    public double[] getAngles() {
        MPU6050Data data = readAll();
        
        double roll = Math.atan2(data.accelY, data.accelZ) * 180 / Math.PI;
        double pitch = Math.atan2(-data.accelX, 
            Math.sqrt(data.accelY * data.accelY + data.accelZ * data.accelZ)) * 180 / Math.PI;
        
        return new double[]{roll, pitch};
    }
    
    public static void main(String[] args) throws InterruptedException {
        var pi4j = Pi4J.newAutoContext();
        
        MPU6050Sensor mpu = new MPU6050Sensor(pi4j, 1);
        
        System.out.println("MPU6050六轴传感器测试");
        System.out.printf("设备ID: 0x%02X%n", mpu.whoAmI());
        
        while (true) {
            MPU6050Data data = mpu.readAll();
            double[] angles = mpu.getAngles();
            
            System.out.printf("加速度: X=%.2f, Y=%.2f, Z=%.2f g%n",
                data.accelX, data.accelY, data.accelZ);
            System.out.printf("角速度: X=%.2f, Y=%.2f, Z=%.2f °/s%n",
                data.gyroX, data.gyroY, data.gyroZ);
            System.out.printf("温度: %.2f°C%n", data.temperature);
            System.out.printf("姿态: Roll=%.2f°, Pitch=%.2f°%n", 
                angles[0], angles[1]);
            System.out.println("---");
            
            Thread.sleep(500);
        }
    }
}

package com.example.sensor;

import com.pi4j.Pi4J;
import com.pi4j.io.gpio.digital.*;

public class HCSR04Sensor {

    private final DigitalOutput trigger;
    private final DigitalInput echo;
    
    private static final double SOUND_SPEED = 34300.0;
    
    public HCSR04Sensor(var pi4j, int triggerPin, int echoPin) {
        this.trigger = pi4j.create(DigitalOutput.newConfigBuilder(pi4j)
                .id("hcsr04-trigger")
                .name("HC-SR04 Trigger")
                .address(triggerPin)
                .shutdown(DigitalState.LOW)
                .initial(DigitalState.LOW)
                .provider("pigpio-digital-output"));
        
        this.echo = pi4j.create(DigitalInput.newConfigBuilder(pi4j)
                .id("hcsr04-echo")
                .name("HC-SR04 Echo")
                .address(echoPin)
                .pull(PullResistance.DOWN)
                .provider("pigpio-digital-input"));
    }
    
    public double measureDistance() {
        trigger.high();
        try { Thread.sleep(0, 10000); } catch (InterruptedException e) {}
        trigger.low();
        
        long startTime = System.nanoTime();
        long timeout = 30000000;
        
        while (echo.state() == DigitalState.LOW) {
            if (System.nanoTime() - startTime > timeout) {
                return -1;
            }
        }
        
        long echoStart = System.nanoTime();
        
        while (echo.state() == DigitalState.HIGH) {
            if (System.nanoTime() - echoStart > timeout) {
                return -1;
            }
        }
        
        long echoEnd = System.nanoTime();
        
        double duration = (echoEnd - echoStart) / 1000000.0;
        
        return (duration * SOUND_SPEED) / 20000.0;
    }
    
    public double measureDistanceAvg(int samples) {
        double sum = 0;
        int valid = 0;
        
        for (int i = 0; i < samples; i++) {
            double distance = measureDistance();
            if (distance > 0) {
                sum += distance;
                valid++;
            }
            try { Thread.sleep(60); } catch (InterruptedException e) {}
        }
        
        return valid > 0 ? sum / valid : -1;
    }
    
    public static void main(String[] args) throws InterruptedException {
        var pi4j = Pi4J.newAutoContext();
        
        HCSR04Sensor hcsr04 = new HCSR04Sensor(pi4j, 17, 18);
        
        System.out.println("HC-SR04超声波测距传感器测试");
        
        while (true) {
            double distance = hcsr04.measureDistanceAvg(5);
            
            if (distance > 0) {
                System.out.printf("距离: %.2f cm%n", distance);
            } else {
                System.out.println("测量超时");
            }
            
            Thread.sleep(500);
        }
    }
}

package com.example.sensor;

import com.example.adc.MCP3008ADC;

public class MQGasSensor {

    private final MCP3008ADC adc;
    private final int channel;
    private final double vref;
    private final double loadResistance;
    private final double roCleanAir;
    private final double[] curve;
    
    public static class GasType {
        public static final double[] MQ2_LPG = {2.3, 0.21, -0.47};
        public static final double[] MQ2_CO = {2.3, 0.72, -0.34};
        public static final double[] MQ2_SMOKE = {2.3, 0.65, -0.44};
        public static final double[] MQ135_CO2 = {2.3, 0.35, -0.28};
    }
    
    public MQGasSensor(MCP3008ADC adc, int channel, double vref, 
                       double loadResistance, double roCleanAir, double[] curve) {
        this.adc = adc;
        this.channel = channel;
        this.vref = vref;
        this.loadResistance = loadResistance;
        this.roCleanAir = roCleanAir;
        this.curve = curve;
    }
    
    public double readRs() {
        double vOut = adc.readVoltage(channel);
        double rs = loadResistance * (vref - vOut) / vOut;
        return rs;
    }
    
    public double readRatio() {
        return readRs() / roCleanAir;
    }
    
    public double readPPM() {
        double ratio = readRatio();
        return curve[0] * Math.pow(ratio / curve[1], curve[2]);
    }
    
    public double calibrate(int samples) {
        double sum = 0;
        for (int i = 0; i < samples; i++) {
            sum += readRs();
            try { Thread.sleep(100); } catch (InterruptedException e) {}
        }
        return sum / samples;
    }
    
    public static void main(String[] args) throws InterruptedException {
        var pi4j = Pi4J.newAutoContext();
        
        MCP3008ADC adc = new MCP3008ADC(pi4j, 0, 0);
        
        MQGasSensor mq2 = new MQGasSensor(adc, 0, 3.3, 10000, 10000, GasType.MQ2_SMOKE);
        
        System.out.println("MQ-2气体传感器测试");
        
        while (true) {
            double ppm = mq2.readPPM();
            double ratio = mq2.readRatio();
            
            System.out.printf("比值: %.3f, 浓度: %.1f ppm%n", ratio, ppm);
            
            Thread.sleep(1000);
        }
    }
}

package com.example.sensor;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;

public class SensorManager {

    public interface Sensor {
        String getName();
        Map<String, Object> read();
    }
    
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(4);
    private final Map<String, Sensor> sensors = new ConcurrentHashMap<>();
    private final Map<String, List<Consumer<Map<String, Object>>>> listeners = new ConcurrentHashMap<>();
    private final Map<String, ScheduledFuture<?>> tasks = new ConcurrentHashMap<>();
    
    public void registerSensor(String id, Sensor sensor) {
        sensors.put(id, sensor);
        listeners.put(id, new CopyOnWriteArrayList<>());
    }
    
    public void startPolling(String sensorId, int intervalMs) {
        Sensor sensor = sensors.get(sensorId);
        if (sensor == null) return;
        
        stopPolling(sensorId);
        
        ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
            try {
                Map<String, Object> data = sensor.read();
                data.put("timestamp", System.currentTimeMillis());
                data.put("sensor", sensorId);
                
                List<Consumer<Map<String, Object>>> sensorListeners = listeners.get(sensorId);
                if (sensorListeners != null) {
                    for (Consumer<Map<String, Object>> listener : sensorListeners) {
                        listener.accept(data);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, 0, intervalMs, TimeUnit.MILLISECONDS);
        
        tasks.put(sensorId, future);
    }
    
    public void stopPolling(String sensorId) {
        ScheduledFuture<?> future = tasks.remove(sensorId);
        if (future != null) {
            future.cancel(false);
        }
    }
    
    public void addListener(String sensorId, Consumer<Map<String, Object>> listener) {
        List<Consumer<Map<String, Object>>> sensorListeners = listeners.get(sensorId);
        if (sensorListeners != null) {
            sensorListeners.add(listener);
        }
    }
    
    public void removeListener(String sensorId, Consumer<Map<String, Object>> listener) {
        List<Consumer<Map<String, Object>>> sensorListeners = listeners.get(sensorId);
        if (sensorListeners != null) {
            sensorListeners.remove(listener);
        }
    }
    
    public Map<String, Object> readOnce(String sensorId) {
        Sensor sensor = sensors.get(sensorId);
        if (sensor == null) return null;
        
        Map<String, Object> data = sensor.read();
        data.put("timestamp", System.currentTimeMillis());
        data.put("sensor", sensorId);
        return data;
    }
    
    public void shutdown() {
        tasks.values().forEach(f -> f.cancel(false));
        scheduler.shutdown();
    }
}