fischertechnik-factory-twin/simulator/devs_models/collector.py

from pypdevs.DEVS import AtomicDEVS
from loguru import logger

class Collector(AtomicDEVS):
    """A simple collector which holds every received item in a list in its state"""
    def __init__(self, name):
        super(Collector, self).__init__(name)
        self.inp = self.addInPort("inp")

        # keep track of items received
        self.state = {
            "items": []
        }

    def get_contents(self):
        """ Get the current contents of the collector """
        return self.state["items"]

    def extTransition(self, inputs):
        if self.inp in inputs:
            new_item = inputs[self.inp]
            self.state["items"].append(new_item)
            logger.trace(f"{type(self).__name__} '{self.name}' received: {new_item}")
        return self.state