Source code for genericroboticarm.sila_server.feature_implementations.labwaretransfermanipulatorcontroller_impl

# Generated by sila2.code_generator; sila2.__version__: 0.12.2
from __future__ import annotations

from datetime import timedelta, datetime
from typing import TYPE_CHECKING, List

from sila2.server import MetadataDict, ObservableCommandInstance
from ..generated.intermediateactionplanning import IntermediateActionPlanningFeature
from ..generated.labwaretransfermanipulatorcontroller import (
    GetLabware_Responses,
    HandoverPosition,
    LabwareTransferManipulatorControllerBase,
    PositionIndex,
    PrepareForInput_Responses,
    PrepareForOutput_Responses,
    PutLabware_Responses,
    InvalidCommandSequence,
)

from genericroboticarm.robo_APIs import InteractiveRobot, InteractiveTransfer

if TYPE_CHECKING:
    from ..server import Server




class LabwareTransferManipulatorControllerImpl(LabwareTransferManipulatorControllerBase):
    robot : InteractiveRobot

    def __init__(self, parent_server: Server) -> None:
        super().__init__(parent_server=parent_server)
        assert isinstance(parent_server.robot, InteractiveTransfer)
        self.robot = parent_server.robot

        # Default lifetime of observable command instances. Possible values:
        # None: Command instance is valid and stored in memory until server shutdown
        # datetime.timedelta: Command instance is deleted after this duration, can be increased during command runtime
        self.PrepareForInput_default_lifetime_of_execution = timedelta(minutes=30)
        self.PrepareForOutput_default_lifetime_of_execution = timedelta(minutes=30)
        self.PutLabware_default_lifetime_of_execution = timedelta(minutes=30)
        self.GetLabware_default_lifetime_of_execution = timedelta(minutes=30)



    def _position_name(self, handover_position: HandoverPosition) -> str:
        """The name of the handover site as the robot knows it, for the console output."""
        return self.robot.site_to_position_identifier(
            handover_position.Position, handover_position.SubPosition - 1
        )




    def get_AvailableHandoverPositions(self, *, metadata: MetadataDict) -> List[HandoverPosition]:
        return self.robot.available_handover_positions




    def get_NumberOfInternalPositions(self, *, metadata: MetadataDict) -> int:
        return self.robot.number_internal_positions




    def get_AvailableIntermediateActions(self, *, metadata: MetadataDict) -> List[str]:
        available = self.robot.available_intermediate_actions
        return available




    def PrepareForInput(
        self,
        HandoverPosition: HandoverPosition,
        InternalPosition: PositionIndex,
        LabwareType: str,
        LabwareUniqueID: str,
        *,
        metadata: MetadataDict,
        instance: ObservableCommandInstance,
    ) -> PrepareForInput_Responses:
        # set execution status from `waiting` to `running`
        with self.parent_server.robot_command():
            instance.begin_execution()
            print(f"{datetime.now()}: Preparing to take plate at {self._position_name(HandoverPosition)}")
            # preparing to receive a plate while one is still held would drop the held
            # plate during the subsequent pick, so refuse it as an invalid command sequence
            if self.robot.currently_gripping_plate:
                self.robot.cancel_transfer()
                raise InvalidCommandSequence(
                    "Cannot prepare for input: the gripper is already holding a plate. "
                    "Recover the held plate before transferring another one."
                )
            # the client may have announced the intermediate actions of the following GetLabware,
            # so that the robot can already do its share of them while the other device prepares.
            # Announcing nothing also clears what a previous transfer announced.
            announcement = metadata.get(IntermediateActionPlanningFeature["PlannedIntermediateActions"])
            self.robot.announce_intermediate_actions(
                announcement.IntermediateActions if announcement else []
            )
            preparation_succeeded = self.robot.prepare_for_input(
                internal_pos=InternalPosition,
                device=HandoverPosition.Position,
                position=HandoverPosition.SubPosition,
                plate_type=LabwareType,
            )
            if not preparation_succeeded:
                self.robot.cancel_transfer()
                raise InvalidCommandSequence("Robot failed to prepare for input.")




    def PrepareForOutput(
        self,
        HandoverPosition: HandoverPosition,
        InternalPosition: PositionIndex,
        *,
        metadata: MetadataDict,
        instance: ObservableCommandInstance,
    ) -> PrepareForOutput_Responses:
        # set execution status from `waiting` to `running`
        with self.parent_server.robot_command():
            instance.begin_execution()
            print(f"{datetime.now()}: Preparing to place plate at {self._position_name(HandoverPosition)}")
            preparation_succeeded = self.robot.prepare_for_output(
                internal_pos=InternalPosition,
                device=HandoverPosition.Position,
                position=HandoverPosition.SubPosition,
            )
            if not preparation_succeeded:
                self.robot.cancel_transfer()
                raise InvalidCommandSequence("Robot failed to prepare for output.")




    def PutLabware(
        self,
        HandoverPosition: HandoverPosition,
        IntermediateActions: List[str],
        *,
        metadata: MetadataDict,
        instance: ObservableCommandInstance,
    ) -> PutLabware_Responses:
        # set execution status from `waiting` to `running`
        with self.parent_server.robot_command():
            instance.begin_execution()
            print(f"{datetime.now()}: Delivering plate at {self._position_name(HandoverPosition)}")

            self.robot.put_labware(
                intermediate_actions=IntermediateActions,
                device=HandoverPosition.Position,
                position=HandoverPosition.SubPosition - 1,
            )




    def GetLabware(
        self,
        HandoverPosition: HandoverPosition,
        IntermediateActions: List[str],
        *,
        metadata: MetadataDict,
        instance: ObservableCommandInstance,
    ) -> GetLabware_Responses:
        # set execution status from `waiting` to `running`
        with self.parent_server.robot_command():
            instance.begin_execution()
            print(f"{datetime.now()}: Taking plate at {self._position_name(HandoverPosition)}")

            self.robot.get_labware(
                intermediate_actions=IntermediateActions,
                device=HandoverPosition.Position,
                position=HandoverPosition.SubPosition - 1,
            )