gpt4 book ai didi

Python电路加入 channel

转载 作者:行者123 更新时间:2023-11-28 18:42:13 27 4
gpt4 key购买 nike

我有一个场景,我正在使用 python 电路框架动态创建具有自己 channel 的新组件。我希望在所有创建的 channel 都触发特定事件时收到通知。

我尝试过使用成功事件,但它会在每个 channel 上独立触发,所以我每个 channel 都有一个。对我来说,使用不同的 channel 是有意义的,因为在不同的数据集上执行相同的任务。

我目前的解决方案是记录创建的 channel 名称,然后监听结束事件(下面的“boom”)并从列表中删除触发 channel 。当列表为空时,我可以停下来。下面是一个例子。它有效,但我觉得应该有一种更优雅的方式在这些 channel 结束后加入它们。

import time
import sys

from circuits import Component, Event
from circuits.core.debugger import Debugger

class boom(Event):
"boom event"

class Start(Component):
def __init__(self, channel="*"):
super(Start, self).__init__(channel=channel)
self._boom_channels = []
return

def started(self, *args):
for i in [1,2]:
channel = 'channel_{}'.format(i)
self._boom_channels.append(channel)

new = Middle(channel=channel).register(self)

def boom(self, event, *args):
new_chans = set(self._boom_channels) - set(event.channels)
self._boom_channels = list(new_chans)
print self._boom_channels
if not self._boom_channels:
sys.exit()

class Middle(Component):
def __init__(self, channel="*"):
super(Middle, self).__init__(channel=channel)
time.sleep(2)
self.fire(boom())
return

if __name__ == '__main__':
(Start() + Debugger()).run()

最佳答案

我相信您的想法是正确的。由于这些是离散事件,因此如果不跟踪它们就无法确定它们何时全部完成(我不认为)。

让您的示例更好的唯一方法是:

#!/usr/bin/env python


from __future__ import print_function


from circuits import Component, Debugger, Event


class boom(Event):
"boom event"


class Start(Component):

channel = "start"

def init(self, channel=channel):
self.boom_channels = []

def started(self, *args):
for i in [1, 2]:
channel = "channel_{}".format(i)
self.boom_channels.append(channel)

Middle(channel=channel).register(self)

def boom(self, middle):
self.boom_channels.remove(middle.channel)

if not self.boom_channels:
raise SystemExit(0)


class Middle(Component):

def registered(self, component, manager):
self.fire(boom(self), manager)
return


def main():
app = (Start() + Debugger())
app.run()


if __name__ == '__main__':
main()

输出:

$ python test.py 
<registered[*] (<Debugger/* 1156:MainThread (queued=0) [S]>, <Start/start 1156:MainThread (queued=2) [R]> )>
<started[start] (<Start/start 1156:MainThread (queued=1) [R]> )>
<registered[channel_1] (<Middle/channel_1 1156:MainThread (queued=0) [S]>, <Start/start 1156:MainThread (queued=2) [R]> )>
<registered[channel_2] (<Middle/channel_2 1156:MainThread (queued=0) [S]>, <Start/start 1156:MainThread (queued=2) [R]> )>
<boom[<Start/start 1156:MainThread (queued=2) [R]>] (<Middle/channel_1 1156:MainThread (queued=0) [S]> )>
<boom[<Start/start 1156:MainThread (queued=1) [R]>] (<Middle/channel_2 1156:MainThread (queued=0) [S]> )>
<stopped[start] (<Start/start 1156:MainThread (queued=0) [S]> )>

编辑:一些注意事项:

  • 显式地将 boom() 事件推送给管理器。
  • 在我们注册“之后”触发 boom() 事件。

关于Python电路加入 channel ,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/24822986/

27 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com