python中的动态类

Question

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

我有一个由微控制器(MCU)、串行接口(SPI)、数字/模拟转换器( DAC )、电极(E)组成的系统。在我的python建模中，每个元素都被定义为一个类。

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

让我们考虑以下几点：

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

我最大的两个问题是: 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个DAC和电极，还有一个现场可编程门阵列，它们都在听同一个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()

Answer 1

我假设您希望保持它是单线程的，并且不使用异步。在这种情况下，您可能希望在实现SPI时使用观察者或pub/sub模式：

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')

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

Answer 2

您可以简单地将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)

正如您在评论中所述，您需要更多关于正在发生的事情的时间线视图。您可以有一个明确的时间表，您的组件与之交互。外部输入(状态更改)可以以相同的方式预先设置。为了对时间行进行排序，我将每次只运行排序，但是使用类似于优先级队列之类的东西可能会更好。

这主要不同于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)

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