智慧环卫动态调度垃圾满溢识别与清运路线优化

2026 年，城市环卫正经历深刻变革：从传统的“固定路线清运”向“动态调度运营”转型。智慧环卫系统的落地，正让垃圾清运变得更智能、更高效。

过去是什么样？固定时间、固定路线、固定车辆，每天按部就班。哪些垃圾桶需要清理、哪些道路需要清扫、哪些点位可能发生满溢，全都依赖人工巡查和经验判断。

这套模式虽然稳定，但也存在两个明显的“痛点”。

一方面，不少垃圾桶还没装满，清运车却已经开过去了，燃油和人力白白浪费；另一方面，人流密集区域的垃圾可能已经冒尖，却无人知晓，直到居民投诉才被动处理。

如今，随着智能垃圾桶、车载定位、图像识别、物联网传感器和调度算法的成熟应用，智慧环卫正式迈入动态调度阶段。

简单来说，系统不再仅仅记录“谁干了什么活”，而是通过分析垃圾桶的满载率、区域人流量、车辆实时位置、道路距离和任务优先级，自动生成最优清运路线，甚至直接给出调度建议。

一、为什么环卫需要动态调度？

城市环卫有一个典型特征——时空差异显著。

商业街、景区、学校、地铁口、居民区和办公区，垃圾产生的节奏截然不同。如果所有区域都采用统一的清运频率，既无法保证效率，也难以保障服务质量。

动态调度系统能够帮助管理者快速回答几个关键问题：

下面用 Python 编写一个简化版的智慧环卫清运调度系统，直观展示这套逻辑的运行过程。

二、基础数据：定义垃圾桶和清运车辆

第一步，定义垃圾桶点位和清运车辆。

每个垃圾桶包含位置坐标、满载率、区域类型和状态；每辆车包含当前位置、容量和状态。

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

class TrashBin:
    def __init__(self, bin_id, area, x, y, area_type):
        self.bin_id = bin_id
        self.area = area
        self.x = x
        self.y = y
        self.area_type = area_type
        self.fill_rate = 0
        self.status = "normal"
        self.updated_at = datetime.now().isoformat()

    def to_dict(self):
        return {
            "bin_id": self.bin_id,
            "area": self.area,
            "x": self.x,
            "y": self.y,
            "area_type": self.area_type,
            "fill_rate": self.fill_rate,
            "status": self.status,
            "updated_at": self.updated_at
        }

class SanitationTruck:
    def __init__(self, truck_id, x, y, capacity):
        self.truck_id = truck_id
        self.x = x
        self.y = y
        self.capacity = capacity
        self.current_load = 0
        self.status = "idle"

    def to_dict(self):
        return {
            "truck_id": self.truck_id,
            "x": self.x,
            "y": self.y,
            "capacity": self.capacity,
            "current_load": self.current_load,
            "status": self.status
        }

垃圾桶和车辆是智慧环卫调度的基础对象。在真实系统中，满载率数据可能来自传感器，也可能源自巡检拍照和图像识别技术。

三、采集垃圾桶状态

第二步，模拟垃圾桶实时状态采集。

不同区域的垃圾增长速度差异明显——商业区和景区更容易出现高满载率。

def collect_bin_status(bin_obj: TrashBin):
    base_rate = {
        "business": 0.75,
        "scenic": 0.7,
        "residential": 0.55,
        "office": 0.5,
        "school": 0.65
    }.get(bin_obj.area_type, 0.5)

    random_factor = random.uniform(-0.25, 0.25)
    bin_obj.fill_rate = round(min(max(base_rate + random_factor, 0), 1), 2)

    if bin_obj.fill_rate >= 0.9:
        bin_obj.status = "overflow_risk"
    elif bin_obj.fill_rate >= 0.75:
        bin_obj.status = "need_collect"
    else:
        bin_obj.status = "normal"
    bin_obj.updated_at = datetime.now().isoformat()
    return bin_obj.to_dict()

满载率是判断清运优先级的重要指标。动态调度的核心，就是让清运工作从“固定模式”转向“按需响应”。

四、清运任务生成

第三步，根据垃圾桶状态生成清运任务。

满载率越高，任务优先级也越高。

def generate_collection_tasks(bin_records):
    tasks = []
    for record in bin_records:
        if record["fill_rate"] < 0.7:
            continue
        if record["fill_rate"] >= 0.9:
            priority = 10
            level = "urgent"
        elif record["fill_rate"] >= 0.8:
            priority = 8
            level = "high"
        else:
            priority = 5
            level = "medium"
        tasks.append({
            "task_id": f"task_{record['bin_id']}",
            "bin_id": record["bin_id"],
            "area": record["area"],
            "x": record["x"],
            "y": record["y"],
            "fill_rate": record["fill_rate"],
            "priority": priority,
            "level": level
        })
    tasks.sort(key=lambda item: item["priority"], reverse=True)
    return tasks

任务生成让系统从状态监测进入到业务执行层面。管理者不仅能看见“哪里满了”，还能获得可调度的任务清单。

五、车辆距离计算

第四步，计算车辆与垃圾桶之间的距离。

def distance(x1, y1, x2, y2):
    return math.sqrt((x1 - x2) ** 2 + (y1 - y2) ** 2)

def score_truck_for_task(truck, task):
    if truck.status != "idle":
        return -1
    remaining_capacity = truck.capacity - truck.current_load
    if remaining_capacity <= 0:
        return -1
    dist = distance(truck.x, truck.y, task["x"], task["y"])
    score = 100
    score -= dist * 2
    score += task["priority"] * 3
    score += remaining_capacity * 0.1
    return round(score, 2)

车辆调度不能只考虑距离。任务紧急程度、车辆剩余容量和车辆工作状态，都应该纳入综合评分体系。

六、任务分配

第五步，将清运任务合理分配给车辆。

def assign_tasks_to_trucks(trucks, tasks):
    assignments = []
    for task in tasks:
        candidates = []
        for truck in trucks:
            score = score_truck_for_task(truck, task)
            if score >= 0:
                candidates.append({"truck": truck, "score": score})
        if not candidates:
            assignments.append({
                "task_id": task["task_id"],
                "bin_id": task["bin_id"],
                "status": "waiting",
                "reason": "暂无可用车辆"
            })
            continue
        candidates.sort(key=lambda item: item["score"], reverse=True)
        selected = candidates[0]["truck"]
        selected.status = "busy"
        selected.current_load += int(task["fill_rate"] * 20)
        assignments.append({
            "task_id": task["task_id"],
            "bin_id": task["bin_id"],
            "truck_id": selected.truck_id,
            "score": candidates[0]["score"],
            "status": "assigned"
        })
    return assignments

通过任务分配，车辆资源得到更高效的利用。高优先级的任务会被优先处理，低风险点位则可以延后或合并清运。

七、区域环卫压力统计

第六步，按区域统计环卫压力状况。

def summarize_area_pressure(bin_records):
    area_stats = defaultdict(lambda: {"bin_count": 0, "risk_count": 0, "a vg_fill_rate": 0})
    for record in bin_records:
        area = record["area"]
        area_stats[area]["bin_count"] += 1
        area_stats[area]["a vg_fill_rate"] += record["fill_rate"]
        if record["status"] in ["overflow_risk", "need_collect"]:
            area_stats[area]["risk_count"] += 1
    results = []
    for area, stat in area_stats.items():
        a vg_fill = stat["a vg_fill_rate"] / stat["bin_count"]
        if stat["risk_count"] >= 2 or a vg_fill >= 0.75:
            pressure_level = "high"
        elif a vg_fill >= 0.55:
            pressure_level = "medium"
        else:
            pressure_level = "normal"
        results.append({
            "area": area,
            "bin_count": stat["bin_count"],
            "risk_count": stat["risk_count"],
            "a vg_fill_rate": round(a vg_fill, 2),
            "pressure_level": pressure_level
        })
    return results

区域压力统计能帮助管理者看清城市环卫的热点——不只是关注单个垃圾桶，而是从宏观上把握各个区域的整体压力水平。

八、运行完整智慧环卫流程

最后，将状态采集、任务生成、车辆调度和区域统计串联起来，形成完整的动态调度闭环。

def run_smart_sanitation_dispatch():
    bins = [
        TrashBin("B001", "人民广场", 10, 20, "business"),
        TrashBin("B002", "人民广场", 12, 25, "business"),
        TrashBin("B003", "滨河公园", 40, 60, "scenic"),
        TrashBin("B004", "幸福小区", 70, 20, "residential"),
        TrashBin("B005", "科技园", 30, 15, "office")
    ]
    trucks = [
        SanitationTruck("T001", 0, 0, 100),
        SanitationTruck("T002", 50, 50, 120),
        SanitationTruck("T003", 80, 20, 90)
    ]
    bin_records = [collect_bin_status(bin_obj) for bin_obj in bins]
    tasks = generate_collection_tasks(bin_records)
    assignments = assign_tasks_to_trucks(trucks, tasks)
    area_pressure = summarize_area_pressure(bin_records)
    report = {
        "report_name": "智慧环卫动态清运调度报告",
        "bin_records": bin_records,
        "tasks": tasks,
        "assignments": assignments,
        "area_pressure": area_pressure,
        "trucks": [truck.to_dict() for truck in trucks],
        "generate_time": datetime.now().isoformat()
    }
    return report

if __name__ == "__main__":
    report = run_smart_sanitation_dispatch()
    print(json.dumps(report, ensure_ascii=False, indent=2))

九、趋势判断

从这套流程可以清晰看出，智慧环卫正在从传统的固定作业模式向动态运营全面转变。

未来，城市环卫将不再依赖固定班次和人工巡检，而是基于满载率、区域压力、车辆实时位置和清运优先级进行智能调度。

环卫系统的价值，也将从单纯的“完成清运任务”升级为“提升城市环境质量与资源利用效率”。

谁能真正把感知数据、车辆调度和区域运营融合在一起，谁就更容易实现精细化城市治理的目标。