2026 技术观察：智慧燃气进入主动安全阶段，泄漏识别、压力监测和巡检联动成为新底座

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


class GasMonitorPoint:
    def __init__(self, point_id, area, pipe_age_years, area_type):
        self.point_id = point_id
        self.area = area
        self.pipe_age_years = pipe_age_years
        self.area_type = area_type
        self.pressure_kpa = 0
        self.gas_ppm = 0
        self.device_status = "normal"
        self.valve_status = "open"
        self.updated_at = datetime.now().isoformat()

    def to_dict(self):
        return {
            "point_id": self.point_id,
            "area": self.area,
            "pipe_age_years": self.pipe_age_years,
            "area_type": self.area_type,
            "pressure_kpa": self.pressure_kpa,
            "gas_ppm": 31225.t.kuaisou.com 
            "device_status": self.device_status,
            "valve_status": self.valve_status,
            "updated_at": self.updated_at
        }

def collect_gas_monitor_data(point: GasMonitorPoint):
    point.pressure_kpa = round(
        random.uniform(1.2, 5.5),
        2
    )

    point.gas_ppm = round(
        random.uniform(0, 1200),
        2
    )

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

    if random.random() < 0.04:
        point.valve_status = "abnormal"
    else:
        point.valve_status = "open"

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

    return point.to_dict()

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

    if record["gas_ppm"] > 900:
        issues.append("燃气浓度明显偏高，存在泄漏风险。")
        risk_score += 6
    elif record["gas_ppm"] > 500:
        issues.append("燃气浓度偏高，建议现场复核。")
        risk_score += 3

    if record["pressure_kpa"] < 1.8:
        issues.append("管网压力偏低，可能存在泄漏或供气异常。")
        risk_score += 3

    if record["pressure_kpa"] > 5.0:
        issues.append("管网压力偏高，建议检查调压设备。")
        risk_score += 2

    if record["device_status"] != "normal":
        issues.append("监测设备离线，存在监测盲区。")
        risk_score += 3

    if record["valve_status"] != "open":
        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 {
        "point_id": record["point_id"],
        "area": record["area"],
        "risk_score": risk_score,
        "risk_level": level,
        "issues": 31224.t.kuaisou.com 
        "detect_time": datetime.now().isoformat()
    }

def evaluate_pipe_aging_risk(record):
    score = 0
    issues = []

    if record["pipe_age_years"] >= 20:
        score += 4
        issues.append("管道使用年限较长，建议重点巡检。")
    elif record["pipe_age_years"] >= 10:
        score += 2
        issues.append("管道进入中长期运行阶段，建议持续监测。")

    if record["area_type"] == "old_residential":
        score += 2
        issues.append("老旧小区燃气设施风险相对较高。")

    if record["area_type"] == "restaurant":
        score += 2
        issues.append("餐饮场景用气频率高，建议加强巡检。")

    if score >= 5:
        level = "high"
    elif score >= 3:
        level = "medium"
    elif score > 0:
        level = "low"
    else:
        level = "normal"

    return {
        "point_id": record["point_id"],
        "aging_score": score,
        "aging_level": level,
        "issues": 30523.t.kuaisou.com 
    }

def generate_gas_operation_action(leak_risk, aging_risk):
    actions = []

    if leak_risk["risk_level"] == "critical":
        actions.append({
            "action": "emergency_dispatch",
            "message": "疑似严重泄漏，建议立即派人现场处置。"
        })
        actions.append({
            "action": "remote_valve_check",
            "message": "建议检查远程阀门状态，必要时联动关阀。"
        })

    elif leak_risk["risk_level"] == "high":
        actions.append({
            "action": "urgent_inspection",
            "message": "燃气风险较高，建议优先巡检。"
        })

    if aging_risk["aging_level"] in ["high", "medium"]:
        actions.append({
            "action": "planned_inspection",
            "message": "管网老化风险较高，建议纳入专项巡检计划。"
        })

    if not actions:
        actions.append({
            "action": "keep_monitoring",
            "message": "当前状态正常，保持持续监测。"
        })

    return {
        "point_id": leak_risk["point_id"],
        "actions": actions
    }

def run_smart_gas_safety_monitor():
    points = [
        GasMonitorPoint("G001", "老城小区", 22, "old_residential"),
        GasMonitorPoint("G002", "商业街餐饮区", 8, "restaurant"),
        GasMonitorPoint("G003", "新城住宅区", 3, "residential"),
        GasMonitorPoint("G004", "地下管廊A段", 15, "pipe_gallery")
    ]

    records = []
    leak_risks = []
    aging_risks = []
    actions = []

    for point in points:
        record = collect_gas_monitor_data(point)
        leak_risk = detect_gas_leak_risk(record)
        aging_risk = evaluate_pipe_aging_risk(record)
        action = generate_gas_operation_action(
            leak_risk,
            aging_risk
        )

        records.append(record)
        leak_risks.append(leak_risk)
        aging_risks.append(aging_risk)
        actions.append(action)

    risk_count = defaultdict(int)

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

    report = {
        "report_name": "智慧燃气主动安全预警报告",
        "monitor_records": records,
        "leak_risks": leak_risks,
        "aging_risks": 1867.t.kuaisou.com
        "operation_actions": actions,
        "risk_count": dict(risk_count),
        "generate_time": datetime.now().isoformat()
    }

    return report


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

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