Dynamic class in python

Question

It is probably the wrong title, but here is my problem.

I have a system comprised of a microcontroller (MCU), a serial interface (SPI), a DAC (Digital / Analog converter), an electrode (E). Each element is defined as a class in my python modelization.

As a first step, I want to monitor the output on the electrode as I input something in the microcontroller.

Let's consider the following:

• Input: 2 mA on the electrode during 1 ms.
• MCU send the new DAC value via the SPI: 30 us
• DAC updates its register and output: 400 us
• MCU send a switch on command to the electrode: 1 us
• The electrode is now outputting.
• 1 ms later, send a switch off command to the electrode: 1us
• The electrode doesn't output anymore.

My 2 biggest issues are 1. How to take into account this time component and 2. How to monitor the SPI line to determine if something has to be done.

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

My system actually has 16 DACs and electrodes and a field-programmable gate array which are all listening to the same SPI. What I described above is a fairly simplified version.

Question is: How to have the components check the value in SPI.msg regularly and act accordingly?

In reality, each component is doing its life. Thus actions are performed in parallel. Since I'm trying to simulate the timeline and the action performed, I do not mind doing everything serially with a timeline variable (attribute) for each element. I just have issues to figure out how to have my classes interact together.

i.e. I can't do the following in python or I will get stuck:

class DAC:
    def __init__(id):
        # init

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

Maybe an event triggering could be used... But I don't know how.

Maybe with multithreading, defining one thread / element?

EDIT: Current state:

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

I'm assuming you want to keep it single-threaded and you don't use asyncio. In this case, you might want to employ observer or pub/sub pattern when implementing the 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!')

So you can use it like this:

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

I haven't done anything to be able to send SPI messages from Device objects, but you can update the abstraction accordingly.

Answer 2

You could move the while loop simply outside:

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)

As stated in your comment you want more a timeline view on what is happening. You can have an explicit timeline with which your components interact. External input (state changes) can be set beforehand in the same manner. To order the timemline I'll just run sort each time but it would probably be more performant to use something like a priority queue.

This mainly differs from Vovanrock2002 answer by not recursing in each timestep and having an explicit timeline.

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)

This way you have an explicit timeline (which you could also "record", just add each popped event to some list) and you can have arbitrarily connected components (e.g. if you want to have multiple SPIs).