用Python模拟Petri网：从标识网到网系统的完整实现（附代码）

用Python模拟Petri网从标识网到网系统的完整实现附代码Petri网作为一种描述离散事件系统的数学模型在计算机科学、自动化控制等领域有着广泛应用。本文将带您用Python从零实现一个完整的Petri网模拟器涵盖标识网定义、变迁规则、状态转移等核心功能。不同于理论讲解我们将聚焦于如何将抽象概念转化为可运行的代码让您在实践中深入理解Petri网的运行机制。1. Petri网基础与Python表示Petri网由四个基本元素构成库所(Place)、变迁(Transition)、流关系(Flow)和标识(Marking)。在Python中我们可以用面向对象的方式来表示这些概念class Place: def __init__(self, name): self.name name self.tokens 0 class Transition: def __init__(self, name): self.name name self.inputs [] self.outputs [] class PetriNet: def __init__(self): self.places {} self.transitions {} self.marking_history []关键数据结构说明Place类表示库所tokens属性记录当前标记数量Transition类表示变迁inputs和outputs分别存储前集和后集PetriNet类作为容器管理整个网络的组件和状态历史2. 标识网的核心实现标识网是Petri网的基础我们需要实现标记的初始化和可视化功能def add_marking(self, place_names, tokens): for name, count in zip(place_names, tokens): if name in self.places: self.places[name].tokens count def visualize(self): print(Current Marking:) for name, place in self.places.items(): print(f{name}: {• * place.tokens if place.tokens 5 else str(place.tokens)})可视化示例输出Current Marking: p1: ••• p2: ••••• p3: p4: •提示当标记数量超过5时改用数字显示以提高可读性3. 变迁规则的代码实现变迁的触发是Petri网动态行为的核心我们需要实现两个关键方法def is_enabled(self, transition_name): transition self.transitions[transition_name] return all(self.places[p.name].tokens 1 for p in transition.inputs) def fire_transition(self, transition_name): if not self.is_enabled(transition_name): raise ValueError(fTransition {transition_name} is not enabled) transition self.transitions[transition_name] # 消耗输入库所的标记 for place in transition.inputs: self.places[place.name].tokens - 1 # 产生输出库所的标记 for place in transition.outputs: self.places[place.name].tokens 1 # 记录当前状态 self.marking_history.append(self.get_current_marking())变迁规则要点检查变迁是否具备发生权所有输入库所都有至少一个标记执行变迁时会消耗输入标记并产生输出标记每次变迁后记录当前状态以便追溯4. 完整网系统模拟案例让我们通过一个生产者-消费者模型的完整实现来演示Petri网的应用# 创建网络实例 net PetriNet() # 添加库所 net.places { producer_ready: Place(producer_ready), buffer_empty: Place(buffer_empty), consumer_ready: Place(consumer_ready), buffer_full: Place(buffer_full) } # 添加变迁 produce Transition(produce) consume Transition(consume) # 设置流关系 produce.inputs [net.places[producer_ready], net.places[buffer_empty]] produce.outputs [net.places[producer_ready], net.places[buffer_full]] consume.inputs [net.places[buffer_full], net.places[consumer_ready]] consume.outputs [net.places[buffer_empty], net.places[consumer_ready]] # 初始标识 net.add_marking([producer_ready, buffer_empty, consumer_ready], [1, 1, 1]) # 模拟运行 net.fire_transition(produce) net.visualize() net.fire_transition(consume) net.visualize()状态转移表变迁producer_readybuffer_emptyconsumer_readybuffer_full初始1110生产1011消费11105. 高级功能扩展基础实现之上我们可以添加更多实用功能来增强模拟器的能力def simulate(self, steps10): for _ in range(steps): enabled [t for t in self.transitions if self.is_enabled(t)] if not enabled: break random.choice(enabled).fire() def find_deadlocks(self): return [t.name for t in self.transitions.values() if not self.is_enabled(t.name) and all(not self.is_enabled(i.name) for i in self.transitions.values())]性能优化技巧使用位运算加速标记计数对大型网络采用稀疏矩阵存储实现增量式状态检查避免全量计算6. 测试与验证策略确保Petri网实现的正确性需要系统化的测试方法import unittest class TestPetriNet(unittest.TestCase): def setUp(self): self.net PetriNet() # 设置测试网络... def test_initial_marking(self): self.assertEqual(self.net.places[p1].tokens, 1) def test_transition_firing(self): self.assertTrue(self.net.is_enabled(t1)) self.net.fire_transition(t1) self.assertEqual(self.net.places[p2].tokens, 1)测试覆盖要点初始标识正确性变迁触发条件验证状态转移结果检查边界条件处理如空标识7. 实际应用场景分析Petri网模拟器可以应用于多种实际场景工作流建模业务流程验证系统瓶颈分析资源分配优化典型应用案例生产线调度系统网络协议验证数字电路设计生物化学反应模拟在实现一个分布式任务调度系统时我们发现Petri网特别适合描述任务之间的依赖关系。通过模拟不同负载下的状态变化能够提前发现潜在的资源竞争问题。