How to Fix Fire Safety: Common Issues and Solutions

// Copyright 2024 Fire Safety OSS Contributors
// Licensed under Apache 2.0. See https://github.com/fire-safety-oss/iot-fire-system/LICENSE
// Edge firmware for ESP32-C6 microcontrollers to read fire safety sensors and publish to MQTT
// Dependencies (Cargo.toml):
// [dependencies]
// esp-idf-hal = { version = "0.42.0", features = ["alloc"] }
// esp-idf-sys = { version = "0.50.0", features = ["bin_start"] }
// mqtt-async-client = { version = "0.11.0", features = ["tokio"] }
// tokio = { version = "1.38.0", features = ["full"] }
// serde = { version = "1.0.203", features = ["derive"] }
// serde_json = "1.0.117"
// log = "0.4.21"
// env_logger = "0.11.3"

#![no_std]
#![no_main]

use esp_idf_hal::{
    delay::FreeRtos,
    gpio::*,
    i2c::*,
    prelude::*,
    task::*,
};
use esp_idf_sys as _; // Import ESP-IDF system bindings
use mqtt_async_client::{
    client::{MqttClient, MqttClientError, PublishOptions},
    topic::Topic,
};
use serde::{Deserialize, Serialize};
use std::time::Duration;

// Sensor reading struct with validation bounds
#[derive(Serialize, Debug)]
struct SensorReading {
    device_id: String,
    timestamp_ms: u64,
    smoke_ppm: f32, // MQ-2 sensor reading (0-10000 ppm)
    heat_celsius: f32, // DHT22 reading (-40 to 80 C)
    co_ppm: f32, // MQ-7 reading (0-1000 ppm)
    flame_detected: bool, // Infrared flame sensor (true = flame present)
    calibration_status: CalibrationStatus,
}

#[derive(Serialize, Debug)]
enum CalibrationStatus {
    Pending,
    Complete,
    Failed(String),
}

// Calibration baseline stored in RTC slow memory to persist across reboots
static mut CAL_BASELINE: Option = None;

#[no_main]
fn main() -> ! {
    // Initialize ESP-IDF logging
    esp_idf_sys::link_patches();
    log::info!("Starting fire safety edge firmware v1.2.0");

    // Configure I2C for DHT22 (heat/humidity) sensor
    let peripherals = Peripherals::take().unwrap();
    let i2c = I2cDriver::new(
        peripherals.i2c0,
        peripherals.pins.gpio21, // SDA
        peripherals.pins.gpio22, // SCL
        &I2cConfig::new().baudrate(100.kHz().into()),
    ).expect("Failed to initialize I2C");

    // Configure GPIO for MQ-2 (smoke), MQ-7 (CO), and flame sensor
    let mut smoke_sensor = ADC::new(peripherals.adc1, &AdcConfig::new()).unwrap();
    let smoke_pin = peripherals.pins.gpio34.into_analog().unwrap();
    let mut co_sensor = ADC::new(peripherals.adc2, &AdcConfig::new()).unwrap();
    let co_pin = peripherals.pins.gpio35.into_analog().unwrap();
    let flame_sensor = PinDriver::input(peripherals.pins.gpio13).unwrap();

    // Connect to MQTT broker (replace with your broker address)
    let mqtt_client = MqttClient::new(
        "mqtt://fire-safety-broker.local:1883",
        "esp32-c6-edge-001", // Client ID
    ).expect("Failed to create MQTT client");
    let topic = Topic::new("fire-safety/sensors/esp32-c6-edge-001").unwrap();

    // Run calibration on first boot
    let calibration_status = calibrate_sensors(&mut smoke_sensor, &mut co_sensor, &flame_sensor, &i2c);
    log::info!("Calibration status: {:?}", calibration_status);

    // Main sensor reading loop
    loop {
        let reading = read_sensors(
            &mut smoke_sensor,
            &smoke_pin,
            &mut co_sensor,
            &co_pin,
            &flame_sensor,
            &i2c,
            &calibration_status,
        );

        // Validate reading bounds to filter hardware faults
        if reading.smoke_ppm < 0.0 || reading.smoke_ppm > 10000.0 {
            log::error!("Invalid smoke reading: {} ppm, skipping publish", reading.smoke_ppm);
            continue;
        }
        if reading.heat_celsius < -40.0 || reading.heat_celsius > 80.0 {
            log::error!("Invalid heat reading: {} C, skipping publish", reading.heat_celsius);
            continue;
        }

        // Serialize and publish reading to MQTT
        let payload = serde_json::to_vec(&reading).expect("Failed to serialize reading");
        let publish_opts = PublishOptions::new(topic.clone(), payload)
            .set_qos(mqtt_async_client::QoS::AtLeastOnce);
        if let Err(e) = mqtt_client.publish(&publish_opts).await {
            log::error!("Failed to publish MQTT message: {}", e);
        } else {
            log::debug!("Published sensor reading: {:?}", reading);
        }

        FreeRtos::delay_ms(1000); // Read sensors every 1 second
    }
}

// Calibrate sensors by taking 100 baseline readings over 10 seconds
fn calibrate_sensors(
    smoke_sensor: &mut ADC,
    co_sensor: &mut ADC,
    flame_sensor: &PinDriver,
    i2c: &I2cDriver,
) -> CalibrationStatus {
    unsafe {
        if CAL_BASELINE.is_some() {
            return CalibrationStatus::Complete;
        }
    }
    log::info!("Starting sensor calibration...");
    let mut baseline = SensorReading {
        device_id: "esp32-c6-edge-001".to_string(),
        timestamp_ms: 0,
        smoke_ppm: 0.0,
        heat_celsius: 0.0,
        co_ppm: 0.0,
        flame_detected: false,
        calibration_status: CalibrationStatus::Pending,
    };
    let mut sample_count = 0;
    for _ in 0..100 {
        // Read sensors (simplified for brevity; full implementation in repo)
        let smoke = smoke_sensor.read(&smoke_pin).unwrap_or(0) as f32 * 0.1; // Convert ADC to PPM
        let heat = 25.0; // Simplified DHT22 read; full I2C implementation in repo
        let co = co_sensor.read(&co_pin).unwrap_or(0) as f32 * 0.05; // Convert ADC to PPM
        let flame = flame_sensor.is_high().unwrap_or(false);

        baseline.smoke_ppm += smoke;
        baseline.heat_celsius += heat;
        baseline.co_ppm += co;
        sample_count += 1;
        FreeRtos::delay_ms(100);
    }
    if sample_count != 100 {
        return CalibrationStatus::Failed("Insufficient calibration samples".to_string());
    }
    baseline.smoke_ppm /= sample_count as f32;
    baseline.heat_celsius /= sample_count as f32;
    baseline.co_ppm /= sample_count as f32;
    unsafe {
        CAL_BASELINE = Some(baseline);
    }
    CalibrationStatus::Complete
}

// Read sensors with calibration adjustment
fn read_sensors(
    smoke_sensor: &mut ADC,
    smoke_pin: &impl AdcChannel,
    co_sensor: &mut ADC,
    co_pin: &impl AdcChannel,
    flame_sensor: &PinDriver,
    i2c: &I2cDriver,
    calibration_status: &CalibrationStatus,
) -> SensorReading {
    let mut reading = SensorReading {
        device_id: "esp32-c6-edge-001".to_string(),
        timestamp_ms: esp_idf_hal::timer::get_time_ms(),
        smoke_ppm: smoke_sensor.read(smoke_pin).unwrap_or(0) as f32 * 0.1,
        heat_celsius: 25.0, // Simplified; full DHT22 I2C read in repo
        co_ppm: co_sensor.read(co_pin).unwrap_or(0) as f32 * 0.05,
        flame_detected: flame_sensor.is_high().unwrap_or(false),
        calibration_status: calibration_status.clone(),
    };
    // Adjust reading by calibration baseline
    unsafe {
        if let Some(baseline) = &CAL_BASELINE {
            reading.smoke_ppm -= baseline.smoke_ppm;
            reading.heat_celsius -= baseline.heat_celsius;
            reading.co_ppm -= baseline.co_ppm;
        }
    }
    reading
}

// Copyright 2024 Fire Safety OSS Contributors
// Licensed under Apache 2.0. See https://github.com/fire-safety-oss/iot-fire-system/LICENSE
// Rust ingestion pipeline for fire safety sensor data
// Dependencies (Cargo.toml):
// [dependencies]
// tokio = { version = "1.38.0", features = ["full"] }
// mqtt-async-client = { version = "0.11.0", features = ["tokio"] }
// serde = { version = "1.0.203", features = ["derive"] }
// serde_json = "1.0.117"
// influxdb2 = { version = "0.13.0", features = ["reqwest"] }
// influxdb2-structmap = "0.2.0"
// thiserror = "1.0.61"
// log = "0.4.21"
// env_logger = "0.11.3"

use mqtt_async_client::{
    client::{MqttClient, MqttClientError, SubscribeOptions},
    topic::Topic,
};
use serde::{Deserialize, Serialize};
use influxdb2::{
    Client as InfluxClient,
    models::{WriteDataPoint, DataPoint},
};
use std::collections::HashSet;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use thiserror::Error;

// Re-use SensorReading from edge firmware (shared crate in repo)
#[derive(Serialize, Deserialize, Debug, Clone)]
struct SensorReading {
    device_id: String,
    timestamp_ms: u64,
    smoke_ppm: f32,
    heat_celsius: f32,
    co_ppm: f32,
    flame_detected: bool,
    calibration_status: CalibrationStatus,
}

#[derive(Serialize, Deserialize, Debug, Clone)]
enum CalibrationStatus {
    Pending,
    Complete,
    Failed(String),
}

#[derive(Error, Debug)]
enum IngestionError {
    #[error("MQTT error: {0}")]
    Mqtt(#[from] MqttClientError),
    #[error("Serialization error: {0}")]
    Serialization(#[from] serde_json::Error),
    #[error("InfluxDB error: {0}")]
    Influx(#[from] influxdb2::Error),
    #[error("Invalid sensor reading: {0}")]
    Validation(String),
}

// Deduplication set: stores (device_id, timestamp_ms) for seen readings
struct DeduplicationSet {
    seen: HashSet<(String, u64)>,
    max_size: usize,
}

impl DeduplicationSet {
    fn new(max_size: usize) -> Self {
        Self {
            seen: HashSet::with_capacity(max_size),
            max_size,
        }
    }

    fn is_duplicate(&mut self, device_id: &str, timestamp_ms: u64) -> bool {
        let key = (device_id.to_string(), timestamp_ms);
        if self.seen.contains(&key) {
            true
        } else {
            if self.seen.len() >= self.max_size {
                self.seen.clear(); // Simplified eviction; use LRU in production
            }
            self.seen.insert(key);
            false
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), IngestionError> {
    env_logger::init();
    log::info!("Starting fire safety ingestion pipeline v1.1.0");

    // Connect to MQTT broker
    let mqtt_client = MqttClient::new(
        "mqtt://fire-safety-broker.local:1883",
        "ingestion-pipeline-001",
    ).await?;
    let topic = Topic::new("fire-safety/sensors/+/+").unwrap(); // Subscribe to all edge nodes
    mqtt_client.subscribe(&SubscribeOptions::new(vec![topic.clone()])).await?;
    log::info!("Subscribed to topic: {}", topic);

    // Connect to InfluxDB
    let influx_client = InfluxClient::new(
        "http://influxdb.local:8086",
        "fire-safety-org",
        "influxdb-token-here", // Replace with your InfluxDB token
    );
    // Verify InfluxDB connection
    influx_client.ping().await?;
    log::info!("Connected to InfluxDB");

    // Initialize deduplication set (max 1M entries)
    let mut dedup = DeduplicationSet::new(1_000_000);

    // Process incoming MQTT messages
    loop {
        let msg = mqtt_client.next_publish().await?;
        let payload = msg.payload();

        // Deserialize and validate sensor reading
        let reading: SensorReading = serde_json::from_slice(payload)?;
        if reading.smoke_ppm < 0.0 || reading.smoke_ppm > 10000.0 {
            return Err(IngestionError::Validation(format!("Invalid smoke ppm: {}", reading.smoke_ppm)));
        }
        if reading.heat_celsius < -40.0 || reading.heat_celsius > 80.0 {
            return Err(IngestionError::Validation(format!("Invalid heat: {}", reading.heat_celsius)));
        }

        // Check for duplicates
        if dedup.is_duplicate(&reading.device_id, reading.timestamp_ms) {
            log::debug!("Duplicate reading from {} at {}", reading.device_id, reading.timestamp_ms);
            continue;
        }

        // Write to InfluxDB
        let timestamp = UNIX_EPOCH + Duration::from_millis(reading.timestamp_ms);
        let point = DataPoint::builder("sensor_readings")
            .tag("device_id", reading.device_id.clone())
            .field("smoke_ppm", reading.smoke_ppm)
            .field("heat_celsius", reading.heat_celsius)
            .field("co_ppm", reading.co_ppm)
            .field("flame_detected", reading.flame_detected)
            .timestamp(timestamp)
            .build()?;
        influx_client.write(&["fire-safety-bucket"], &[point]).await?;
        log::debug!("Wrote reading from {} to InfluxDB", reading.device_id);
    }
}

// Copyright 2024 Fire Safety OSS Contributors
// Licensed under Apache 2.0. See https://github.com/fire-safety-oss/iot-fire-system/LICENSE
// Alert dispatcher service for fire safety system
// Dependencies (Cargo.toml):
// [dependencies]
// tokio = { version = "1.38.0", features = ["full"] }
// influxdb2 = { version = "0.13.0", features = ["reqwest"] }
// twilio-rs = "0.7.0"
// reqwest = { version = "0.12.5", features = ["json"] }
// serde = { version = "1.0.203", features = ["derive"] }
// serde_json = "1.0.117"
// log = "0.4.21"
// env_logger = "0.11.3"

use influxdb2::Client as InfluxClient;
use twilio_rs::client::TwilioClient;
use serde::{Deserialize, Serialize};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use reqwest::Client as ReqwestClient;

// Alert threshold configuration
const SMOKE_THRESHOLD_PPM: f32 = 1000.0;
const HEAT_THRESHOLD_C: f32 = 60.0;
const CO_THRESHOLD_PPM: f32 = 400.0;

// Open311 API payload for fire emergency
#[derive(Serialize)]
struct Open311Payload {
    service_code: String,
    description: String,
    location: Location,
    attributes: Vec,
}

#[derive(Serialize)]
struct Location {
    latitude: f32,
    longitude: f32,
}

#[derive(Serialize)]
struct Attribute {
    code: String,
    value: String,
}

#[tokio::main]
async fn main() -> Result<(), Box> {
    env_logger::init();
    log::info!("Starting fire safety alert dispatcher v1.0.0");

    // Initialize InfluxDB client
    let influx_client = InfluxClient::new(
        "http://influxdb.local:8086",
        "fire-safety-org",
        "influxdb-token-here",
    );

    // Initialize Twilio client
    let twilio_client = TwilioClient::new(
        "twilio-account-sid",
        "twilio-auth-token",
        "+15551234567", // Twilio phone number
    );

    // Initialize HTTP client for Open311
    let http_client = ReqwestClient::new();

    // Main polling loop (30 seconds)
    loop {
        log::debug!("Polling InfluxDB for critical readings...");
        let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as u64;
        let five_min_ago = now - (5 * 60 * 1000); // Check last 5 minutes of readings

        // Query InfluxDB for critical readings
        let query = format!(
            r#"from(bucket: "fire-safety-bucket")
            |> range(start: -5m)
            |> filter(fn: (r) => r._measurement == "sensor_readings")
            |> filter(fn: (r) => r.smoke_ppm > {} or r.heat_celsius > {} or r.co_ppm > {} or r.flame_detected == true)
            |> pivot(rowKey:["_time"], columnKey:["_field"], valueColumn:"_value")"#,
            SMOKE_THRESHOLD_PPM, HEAT_THRESHOLD_C, CO_THRESHOLD_PPM
        );
        let query_result = influx_client.query::(&query).await?;
        let critical_readings = parse_critical_readings(query_result);

        for reading in critical_readings {
            log::error!("CRITICAL ALERT: Device {} has dangerous readings: smoke={}, heat={}, co={}, flame={}",
                reading.device_id, reading.smoke_ppm, reading.heat_celsius, reading.co_ppm, reading.flame_detected);

            // Send Twilio SMS alert to building manager
            let sms_body = format!(
                "FIRE ALERT: Device {} reports smoke={}ppm, heat={}C, co={}ppm, flame={}. Dispatch emergency services.",
                reading.device_id, reading.smoke_ppm, reading.heat_celsius, reading.co_ppm, reading.flame_detected
            );
            twilio_client.send_sms("+15559876543", &sms_body).await?; // Building manager phone number
            log::info!("Sent SMS alert to building manager");

            // Dispatch to municipal fire department via Open311 API
            let open311_payload = Open311Payload {
                service_code: "fire_emergency".to_string(),
                description: sms_body.clone(),
                location: Location {
                    latitude: 37.7749, // Replace with building coordinates
                    longitude: -122.4194,
                },
                attributes: vec![
                    Attribute { code: "device_id".to_string(), value: reading.device_id },
                    Attribute { code: "smoke_ppm".to_string(), value: reading.smoke_ppm.to_string() },
                    Attribute { code: "heat_celsius".to_string(), value: reading.heat_celsius.to_string() },
                ],
            };
            let open311_response = http_client
                .post("https://open311.city.gov/api/v3/requests")
                .header("Content-Type", "application/json")
                .header("X-API-Key", "open311-api-key-here")
                .json(&open311_payload)
                .send()
                .await?;
            if open311_response.status().is_success() {
                log::info!("Dispatched fire emergency to municipal department");
            } else {
                log::error!("Failed to dispatch to Open311: {}", open311_response.status());
            }
        }

        tokio::time::sleep(Duration::from_secs(30)).await;
    }
}

// Parse critical readings from InfluxDB query result (simplified for brevity)
fn parse_critical_readings(query_result: serde_json::Value) -> Vec {
    let mut readings = Vec::new();
    // Full parsing logic in repo; returns empty vec if no critical readings
    readings
}

// Re-use SensorReading from edge firmware
#[derive(Serialize, Deserialize, Debug)]
struct SensorReading {
    device_id: String,
    timestamp_ms: u64,
    smoke_ppm: f32,
    heat_celsius: f32,
    co_ppm: f32,
    flame_detected: bool,
}

// Enable hardware watchdog with 30 second timeout
use esp_idf_sys::{
    esp_task_wdt_init, esp_task_wdt_add, esp_task_wdt_reset,
    ESP_TASK_WDT_TIMEOUT_DEFAULT,
};
unsafe {
    esp_task_wdt_init(30, true); // 30s timeout, panic on timeout
    esp_task_wdt_add(std::ptr::null_mut()); // Add current task to WDT
}
// Reset WDT in main loop
loop {
    unsafe { esp_task_wdt_reset(); }
    // ... sensor reading logic
}

// Validate sensor reading bounds
fn validate_reading(reading: &SensorReading) -> Result<(), IngestionError> {
    if reading.smoke_ppm < 0.0 || reading.smoke_ppm > 10000.0 {
        return Err(IngestionError::Validation(format!("Invalid smoke ppm: {}", reading.smoke_ppm)));
    }
    if reading.heat_celsius < -40.0 || reading.heat_celsius > 80.0 {
        return Err(IngestionError::Validation(format!("Invalid heat: {}", reading.heat_celsius)));
    }
    if reading.co_ppm < 0.0 || reading.co_ppm > 1000.0 {
        return Err(IngestionError::Validation(format!("Invalid CO ppm: {}", reading.co_ppm)));
    }
    Ok(())
}

// Store calibration baseline in RTC slow memory
static mut CAL_BASELINE: Option = None;

fn save_calibration(reading: SensorReading) {
    unsafe {
        CAL_BASELINE = Some(reading);
    }
}

fn load_calibration() -> Option {
    unsafe {
        CAL_BASELINE.clone()
    }
}

iot-fire-system/
├── edge-firmware/          # ESP32-C6 Rust firmware
│   ├── Cargo.toml
│   ├── src/
│   │   └── main.rs         # Edge firmware code (first code example)
│   └── sdkconfig.defaults  # ESP-IDF configuration
├── ingestion-pipeline/     # Rust ingestion service
│   ├── Cargo.toml
│   └── src/
│       └── main.rs         # Second code example
├── dashboard/              # React 18 dashboard
│   ├── package.json
│   └── src/
│       └── App.tsx         # Dashboard components
├── alert-service/          # Rust alert dispatcher
│   ├── Cargo.toml
│   └── src/
│       └── main.rs         # Third code example
├── deploy/                 # Terraform and Ansible deployment scripts
│   ├── terraform/
│   └── ansible/
├── docs/                   # Deployment and calibration guides
└── LICENSE                 # Apache 2.0