关于独立开发并维护一个中小型软件项目的一些总结
为 Python 设计的状态机库 Transitions 开发文档 (中文翻译)
Transitions 是一个轻量级的、面向对象的 Python 状态机实现，包含许多扩展功能。兼容 Python 2.7+ 和 3.0+。 安装 pip install transitions 快速入门 俗话说，一个好的例子抵得上 100 页 API 文档、一百万个指令或一千句话。 好吧，说这话的人可能有点夸张... 但不管怎样，这里有一个例子： from transitions import Machine import random class 嗜睡超级英雄(object): # 定义一些状态。大多数时候，嗜睡超级英雄就像普通人一样。 # 除了... states = ['睡觉中', '闲逛中', '饥饿中', '满身汗', '拯救世界'] def __init__(self, name): # 在我的地盘上不允许有无名的超级英雄！每个嗜睡超级英雄都必须有名字。 # 任何名字都行。瞌睡侠、梦游女。你懂的。 self.name = name # 今天我们救了啥？ self.kittens_rescued = 0 # 初始化状态机 self.machine = Machine(model=self, states=嗜睡超级英雄.states, initial='睡觉中') # 添加一些转换。我们也可以像上面定义状态那样，用一个字典的静态列表来定义它们， # 然后将这个列表作为 transitions= 参数传给 Machine 的初始化器。 # 某个时刻，每个超级英雄都必须起床亮相。 self.machine.add_transition(trigger='唤醒', source='睡觉中', dest='闲逛中') # 超级英雄需要保持身材。 self.machine.add_transition('锻炼', '闲逛中', '饥饿中') # 消耗的卡路里不会自己补充！ self.machine.add_transition('吃', '饥饿中', '闲逛中') # 超级英雄随时待命。永远待命。但他们并不总是穿着适合工作的服装。 self.machine.add_transition('求救信号', '*', '拯救世界', before='换上超级秘密战衣') # 他们下班时，浑身是汗，脏兮兮的。但在做其他事情之前， # 他们必须仔细记录他们最新的冒险经历。因为法律部门这么说。 self.machine.add_transition('完成任务', '拯救世界', '满身汗', after='更新日志') # 汗水是一种可以通过水解决的状况。 # 除非你度过了特别漫长的一天，那样的话... 就该睡觉了！ self.machine.add_transition('清洁', '满身汗', '睡觉中', conditions=['筋疲力尽了吗']) self.machine.add_transition('清洁', '满身汗', '闲逛中') # 我们的嗜睡超级英雄几乎可以在任何时间睡着。 self.machine.add_transition('小睡', '*', '睡觉中') def 更新日志(self): """ 亲爱的日记，今天我救了胡须先生。又一次。 """ self.kittens_rescued += 1 @property def 筋疲力尽了吗(self): """ 基本上就是抛硬币。 """ return random.random() < 0.5 def 换上超级秘密战衣(self): print("帅吧？") 好了，现在你已经把一个状态机烘焙进了 嗜睡超级英雄。让我们带他/她/它出去遛遛... >>> 蝙蝠侠 = 嗜睡超级英雄("蝙蝠侠") >>> 蝙蝠侠.state '睡觉中' >>> 蝙蝠侠.唤醒() >>> 蝙蝠侠.state '闲逛中' >>> 蝙蝠侠.小睡() >>> 蝙蝠侠.state '睡觉中' >>> 蝙蝠侠.清洁() MachineError: "无法从状态 '睡觉中' 触发事件 '清洁'！" >>> 蝙蝠侠.唤醒() >>> 蝙蝠侠.锻炼() >>> 蝙蝠侠.state '饥饿中' # 蝙蝠侠还没干任何有用的事... >>> 蝙蝠侠.kittens_rescued 0 # 我们现在为您直播一场可怕的小猫被困事件现场... >>> 蝙蝠侠.求救信号() '帅吧？' >>> 蝙蝠侠.state '拯救世界' # 回基地。 >>> 蝙蝠侠.完成任务() >>> 蝙蝠侠.state '满身汗' >>> 蝙蝠侠.清洁() >>> 蝙蝠侠.state '睡觉中' # 太累了，澡都不洗了！ # 又是高效的一天，阿尔弗雷德。 >>> 蝙蝠侠.kittens_rescued 1 虽然我们无法读懂真正蝙蝠侠的心思，但我们肯定可以可视化我们的 嗜睡超级英雄 的当前状态。 如果你想了解如何做到，可以看看 图表 扩展。 非快速入门 状态机是行为的一个模型，由有限数量的状态和这些状态之间的转换组成。在每个状态和转换中，都可以执行一些动作。状态机需要从某个初始状态开始。使用 transitions 时，状态机可能由多个对象组成，其中一些（机器）包含用于操作其他（模型）对象的定义。下面，我们将看一些核心概念以及如何使用它们。 一些关键概念 状态。状态表示状态机中一个特定的条件或阶段。它是一个行为的不同模式或过程中的一个阶段。 转换。这是导致状态机从一个状态改变到另一个状态的过程或事件。 模型。实际的有状态结构。它是在转换过程中被更新的实体。它还可以定义在转换期间执行的动作。例如，就在转换之前，或当进入或退出一个状态时。 机器。这是管理并控制模型、状态、转换和动作的实体。它是协调状态机整个过程的指挥者。 触发器。这是启动转换的事件，即发送信号开始转换的方法。 动作。在进入某个状态、退出某个状态或转换期间执行的特定操作或任务。动作通过回调实现，回调是在某些事件发生时执行的函数。 基本初始化 启动并运行一个状态机相当简单。假设你有一个对象 lump（Matter 类的一个实例），并且你想管理它的状态： class 物质(object): pass lump = 物质() 你可以像这样初始化一个绑定到模型 lump 的（最小化）工作状态机： from transitions import Machine machine = Machine(model=lump, states=['固态', '液态', '气态', '等离子态'], initial='固态') # Lump 现在有了一个新的 state 属性！ lump.state >>> '固态' 另一种方法是不显式地将模型传递给 Machine 初始化器： machine = Machine(states=['固态', '液态', '气态', '等离子态'], initial='固态') # 此时 machine 实例本身充当模型 machine.state >>> '固态' 请注意，这次我没有将 lump 模型作为参数传递。传递给 Machine 的第一个参数充当模型。所以当我在那里传递了某些东西时，所有便捷函数都将添加到该对象上。如果没有提供模型，则 machine 实例本身充当模型。 一开始我说“最小化”，是因为虽然这个状态机在技术上可以运行，但它实际上什么也不做。它从 '固态' 状态开始，但永远不会移动到另一个状态，因为还没有定义任何转换... 目前！ 让我们再试一次。 # 状态 states=['固态', '液态', '气态', '等离子态'] # 以及状态之间的一些转换。我们很懒，所以会省略 # 逆相变（凝固、凝结等）。 transitions = [ { 'trigger': '融化', 'source': '固态', 'dest': '液态' }, { 'trigger': '蒸发', 'source': '液态', 'dest': '气态' }, { 'trigger': '升华', 'source': '固态', 'dest': '气态' }, { 'trigger': '电离', 'source': '气态', 'dest': '等离子态' } ] # 初始化 machine = Machine(lump, states=states, transitions=transitions, initial='液态') # 现在 lump 维护状态... lump.state >>> '液态' # 并且状态可以改变... # 可以调用闪亮的新触发器方法 lump.蒸发() lump.state >>> '气态' # 或者直接调用 trigger 方法 lump.trigger('电离') lump.state >>> '等离子态' 注意那些附加到 物质 实例的闪亮的新方法（蒸发()、电离() 等）。 每个方法触发相应的转换。 转换也可以通过调用提供的 trigger() 方法并传入转换名称来动态触发，如上所示。 更多内容见 触发转换 部分。 状态 任何好的状态机（以及许多不好的状态机，毫无疑问）的灵魂都是一组状态。上面，我们通过传递一个字符串列表给 Machine 初始化器来定义有效的模型状态。但在内部，状态实际上表示为 State 对象。 你可以通过多种方式初始化和修改 State。具体来说，你可以： 传递一个字符串给 Machine 初始化器，给出状态的名称，或者 直接初始化每个新的 State 对象，或者 传递一个带有初始化参数的字典 以下代码片段说明了实现相同目标的几种方法： # 导入 Machine 和 State 类 from transitions import Machine, State # 创建包含 3 个状态的列表，传递给 Machine 初始化器。 # 我们可以混合类型；在这个例子中， # 我们传递一个 State，一个字符串，和一个 dict。 states = [ State(name='固态'), '液态', { 'name': '气态'} ] machine = Machine(lump, states) # 这个替代例子展示了更显式的 # 状态和状态回调的添加方式，但最终 # 结果与上面相同。 machine = Machine(lump) solid = State('固态') liquid = State('液态') gas = State('气态') machine.add_states([solid, liquid, gas]) 状态在添加到机器时被初始化一次，并将持续存在直到从机器中移除。换句话说：如果你更改状态对象的属性，这种更改在下次进入该状态时将不会重置。如果你需要其他行为，请查看如何 扩展状态特性。 回调 但仅仅有状态并且能够在它们之间移动（转换）本身并不是很有用。如果你想在进入或退出一个状态时做点事情，执行一些动作呢？这就是回调的用武之地。 一个 State 还可以与 enter 和 exit 回调列表相关联，每当状态机进入或离开该状态时，这些回调就会被调用。你可以在初始化期间通过将它们传递给 State 对象构造函数、在状态属性字典中传递它们来指定回调，或者稍后添加它们。 为方便起见，每当一个新的 State 被添加到 Machine 时，方法 on_enter_«state name» 和 on_exit_«state name» 会在 Machine 上（而不是模型上！）动态创建，允许你稍后根据需要动态添加新的进入和退出回调。 # 我们旧的物质类，现在有了几个新方法， # 我们可以在进入或退出状态时触发它们。 class 物质(object): def 问好(self): print("你好，新状态！") def 道别(self): print("再见，旧状态！") lump = 物质() # 状态与上面相同，但现在我们给 StateA 一个退出回调 states = [ State(name='固态', on_exit=['道别']), '液态', { 'name': '气态', 'on_exit': ['道别']} ] machine = Machine(lump, states=states) machine.add_transition('升华', '固态', '气态') # 回调也可以在初始化后使用动态添加的 # on_enter_ 和 on_exit_ 方法添加。 # 注意，添加回调的初始调用是在 Machine 上进行的， # 而不是在模型上。 machine.on_enter_气态('问好') # 测试回调... machine.set_state('固态') lump.升华() >>> '再见，旧状态！' >>> '你好，新状态！' 请注意，当 Machine 首次初始化时，on_enter_«state name» 回调将不会触发。例如，如果你定义了一个 on_enter_A() 回调，并且用 initial='A' 初始化 Machine，那么 on_enter_A() 在下次进入状态 A 之前不会触发。（如果你需要确保 on_enter_A() 在初始化时触发，你可以简单地创建一个虚拟的初始状态，然后在 __init__ 方法中显式调用 to_A()。） 除了在初始化 State 时传入回调或动态添加它们外，还可以在模型类本身中定义回调，这可能会提高代码清晰度。例如： class 物质(object): def 问好(self): print("你好，新状态！") def 道别(self): print("再见，旧状态！") def on_enter_A(self): print("我们刚进入了状态 A！") lump = 物质() machine = Machine(lump, states=['A', 'B', 'C']) 现在，任何时候 lump 转换到状态 A，在 物质 类中定义的 on_enter_A() 方法就会触发。 你可以使用 on_final 回调，当进入一个 final=True 的状态时，它们将被触发。 from transitions import Machine, State states = [State(name='待命中'), State(name='救援小猫中'), State(name='坏人已逃', final=True), State(name='坏人被捕', final=True)] transitions = [["接到呼叫", "待命中", "救援小猫中"], # 一叫我们就来 {"trigger": "干预", "source": "救援小猫中", "dest": "坏人已逃", # 我们 "conditions": "坏人更快"}, # 除非他们更快 ["干预", "救援小猫中", "坏人被捕"]] class 最终版超级英雄(object): def __init__(self, speed): self.machine = Machine(self, states=states, transitions=transitions, initial="待命中", on_final="宣告成功") self.speed = speed def 坏人更快(self, offender_speed): return self.speed < offender_speed def 宣告成功(self, **kwargs): print("小猫安全了。") hero = 最终版超级英雄(speed=10) # 我们今天状态不好 hero.接到呼叫() assert hero.is_救援小猫中() hero.干预(offender_speed=15) # >>> '小猫安全了' assert hero.machine.get_state(hero.state).final # 结束了 assert hero.is_坏人已逃() # 也许下次... 检查状态 你可以随时通过以下任一方式检查模型的当前状态： 检查 .state 属性，或者 调用 is_«state name»() 如果你想要获取当前状态的实际 State 对象，可以通过 Machine 实例的 get_state() 方法来完成。 lump.state >>> '固态' lump.is_气态() >>> False lump.is_固态() >>> True machine.get_state(lump.state).name >>> '固态' 如果你愿意，可以在初始化 Machine 时通过传递 model_attribute 参数来选择自己的状态属性名称。这也会将 is_«state name»() 的名称改为 is_«model_attribute»_«state name»()。同样，自动转换将被命名为 to_«model_attribute»_«state name»() 而不是 to_«state name»()。这样做是为了允许多个机器在同一个模型上工作，每个都有独立的状态属性名称。 lump = 物质() machine = Machine(lump, states=['固态', '液态', '气态'], model_attribute='物质状态', initial='固态') lump.物质状态 >>> '固态' # 使用自定义的 'model_attribute'，也可以这样检查状态： lump.is_物质状态_固态() >>> True lump.to_物质状态_气态() >>> True 枚举 到目前为止，我们已经看到了如何给出状态名称并使用这些名称来处理我们的状态机。 如果你偏爱更严格的类型和更多的 IDE 代码补全（或者你就是打不出 'sesquipedalophobia' 这个词了，因为这个词吓到你了），使用枚举可能正是你想要的： import enum # Python 2.7 用户需要安装 'enum34' from transitions import Machine class 状态(enum.Enum): 错误 = 0 红 = 1 黄 = 2 绿 = 3 transitions = [['前进', 状态.红, 状态.黄], ['前进', 状态.黄, 状态.绿], ['出错', '*', 状态.错误]] m = Machine(states=状态, transitions=transitions, initial=状态.红) assert m.is_红() assert m.state is 状态.红 state = m.get_state(状态.红) # 获取 transitions.State 对象 print(state.name) # >>> 红 m.前进() m.前进() assert m.is_绿() m.出错() assert m.state is 状态.错误 如果你愿意，可以混合使用枚举和字符串（例如 [状态.红, '橙色', 状态.黄, 状态.绿]），但请注意，在内部，transitions 仍然会按名称（enum.Enum.name）处理状态。 因此，不能同时拥有状态 '绿' 和 状态.绿。 转换 上面的一些例子已经顺便说明了转换的用法，但在这里我们将更详细地探讨它们。 与状态一样，每个转换在内部都表示为其自身的对象——Transition 类的一个实例。初始化一组转换的最快方法是向 Machine 初始化器传递一个字典或字典列表。我们上面已经看到了： transitions = [ { 'trigger': '融化', 'source': '固态', 'dest': '液态' }, { 'trigger': '蒸发', 'source': '液态', 'dest': '气态' }, { 'trigger': '升华', 'source': '固态', 'dest': '气态' }, { 'trigger': '电离', 'source': '气态', 'dest': '等离子态' } ] machine = Machine(model=物质(), states=states, transitions=transitions) 用字典定义转换的好处是清晰，但可能很繁琐。如果你追求简洁，你可以选择使用列表来定义转换。只需确保每个列表中的元素顺序与 Transition 初始化中的位置参数顺序相同（即，trigger、source、destination 等）。 下面的列表的列表在功能上等同于上面的字典列表： transitions = [ ['融化', '固态', '液态'], ['蒸发', '液态', '气态'], ['升华', '固态', '气态'], ['电离', '气态', '等离子态'] ] 或者，你也可以在初始化后向 Machine 添加转换： machine = Machine(model=lump, states=states, initial='固态') machine.add_transition('融化', source='固态', dest='液态') 触发转换 要执行一个转换，需要某个事件来触发它。有两种方法可以做到这一点： 使用自动附加到基本模型的方法： >>> lump.融化() >>> lump.state '液态' >>> lump.蒸发() >>> lump.state '气态' 注意，你不需要在任何地方显式定义这些方法；每个转换的名称都绑定到传递给 Machine 初始化器的模型（在这个例子中是 lump）。这也意味着你的模型不应该已经包含与事件触发器同名的方法，因为 transitions 只会在该位置未被占用时才会将便捷方法附加到你的模型上。如果你想修改这种行为，请查看 FAQ。 使用现在已附加到你的模型上的 trigger 方法（如果之前没有的话）。这个方法让你可以按名称执行转换，以防需要动态触发： >>> lump.trigger('融化') >>> lump.state '液态' >>> lump.trigger('蒸发') >>> lump.state '气态' 触发无效转换 默认情况下，触发无效转换将引发异常： >>> lump.to_气态() >>> # 这行不通，因为只有处于固态的对象才能融化 >>> lump.融化() transitions.core.MachineError: "无法从状态 '气态' 触发事件 '融化'！" 这种行为通常是可取的，因为它有助于提醒你代码中的问题。但在某些情况下，你可能希望静默忽略无效触发器。你可以通过设置 ignore_invalid_triggers=True 来实现这一点（可以在逐个状态的基础上设置，也可以全局为所有状态设置）： >>> # 全局抑制无效触发器异常 >>> m = Machine(lump, states, initial='固态', ignore_invalid_triggers=True) >>> # ...或者只为某一组状态抑制 >>> states = ['新状态1', '新状态2'] >>> m.add_states(states, ignore_invalid_triggers=True) >>> # ...甚至只为一个状态。在这里，只有当当前状态是 A 时才会抑制异常。 >>> states = [State('A', ignore_invalid_triggers=True), 'B', 'C'] >>> m = Machine(lump, states) >>> # ...如果只有一个状态应该引发异常，也可以反过来 >>> # 因为机器的全局值不会应用于先前初始化的状态。 >>> states = ['A', 'B', State('C')] # 'ignore_invalid_triggers' 的默认值是 False >>> m = Machine(lump, states, ignore_invalid_triggers=True) 如果你需要知道从某个状态哪些转换是有效的，你可以使用 get_triggers： m.get_triggers('固态') >>> ['融化', '升华'] m.get_triggers('液态') >>> ['蒸发'] m.get_triggers('等离子态') >>> [] # 你也可以一次查询多个状态 m.get_triggers('固态', '液态', '气态', '等离子态') >>> ['融化', '蒸发', '升华', '电离'] 如果你从一开始就关注此文档，你会注意到 get_triggers 实际上返回的触发器比上面显示的显式定义的触发器更多，例如 to_液态 等等。 这些被称为 自动转换，将在下一节介绍。 所有状态的自动转换 除了显式添加的任何转换外，每当一个状态被添加到 Machine 实例时，都会自动创建一个 to_«state»() 方法。这个方法会转换到目标状态，无论机器当前处于哪个状态： lump.to_液态() lump.state >>> '液态' lump.to_固态() lump.state >>> '固态' 如果你希望，可以通过在 Machine 初始化器中设置 auto_transitions=False 来禁用此行为。 从多个状态转换 一个给定的触发器可以附加到多个转换，其中一些可能开始或结束于相同的状态。例如： machine.add_transition('嬗变', ['固态', '液态', '气态'], '等离子态') machine.add_transition('嬗变', '等离子态', '固态') # 下一个转换永远不会执行 machine.add_transition('嬗变', '等离子态', '气态') 在这种情况下，调用 嬗变() 如果模型当前处于 '等离子态'，则将模型的状态设置为 '固态'，否则将其设置为 '等离子态'。（注意，只有第一个匹配的转换会执行；因此，上面最后一行定义的转换不会做任何事情。） 你也可以让一个触发器从所有状态转换到特定目的地，使用 '*' 通配符： machine.add_transition('to_液态', '*', '液态') 请注意，通配符转换只适用于在调用 add_transition() 时存在的状态。当模型处于添加转换之后才定义的状态时，调用基于通配符的转换将引发无效转换消息，并且不会转换到目标状态。 来自多个状态的自反转换 自反触发器（源状态和目标状态相同的触发器）可以很容易地通过指定 = 作为目的地来添加。 如果同一个自反触发器需要添加到多个状态，这很方便。 例如： machine.add_transition('触摸', ['液态', '气态', '等离子态'], '=', after='改变形状') 这将为所有三个状态添加自反转换，以 触摸() 作为触发器，并在每次触发后执行 改变形状。 内部转换 与自反转换相反，内部转换实际上永远不会离开状态。 这意味着诸如 before 或 after 等与转换相关的回调将被处理，而状态相关的回调 exit 或 enter 则不会。 要将转换定义为内部转换，将目标设置为 None。 machine.add_transition('内部', ['液态', '气态'], None, after='改变形状') 有序转换 一个常见的需求是状态转换遵循严格的线性顺序。例如，给定状态 ['A', 'B', 'C']，你可能只想要 A → B、B → C 和 C → A 的有效转换（但没有其他配对）。 为了便于实现这种行为，Transitions 在 Machine 类中提供了 add_ordered_transitions() 方法： states = ['A', 'B', 'C'] # 查看“替代初始化”部分，了解 init 第一个参数的说明 machine = Machine(states=states, initial='A') machine.add_ordered_transitions() machine.next_state() print(machine.state) >>> 'B' # 我们也可以定义不同的转换顺序 machine = Machine(states=states, initial='A') machine.add_ordered_transitions(['A', 'C', 'B']) machine.next_state() print(machine.state) >>> 'C' # 条件也可以传递给 'add_ordered_transitions' # 如果传递了一个条件，它将用于所有转换 machine = Machine(states=states, initial='A') machine.add_ordered_transitions(conditions='检查') # 如果传递了一个列表，它必须包含与机器包含的状态数完全相同的元素 # (A->B, ..., X->A) machine = Machine(states=states, initial='A') machine.add_ordered_transitions(conditions=['检查_A2B', ..., '检查_X2A']) # 条件总是从初始状态开始应用 machine = Machine(states=states, initial='B') machine.add_ordered_transitions(conditions=['检查_B2C', ..., '检查_A2B']) # 使用 `loop=False`，从最后一个状态到第一个状态的转换将被省略（例如 C->A） # 当你也传递条件时，你需要少传递一个条件（len(states)-1） machine = Machine(states=states, initial='A') machine.add_ordered_transitions(loop=False) machine.next_state() machine.next_state() machine.next_state() # transitions.core.MachineError: "无法从状态 C 触发事件 next_state！" 队列转换 Transitions 中的默认行为是即时处理事件。这意味着在 on_enter 方法内的事件将在绑定到 after 的回调被调用之前就被处理。 def 去C(): global machine machine.to_C() def after_advance(): print("我现在在状态 B 了！") def entering_C(): print("我现在在状态 C 了！") states = ['A', 'B', 'C'] machine = Machine(states=states, initial='A') # 我们想在状态转换到 B 完成时得到一条消息 machine.add_transition('前进', 'A', 'B', after=after_advance) # 调用从状态 B 到状态 C 的转换 machine.on_enter_B(去C) # 我们也想在进入状态 C 时得到一条消息 machine.on_enter_C(entering_C) machine.前进() >>> '我现在在状态 C 了！' >>> '我现在在状态 B 了！' # 啥？ 这个例子的执行顺序是 准备 -> before -> on_enter_B -> on_enter_C -> after。 如果启用了队列处理，一个转换将在下一个转换被触发之前完成： machine = Machine(states=states, queued=True, initial='A') ... machine.前进() >>> '我现在在状态 B 了！' >>> '我现在在状态 C 了！' # 这样好多了！ 这导致 准备 -> before -> on_enter_B -> queue(to_C) -> after -> on_enter_C。 重要提示： 当在队列中处理事件时，触发器调用将始终返回 True，因为在排队时无法确定涉及排队调用的转换最终是否会成功完成。即使只处理单个事件也是如此。 machine.add_transition('跳跃', 'A', 'C', conditions='将会失败') ... # queued=False machine.跳跃() >>> False # queued=True machine.跳跃() >>> True 当模型从机器中移除时，transitions 也会从队列中移除所有相关事件。 class Model: def on_enter_B(self): self.to_C() # 添加事件到队列 ... self.machine.remove_model(self) # 然后... 没了 条件转换 有时你只想在特定条件出现时才执行某个转换。你可以通过传递一个方法或方法列表到 conditions 参数来实现： # 我们的物质类，现在有一堆返回布尔值的方法。 class 物质(object): def 是可燃的(self): return False def 真的很热(self): return True machine.add_transition('加热', '固态', '气态', conditions='是可燃的') machine.add_transition('加热', '固态', '液态', conditions=['真的很热']) 在上面的例子中，当模型处于状态 '固态' 时调用 加热()，如果 是可燃的 返回 True，将转换到状态 '气态'。否则，如果 真的很热 返回 True，它将转换到状态 '液态'。 为方便起见，还有一个 'unless' 参数，其行为与 conditions 完全相同，但取反： machine.add_transition('加热', '固态', '气态', unless=['是可燃的', '真的很热']) 在这种情况下，只要 加热() 触发，模型就会从固态转换到气态，前提是 是可燃的() 和 真的很热() 都返回 False。 请注意，条件检查方法将被动接收传递给触发方法的可选参数和/或数据对象。例如，以下调用： lump.加热(temp=74) # 等价于 lump.trigger('加热', temp=74) ... 将把 temp=74 可选关键字参数传递给 是可燃的() 检查（可能包装在 EventData 实例中）。更多关于此的内容，请参见下面的 传递数据 部分。 检查转换 如果你想在继续之前确保转换是可能的，你可以使用已添加到你的模型中的 may_<trigger_name> 函数。 你的模型还包含 may_trigger 函数来按名称检查触发器： # 检查当前温度是否足够热以触发转换 if lump.may_加热(): # 如果 lump.may_trigger("加热"): lump.加热() 这将执行所有 准备 回调并评估分配给潜在转换的条件。 当转换的目标不可用（尚未）时，也可以使用转换检查： machine.add_transition('升华', '固态', '精神层面') assert not lump.may_升华() # 还没准备好 :( assert not lump.may_trigger("升华") # 通过触发器名称检查的结果相同 回调 你不仅可以将回调附加到状态，还可以附加到转换。每个转换都有 'before' 和 'after' 属性，其中包含在转换执行之前和之后要调用的方法列表： class 物质(object): def 发出嘶嘶声(self): print("嘶嘶嘶嘶嘶嘶嘶嘶嘶嘶") def 消失(self): print("所有液体都去哪儿了？") transitions = [ { 'trigger': '融化', 'source': '固态', 'dest': '液态', 'before': '发出嘶嘶声'}, { 'trigger': '蒸发', 'source': '液态', 'dest': '气态', 'after': '消失' } ] lump = 物质() machine = Machine(lump, states, transitions=transitions, initial='固态') lump.融化() >>> "嘶嘶嘶嘶嘶嘶嘶嘶嘶嘶" lump.蒸发() >>> "所有液体都去哪儿了？" 还有一个 'prepare' 回调，它会在转换一开始就执行，在任何 'conditions' 检查或其他回调执行之前。 class 物质(object): heat = False attempts = 0 def 计数尝试(self): self.attempts += 1 def 加热(self): self.heat = random.random() < 0.25 def 统计(self): print('你花了 %i 次尝试才融化这块东西！' %self.attempts) @property def 真的很热(self): return self.heat states=['固态', '液态', '气态', '等离子态'] transitions = [ { 'trigger': '融化', 'source': '固态', 'dest': '液态', 'prepare': ['加热', '计数尝试'], 'conditions': '真的很热', 'after': '统计'}, ] lump = 物质() machine = Machine(lump, states, transitions=transitions, initial='固态') lump.融化() lump.融化() lump.融化() lump.融化() >>> "你花了 4 次尝试才融化这块东西！" 请注意，除非当前状态是命名转换的有效源，否则 prepare 不会被调用。 意为在每次转换之前或之后执行的默认动作可以分别通过 before_state_change 和 after_state_change 在初始化期间传递给 Machine： class 物质(object): def 发出嘶嘶声(self): print("嘶嘶嘶嘶嘶嘶嘶嘶嘶嘶") def 消失(self): print("所有液体都去哪儿了？") states=['固态', '液态', '气态', '等离子态'] lump = 物质() m = Machine(lump, states, before_state_change='发出嘶嘶声', after_state_change='消失') lump.to_气态() >>> "嘶嘶嘶嘶嘶嘶嘶嘶嘶嘶" >>> "所有液体都去哪儿了？" 还有两个关键字用于回调，这些回调应该独立于 a) 有多少转换是可能的，b) 任何转换是否成功以及 c) 即使在其他回调执行期间引发错误，也应该执行。 通过 prepare_event 传递给 Machine 的回调将在处理可能的转换（及其各自的 prepare 回调）之前执行一次。 finalize_event 的回调将无论已处理转换的成功与否都会执行。 请注意，如果发生错误，它将被附加到 event_data 作为 error，并且可以通过 send_event=True 检索。 from transitions import Machine class 物质(object): def 引发错误(self, event): raise ValueError("哦不") def 准备(self, event): print("我准备好了！") def 最终处理(self, event): print("结果：", type(event.error), event.error) states=['固态', '液态', '气态', '等离子态'] lump = 物质() m = Machine(lump, states, prepare_event='准备', before_state_change='引发错误', finalize_event='最终处理', send_event=True) try: lump.to_气态() except ValueError: pass print(lump.state) # >>> 我准备好了！ # >>> 结果： <class 'ValueError'> 哦不 # >>> 初始状态 有时事情就是不如预期，我们需要处理异常并清理混乱以保持事情进行。 我们可以将回调传递给 on_exception 来做到这一点： from transitions import Machine class 物质(object): def 引发错误(self, event): raise ValueError("哦不") def 处理错误(self, event): print("修复中...") del event.error # 如果我们看不到它，那它就没发生... states=['固态', '液态', '气态', '等离子态'] lump = 物质() m = Machine(lump, states, before_state_change='引发错误', on_exception='处理错误', send_event=True) try: lump.to_气态() except ValueError: pass print(lump.state) # >>> 修复中... # >>> 初始状态 可调用对象解析 你可能已经意识到，将可调用对象传递给状态、条件和转换的标准方式是通过名称。当处理回调和条件时，transitions 将使用它们的名称从模型中检索相关的可调用对象。如果无法检索到该方法并且它包含点，transitions 将把该名称视为模块函数的路径并尝试导入它。或者，你可以传递属性或特性的名称。它们将被包装成函数，但显然不能接收事件数据。你也可以直接传递可调用对象，例如（绑定的）函数。如前所述，你也可以将可调用对象名称的列表/元组传递给回调参数。回调将按照它们被添加的顺序执行。 from transitions import Machine from mod import 导入的函数 import random class 模型(object): def 一个回调(self): 导入的函数() @property def 一个特性(self): """ 基本上是抛硬币。 """ return random.random() < 0.5 一个属性 = False model = 模型() machine = Machine(model=model, states=['A'], initial='A') machine.add_transition('按名称', 'A', 'A', conditions='一个特性', after='一个回调') machine.add_transition('按引用', 'A', 'A', unless=['一个特性', '一个属性'], after=model.一个回调) machine.add_transition('导入的', 'A', 'A', after='mod.导入的函数') model.按名称() model.按引用() model.导入的() 可调用对象的解析在 Machine.resolve_callable 中完成。 如果需要更复杂的可调用对象解析策略，可以重写此方法。 示例 class 自定义机器(Machine): @staticmethod def resolve_callable(func, event_data): # 在这里操作参数，并返回 func，或者如果不进行操作，则返回 super()。 super(自定义机器, 自定义机器).resolve_callable(func, event_data) 回调执行顺序 总结一下，目前有三种触发事件的方式。你可以调用模型的便捷函数，如 lump.融化()， 通过名称执行触发器，如 lump.trigger("融化")，或者使用 machine.dispatch("融化") 在多个模型上分派事件 （请参阅 替代初始化模式 中关于多个模型的部分）。 转换上的回调将按以下顺序执行： 回调 当前状态 说明 'machine.prepare_event' 源 在开始处理各个转换之前执行一次 'transition.prepare' 源 转换一开始就执行 'transition.conditions' 源 条件可能失败并停止转换 'transition.unless' 源 条件可能失败并停止转换 'machine.before_state_change' 源 在模型上声明的默认回调 'transition.before' 源 'state.on_exit' 源 在源状态上声明的回调 <状态改变> 'state.on_enter' 目标 在目标状态上声明的回调 'transition.after' 目标 'machine.on_final' 目标 子级的回调将首先被调用 'machine.after_state_change' 目标 在模型上声明的默认回调；内部转换之后也会调用 'machine.on_exception' 源/目标 当引发异常时将执行回调 'machine.finalize_event' 源/目标 即使没有发生转换或引发了异常，也会执行回调 如果任何回调引发异常，回调的处理将不会继续。这意味着当转换之前发生错误（在 state.on_exit 或更早），转换会被停止。如果在转换已经进行之后发生错误（在 state.on_enter 或更晚），状态更改会持久化，不会发生回滚。除非异常是由最终化回调本身引发的，否则在 machine.finalize_event 中指定的回调将始终执行。请注意，每个回调序列必须在下一个阶段执行之前完成。阻塞回调将停止执行顺序，从而阻塞 trigger 或 dispatch 调用本身。如果你希望回调并行执行，可以查看 扩展 中的 AsyncMachine 用于异步处理或 LockedMachine 用于线程。 传递数据 有时你需要将一些数据传递给在机器初始化时注册的回调函数，这些数据反映了模型的当前状态。 Transitions 允许你以两种不同的方式做到这一点。 首先（默认），你可以将任何位置参数或关键字参数直接传递给触发器方法（在你调用 add_transition() 时创建的方法）： class 物质(object): def __init__(self): self.设置环境() def 设置环境(self, temp=0, pressure=101.325): self.temp = temp self.pressure = pressure def 打印温度(self): print("当前温度是 %d 摄氏度。" % self.temp) def 打印压力(self): print("当前压力是 %.2f 千帕。" % self.pressure) lump = 物质() machine = Machine(lump, ['固态', '液态'], initial='固态') machine.add_transition('融化', '固态', '液态', before='设置环境') lump.融化(45) # 位置参数； # 等价于 lump.trigger('融化', 45) lump.打印温度() >>> '当前温度是 45 摄氏度。' machine.set_state('固态') # 重置状态以便我们能再次融化 lump.融化(pressure=300.23) # 关键字参数也有效 lump.打印压力() >>> '当前压力是 300.23 千帕。' 你可以向触发器传递任意数量的参数。 这种方法有一个重要的限制：由状态转换触发的每个回调函数都必须能够处理所有参数。如果每个回调期望的数据有些不同，这可能会引起问题。 为了解决这个问题，Transitions 支持一种替代的发送数据方法。如果你在 Machine 初始化时设置 send_event=True，所有触发器的参数都将包装在一个 EventData 实例中并传递给每个回调。（EventData 对象还在内部维护对事件的源状态、模型、转换、机器和触发器的引用，以防你需要访问这些内容做任何事情。） class 物质(object): def __init__(self): self.temp = 0 self.pressure = 101.325 # 注意，唯一的参数现在是 EventData 实例。 # 这个对象存储传递给触发器方法的位置参数在 .args 属性中， # 存储关键字参数在 .kwargs 字典中。 def 设置环境(self, event): self.temp = event.kwargs.get('temp', 0) self.pressure = event.kwargs.get('pressure', 101.325) def 打印压力(self): print("当前压力是 %.2f 千帕。" % self.pressure) lump = 物质() machine = Machine(lump, ['固态', '液态'], send_event=True, initial='固态') machine.add_transition('融化', '固态', '液态', before='设置环境') lump.融化(temp=45, pressure=1853.68) # 关键字参数 lump.打印压力() >>> '当前压力是 1853.68 千帕。' 替代初始化模式 到目前为止的所有例子中，我们都将一个新的 Machine 实例附加到一个单独的模型（lump，物质 类的一个实例）。虽然这种分离保持了整洁（因为你不需要将一大堆新方法打补丁到 物质 类中），但它也可能变得烦人，因为它要求你跟踪哪些方法是在状态机上调用的，哪些是在状态机绑定的模型上调用的（例如，lump.on_enter_StateA() 与 machine.add_transition()）。 幸运的是，Transitions 很灵活，支持另外两种初始化模式。 首先，你可以创建一个独立的状态机，完全不需要另一个模型。只需在初始化时省略模型参数： machine = Machine(states=states, transitions=transitions, initial='固态') machine.融化() machine.state >>> '液态' 如果你以这种方式初始化机器，那么你可以将所有触发事件（如 蒸发()、升华() 等）和所有回调函数直接附加到 Machine 实例。 这种方法的好处是将所有状态机功能集中在一个地方，但如果你认为状态逻辑应该包含在模型本身内，而不是在一个单独的控制器中，可能会感觉有点不自然。 另一种（可能更好的）方法是让模型继承 Machine 类。Transitions 设计为无缝支持继承。（只需确保重写类 Machine 的 __init__ 方法！）： class 物质(Machine): def 问好(self): print("你好，新状态！") def 道别(self): print("再见，旧状态！") def __init__(self): states = ['固态', '液态', '气态'] Machine.__init__(self, states=states, initial='固态') self.add_transition('融化', '固态', '液态') lump = 物质() lump.state >>> '固态' lump.融化() lump.state >>> '液态' 在这里，你可以将所有状态机功能整合到你现有的模型中，这通常比把所有我们想要的功能放在一个单独的独立 Machine 实例中感觉更自然。 一台机器可以处理多个模型，这些模型可以作为列表传递，如 Machine(model=[model1, model2, ...])。 如果你想添加模型以及机器实例本身，你可以在初始化时传递类变量占位符（字符串） Machine.self_literal，如 Machine(model=[Machine.self_literal, model1, ...])。 你也可以创建一个独立的机器，并通过 machine.add_model 动态注册模型，方法是将 model=None 传递给构造函数。 此外，你可以使用 machine.dispatch 在所有当前添加的模型上触发事件。 如果机器是长期存在的，而你的模型是临时的并且应该被垃圾回收，请记住调用 machine.remove_model： class 物质(): pass lump1 = 物质() lump2 = 物质() # 将 'model' 设置为 None 或传递一个空列表将初始化一个没有模型的机器 machine = Machine(model=None, states=states, transitions=transitions, initial='固态') machine.add_model(lump1) machine.add_model(lump2, initial='液态') lump1.state >>> '固态' lump2.state >>> '液态' # 自定义事件以及自动转换都可以被分派到所有模型 machine.dispatch("to_等离子态") lump1.state >>> '等离子态' assert lump1.state == lump2.state machine.remove_model([lump1, lump2]) del lump1 # lump1 被垃圾回收 del lump2 # lump2 被垃圾回收 如果你不在状态机构造函数中提供初始状态，transitions 将创建并添加一个名为 '初始' 的默认状态。 如果你不想要默认的初始状态，你可以传递 initial=None。 但是，在这种情况下，每次添加模型时都需要传递一个初始状态。 machine = Machine(model=None, states=states, transitions=transitions, initial=None) machine.add_model(物质()) >>> "MachineError: 机器未配置初始状态，必须在添加模型时指定。" machine.add_model(物质(), initial='液态') 具有多个状态的模型可以使用不同的 model_attribute 值附加多个机器。如 检查状态 中所述，这将添加自定义的 is/to_<model_attribute>_<state_name> 函数： lump = 物质() 物质机器 = Machine(lump, states=['固态', '液态', '气态'], initial='固态') # 向同一个模型添加第二个机器，但分配不同的状态属性 运输机器 = Machine(lump, states=['已交付', '运输中'], initial='已交付', model_attribute='运输状态') lump.state >>> '固态' lump.is_固态() # 检查默认字段 >>> True lump.运输状态 >>> '已交付' lump.is_运输状态_已交付() # 检查自定义字段。 >>> True lump.to_运输状态_运输中() >>> True lump.is_运输状态_已交付() >>> False 日志记录 Transitions 包含非常基本的日志记录功能。一些事件——即状态更改、转换触发和条件检查——使用标准的 Python logging 模块作为 INFO 级别事件记录。这意味着你可以在脚本中轻松配置日志记录到标准输出： # 设置日志记录；基本日志级别将是 DEBUG import logging logging.basicConfig(level=logging.DEBUG) # 将 transitions 的日志级别设置为 INFO；将省略 DEBUG 消息 logging.getLogger('transitions').setLevel(logging.INFO) # 照常进行 machine = Machine(states=states, transitions=transitions, initial='固态') ... （重新）存储机器实例 机器是可 pickle 的，可以使用 pickle 存储和加载。对于 Python 3.3 及更早版本，需要 dill。 import dill as pickle # 仅适用于 Python 3.3 及更早版本 m = Machine(states=['A', 'B', 'C'], initial='A') m.to_B() m.state >>> B # 存储机器 dump = pickle.dumps(m) # 再次加载 Machine 实例 m2 = pickle.loads(dump) m2.state >>> B m2.states.keys() >>> ['A', 'B', 'C'] 类型支持 你可能已经注意到，transitions 使用了一些 Python 的动态特性，为你提供了处理模型的便捷方法。然而，静态类型检查器不喜欢在运行时之前未知的模型属性和方法。历史上，transitions 也不会分配模型上已经定义的便捷方法，以防止意外覆盖。 但别担心！你可以使用机器构造参数 model_override 来改变模型的装饰方式。如果你设置 model_override=True，transitions 将只覆盖已经定义的方法。这可以防止新方法在运行时出现，并且也允许你定义你想使用哪些辅助方法。 from transitions import Machine # 动态分配 class 模型: pass model = 模型() default_machine = Machine(model, states=["A", "B"], transitions="go", "A", "B", initial="A") print(model.__dict__.keys()) # 所有便捷函数都已被分配 # >> dict_keys(['trigger', 'to_A', 'may_to_A', 'to_B', 'may_to_B', 'go', 'may_go', 'is_A', 'is_B', 'state']) assert model.is_A() # 类 '模型' 的未解析属性引用 'is_A' # 预定义分配：我们只对调用我们的 'go' 事件感兴趣，其他事件将通过名称触发 class 预定义模型: # state（或者如果你设置了 'model_attribute' 则是另一个参数）无论如何都会被分配 # 因为我们需要跟踪模型的状态 state: str def go(self) -> bool: raise RuntimeError("应该被覆盖！") def trigger(self, trigger_name: str) -> bool: raise RuntimeError("应该被覆盖！") model = 预定义模型() override_machine = Machine(model, states=["A", "B"], transitions="go", "A", "B", initial="A", model_override=True) print(model.__dict__.keys()) # >> dict_keys(['trigger', 'go', 'state']) model.trigger("to_B") assert model.state == "B" 如果你想使用所有便捷函数并加入一些回调，当你有很多状态和转换时，定义一个模型会变得相当复杂。 transitions 中的方法 generate_base_model 可以从机器配置生成一个基础模型来帮助你。 from transitions.experimental.utils import generate_base_model simple_config = { "states": ["A", "B"], "transitions": [ ["go", "A", "B"], ], "initial": "A", "before_state_change": "call_this", "model_override": True, } class_definition = generate_base_model(simple_config) with open("base_model.py", "w") as f: f.write(class_definition) # ... 在另一个文件中 from transitions import Machine from base_model import BaseModel class Model(BaseModel): # call_this 在 BaseModel 中将是抽象方法 def call_this(self) -> None: # 做些事情 model = Model() machine = Machine(model, **simple_config) 定义将被覆盖的模型方法会增加一些额外的工作。 确保事件名称拼写正确可能很麻烦，尤其是在状态和转换在你的模型之前或之后在列表中定义的情况下。你可以通过将状态定义为枚举来减少样板文件和工作字符串的不确定性。你还可以借助 add_transitions 和 event 在模型类中直接定义转换。 取决于你喜欢的代码风格，你可以使用函数装饰器 add_transitions 还是 event 来为属性赋值。 它们的工作方式相同，具有相同的签名，并且应该产生（几乎）相同的 IDE 类型提示。 由于这仍然是一项正在进行的工作，你需要创建一个自定义的 Machine 类，并使用 with_model_definitions 来让 transitions 检查以这种方式定义的转换。 from enum import Enum from transitions.experimental.utils import with_model_definitions, event, add_transitions, transition from transitions import Machine class State(Enum): A = "A" B = "B" C = "C" class Model: state: State = State.A @add_transitions(transition(source=State.A, dest=State.B), [State.C, State.A]) @add_transitions({"source": State.B, "dest": State.A}) def foo(self): ... bar = event( {"source": State.B, "dest": State.A, "conditions": lambda: False}, transition(source=State.B, dest=State.C) ) @with_model_definitions # 别忘了用这个装饰器定义你的模型！ class MyMachine(Machine): pass model = Model() machine = MyMachine(model, states=State, initial=model.state) model.foo() model.bar() assert model.state == State.C model.foo() assert model.state == State.A 扩展 尽管 transitions 的核心保持轻量级，但有各种各样的 MixIns 来扩展其功能。目前支持的有： 分层状态机 用于嵌套和重用 图表 用于可视化机器的当前状态 线程安全锁 用于并行执行 异步回调 用于异步执行 自定义状态 用于扩展与状态相关的行为 有两种机制来检索具有所需功能的启用状态机实例。 第一种方法是使用便捷 factory，四个参数 graph、nested、locked 或 asyncio 在需要该功能时设置为 True： from transitions.extensions import MachineFactory # 创建带有混合类的机器 diagram_cls = MachineFactory.get_predefined(graph=True) nested_locked_cls = MachineFactory.get_predefined(nested=True, locked=True) async_machine_cls = MachineFactory.get_predefined(asyncio=True) # 从这些类创建实例 # 实例可以像简单机器一样使用 machine1 = diagram_cls(model, state, transitions) machine2 = nested_locked_cls(model, state, transitions) 这种方法针对实验性使用，因为在这种情况下不需要知道底层类。 但是，类也可以直接从 transitions.extensions 导入。命名方案如下： 图表 嵌套 线程安全 异步 Machine ✘ ✘ ✘ ✘ GraphMachine ✓ ✘ ✘ ✘ HierarchicalMachine ✘ ✓ ✘ ✘ LockedMachine ✘ ✘ ✓ ✘ HierarchicalGraphMachine ✓ ✓ ✘ ✘ LockedGraphMachine ✓ ✘ ✓ ✘ LockedHierarchicalMachine ✘ ✓ ✓ ✘ LockedHierarchicalGraphMachine ✓ ✓ ✓ ✘ AsyncMachine ✘ ✘ ✘ ✓ AsyncGraphMachine ✓ ✘ ✘ ✓ HierarchicalAsyncMachine ✘ ✓ ✘ ✓ HierarchicalAsyncGraphMachine ✓ ✓ ✘ ✓ 要使用功能丰富的状态机，可以这样写： from transitions.extensions import LockedHierarchicalGraphMachine as LHGMachine machine = LHGMachine(model, states, transitions) 分层状态机 (HSM) Transitions 包含一个允许嵌套状态的扩展模块。 这使我们能够创建上下文，并模拟状态与状态机中某些子任务相关的情况。 要创建嵌套状态，可以从 transitions 导入 NestedState，或者使用带有初始化参数 name 和 children 的字典。 可选地，可以使用 initial 来定义当进入嵌套状态时要转换到的子状态。 from transitions.extensions import HierarchicalMachine states = ['站立', '行走', {'name': '咖啡因兴奋', 'children':['犹豫不决', '奔跑']}] transitions = [ ['走', '站立', '行走'], ['停', '行走', '站立'], ['喝', '*', '咖啡因兴奋'], ['走', ['咖啡因兴奋', '咖啡因兴奋_犹豫不决'], '咖啡因兴奋_奔跑'], ['放松', '咖啡因兴奋', '站立'] ] machine = HierarchicalMachine(states=states, transitions=transitions, initial='站立', ignore_invalid_triggers=True) machine.走() # 现在正在行走 machine.停() # 停一下 machine.喝() # 咖啡时间 machine.state >>> '咖啡因兴奋' machine.走() # 我们得走快点 machine.state >>> '咖啡因兴奋_奔跑' machine.停() # 停不下来！ machine.state >>> '咖啡因兴奋_奔跑' machine.放松() # 离开嵌套状态 machine.state # 呼，这一趟真够呛 >>> '站立' # machine.on_enter_咖啡因兴奋_奔跑('回调方法') 使用 initial 的配置可能如下所示： # ... states = ['站立', '行走', {'name': '咖啡因兴奋', 'initial': '犹豫不决', 'children': ['犹豫不决', '奔跑']}] transitions = [ ['走', '站立', '行走'], ['停', '行走', '站立'], # 这个转换将结束于 '咖啡因兴奋_犹豫不决'... ['喝', '*', '咖啡因兴奋'], # ... 这就是为什么我们不再需要在这里指定 '咖啡因兴奋' 了 ['走', '咖啡因兴奋_犹豫不决', '咖啡因兴奋_奔跑'], ['放松', '咖啡因兴奋', '站立'] ] # ... HierarchicalMachine 构造函数的 initial 关键字接受嵌套状态（例如 initial='咖啡因兴奋_奔跑'）、一个被视为并行状态的列表（例如 initial=['A', 'B']）或另一个模型的当前状态（initial=model.state），该状态应该有效地是前面提到的一个选项。请注意，当传递字符串时，transition 将检查目标状态是否有 initial 子状态，并将其用作入口状态。这将递归进行，直到子状态没有提到初始状态。并行状态或作为列表传递的状态将“按原样”使用，不会进行进一步的初始评估。 请注意，你先前创建的状态对象必须是 NestedState 或其派生类。 简单 Machine 实例中使用的标准 State 类缺乏嵌套所需的功能。 from transitions.extensions.nesting import HierarchicalMachine, NestedState from transitions import State m = HierarchicalMachine(states=['A'], initial='初始') m.add_state('B') # 好的 m.add_state({'name': 'C'}) # 也没问题 m.add_state(NestedState('D')) # 同样没问题 m.add_state(State('E')) # 不行！ 使用嵌套状态时需要考虑的一些事项：状态名称与 NestedState.separator连接。 目前分隔符设置为下划线 ('_')，因此行为类似于基本机器。 这意味着状态 foo 的子状态 bar 将以 foo_bar 为人所知。bar 的子状态 baz 将被称为 foo_bar_baz，依此类推。 当进入一个子状态时，将为所有父状态调用 enter。退出子状态时也是如此。 第三，嵌套状态可以覆盖其父状态的转换行为。 如果当前状态不知道某个转换，它将委托给其父级。 这意味着，在标准配置中，HSM 中的状态名称绝不能包含下划线。 对于 transitions 来说，无法判断 machine.add_state('state_name') 是应该添加一个名为 state_name 的状态，还是向状态 state 添加一个子状态 name。 但在某些情况下，这还不够。 例如，如果状态名称由多个单词组成，并且你希望/需要使用下划线而不是 CamelCase 来分隔它们。 为了解决这个问题，你可以很容易地更改用于分隔的字符。 如果你使用 Python 3，你甚至可以使用花哨的 unicode 字符。 将分隔符设置为下划线以外的其他东西会改变一些行为（自动转换和设置回调），不过： from transitions.extensions import HierarchicalMachine from transitions.extensions.nesting import NestedState NestedState.separator = '↦' states = ['A', 'B', {'name': 'C', 'children':['1', '2', {'name': '3', 'children': ['a', 'b', 'c']} ]} ] transitions = [ ['重置', 'C', 'A'], ['重置', 'C↦2', 'C'] # 覆盖父级的重置 ] # 我们依赖自动转换 machine = HierarchicalMachine(states=states, transitions=transitions, initial='A') machine.to_B() # 退出状态 A，进入状态 B machine.to_C() # 退出 B，进入 C machine.to_C.s3.a() # 进入 C↦a；进入 C↦3↦a； machine.state >>> 'C↦3↦a' assert machine.is_C.s3.a() machine.to('C↦2') # 不是交互式的；退出 C↦3↦a，退出 C↦3，进入 C↦2 machine.重置() # 退出 C↦2；C↦3 的重置已被覆盖 machine.state >>> 'C' machine.重置() # 退出 C，进入 A machine.state >>> 'A' # s.on_enter('C↦3↦a', '回调方法') 自动转换被调用为 to_C.s3.a() 而不是 to_C_3_a()。如果你的子状态以数字开头，transitions 会为自动转换 FunctionWrapper 添加一个前缀 's'（'3' 变成 's3'），以符合 Python 的属性命名方案。 如果不需要交互式补全，可以直接调用 to('C↦3↦a')。此外，on_enter/exit_<<state name>> 被替换为 on_enter/exit(state_name, callback)。状态检查可以类似地进行。可以使用 FunctionWrapper 变体 is_C.s3.a() 而不是 is_C_3_a()。 要检查当前状态是否是特定状态的子状态，is_state 支持关键字 allow_substates： machine.state >>> 'C.2.a' machine.is_C() # 检查特定状态 >>> False machine.is_C(allow_substates=True) >>> True assert machine.is_C.s2() is False assert machine.is_C.s2(allow_substates=True) # FunctionWrapper 也支持 allow_substate 你也可以在 HSM 中使用枚举，但请记住 Enum 是按值比较的。 如果你在状态树中有一个值出现多次，那么这些状态就无法区分。 states = [States.RED, States.YELLOW, {'name': States.GREEN, 'children': ['tick', 'tock']}] states = ['A', {'name': 'B', 'children': states, 'initial': States.GREEN}, States.GREEN] machine = HierarchicalMachine(states=states) machine.to_B() machine.is_GREEN() # 返回 True，即使实际状态是 B_GREEN HierarchicalMachine 已从头重写，以支持并行状态和更好的嵌套状态隔离。 这涉及一些基于社区反馈的调整。 要了解处理顺序和配置，请查看以下示例： from transitions.extensions.nesting import HierarchicalMachine import logging states = ['A', 'B', {'name': 'C', 'parallel': [{'name': '1', 'children': ['a', 'b', 'c'], 'initial': 'a', 'transitions': 'go', 'a', 'b'}, {'name': '2', 'children': ['x', 'y', 'z'], 'initial': 'z'}], 'transitions': 'go', '2_z', '2_x'}] transitions = 'reset', 'C_1_b', 'B' logging.basicConfig(level=logging.INFO) machine = HierarchicalMachine(states=states, transitions=transitions, initial='A') machine.to_C() # INFO:transitions.extensions.nesting:退出状态 A # INFO:transitions.extensions.nesting:进入状态 C # INFO:transitions.extensions.nesting:进入状态 C_1 # INFO:transitions.extensions.nesting:进入状态 C_2 # INFO:transitions.extensions.nesting:进入状态 C_1_a # INFO:transitions.extensions.nesting:进入状态 C_2_z machine.go() # INFO:transitions.extensions.nesting:退出状态 C_1_a # INFO:transitions.extensions.nesting:进入状态 C_1_b # INFO:transitions.extensions.nesting:退出状态 C_2_z # INFO:transitions.extensions.nesting:进入状态 C_2_x machine.reset() # INFO:transitions.extensions.nesting:退出状态 C_1_b # INFO:transitions.extensions.nesting:退出状态 C_2_x # INFO:transitions.extensions.nesting:退出状态 C_1 # INFO:transitions.extensions.nesting:退出状态 C_2 # INFO:transitions.extensions.nesting:退出状态 C # INFO:transitions.extensions.nesting:进入状态 B 当使用 parallel 而不是 children 时，transitions 将同时进入传递列表的所有状态。 进入哪个子状态由 initial 定义，它应该始终指向一个直接的子状态。 一个新特性是通过在状态定义中传递 transitions 关键字来定义局部转换。 上面定义的转换 ['go', 'a', 'b'] 只在 C_1 中有效。 虽然你可以像在 ['go', '2_z', '2_x'] 中那样引用子状态，但不能在局部定义的转换中直接引用父状态。 当父状态退出时，其子级也将退出。 除了从 Machine 中已知的转换处理顺序（转换按照它们添加的顺序被考虑）外，HierarchicalMachine 也会考虑层次结构。 在子状态中定义的转换将首先被评估（例如，C_1_a 在 C_2_z 之前离开），并且使用通配符 * 定义的转换将（目前）只添加转换到根状态（在这个例子中是 A、B、C） 从 0.8.0 开始，可以直接添加嵌套状态，并且将即时创建父状态： m = HierarchicalMachine(states=['A'], initial='A') m.add_state('B_1_a') m.to_B_1() assert m.is_B(allow_substates=True) 0.9.1 中的实验性功能： 你可以在状态或 HSM 本身上使用 on_final 回调。回调将在以下情况下触发：a) 状态本身被标记为 final 并且刚被进入，或者 b) 所有子状态都被认为是 final 并且至少一个子状态刚进入 final 状态。在情况 b) 下，如果条件 b) 对它们成立，所有父级也将被认为是 final。这在处理并行进行并且你的 HSM 或任何父状态应该在所有子状态都达到 final 状态时被通知的情况下可能很有用： from transitions.extensions import HierarchicalMachine from functools import partial # 我们以状态 A 初始化这个并行 HSM： # / X # / / yI # A -> B - Y - yII [最终] # \ Z - zI # \ zII [最终] def final_event_raised(name): print("{} 是最终状态！".format(name)) states = ['A', {'name': 'B', 'parallel': [{'name': 'X', 'final': True, 'on_final': partial(final_event_raised, 'X')}, {'name': 'Y', 'transitions': 'final_Y', 'yI', 'yII', 'initial': 'yI', 'on_final': partial(final_event_raised, 'Y'), 'states': ['yI', {'name': 'yII', 'final': True}] }, {'name': 'Z', 'transitions': 'final_Z', 'zI', 'zII', 'initial': 'zI', 'on_final': partial(final_event_raised, 'Z'), 'states': ['zI', {'name': 'zII', 'final': True}] }, ], "on_final": partial(final_event_raised, 'B')}] machine = HierarchicalMachine(states=states, on_final=partial(final_event_raised, 'Machine'), initial='A') # X 将立即发出最终事件 machine.to_B() # >>> X 是最终状态！ print(machine.state) # >>> ['B_X', 'B_Y_yI', 'B_Z_zI'] # Y 的子状态现在是最終的，將觸發 Y 的 'on_final' machine.final_Y() # >>> Y 是最终状态！ print(machine.state) # >>> ['B_X', 'B_Y_yII', 'B_Z_zI'] # Z 的子状态变成最终状态，这也使得 B 的所有子级都是最终的，因此机器本身也是最终的 machine.final_Z() # >>> Z 是最终状态！ # >>> B 是最终状态！ # >>> Machine 是最终状态！ 重用先前创建的 HSM 除了语义顺序，嵌套状态在你想要为特定任务指定状态机并计划重用它们时非常方便。 在 0.8.0 之前，HierarchicalMachine 不会集成机器实例本身，而是通过创建它们的副本来集成状态和转换。 然而，从 0.8.0 开始，(Nested)State 实例只是被引用，这意味着一个机器的状态和事件集合中的更改将影响另一个机器实例。不过，模型及其状态不会被共享。 请注意，事件和转换也是通过引用复制的，如果不使用 remap 关键字，它们将被两个实例共享。 进行此更改是为了更符合 Machine，它也通过引用使用传递的 State 实例。 count_states = ['1', '2', '3', '完成'] count_trans = [ ['增加', '1', '2'], ['增加', '2', '3'], ['减少', '3', '2'], ['减少', '2', '1'], ['完成', '3', '完成'], ['重置', '*', '1'] ] counter = HierarchicalMachine(states=count_states, transitions=count_trans, initial='1') counter.增加() # 喜欢我的计数器 states = ['等待中', '收集中', {'name': '计数中', 'children': counter}] transitions = [ ['收集', '*', '收集中'], ['等待', '*', '等待中'], ['计数', '收集中', '计数中'] ] collector = HierarchicalMachine(states=states, transitions=transitions, initial='等待中') collector.收集() # 收集中 collector.计数() # 看看我们得到了什么；计数中_1 collector.增加() # 计数中_2 collector.增加() # 计数中_3 collector.完成() # collector.state == 计数中_完成 collector.等待() # collector.state == 等待中 如果通过 children 关键字传递一个 HierarchicalMachine，该机器的初始状态将被分配给新的父状态。 在上面的例子中，我们看到进入 计数中 也会进入 计数中_1。 如果这是不希望的行为，并且机器应该停止在父状态，用户可以通过传递 initial 为 False 来实现，如 {'name': '计数中', 'children': counter, 'initial': False}。 有时你希望这样的嵌入式状态集合“返回”，这意味着在它完成后，它应该退出并转换到你的超级状态之一。 为了实现这种行为，你可以重新映射状态转换。 在上面的例子中，我们希望计数器在达到状态 完成 时返回。 这样做如下： states = ['等待中', '收集中', {'name': '计数中', 'children': counter, 'remap': {'完成': '等待中'}}] ... # 与上面相同 collector.增加() # 计数中_3 collector.完成() collector.state >>> '等待中' # 注意 '计数中_完成' 将从状态机中移除 如上所述，使用 remap 将复制事件和转换，因为它们在原始状态机中可能无效。 如果一个被重用的状态机没有最终状态，你当然可以手动添加转换。 如果 'counter' 没有 '完成' 状态，我们可以添加 ['完成', 'counter_3', '等待中'] 来实现相同的行为。 如果你希望状态和转换按值而不是按引用复制（例如，如果你想保持 0.8 之前的行为），你可以通过创建一个 NestedState 并将机器的状态和事件的深拷贝分配给它来实现。 from transitions.extensions.nesting import NestedState from copy import deepcopy # ... 配置和创建 counter counting_state = NestedState(name="计数中", initial='1') counting_state.states = deepcopy(counter.states) counting_state.events = deepcopy(counter.events) states = ['等待中', '收集中', counting_state] 对于复杂的状态机，共享配置而不是实例化的机器可能更可行。 特别是因为实例化的机器必须派生自 HierarchicalMachine。 这样的配置可以通过 JSON 或 YAML 轻松存储和加载（请参阅 FAQ）。 HierarchicalMachine 允许使用关键字 children 或 states 定义子状态。 如果两者都存在，只考虑 children。 counter_conf = { 'name': '计数中', 'states': ['1', '2', '3', '完成'], 'transitions': [ ['增加', '1', '2'], ['增加', '2', '3'], ['减少', '3', '2'], ['减少', '2', '1'], ['完成', '3', '完成'], ['重置', '*', '1'] ], 'initial': '1' } collector_conf = { 'name': '收集器', 'states': ['等待中', '收集中', counter_conf], 'transitions': [ ['收集', '*', '收集中'], ['等待', '*', '等待中'], ['计数', '收集中', '计数中'] ], 'initial': '等待中' } collector = HierarchicalMachine(**collector_conf) collector.收集() collector.计数() collector.增加() assert collector.is_计数中_2() 图表 附加关键字： title（可选）：设置生成图像的标题。 show_conditions（默认 False）：在转换边上显示条件 show_auto_transitions（默认 False）：在图中显示自动转换 show_state_attributes（默认 False）：在图中显示回调（进入、退出）、标签和超时 Transitions 可以生成基本的状态图，显示所有状态之间的有效转换。 基本的图表支持生成 mermaid 状态机定义，可与 mermaid 的 在线编辑器、GitLab 或 GitHub 中的 markdown 文件以及其他网络服务一起使用。 例如，此代码： from transitions.extensions.diagrams import HierarchicalGraphMachine import pyperclip states = ['A', 'B', {'name': 'C', 'final': True, 'parallel': [{'name': '1', 'children': ['a', {"name": "b", "final": True}], 'initial': 'a', 'transitions': 'go', 'a', 'b'}, {'name': '2', 'children': ['a', {"name": "b", "final": True}], 'initial': 'a', 'transitions': 'go', 'a', 'b'}]}] transitions = [['reset', 'C', 'A'], ["init", "A", "B"], ["do", "B", "C"]] m = HierarchicalGraphMachine(states=states, transitions=transitions, initial="A", show_conditions=True, title="Mermaid", graph_engine="mermaid", auto_transitions=False) m.init() pyperclip.copy(m.get_graph().draw(None)) # 为了方便使用 pyperclip print("图表已复制到剪贴板！") 生成此图表（检查文档源以查看 markdown 符号）： --- Mermaid Graph --- stateDiagram-v2 direction LR classDef s_default fill:white,color:black classDef s_inactive fill:white,color:black classDef s_parallel color:black,fill:white classDef s_active color:red,fill:darksalmon classDef s_previous color:blue,fill:azure state "A" as A Class A s_previous state "B" as B Class B s_active state "C" as C C --> [*] Class C s_default state C { state "1" as C_1 state C_1 { [*] --> C_1_a state "a" as C_1_a state "b" as C_1_b C_1_b --> [*] } -- state "2" as C_2 state C_2 { [*] --> C_2_a state "a" as C_2_a state "b" as C_2_b C_2_b --> [*] } } C --> A: reset A --> B: init B --> C: do C_1_a --> C_1_b: go C_2_a --> C_2_b: go [*] --> A 要使用更复杂的绘图功能，你需要安装 graphviz 和/或 pygraphviz。 要使用 graphviz 包生成图形，你需要手动或通过包管理器安装 Graphviz。 sudo apt-get install graphviz graphviz-dev # Ubuntu 和 Debian brew install graphviz # MacOS conda install graphviz python-graphviz # (Ana)conda 现在你可以安装实际的 Python 包 pip install graphviz pygraphviz # 手动安装 graphviz 和/或 pygraphviz... pip install transitions[diagrams] # ... 或者安装带有 'diagrams' 附加功能的 transitions，它目前依赖于 pygraphviz 目前，GraphMachine 在可用时会使用 pygraphviz，并在找不到 pygraphviz 时回退到 graphviz。 如果 graphviz 也不可用，则将使用 mermaid。 可以通过向构造函数传递 graph_engine="graphviz"（或 "mermaid"）来覆盖此设置。 请注意，此默认值将来可能会更改，并且可能放弃对 pygraphviz 的支持。 使用 Model.get_graph() 可以获取当前图形或感兴趣的区域 (roi) 并像这样绘制它： # import transitions from transitions.extensions import GraphMachine m = Model() # 没有进一步参数时，将使用 pygraphviz machine = GraphMachine(model=m, ...) # 当你想要显式使用 graphviz 时 machine = GraphMachine(model=m, graph_engine="graphviz", ...) # 在应该显示自动转换的情况下 machine = GraphMachine(model=m, show_auto_transitions=True, ...) # 绘制整个图形 ... m.get_graph().draw('my_state_diagram.png', prog='dot') # ... 或者只绘制感兴趣的区域 # (先前状态、活动状态和所有可达状态) roi = m.get_graph(show_roi=True).draw('my_state_diagram.png', prog='dot') 这会产生类似这样的东西： 无论你使用哪个后端，draw 函数也接受文件描述符或二进制流作为第一个参数。如果你将此参数设置为 None，将返回字节流： import io with open('a_graph.png', 'bw') as f: # 当你传递对象而不是文件名时，需要传递格式。 m.get_graph().draw(f, format="png", prog='dot') # 你也可以传递一个（二进制）流 b = io.BytesIO() m.get_graph().draw(b, format="png", prog='dot') # 或者干脆自己处理二进制字符串 result = m.get_graph().draw(None, format="png", prog='dot') assert result == b.getvalue() 作为回调传递的引用和部分函数将尽可能解析： from transitions.extensions import GraphMachine from functools import partial class Model: def clear_state(self, deep=False, force=False): print("Clearing state ...") return True model = Model() machine = GraphMachine(model=model, states=['A', 'B', 'C'], transitions=[ {'trigger': 'clear', 'source': 'B', 'dest': 'A', 'conditions': model.clear_state}, {'trigger': 'clear', 'source': 'C', 'dest': 'A', 'conditions': partial(model.clear_state, False, force=True)}, ], initial='A', show_conditions=True) model.get_graph().draw('my_state_diagram.png', prog='dot') 这应该产生类似这样的东西： 如果引用的格式不符合你的需求，你可以覆盖静态方法 GraphMachine.format_references。如果你想完全跳过引用，只需让 GraphMachine.format_references 返回 None。 另外，看看我们的 示例 IPython/Jupyter notebook，了解关于如何使用和编辑图形的更详细示例。 线程安全（-ish）状态机 在事件分派在线程中完成的情况下，可以使用 LockedMachine 或 LockedHierarchicalMachine，其中函数访问（!sic）使用可重入锁保护。 这并不能防止你通过篡改模型或状态机的成员变量来破坏你的机器。 from transitions.extensions import LockedMachine from threading import Thread import time states = ['A', 'B', 'C'] machine = LockedMachine(states=states, initial='A') # 让我们假设进入 B 需要一些时间 thread = Thread(target=machine.to_B) thread.start() time.sleep(0.01) # 线程需要一些时间来启动 machine.to_C() # 同步访问；在线程完成之前不会执行 # 直接访问属性 thread = Thread(target=machine.to_B) thread.start() machine.new_attrib = 42 # 不同步！会打乱执行顺序 任何 Python 上下文管理器都可以通过 machine_context 关键字参数传入： from transitions.extensions import LockedMachine from threading import RLock states = ['A', 'B', 'C'] lock1 = RLock() lock2 = RLock() machine = LockedMachine(states=states, initial='A', machine_context=[lock1, lock2]) 任何通过 machine_model 的上下文将在所有向 Machine 注册的模型之间共享。 也可以添加每个模型的上下文： lock3 = RLock() machine.add_model(model, model_context=lock3) 重要的是所有用户提供的上下文管理器都必须是可重入的，因为状态机将多次调用它们，即使在单个触发器调用的上下文中也是如此。 使用异步回调 如果你使用的是 Python 3.7 或更高版本，你可以使用 AsyncMachine 来处理异步回调。 如果你愿意，可以混合使用同步和异步回调，但这可能会有不良的副作用。 请注意，事件需要被 await，并且事件循环也必须由你处理。 from transitions.extensions.asyncio import AsyncMachine import asyncio import time class AsyncModel: def prepare_model(self): print("我是同步的。") self.start_time = time.time() async def before_change(self): print("我是异步的，现在将阻塞 100 毫秒。") await asyncio.sleep(0.1) print("我等完了。") def sync_before_change(self): print("我是同步的，将阻塞事件循环（我可能不应该这样做）") time.sleep(0.1) print("我同步地等完了。") def after_change(self): print(f"我又变成同步的了。执行花了 {int((time.time() - self.start_time) * 1000)} 毫秒。") transition = dict(trigger="start", source="Start", dest="Done", prepare="prepare_model", before=["before_change"] * 5 + ["sync_before_change"], after="after_change") # 异步执行 before 函数 5 次 model = AsyncModel() machine = AsyncMachine(model, states=["Start", "Done"], transitions=[transition], initial='Start') asyncio.get_event_loop().run_until_complete(model.start()) # >>> 我是同步的。 # 我是异步的，现在将阻塞 100 毫秒。 # 我是异步的，现在将阻塞 100 毫秒。 # 我是异步的，现在将阻塞 100 毫秒。 # 我是异步的，现在将阻塞 100 毫秒。 # 我是异步的，现在将阻塞 100 毫秒。 # 我是同步的，将阻塞事件循环（我可能不应该这样做） # 我同步地等完了。 # 我等完了。 # 我等完了。 # 我等完了。 # 我等完了。 # 我等完了。 # 我又变成同步的了。执行花了 101 毫秒。 assert model.is_Done() 那么，为什么你需要使用 Python 3.7 或更高版本呢？你可能会问。 异步支持早已引入。 AsyncMachine 利用 contextvars 来处理在新的转换完成之前到达的新事件： async def await_never_return(): await asyncio.sleep(100) raise ValueError("那花了太长时间！") async def fix(): await m2.fix() m1 = AsyncMachine(states=['A', 'B', 'C'], initial='A', name="m1") m2 = AsyncMachine(states=['A', 'B', 'C'], initial='A', name="m2") m2.add_transition(trigger='go', source='A', dest='B', before=await_never_return) m2.add_transition(trigger='fix', source='A', dest='C') m1.add_transition(trigger='go', source='A', dest='B', after='go') m1.add_transition(trigger='go', source='B', dest='C', after=fix) asyncio.get_event_loop().run_until_complete(asyncio.gather(m2.go(), m1.go())) assert m1.state == m2.state 这个例子实际上说明了两件事： 首先，在 m1 从 A 到 B 的转换中调用的 'go' 没有被取消，其次，调用 m2.fix() 将通过执行从 A 到 C 的 'fix' 来停止 m2 从 A 到 B 的转换尝试。 没有 contextvars，这种分离是不可能的。 请注意，prepare 和 conditions 不被视为正在进行的转换。 这意味着在 conditions 被评估之后，即使另一个事件已经发生，转换也会被执行。 只有作为 before 回调或之后运行时，任务才会被取消。 AsyncMachine 具有一个模型特定的队列模式，当 queued='model' 传递给构造函数时可以使用。 使用模型特定的队列，事件只有在属于同一个模型时才会被排队。 此外，引发的异常只会清除引发该异常的模型的事件队列。 为简单起见，让我们假设下面 asyncio.gather 中的每个事件不是同时触发，而是稍微延迟的： asyncio.gather(model1.event1(), model1.event2(), model2.event1()) # 使用 AsyncMachine(queued=True) 的执行顺序 # model1.event1 -> model1.event2 -> model2.event1 # 使用 AsyncMachine(queued='model') 的执行顺序 # (model1.event1, model2.event1) -> model1.event2 asyncio.gather(model1.event1(), model1.error(), model1.event3(), model2.event1(), model2.event2(), model2.event3()) # 使用 AsyncMachine(queued=True) 的执行顺序 # model1.event1 -> model1.error # 使用 AsyncMachine(queued='model') 的执行顺序 # (model1.event1, model2.event1) -> (model1.error, model2.event2) -> model2.event3 请注意，队列模式在机器构造后不得更改。 向状态添加特性 如果你的超级英雄需要一些自定义行为，你可以通过装饰机器状态来添加一些额外的功能： from time import sleep from transitions import Machine from transitions.extensions.states import add_state_features, Tags, Timeout @add_state_features(Tags, Timeout) class CustomStateMachine(Machine): pass class 社交超级英雄(object): def __init__(self): self.随从 = 0 def on_enter_等待中(self): self.随从 += 1 states = [{'name': '准备中', 'tags': ['在家', '忙碌']}, {'name': '等待中', 'timeout': 1, 'on_timeout': '出发'}, {'name': '离开'}] # 城市需要我们！ transitions = [['完成', '准备中', '等待中'], ['加入', '等待中', '等待中'], # 再次进入等待中会增加我们的随从 ['出发', '等待中', '离开']] # 好的，我们走吧 hero = 社交超级英雄() machine = CustomStateMachine(model=hero, states=states, transitions=transitions, initial='准备中') assert hero.state == '准备中' # 为夜班做准备 assert machine.get_state(hero.state).is_忙碌 # 我们在家并且忙碌 hero.完成() assert hero.state == '等待中' # 等待其他超级英雄加入我们 assert hero.随从 == 1 # 目前只有我们 sleep(0.7) # 等待... hero.加入() # 耶，我们有伴了 sleep(0.5) # 等待... hero.加入() # 更多伙伴 \o/ sleep(2) # 等待... assert hero.state == '离开' # 不耐烦的超级英雄已经离开了大楼 assert machine.get_state(hero.state).is_在家 is False # 是的，不再在家了 assert hero.随从 == 3 # 至少他不是一个人 目前，transitions 配备了以下状态特性： 超时 -- 在一段时间后触发一个事件 关键字：timeout（int，可选） -- 如果传递，进入的状态将在 timeout 秒后超时 关键字：on_timeout（字符串/可调用对象，可选） -- 当达到超时时间时将被调用 当设置了 timeout 但未设置 on_timeout 时将引发 AttributeError 注意：超时在线程中触发。这意味着有几个限制（例如捕获超时中引发的异常）。对于更复杂的应用程序，请考虑事件队列。 标签 -- 向状态添加标签 关键字：tags（列表，可选） -- 为状态分配标签 当状态被标记为 tag_name 时，State.is_<tag_name> 将返回 True，否则返回 False 错误 -- 当状态无法离开时引发 MachineError 继承自 Tags（如果使用 Error 则不要使用 Tags） 关键字：accepted（bool，可选） -- 将状态标记为 accepted 或者可以传递关键字 tags，包含 'accepted' 注意：只有当 auto_transitions 设置为 False 时才会引发错误。否则每个状态都可以通过 to_<state> 方法退出。 易失性 -- 每次进入状态时初始化一个对象 关键字：volatile（类，可选） -- 每次进入状态时，将分配一个类类型的对象给模型。属性名称由 hook 定义。如果省略，将创建一个空的 VolatileObject。 关键字：hook（字符串，默认='scope'） -- 临时对象的模型属性名称。 你可以编写自己的 State 扩展，并以相同的方式添加它们。只需注意 add_state_features 期望混合类。这意味着你的扩展应始终调用被重写的方法 __init__、enter 和 exit。你的扩展可以继承自State，但没有它也能工作。 使用 @add_state_features 有一个缺点，那就是被装饰的机器不能被 pickle（更准确地说，动态生成的 CustomState 不能被 pickle）。 这可能是编写专用自定义状态类的原因。 根据所选的状态机，你的自定义状态类可能需要提供某些状态特性。例如，HierarchicalMachine 要求你的自定义状态是 NestedState 的实例（仅 State 是不够的）。要注入你的状态，你可以将它们分配给你的 Machine 的类属性 state_cls，或者覆盖 Machine.create_state，以防在创建状态时需要一些特定的过程： from transitions import Machine, State class 我的状态(State): pass class 自定义机器(Machine): # 使用 我的状态 作为状态类 state_cls = 我的状态 class 详细机器(Machine): # `Machine._create_state` 是一个类方法，但我们可以 # 覆盖它成为一个实例方法 def _create_state(self, *args, **kwargs): print("用机器 '{0}' 创建新状态".format(self.name)) return 我的状态(*args, **kwargs) 如果你想在你的 AsyncMachine 中完全避免线程，你可以用 asyncio 扩展中的 AsyncTimeout 替换 Timeout 状态特性： import asyncio from transitions.extensions.states import add_state_features from transitions.extensions.asyncio import AsyncTimeout, AsyncMachine @add_state_features(AsyncTimeout) class 超时机器(AsyncMachine): pass states = ['A', {'name': 'B', 'timeout': 0.2, 'on_timeout': 'to_C'}, 'C'] m = 超时机器(states=states, initial='A', queued=True) # 参见下面的说明 asyncio.run(asyncio.wait([m.to_B(), asyncio.sleep(0.1)])) assert m.is_B() # 超时不应该触发 asyncio.run(asyncio.wait([m.to_B(), asyncio.sleep(0.3)])) assert m.is_C() # 现在超时应该已经被处理了 你应该考虑将 queued=True 传递给 超时机器 构造函数。这将确保事件被顺序处理，并避免在超时和事件在接近时间发生时可能出现的异步竞争条件。 将 transitions 与 Django 一起使用 你可以看看 FAQ 来获取一些灵感，或者查看 django-transitions。 它是由 Christian Ledermann 开发的，也托管在 Github 上。 文档 包含一些用法示例。
《学习实践中重复间隔的优化》部分翻译与简易应用
本文针对知识保留水平为 95% 的配对联想学习，确定了一个计算重复间隔的通用公式。 据称，该公式可广泛应用于各种学科的学习实践中，且不受个体学习者能力的限制。 翻译对原文的表述做了一些改动 关于此文章 此文章所使用的论文的原标题是: Optimization of repetition spacing in the practice of learning 翻译对原文的表述做了一些改动 原始 PDF 文档在 此处 可以下载 (镜像) 论文原始作者是 Piotr A. Woiniak 和 Edward J. Gorzelanczyk (来自波兹南医学院组织胚胎学系) 这篇文章的相关原理可被运用到辅助记忆软件中. 此文章虽未提及, 但与 SM-2 算法密切相关. 作为主流记忆软件的理论基石之一, 此文章或许可作为改进变体的理论依据. 摘要 本文针对知识保留水平为 95% 的配对联想学习，确定了一个计算重复间隔的通用公式。 据称，该公式可广泛应用于各种学科的学习实践中，且不受个体学习者能力的限制。 研究思路 许多出版物都广泛讨论了不同重复间隔对学习效果的影响。特别是，间隔效应被认为是一种普遍现象。间隔效应是指，如果重复的间隔是分散/稀疏的，而不是集中重复，那么学习任务的表现会更好。因此，有观点提出，学习中使用的最佳重复间隔是最长的、但不会导致遗忘的间隔。然而，遗忘具有随机性，因此无法预测它何时会在特定情况下发生。所以，尽管进行了多次尝试，但很少有研究提出可在学习实践中使用的重复间隔。 在本研究中，我们采用了一种随机方法。我们通过将最佳间隔定义为导致少量（预先确定）学习材料被遗忘的间隔来计算最佳重复间隔。我们使用了一个计算机程序来监控学习过程，确保在重复时只有 5% 的待记忆项目未被记住。该程序采用了一种优化算法，当遗忘项目的比例低于或高于所需 5% 的水平时，它会自动延长或缩短重复间隔。 此外，在学习过程中，待记忆项目根据受试者的表现自动分为不同的难度类别。对于每个难度类别，我们都应用了不同的重复间隔。 实验过程 参与实验的受试者是波兹南工业大学的 7 名计算机科学系本科生，他们是无偿志愿者。在 18 个月的时间里，他们总共记忆和重复了超过 35000 个他们选择的项目（这些项目是波兰语-英语单词对）。这些记忆的项目是受试者在实验前不认识的。受试者在 2 个月内使用下述算法分批掌握了所有材料，并在接下来的 16 个月内继续重复。 算法中的每次重复过程如下： 呈现问题 受试者尝试回答 将回答与正确答案进行比较 受试者进行 0-5 分的自我评估（0 分 - 非常差，5 分 - 非常好） 用于确定最佳重复间隔的函数（最佳间隔函数）以表格形式的最佳因子矩阵表示. (1) 式: $$ I(EF, 1) = OF(EF, 1) $$ (2) 式: $$ I(EF, R) = I(EF, R - 1) * OF(EF, R) $$ 公式中: $ EF $ - 简单度因子，用于反映该项目被记忆的难易程度 $ I(EF,R) $ - 间隔 (单位: 天)，对于简单度因子确定为 EF 的项目，在第 R 次重复之前的间隔 $ OF(EF, R) $ - 最优间隔函数 为了更好地拟合最优间隔因子，我们使用了一种优化算法来修改矩阵 OF 的初始值，主要判据是稳定的知识保留率达到95%。 使用了两类不同的OF矩阵： 单一值矩阵：所有条目初始化为相同的值，即对于所有的EF和R，OF(EF,R) := 1.5。 预定值矩阵：OF条目根据不同的难度类别（Woiniak和Biedalak 1992）而变化：对于所有的EF，OF(EF, 1) := 5；对于所有的EF和R，OF(EF,R) := EF；其中EF的选择范围是1.3, 1.4, 1.5, . . . , 3.2（参见表一）。 预定矩阵OF旨在确保OF条目更快地收敛到其最优值，而单一值矩阵OF则用于验证优化算法的有效性。15%的项目根据单一值OF矩阵进行优化，85%的项目则使用预定OF矩阵进行安排。 使用了以下学习算法（请注意，重复可以无限期进行，并且OF矩阵的实验值可以随时获取，越晚越好）： 记忆一个新项目i：以自定进度的方式学习问题和答案之间的关联。 设置初始EF值：EFi := 2.5（设置对应于项目i的EF因子的初始值）。 设置重复次数：Ri := 1（将重复次数设置为1）。 确定首次重复日期：使用OF矩阵根据公式（1）确定首次重复的日期。 进行重复：在预定的重复日期，通过回答相关问题来重复该项目，并以0-5的等级评估回答的质量（表现良好则等级高）。 继续学习或停止：如果学习者希望继续学习，则Ri := Ri + 1；否则，停止。 调整EFi值：如果质量低于4，则降低EFi值；否则，增加。学习中出现问题的项目将被归类为更困难，因此会采用不同的重复间隔（详见Woiniak和Biedalak 1992）。 调整OF(EFi, Ri - 1)值：如果其应用导致质量低于4，则降低OF(EFi, Ri - 1)值；否则，增加。因此，不成功的重复会缩短学习中使用的间隔（调整变化速度以在重复中获得95%的稳定保留率；保留率下降将倾向于OF值较慢的向上变化和较快的向下变化，从而缩短重复中使用的间隔长度；详见Woiniak和Biedalak 1992）。 安排下一次重复或重置：如果质量大于或等于3，则根据公式（2）安排下一次重复并转到步骤5。如果质量低于3，则转到步骤3（该项目被视为遗忘）。 实验过程中产生的示例OF矩阵如表二所示（该矩阵来源于单一值OF矩阵）。 数据处理 完成数据收集过程后，结果通过一系列逼近程序进行处理，这些程序旨在获得最适合的、可在课堂学习实践中使用的最优间隔累积函数. 针对95%知识保留率的最佳重复间隔函数，其通用公式已确定如下： (3) 式: $$ OI(EF, 1) = 5 $$ (4) 式: $$ OI(EF, R) = OI(EF, R - 1) * (EF - 0.1 + e^{-2.3*R + 5}) $$ 公式中： $ EF $ - 表示一个待记忆项目的简单度因子。这个值通常在1.3（表示最难的项目）到2.8（表示最简单的项目）之间。 $ R $ - 代表重复的次数。 $ OI(EF, R) $ - 代表在第R次重复之前，针对难度系数为EF的项目的最佳间隔时长（单位：天）。 在公式 (3) 和 (4) 中，针对不同主题的函数参数之间的统计差异被认为不显著。同样，从单一值和预定值矩阵推导出的矩阵之间的差异也微不足道。这些结果明确表明，渐进式间隔优于集中式或分布式间隔。这在学习实践中具有应用价值，能够最大限度地减少记忆和保留所学材料所需的时间。 这项研究中使用的软件可免费从作者处获取，只需寄送一张3.5英寸、1.44 MB的软盘即可。 现在估计不行了 ( 应用与实现 这是绘制的图像:
关于电子邮件的"来源伪造"攻击分析
起因是收到了一封来自"自己"的难绷诈骗邮件 以下是原文: (迫 真 演 技) Hello pervert, I've sent thіs message from your Microsoft account. I want to іnform you about a very bad sіtuatіon for you. However, you can benefіt from іt, іf you wіll act wіsely. Have you heard of Pegasus? Thіs іs a spyware program that іnstalls on computers and smartphones and allows hackers to monіtor the actіvіty of devіce owners. It provіdes access to your webcam, messengers, emaіls, call records, etc. It works well on Androіd, іOS, macOS and Wіndows. I guess, you already fіgured out where I’m gettіng at. It’s been a few months sіnce I іnstalled іt on all your devісes because you were not quіte choosy about what lіnks to clіck on the іnternet. Durіng thіs perіod, I’ve learned about all aspects of your prіvate lіfe, but one іs of specіal sіgnіfіcance to me. I’ve recorded many vіdeos of you jerkіng off to hіghly controversіal рorn vіdeos. Gіven that the “questіonable” genre іs almost always the same, I can conclude that you have sіck рerversіon. I doubt you’d want your frіends, famіly and co-workers to know about іt. However, I can do іt іn a few clіcks. Every number іn your contact Iіst wіll suddenly receіve these vіdeos – on WhatsApp, on Telegram, on Instagram, on Facebook, on emaіl – everywhere. It іs goіng to be a tsunamі that wіll sweep away everythіng іn іts path, and fіrst of all, your former lіfe. Don’t thіnk of yourself as an іnnocent vіctіm. No one knows where your рerversіon mіght lead іn the future, so consіder thіs a kіnd of deserved рunіshment to stop you. I’m some kіnd of God who sees everythіng. However, don’t panіc. As we know, God іs mercіful and forgіvіng, and so do I. But my merсy іs not free. Transfer 1700$ to my Lіtecoіn (LTC) wallet: ltc1quvupuw5hz3pr9aguqtevhd342azs6gz2c04zl2 Once I receіve confіrmatіon of the transactіon, I wіll рermanently delete all vіdeos compromіsіng you, unіnstall Pegasus from all of your devіces, and dіsappear from your lіfe. You can be sure – my benefіt іs only money. Otherwіse, I wouldn’t be wrіtіng to you, but destroy your lіfe wіthout a word іn a second. I’ll be notіfіed when you open my emaіl, and from that moment you have exactly 48 hours to send the money. If cryptocurrencіes are unchartered waters for you, don’t worry, іt’s very sіmple. Just google "crypto exchange" or "buy Litecoin" and then іt wіll be no harder than buyіng some useless stuff on Amazon. I strongly warn you agaіnst the followіng: * Do not reply to thіs emaіl. I've sent іt from your Mіcrosoft account. * Do not contact the polіce. I have access to all your devісes, and as soon as I fіnd out you ran to the cops, vіdeos wіll be publіshed. * Don’t try to reset or destroy your devісes. As I mentіoned above: I’m monіtorіng all your actіvіty, so you eіther agree to my terms or the vіdeos are рublіshed. Also, don’t forget that cryptocurrencіes are anonymous, so іt’s іmpossіble to іdentіfy me usіng the provіded address. Good luck, my perverted frіend. I hope thіs іs the last tіme we hear from each other. And some frіendly advіce: from now on, don’t be so careless about your onlіne securіty. 问题分析 这是 KMail 给出的信件头部源代码(有删减): Authentication-Results: spf=softfail (sender IP is 195.191.99.128) smtp.mailfrom=outlook.com; dkim=none (message not signed) header.d=none;dmarc=fail action=none header.from=outlook.com; Received-SPF: SoftFail (protection.outlook.com: domain of transitioning outlook.com discourages use of 195.191.99.128 as permitted sender) X-Sender-IP: 195.191.99.128 X-SID-PRA: (已模糊邮箱地址) X-SID-Result: FAIL (注意此处) X-MS-Exchange-Organization-SCL: 5 Received-SPF: SoftFail (protection.outlook.com: domain of transitioning outlook.com discourages use of 195.191.99.128 as permitted sender) Received: from sharprealtymiami.com (195.191.99.128) by AMS0EPF00000191.mail.protection.outlook.com (10.167.16.216) with Microsoft SMTP Server (version=TLS1_2, cipher=TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384) id 15.20.8722.18 via Frontend Transport; Fri, 9 May 2025 07:42:56 +0000 .......... 一封正常的自我标注信件应该是这样的: Authentication-Results: spf=pass (sender IP is 40.92.44.10) smtp.mailfrom=outlook.com; dkim=pass (signature was verified) header.d=outlook.com;dmarc=pass action=none header.from=outlook.com;compauth=pass reason=100 X-SID-Result: PASS X-MS-Exchange-Organization-SCL: 1 发件者朔源 以下是真正的服务器信息: 发件服务器: sharprealtymiami.com (IP 地址: 195.191.99.128) 中转服务器: AMS0EPF00000191.eurprd05.prod.outlook.com 接收服务器: PH0PR19MB4874.namprd19.prod.outlook.com (发送者的ip分析, 甚至是代理地址, 十分甚至九分谨慎((( ) 预防措施 可见邮件头中显示 SPF 结果为 "softfail"，说明发件人 IP（195.191.99.128）没有被 outlook.com 授权为合法的发送者, 表明邮件可能不是来自声称的发件人。 DKIM 和 DMARC: 邮件头中显示 DKIM 为 "none"，这意味着邮件没有被签名，无法验证其完整性. DMARC 结果为 "fail"，表明邮件未通过域名的验证策略 问题解决, 只要注意 Authentication-Results 即可, 难绷的是, outlook 和微软宁愿把它移动到垃圾邮件里也不愿提醒下用户, 堪称产品设计的典范 后记 探索一下 sharprealtymiami.com 这个神秘的地址:
如题 服务器部署 使用 Docker Compose 部署: #version: '3' # 旧版兼容 services: registry: image: registry:2 ports: - "5000:5000" # 可以把前面的 5000 换成想要的端口 environment: REGISTRY_PROXY_REMOTEURL: https://registry-1.docker.io # 上游 REGISTRY_STORAGE_CACHE_BLOBDESCRIPTOR: inmemory # 内存缓存 volumes: - ./data:/var/lib/registry 然后可以用 Nginx 反向代理端口, 添加证书到 HTTPS 就行了. 本地镜像站配置 在客户端上 echo '{"registry-mirrors": ["https://镜像站域名"]}' | sudo tee /etc/docker/daemon.json > /dev/null
如题, 有 Dijkstra 和 Floyd 算法 /* * @Author: wangzhiyu * @Date: 2023-09-03 02:29:26 * @Last Modified by: wangzhiyu * @Last Modified time: 2024-07-30 22:47:07 */ #include <cmath> #include <cstring> #include <time.h> #include <iostream> #include <vector> #include <unistd.h> // for linux // #include <windows.h> // for windows using namespace std; using ll = long long; constexpr int LIM1 = 205; // 预分配阙值 constexpr int INF = 0x3f3f3f3f; // 近似无限大 /// @brief 图论相关 namespace graph { int n; /// @brief 节点数据结构 struct Node { int nxt; int w; }; vector<Node> edges[LIM1]; /// @brief 增加一条边 /// @param from /// @param to /// @param w 权值 void addedge(int from, int to, int w) { edges[from].push_back({to, w}); } /// @brief 删除某边 /// @param from /// @param to void deledge(int from, int to) { // delete an edge for (auto i = edges[from].begin(); i != edges[from].end();) { if ((*i).nxt == to) { i = edges[from].erase(i); } else { i++; } } } /// @brief 查询某边的权值 /// @param from /// @param to /// @return 权值 int queryedge(int from, int to) { // to query an edge's weight if (from == to) return 0; for (auto i = edges[from].begin(); i != edges[from].end(); i++) { if ((*i).nxt == to) { return (*i).w; } } return INF; } /* void dfs_launcher(){ for(int i = 1; i <= n; i++){ dfs(i); } } void dfs(int begin, ){ cerr << "Access: " << begin; dfs(edges[begin].nxt); } */ /// @brief Floyd 最短路算法 /// @param spec 待展示的点标号 void sp_f(int spec) { // Floyd algorithm int sp[LIM1][LIM1]; for (int i = 1; i <= n; i++) { for (int j = 1; j <= n; j++) { sp[i][j] = queryedge(i, j); } } for (int i = 1; i <= n; i++) { for (int j = 1; j <= n; j++) { for (int k = 1; k <= n; k++) { sp[i][j] = min(sp[i][j], sp[i][k] + sp[k][j]); } } } for (int i = 1; i <= n; i++) { string dist = to_string(sp[spec][i]); if (dist == to_string(INF)) dist = "∞"; // unable to reach cerr << spec << " -[" << dist << "]-> " << i << " (floyd)" << endl; } } /// @brief 输出某点详情 /// @param idx 特定点索引 void print(int idx) { // to print a particular vertex's detail information cout << "Print: " << idx << endl << " > "; for (auto &i : edges[idx]) { cout << i.nxt << "(" << i.w << ")" << " "; } for (int i = 1; i <= idx ; i++){ } cout << endl; } /// @brief Dijkstra 单源最短路径算法 /// @param begin 源点标号 void sp_dij(int begin) { // Dijkstra algorithm bool vis[LIM1]; // visiting status fill(vis + 1, vis + n + 1, 0); // clear status int sp[LIM1]; // shortest path's distance for (int i = 1; i <= n; i++) { // initalize distance sp[i] = queryedge(begin, i); } vis[begin] = 1; // mark beginner's visit for (int k = 1; k <= n - 1; k++) { // this loop makes every vertex // successfully marked (except beginner) int best_transfer; // best transfer vertex int mindistance = INF; // temp min distance for (int i = 1; i <= n; i++) { // calc best_transfer if (queryedge(begin, i) < mindistance && vis[i] != 1) { mindistance = queryedge(begin, i); best_transfer = i; } } // once best_transfer confirmed, mark its visit vis[best_transfer] = 1; for (int i = 1; i <= n; i++) { // recalc transferred distance if (vis[i] != 1) { if (queryedge(begin, best_transfer) + queryedge(best_transfer, i) < queryedge(begin, i)) { sp[i] = queryedge(begin, best_transfer) + queryedge(best_transfer, i); } } } } // print result for (int i = 1; i <= n; i++) { string dist = to_string(sp[i]); if (dist == to_string(INF)) dist = "∞"; // unable to reach cerr << begin << " -[" << dist << "]-> " << i << " (dijkstra)" << endl; } } /// @brief 自然地初始化图 /// @param data 格式化的图数据, 参照 https://csacademy.com/app/graph_editor/, /// 第一行和最后一行必须是换行符 void naturally_init(string data) { // raw string feature required, last line must be \n; bool flag = 0; // dotmode/edgemode string formatted[3]; int k = 0; while (*(data.begin()) != '\n') data.erase(data.begin()); // erase beginner data.erase(data.begin()); for (char i : data) { if (!flag) { if (i != '\n') formatted[0].push_back(i); else { n++; formatted[0] = ""; } } if (i == ' ') flag = 1; if (flag) { if (i != ' ' && i != '\n') formatted[k % 3].push_back(i); else k++; if (i == '\n') { addedge(stoi(formatted[0]), stoi(formatted[1]), stoi(formatted[2])); formatted[0] = ""; formatted[1] = ""; formatted[2] = ""; } } } cerr << "Automatically initialization finished" << endl; cerr << "Node number: " << n << endl; } } // namespace graph /// @brief 实用函数与性能分析 namespace utils { /// @brief 输出分割符 void spilt() { cerr << "------------------------" << endl; } time_t lasttime = INF, currtime = 0; /// @brief 自动计时器 void autoclock() { cerr << clock(); cerr << fixed << endl; cerr << "\033[31m"; if(lasttime == INF){ cerr << "[TIMG GAP] INIT\033[0m" << endl; lasttime = clock(); return; } currtime = clock(); double dur; dur = (double)(currtime - lasttime); cerr << "[TIME GAP] "<< dur/CLOCKS_PER_SEC << endl; lasttime = clock(); cerr << "\033[0m"; } } // namespace utils int main() { string original_graph = R"( 格式化的图数据: 1 2 3 4 5 6 7 8 9 10 6 2 9 2 4 3 4 5 2 1 4 2 1 5 9 2 3 2 2 7 6 1 9 8 1 3 4 9 10 4 4 6 3 2 5 8 2 8 2 2 10 2 8 9 7 3 7 6 )"; graph::naturally_init(original_graph); cout << fixed; // 禁用科学计数法输出 cout << graph::queryedge(1, 3) << endl; cout << graph::queryedge(1, 3) << endl; cout << graph::queryedge(4, 9) << endl; graph::print(1); graph::print(2); utils::spilt(); utils::autoclock(); graph::sp_f(2); utils::autoclock(); graph::sp_dij(2); utils::autoclock(); utils::spilt(); graph::sp_f(3); graph::sp_dij(3); return 0; }
关于 FileZilla 链接 FTP 超时无法列出目录的解决方案
如题 不要使用快速连接 在站点管理器里把模式切换到主动 OK
如题, 测试机是装载 Ubuntu Touch 16.04 的 MIX 3 不要用 amixer 和 alsamixer, 会出错: phablet@ubuntu-phablet:~$ amixer amixer: Mixer default load error: No such device phablet@ubuntu-phablet:~$ alsamixer cannot load mixer controls: No such device 使用 pactl: pactl set-sink-volume 0 20% # 调到 20 % pactl set-sink-volume 0 +20% # 增加 20 % pactl set-sink-volume 0 -20% # 减少 20 %
自费购买云服务器绝对是一种经济负担, 使用免费服务器资源无不是一种很有吸引力的选择, 问题在于免费的服务器资源一般都要续期 所以我写了个脚本来自动白嫖免费服务器... 阅读之前 请确保你有这些东西: 一个手机号 1 到 3 元钱 (用于实名认证) 对数据随时丢失 & 服务不稳定的心理准备 不能搭建 http(s) 网站的心理准备 (除非你有一个已经备案的域名, 三丰云的免备案简直没法用, 但是 tcp 连接不受限制) 基本的 Linux 服务器管理技能 第一步：注册并登录三丰云 此过程略 记得设置管理员密码 系统选择 Ubuntu 16.04(这是三丰云提供的最新系统版本, 凑合用罢). 不要选 Windows, 太耗资源了. 此时你已经可以通过 ssh 管理服务器了, 关于管理服务器, 本文不再赘述 另外第一次续期时间点设置很反人类, 先手动过一下, 后面就可以自动化了 第二步：使用自动续期脚本 先说明一下, 这个脚本虽然理论上可以在你的免费服务器运行, 但是不建议这么做 首先, 因为系统版本低的缘故, 你要手动编译安装 Python 3.7.5, 因为系统自带的最高版本是 Python 3.5, 而且就算你搞好了 Python 环境, 你还要处理 Geckodriver 和老旧火狐浏览器版本一类的破事, 是真的麻烦. 其次, 就算你手动或者用 docker 啥的配好了所有环境, 开心跑脚本提交续期的时候, 你会发现你的续期图片的 IP 地址是北京什么公司, 过不了审核的 个人建议可以在自己的电脑/手机/家庭服务器上整个 Linux 环境的计划任务运行脚本 (理论上 Windows 也可以运行, 但是我没做测试) 如果实在没有设备, 还可以借助 Vercel 的部署脚本在你的免费服务器上整个计划 git commit 运行, 但是这需要你的配置 第三步：部署脚本并运行 从软件源安装 FireFox Python 3.7 或更高 (记得安装 python3-dev python3-pip) (debian 上貌似只有 firefox-esr, 也能用) 如果你要搞计划任务还要安装 cron, 本文不再赘述 使用 pip3 安装依赖包: selenium 创建一个目录, 按文件名放置脚本文件 # run.sh rm -rf result.png python3 webposter.py && python3 imgeditor.py && python3 webuploader.py rm -rf scrst.png rm -rf upload.png rm -rf geckodriver.log # imgeditor.py from PIL import Image def crop_image(input_path, output_path, left, upper, right, lower): try: image = Image.open(input_path) cropped_image = image.crop((left, upper, right, lower)) cropped_image.save(output_path) print("[LOG] 成功生成上传图片") except FileNotFoundError: print("[ERROR] 找不到输入图片") except Exception as e: print("[ERROR] 裁剪图片时出错", str(e)) print("+-- IMGEDITOR --+") print("执行任务: 编辑图片文件") print("任务目标: 生成上传图片") input_path = "scrst.png" output_path = "upload.png" left = 180 upper = 13 right = 1131 lower = 309 crop_image(input_path, output_path, left, upper, right, lower) print("[LOG] 成功生成上传图片") # webposter.py import random from time import sleep from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.firefox.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC print("+-- WEBPOSTER --+") print("执行任务: 发送续期信息") print("任务目标: 生成原始截图") URL = "http://www.adminxy.com/showidc-4335.asp" NAMEAREA_XPATH = "/html/body/div[2]/div[2]/div[1]/div[4]/div[4]/div[3]/form/table/tbody/tr[1]/td[2]/input" TEXTAREA_XPATH = "/html/body/div[2]/div[2]/div[1]/div[4]/div[4]/div[3]/form/table/tbody/tr[4]/td[2]/textarea" A1_XPATH = "/html/body/div[2]/div[2]/div[1]/div[4]/div[4]/div[3]/form/table/tbody/tr[2]/td[2]/table/tbody/tr[1]/td[2]/div/ul/li[5]/a" A2_XPATH = "/html/body/div[2]/div[2]/div[1]/div[4]/div[4]/div[3]/form/table/tbody/tr[2]/td[2]/table/tbody/tr[1]/td[1]/div/ul/li[5]/a" A3_XPATH = "/html/body/div[2]/div[2]/div[1]/div[4]/div[4]/div[3]/form/table/tbody/tr[2]/td[2]/table/tbody/tr[2]/td[1]/div/ul/li[5]/a" A4_XPATH = "/html/body/div[2]/div[2]/div[1]/div[4]/div[4]/div[3]/form/table/tbody/tr[2]/td[2]/table/tbody/tr[2]/td[2]/div/ul/li[5]/a" BTN1_XPATH = "/html/body/div[2]/div[2]/div[1]/div[4]/div[4]/div[3]/form/table/tbody/tr[5]/td[2]/input" TEXT_LIST = [ # 这是 GPT 生成的广告词, 可以自己多搞几条 "三丰云的免费云服务器真是太棒了！连接速度快，稳定性和性能出色，完全满足我的需求。而且，使用起来也非常方便，还永久免费！强烈推荐给需要高质量云服务器的人们！", "我尝试了三丰云的免费云服务器，它真是太赞了！连接速度快得让我惊艳，而且使用起来非常方便。最重要的是，它永久免费！三丰云的服务质量真是一流，让我非常满意！", "免费云服务器果然名不虚传！我用了三丰云的产品，速度快得令人难以置信！能够永久免费使用，真的太实惠了！三丰云的免费虚拟主机和云服务器绝对是必备工具，推荐给大家！", "三丰云的免费云服务器真是太好用了！我用它托管我的网站和应用，连接速度超快，且非常稳定。而且，它的免费性质让我感到惊喜！无论个人用户还是企业用户，都值得选择三丰云！", "三丰云的免费云服务器简直是太棒了！我用它来托管我的网站和应用，连接速度超快，绝对能满足我的需求。同时，它还是永久免费的，物超所值！大家一定要抓住这个机会，亲自体验一下！", "三丰云的免费云服务器真是太实用了！我用它来测试我的应用，连接速度快得让我惊叹不已。而且，它还是永久免费的，非常适合个人用户和企业用户。选择三丰云，你绝对不会后悔！", "作为免费云服务器的提供商，三丰云真是值得称赞！它的连接速度极快，简直令人难以置信。而且，使用起来非常方便，而且永久免费！三丰云的产品真的是个人用户和企业用户的不二选择！", "三丰云的免费云服务器真是太强大了！连接速度快，性能出色，能够满足各种需求。而且，它免费使用，简直太实惠了！如果你需要云服务器，一定不能错过三丰云！", "三丰云的免费云服务器真是让我无法抗拒！连接速度极快，稳定性也非常好。它的免费特性更是令人惊喜！使用起来非常方便，真的是个人用户和企业用户的不二选择！", "三丰云的产品让我感受到了极致的便利和高效。不仅连接速度飞快，而且免费！这是一款真正物超所值的产品，你一定不要错过！", "三丰云的免费云服务器真是太厉害了！我使用它来托管我的网站和应用，连接速度快得让我震惊！而且，它是永久免费的！选择三丰云，你绝对会感到满意！", "我对三丰云的免费云服务器赞不绝口！连接速度快得让人难以置信，使用起来也非常方便！是个人用户和企业用户的最佳选择！选择三丰云，你一定会获得绝佳的体验！", "三丰云的免费云服务器真是太强大了！我使用它来托管我的网站和应用，效果非常好！连接速度快，稳定性和性能都很出色。而且，它的免费性质让我更加喜爱！强烈推荐给大家！", "作为永久免费服务器的提供商，三丰云真是无可挑剔！我使用它的免费云服务器，速度超快，稳定性很好。而且，它的免费性质真的让我感到惊喜！三丰云绝对是你值得选择的品牌！" ] options = Options() options.add_argument("--headless") # 无头模式 driver = webdriver.Firefox(options=options) driver.get(URL) name_input = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH, NAMEAREA_XPATH))) name_input.clear() name_input.send_keys("你的发帖名字") text_input = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH, TEXTAREA_XPATH))) text_input.send_keys(random.choice(TEXT_LIST).replace("三丰云", "三.丰.云.")) ele1 = driver.find_element(By.CSS_SELECTOR,"#stars1 .five-stars"); ele2 = driver.find_element(By.CSS_SELECTOR,"#stars2 .five-stars"); ele3 = driver.find_element(By.CSS_SELECTOR,"#stars3 .five-stars"); ele4 = driver.find_element(By.CSS_SELECTOR,"#stars4 .five-stars"); driver.execute_script("arguments[0].click();", ele1) driver.execute_script("arguments[0].click();", ele2) driver.execute_script("arguments[0].click();", ele3) driver.execute_script("arguments[0].click();", ele4) button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, BTN1_XPATH))) button.click() print("[LOG] 消息发送成功") # 截图 driver.get(URL) driver.execute_script("window.scrollBy(0, 640);") driver.save_screenshot("scrst.png") print("[LOG] 原始截图获取成功") driver.quit() # webuploader.py import os from time import sleep from selenium.webdriver.common.by import By from selenium import webdriver from selenium.webdriver.firefox.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.common.action_chains import ActionChains print("+-- WEBUPLOADER --+") print("执行任务: 上传图片文件") print("任务目标: 完成自动续期") # 定义需要的常量 NAMEAREA_XPATH = "/html/body/div[4]/div[1]/div[1]/div/div/div[2]/form/div[1]/input" PWDAREA_XPATH = "/html/body/div[4]/div[1]/div[1]/div/div/div[2]/form/div[2]/input" # 初始化Firefox驱动 options = Options() options.add_argument("--headless") # 无头模式 driver = webdriver.Firefox(options=options) actions = ActionChains(driver) # 登录 print("[LOG] 准备登录") driver.get("https://www.sanfengyun.com/login") # 在页面上随意滑动鼠标 action_chains = ActionChains(driver) action_chains.move_by_offset(50, 50).perform() action_chains.move_by_offset(50, 50).perform() action_chains.move_by_offset(10, 53).perform() sleep(3) jscmd = """ document.getElementById("userName").value = "你的注册手机号"; document.getElementById("passwordInput").value = "你的三丰云密码"; document.getElementById("loginSubmit").click(); """ driver.execute_script(jscmd) sleep(3) # 自动延期 driver.get("三丰云服务器的控制面板页面链接") # 像这样"https://www.sanfengyun.com/control/#/freeServer/一串数字" print("[LOG] 准备自动延期") link1 = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "/html/body/div/div[2]/div[2]/div[2]/div/div[4]/div[1]/div/div/div/div[8]"))) actions.move_to_element(link1).click().perform() try: print("[LOG] 开始自动续期") addr_input = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH, "/html/body/div/div[2]/div[2]/div[2]/div/div[4]/div[2]/div[8]/form[1]/div/div/div/input"))) addr_input.clear() addr_input.send_keys("http://www.adminxy.com/showidc-4335.asp") driver.find_element(By.ID,"fileImg1").send_keys(os.path.abspath("upload.png")) button = WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, "/html/body/div/div[2]/div[2]/div[2]/div/div[4]/div[2]/div[8]/form[2]/div/div/button"))) sleep(5) button.click() sleep(5) driver.save_screenshot("result.png") print("[LOG] 已生成网页结果截图") driver.quit() except: try: notfoundinf = driver.find_element(By.XPATH,"/html/body/div/div[2]/div[2]/div[2]/div/div[4]/div[2]/div[8]/form[2]/div/div/div/p[2]") print("[LOG] 未到时间") sleep(4) print(notfoundinf.text) sleep(4) driver.quit() exit() except: print("[ERROR] 程序出错") driver.quit() exit() 在 webposter.py 里把"你的发帖名字"改成你想要的网名. 在 webupload.py 里把账号信息和面板链接填完整 执行 run.sh 试试, 没问题就设置计划任务吧 :)
记 vector 容器初始化长度为0导致的段错误ub
初始化 std::vector 时, 有时这么写: std::vector vec(length); 编译可正常通过, 然而当把 length 设置为 0 时, 程序发生段错误, 并直接崩溃 因此当 length 为变量时, 应作特判, 避免测试点 RE. 例如 USACO1.3 混合牛奶 Mixing Milk 的测试点 #2. 如果不想特判, 还有一个解决办法: 使用 resize 方法替代初始化长度 #include <bits/stdc++.h> using namespace std; int main(){ int n; cin >> n; vector<int> vec; vec.resize(n); for(int &i:vec) cout << i << endl; return 0; } 如果对这段代码输入 0, 程序不会输出任何值, 立即退出, 这也说明了 STL 方法一般是安全的. 初始化一个数组长度为 0 似乎并不会出问题, 估计指向的是nullptr罢...