代码之家  ›  专栏  ›  技术社区  ›  Mathieu

python中的动态类

  •  0
  • Mathieu  · 技术社区  · 8 年前

    这可能是错误的标题,但这是我的问题。

    我有一个由 微控制器 (MCU) 串行接口 (SPI) 数模转换器 (数字/模拟转换器),一个 电极 (e)。在我的Python建模中,每个元素都被定义为一个类。

    作为第一步,当我在微控制器中输入东西时,我想监控电极上的输出。

    让我们考虑一下:

    • 输入:1 ms内电极上2 mA。
    • MCU通过SPI发送新的DAC值:30 us
    • DAC更新其寄存器并输出:400 us
    • MCU向电极发送一个接通命令:1us
    • 电极现在正在输出。
    • 1毫秒后,向电极发送关闭命令:1us
    • 电极不再输出。

    我最大的两个问题是1。如何考虑这个时间分量和2。如何监视SPI线以确定是否必须执行某些操作。

    class Electrode:
        def __init__(self, id):
            self.id = id
            self.switch = False
            self.value = 0
    
        def output(self):
            if self.switch:
                return self.value
            else:
                return 0
    
    class SPI:
        def __init__(self):
            self.msg = None
    
    class MCU:
        def __init__(self):
            self.name = "MicroController"
    
        def send_SPI_msg(self, SPI, msg):
            SPI.msg = msg
    
    class DAC:
        def __init__(id):
            self.id = id
            self.cs = 1
            self.register = None
            self.output = None
    
        def read_SPI_msg(self, SPI):
            message = SPI.msg
            # update register and output
    

    我的系统实际上有16个数模转换器和电极,以及一个现场可编程门阵列,它们都在监听同一个SPI。我上面描述的是一个相当简单的版本。

    问题是:如何让组件检查值 SPI.msg 定期采取相应行动?

    实际上,每个组件都在工作。因此,动作是并行执行的。因为我试图模拟时间线和执行的操作,所以我不介意为每个元素使用时间线变量(属性)连续地执行所有操作。我只是想弄清楚如何让我的课程相互作用。

    也就是说,我不能在python中执行以下操作,否则会卡住:

    class DAC:
        def __init__(id):
            # init
    
        def read_SPI_msg(self, SPI):
            while True:        
                message = SPI.msg
                # update register and output if needed
    

    也许可以使用事件触发…但我不知道怎么做。

    可能使用多线程,定义一个线程/元素?

    编辑 :当前状态:

    class SPI:
        def __init__(self):
            self.attached_dacs = []
            self.attached_fpga = []
            self.attached_mcu = []
    
        def attach_device(self, device):
            if type(device) == DAC:
                self.attached_dacs.append(device)
            elif type(device) == FPGA:
                self.attached_fpga.append(device)
            elif type(device) == MCU:
                self.attached_mcu.append(device)
    
        def send_message(self, msg):
            for device in self.attached_dacs + self.attached_fpga:
                device.on_spi_message(self, msg)
    
    class SpiAttachableDevice:
        def on_spi_message(self, SPI, message):
            if self.cs:
                self.execute_SPI_message(message)
            else:
                return None
    
    class DAC(SpiAttachableDevice):
        def __init__(self, id):
            self.id = id
            self.cs = False # Not listening
    
        def execute_SPI_message(message):
            # Do stuff
    
    class FPGA(SpiAttachableDevice):
        def __init__(self):
            self.electrodes = list()
            self.cs = False # Not listening
    
        def execute_SPI_message(message):
            # Do stuff
    
    class MCU:
        def __init__(self):
            self.electrodes = list()
    
    2 回复  |  直到 8 年前
        1
  •  1
  •   u354356007    8 年前

    我假设您希望保持它是单线程的,并且不使用Asyncio。在这种情况下,你可能想雇用 观察者 发布/订阅 实施SPI时的模式:

    class SPI:
        def __init__(self):
            self.attached_devices = []
    
        def attach_device(self, device):
            self.attached_devices.append(device)
    
        def send_message(self, msg):
            for device in self.attached_devices:
                device.on_spi_message(self, msg)
    
    class SpiAttachableDevice:
        def on_spi_message(self, spi_instance, message):
            raise NotImplementedError('subclass me!')
    

    所以你可以这样使用它:

    spi = SPI()
    device_1 = Device()
    device_2 = Device()
    spi.attach_device(device_1)
    spi.attach_device(device_2)
    spi.send_message('hello')
    

    我没有做任何事情来发送SPI消息 Device 对象,但是您可以相应地更新抽象。

        2
  •  1
  •   syntonym    8 年前

    你可以移动 while 仅在外部循环:

    class SPI:
    
        def __init__(self, msg):
            self.msg = msg
    
    class Component:
    
        def __init__(self, spi):
            self.spi = spi
    
        def tick(self, t):
            msg = self.spi.msg
            if msg = "...":
                ...
    spi = SPI()
    components = [Component(spi), ...]
    
    for t in range(TOTAL_TIME):
       for component in components:
           component.tick(t)
    

    正如您的评论中所述,您需要更多关于正在发生的事情的时间线视图。您可以有一个明确的时间表,组件与之交互。外部输入(状态更改)可以以相同的方式预先设置。要订购Timemline,我将每次运行sort,但使用类似于 priority queue .

    这主要不同于vovanrock2002的答案,它不是在每个时间步骤中递归,而是有一个明确的时间线。

    class Component:
       def __init__(self, timeline):
           self._timeline = timeline
           self._out = [] #all connected components
    
       def poke(self, changed_object, time):
           return []
    
    class Clock(Component):
    
        def __init__(self, timeline):
            Component.__init__(self, timeline)
            self._out.append(self)
            self.msg = "tick"
            self._timeline.append((200, self, msg))
    
       def poke(self, time, changed_object, msg):
           self._timeline.append((time + 200, self, self.msg))
    
    timeline = []
    spi = SPI(timeline)
    components = [spi, Clock(timeline), ComponentA(timeline), ...]
    
    timeline.append((500, spi, "new DAC value"))
    
    while timeline:
        timeline.sort(key=lambda event: event[0], reverse=True)
        event = timeline.pop()
        time, changed_component, msg:
        for connected_component in changed_component._out:
            connected_component.poke(time, changed_component, msg)
    

    这样,您就有了一个明确的时间线(您也可以“记录”,只需将每个弹出的事件添加到某个列表中),并且您可以拥有任意连接的组件(例如,如果您希望有多个SPI)。

    推荐文章