从‘校门外的树’到地铁规划：用Python模拟现实中的区间占用与资源统计

从‘校门外的树’到地铁规划：用Python模拟现实中的区间占用与资源统计

想象一下，你负责的城市地铁施工需要占用部分道路资源，而道路两旁原本种满了树木。施工方提供了多个施工区间，这些区间可能有重叠。你需要快速计算出施工结束后，道路上还剩下多少棵树。这听起来像是一个具体的工程问题，但它的核心逻辑与经典的"校门外的树"算法题如出一辙。本文将带你用Python从零开始构建这个问题的解决方案，并扩展到更丰富的现实场景。

1. 问题拆解与基础模型

让我们先回到"校门外的树"这个经典问题。题目描述了一条长度为L的马路上每隔1米种一棵树，然后给出多个需要移走树木的区间（可能重叠），最终计算剩余树木数量。这个抽象模型可以完美映射到许多现实场景：

• 地铁施工：施工区间对应需要移走树木的区域
• 活动日程安排：活动占用时间区间对应资源被占用的时段
• 服务器端口管理：被占用的端口区间对应不可用的资源

用Python表示初始的树木集合非常简单：

L = 500 # 马路长度 trees = set(range(L + 1)) # 包含0到L的所有整数点

这里使用set而不是list是因为集合的差集运算能更高效地处理区间重叠问题。接下来我们看看如何处理施工区间。

2. 处理区间操作的三种方法

2.1 基础循环法

最直观的方法是遍历每个区间内的所有点，逐个从集合中移除：

def remove_trees_loop(trees, intervals): for start, end in intervals: for i in range(start, end + 1): trees.discard(i) # 比remove更安全，不存在时不会报错 return len(trees)

这种方法简单直接，但当区间范围很大时（比如1到10000），效率会明显下降。时间复杂度为O(M*K)，其中M是区间数，K是平均区间长度。

2.2 区间合并优化

更高效的方法是先合并所有重叠或相邻的区间，再计算被移除的树木总数：

def merge_intervals(intervals): if not intervals: return [] sorted_intervals = sorted(intervals, key=lambda x: x[0]) merged = [sorted_intervals[0]] for current in sorted_intervals[1:]: last = merged[-1] if current[0] <= last[1] + 1: # 重叠或相邻 merged[-1] = (last[0], max(last[1], current[1])) else: merged.append(current) return merged def count_removed_trees(intervals): merged = merge_intervals(intervals) return sum(end - start + 1 for start, end in merged)

这种方法的时间复杂度主要取决于排序操作O(M log M)，后续合并和计算都是线性时间。

2.3 集合差集法

利用Python集合的差集运算特性，我们可以写出更简洁的代码：

def calculate_remaining_trees(L, intervals): trees = set(range(L + 1)) removed = set() for start, end in intervals: removed.update(range(start, end + 1)) return len(trees - removed)

这种方法在代码可读性上表现最好，虽然空间复杂度略高，但对于中等规模的数据（L≤10000）完全够用。

3. 扩展到现实场景：地铁施工资源统计

让我们把这个问题具体化到地铁施工场景。假设我们有以下施工计划：

用Python处理这个案例：

L = 500 intervals = [(150, 300), (100, 200), (470, 471)] # 使用方法2.3的集合差集法 remaining = calculate_remaining_trees(L, intervals) print(f"施工后剩余树木数量: {remaining}") # 输出: 298

我们可以进一步可视化剩余树木的分布：

def visualize_trees(L, intervals): trees = set(range(L + 1)) removed = set() for start, end in intervals: removed.update(range(start, end + 1)) remaining_trees = trees - removed visualization = ['·'] * (L + 1) for pos in remaining_trees: visualization[pos] = '🌲' # 每50米显示一行 for i in range(0, L + 1, 50): print(f"{i:3d}: {''.join(visualization[i:i+50])}") visualize_trees(500, intervals)

这将输出类似如下的可视化结果：

0: ·········································· 50: ·········································· 100: 🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲 150: ·········································· 200: 🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲 250: ·········································· 300: 🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲🌲 350: ·········································· 400: ·········································· 450: ··································🌲🌲····

4. 进阶应用：多场景问题解决

4.1 活动日程冲突检测

同样的逻辑可以用于检测活动时间冲突。假设我们有一个会议室，一天的时间被划分为480分钟（8小时工作制），每个活动占用一个时间区间：

time_slots = set(range(480)) # 8:00-16:00, 每分钟一个slot meetings = [ (30, 90), # 8:30-9:30 (60, 120), # 9:00-10:00 (300, 360) # 13:00-14:00 ] # 找出所有冲突的时间点 all_meeting_times = set() conflicts = set() for start, end in meetings: current = set(range(start, end + 1)) conflicts.update(all_meeting_times & current) all_meeting_times.update(current) print(f"总冲突分钟数: {len(conflicts)}") # 输出30分钟(9:00-9:30)

4.2 服务器端口占用检查

在服务器管理中，我们需要检查哪些端口被占用：

all_ports = set(range(1, 65536)) used_ports = {(80, 80), (443, 443), (8000, 8100), (8080, 8080)} # 展开所有被占用的端口号 used_port_numbers = set() for start, end in used_ports: used_port_numbers.update(range(start, end + 1)) available_ports = all_ports - used_port_numbers print(f"可用端口数量: {len(available_ports)}")

4.3 性能对比与选择指南

不同方法的性能特点对比：

选择建议：

1. 当L<10000且需要快速实现时，优先选择集合差集法
2. 当区间范围很大(K很大)时，使用区间合并法
3. 当需要处理动态增减区间时，可以考虑使用专门的区间树结构

5. 工程实践中的优化技巧

在实际项目中，我们还需要考虑更多细节：

内存优化：对于特别大的L值(如L>1e6)，可以使用位图(bitmap)代替集合：

import array def bitmap_solution(L, intervals): bitmap = array.array('B', [0]) * ((L + 8) // 8) # 每个bit代表一个位置 for start, end in intervals: byte_start, bit_start = divmod(start, 8) byte_end, bit_end = divmod(end, 8) if byte_start == byte_end: mask = ((1 << (bit_end - bit_start + 1)) - 1) << bit_start bitmap[byte_start] |= mask else: # 处理跨字节的情况 pass return sum(bin(byte).count('1') for byte in bitmap)

并行处理：对于超大规模数据，可以将区间分片并行处理：

from concurrent.futures import ThreadPoolExecutor def parallel_remove_trees(L, intervals, chunks=4): chunk_size = (L + 1) // chunks results = [] def process_chunk(start, end): chunk_trees = set(range(start, end + 1)) for i_start, i_end in intervals: overlap_start = max(start, i_start) overlap_end = min(end, i_end) if overlap_start <= overlap_end: for x in range(overlap_start, overlap_end + 1): chunk_trees.discard(x) return len(chunk_trees) with ThreadPoolExecutor() as executor: futures = [] for i in range(chunks): chunk_start = i * chunk_size chunk_end = (i + 1) * chunk_size - 1 if i != chunks - 1 else L futures.append(executor.submit(process_chunk, chunk_start, chunk_end)) results = [f.result() for f in futures] return sum(results)

持久化存储：对于需要频繁查询的场景，可以将处理结果存入数据库：

import sqlite3 def save_to_db(L, intervals, db_path='trees.db'): conn = sqlite3.connect(db_path) cursor = conn.cursor() cursor.execute("CREATE TABLE IF NOT EXISTS trees (position INTEGER PRIMARY KEY)") cursor.execute("CREATE TABLE IF NOT EXISTS removed_trees (position INTEGER PRIMARY KEY)") # 批量插入数据 cursor.execute("DELETE FROM trees") cursor.execute("DELETE FROM removed_trees") cursor.executemany("INSERT INTO trees VALUES (?)", [(x,) for x in range(L + 1)]) removed = set() for start, end in intervals: removed.update(range(start, end + 1)) cursor.executemany("INSERT INTO removed_trees VALUES (?)", [(x,) for x in removed]) conn.commit() cursor.execute("SELECT COUNT(*) FROM trees WHERE position NOT IN (SELECT position FROM removed_trees)") remaining = cursor.fetchone()[0] conn.close() return remaining