2026 技术观察：智慧消防进入主动预警阶段，烟感、水压和疏散联动成为城市安全新底座

import json
import random
from datetime import datetime
from collections import defaultdict


class FireZone:
    def __init__(self, zone_id, building, floor, zone_type):
        self.zone_id = zone_id
        self.building = building
        self.floor = floor
        self.zone_type = zone_type
        self.smoke_ppm = 0
        self.temperature = 0
        self.water_pressure = 0
        self.people_count = 0
        self.device_status = "normal"
        self.exit_blocked = False
        self.updated_at = datetime.now().isoformat()

    def to_dict(self):
        return {
            "zone_id": self.zone_id,
            "building": self.building,
            "floor": self.floor,
            "zone_type": self.zone_type,
            "smoke_ppm": self.smoke_ppm,
            "temperature": self.temperature,
            "water_pressure": self.water_pressure,
            "people_count": self.people_count,
            "device_status": self.device_status,
            "exit_blocked": self.exit_blocked,
            "updated_at": self.updated_at
        }

def collect_fire_zone_data(zone: FireZone):
    zone.smoke_ppm = round(
        random.uniform(0, 120),
        2
    )

    zone.temperature = round(
        random.uniform(18, 75),
        2
    )

    zone.water_pressure = round(
        random.uniform(0.1, 0.8),
        2
    )

    zone.people_count = random.randint(0, 120)

    zone.device_status = "normal" if random.random() > 0.08 else "offline"

    zone.exit_blocked = random.random() < 0.1

    zone.updated_at = datetime.now().isoformat()

    return zone.to_dict()

def detect_fire_risk(record):
    issues = []
    risk_score = 0

    if record["smoke_ppm"] > 80:
        issues.append("烟雾浓度明显升高。")
        risk_score += 5
    elif record["smoke_ppm"] > 40:
        issues.append("烟雾浓度偏高，建议关注。")
        risk_score += 3

    if record["temperature"] > 60:
        issues.append("区域温度异常升高。")
        risk_score += 5
    elif record["temperature"] > 45:
        issues.append("区域温度偏高。")
        risk_score += 2

    if record["device_status"] != "normal":
        issues.append("消防监测设备离线。")
        risk_score += 3

    if risk_score >= 8:
        level = "critical"
    elif risk_score >= 5:
        level = "high"
    elif risk_score >= 2:
        level = "medium"
    else:
        level = "normal"

    return {
        "zone_id": record["zone_id"],
        "building": record["building"],
        "floor": record["floor"],
        "risk_score": risk_score,
        "risk_level": level,
        "issues": 31220.t.kuaisou.com 
        "detect_time": datetime.now().isoformat()
    }

def check_facility_risk(record):
    issues = []
    risk_score = 0

    if record["water_pressure"] < 0.25:
        issues.append("喷淋水压偏低，可能影响灭火能力。")
        risk_score += 4

    if record["exit_blocked"]:
        issues.append("疑似疏散通道被占用或堵塞。")
        risk_score += 4

    if record["people_count"] > 80 and record["exit_blocked"]:
        issues.append("人员密集且通道异常，疏散风险较高。")
        risk_score += 3

    if risk_score >= 6:
        level = "high"
    elif risk_score >= 3:
        level = "medium"
    elif risk_score > 0:
        level = "low"
    else:
        level = "normal"

    return {
        "zone_id": record["zone_id"],
        "facility_risk_score": risk_score,
        "facility_risk_level": level,
        "issues": 31221.t.kuaisou.com 
    }

def generate_fire_response_policy(fire_risk, facility_risk):
    actions = []

    if fire_risk["risk_level"] == "critical":
        actions.append("启动声光报警。")
        actions.append("通知消防控制室立即确认。")
        actions.append("联动广播提示人员疏散。")

    elif fire_risk["risk_level"] == "high":
        actions.append("通知值班人员现场复核。")
        actions.append("提高该区域监测频率。")

    if facility_risk["facility_risk_level"] == "high":
        actions.append("生成消防设施巡检工单。")

    if any("疏散通道" in issue for issue in facility_risk["issues"]):
        actions.append("通知物业清理疏散通道。")

    if not actions:
        actions.append("保持常规监测。")

    return {
        "zone_id": fire_risk["zone_id"],
        "actions": actions
    }

def run_smart_fire_monitor():
    zones = [
        FireZone("Z001", "A座", 3, "office"),
        FireZone("Z002", "A座", 5, "equipment_room"),
        FireZone("Z003", "B座", 1, "lobby"),
        FireZone("Z004", "B座", -1, "parking")
    ]

    records = []
    fire_risks = []
    facility_risks = []
    policies = []

    for zone in zones:
        record = collect_fire_zone_data(zone)
        fire_risk = detect_fire_risk(record)
        facility_risk = check_facility_risk(record)
        policy = generate_fire_response_policy(
            fire_risk,
            facility_risk
        )

        records.append(record)
        fire_risks.append(fire_risk)
        facility_risks.append(facility_risk)
        policies.append(policy)

    risk_count = defaultdict(int)

    for risk in fire_risks:
        risk_count[risk["risk_level"]] += 1

    report = {
        "report_name": "智慧消防主动预警报告",
        "zone_records": records,
        "fire_risks": fire_risks,
        "facility_risks": facility_risks,
        "response_policies": 31222.t.kuaisou.com 
        "risk_count": dict(risk_count),
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_smart_fire_monitor()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))