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

from pypdevs.DEVS import AtomicDEVS
from pypdevs.infinity import INFINITY
from dataclasses import dataclass
from loguru import logger
from utils.timed_phase_enum import TimedPhaseEnum
from data_models.workpiece import Workpiece
from data_models.mqtt_message import MqttMessage
from utils.get_timestamp import get_timestamp


class BakeOvenSteps(TimedPhaseEnum):
    """ Group steps together with the time it takes to complete them"""
    IDLE = ('IDLE' ,INFINITY)
    DETECTION = ('DETECTION', 14.2565) # artificial delay in starting
    OPEN_DOOR = ('OPEN_DOOR', 2.584)
    SLIDE_ENTER = ('SLIDE_ENTER', 2.758)
    CLOSE_DOOR = ('CLOSE_DOOR', 0.994) # close doors & notify gripper to come
    BAKE = ('BAKE', 6.513)
    OPEN_DOOR_EXIT = ('OPEN_DOOR_EXIT', 0.280)
    SLIDE_EXIT = ('SLIDE_EXIT', 2.568)
    AWAIT_PICKUP = ('AWAIT_PICKUP', INFINITY)
    OUTPUT_WP = ('OUTPUT_WP', 0.0) # output workpiece to the gripper
    CLOSE_DOOR_QUIT = ('CLOSE_DOOR_QUIT', 4.0) # this seems to happen after a very long delay? maybe not intentional


@dataclass
class BakeOvenState:
    delta_t: float = INFINITY
    phase: BakeOvenSteps = BakeOvenSteps.IDLE
    workpiece: Workpiece | None = None
    status_requested: bool = True # if current status needs to be sent
    visual_update_pending: bool = False

def get_mpo_started_msg() -> MqttMessage:
    """ Get the 'MPO_STARTED' mqtt message """
    message = MqttMessage()
    message.topic = "fl/mpo/ack"
    message.payload = {
        "code": 1,
        "ts": get_timestamp(),
    }
    return message


class MPOOven(AtomicDEVS):
    """ A bake oven which bakes a workpiece. Has a slide to pull it in, and a door to open/close """

    def __init__(self, name: str):
        # name needs to be unique to refer to it
        super(MPOOven, self).__init__(name)
        self.vgr_in = self.addInPort("vgr_in")
        # gripper is transport between oven and table saw
        self.gripper_in = self.addInPort("gripper_in")
        self.gripper_out = self.addOutPort("gripper_out")

        self.mqtt_in = self.addInPort("mqtt_in")  # input for MQTT messages (e.g. visual refresh request)
        self.mqtt_out = self.addOutPort("mqtt_out")

        self.state = BakeOvenState()

    def change_phase(self, new_phase: BakeOvenSteps):
        """ Wrapper for changing the phase and time associated with it, helps with logging """
        self.state.phase = new_phase
        self.state.delta_t = new_phase.timing
        logger.trace(f"{type(self).__name__} '{self.name}' phase changed to {new_phase}")
        self.state.visual_update_pending = True

    def get_visual_update_data(self) -> MqttMessage:
        """ Get visual update data for the animation, contains the action taken and the duration of that action left """
        message = MqttMessage()
        message.topic = "visualization/mpo/oven"
        duration = self.state.delta_t
        if duration == INFINITY: duration = None
        message.payload = {
            "action": self.state.phase.value,
            "duration": duration,  # should be equal to the timing of the phase
            "workpiece": self.state.workpiece.to_dict() if self.state.workpiece else None,
        }
        return message

    def get_status_message(self, use_next_phase=True) -> MqttMessage:
        phase = self.state.phase.next() if use_next_phase else self.state.phase
        active = 0 if (phase == BakeOvenSteps.IDLE) else 1
        code = 2 if active == 1 else 1

        message = MqttMessage()
        message.topic = "f/i/state/mpo"
        message.payload = {
            "active": active,
            "code": code,
            "description": '',
            "station": 'mpo',
            "ts": get_timestamp(),
        }
        return message

    def extTransition(self, inputs):
        self.state.delta_t -= self.elapsed

        if self.mqtt_in in inputs:
            msg = inputs[self.mqtt_in][0]
            if msg.topic == "simulation/ctrl/all" and msg.payload.get("action") == "refresh":
                self.state.status_requested = True
                self.state.visual_update_pending = True
            return self.state

        logger.trace(f"BakeOven '{self.name}' received: {inputs}")

        if not (self.state.phase == BakeOvenSteps.IDLE or self.state.phase == BakeOvenSteps.AWAIT_PICKUP):
          logger.error(f"{type(self).__name__} '{self.name}' received workpiece whilst not IDLE, workpiece will be dropped")
          return self.state

        elif self.vgr_in in inputs:
            new_workpiece = inputs[self.vgr_in][0]
            logger.trace(f"BakeOven '{self.name}' received: {new_workpiece}")
            self.state.workpiece = new_workpiece
            # update step
            self.change_phase(BakeOvenSteps.DETECTION)

        elif self.gripper_in in inputs:
            # gripper is here to pick up the workpiece
            self.change_phase(BakeOvenSteps.OUTPUT_WP)

        return self.state  # important, return state

    def timeAdvance(self):
        if self.state.visual_update_pending or self.state.status_requested:
            return 0.0
        return self.state.delta_t

    def outputFnc(self):
        if self.state.visual_update_pending:
            return {self.mqtt_out: [self.get_visual_update_data()]}

        if self.state.status_requested:
            return {self.mqtt_out: [self.get_status_message(use_next_phase=False)]}

        if self.state.phase == BakeOvenSteps.DETECTION:
            return {self.mqtt_out: [get_mpo_started_msg(), self.get_status_message()]} # output that we've started

        elif self.state.phase == BakeOvenSteps.CLOSE_DOOR:
            return {self.gripper_out: {}, self.mqtt_out: [self.get_status_message()]} # signal gripper to come here

        elif self.state.phase == BakeOvenSteps.OUTPUT_WP:
            return {self.gripper_out: [self.state.workpiece], self.mqtt_out: [self.get_status_message()]}

        return {self.mqtt_out: [self.get_status_message()]}

    def intTransition(self):
        if self.state.visual_update_pending:
            self.state.visual_update_pending = False
            return self.state

        if self.state.status_requested:
            self.state.status_requested = False
            return self.state # no transition

        if self.state.phase == BakeOvenSteps.OUTPUT_WP:
            self.state.workpiece = None  # sent to output

        elif self.state.phase == BakeOvenSteps.BAKE:
            self.state.workpiece.state = "BAKED"  # TODO: change

        # Go to the next phase
        self.change_phase(self.state.phase.next())
        return self.state