import inspect from typing import Callable from UniTAP.dev.modules.opf.handlers.base import OpfHandlerBase, WrongOPFFunctionSignature from UniTAP.dev.modules.opf.parsers import OPFParametersParser from .opf_functions import * class OpfHandlerInternal(OpfHandlerBase): __FUNCTIONS_MAP = { 1: OPFFunctions.opf_1_handler, 2: OPFFunctions.opf_2_handler, 3: OPFFunctions.opf_3_handler, 4: OPFFunctions.opf_4_handler, 5: OPFFunctions.opf_5_handler, 6: OPFFunctions.opf_6_handler, 7: OPFFunctions.opf_7_handler, 8: OPFFunctions.opf_8_handler, 9: OPFFunctions.opf_9_handler, 10: OPFFunctions.opf_10_handler, 11: OPFFunctions.opf_11_handler, 12: OPFFunctions.opf_12_handler, 13: OPFFunctions.opf_13_handler, 14: OPFFunctions.opf_14_handler, 15: OPFFunctions.opf_15_handler, 16: OPFFunctions.opf_16_handler, 17: OPFFunctions.opf_17_handler, 18: OPFFunctions.opf_18_handler, 19: OPFFunctions.opf_19_handler, 20: OPFFunctions.opf_20_handler, 21: OPFFunctions.opf_21_handler, 101: OPFFunctions.opf_101_handler, 102: OPFFunctions.opf_102_handler, 103: OPFFunctions.opf_103_handler, 104: OPFFunctions.opf_104_handler, 105: OPFFunctions.opf_105_handler, 106: OPFFunctions.opf_106_handler, 107: OPFFunctions.opf_107_handler, 120: OPFFunctions.opf_120_handler, 121: OPFFunctions.opf_121_handler, 122: OPFFunctions.opf_122_handler, 123: OPFFunctions.opf_123_handler, 140: OPFFunctions.opf_140_handler, 141: OPFFunctions.opf_141_handler, 142: OPFFunctions.opf_142_handler, 143: OPFFunctions.opf_143_handler, 144: OPFFunctions.opf_144_handler, 145: OPFFunctions.opf_145_handler, 150: OPFFunctions.opf_150_handler, 161: OPFFunctions.opf_161_handler } __FUNCTIONS_MAP_AFTER = { 1: OPFFunctions.opf_1_after_handler } def __init__(self, port_tx, port_rx): super().__init__() self.__parser = OPFParametersParser() self.__tx = port_tx self.__rx = port_rx self.__callback_before_handling = {} self.__callback_after_handling = {} def handle(self, opf_id, reply_answer, *args) -> OPFDialogAnswer: opf_params = self.__parser.parse(opf_id, *args) user_opf_before_callback = self.__callback_before_handling.get(opf_id, None) user_opf_after_callback = self.__callback_after_handling.get(opf_id, None) internal_opf_after_callback = self.__FUNCTIONS_MAP_AFTER.get(opf_id, None) # # Callback before main callback # if user_opf_before_callback is not None: opf_before_result = user_opf_before_callback(*opf_params) # Interrupt OPF in case of fail "before" OPF if opf_before_result not in [OPFDialogAnswer.PASS, OPFDialogAnswer.PROCEED]: return reply_answer(opf_before_result) # # Main callback # OPFFunctions.dsc_content_library_path = self._dsc_content_library_path opf_internal_result = self.__FUNCTIONS_MAP.get(opf_id)(self.__tx, self.__rx, *opf_params) reply_answer(opf_internal_result) # # Callback after main callback # if user_opf_after_callback is not None: user_opf_after_callback(*opf_params) elif internal_opf_after_callback is not None: internal_opf_after_callback(self.__tx, self.__rx, *opf_params) return opf_internal_result def assign_function_before_handling_opf_id(self, opf_id: int, func: Callable): signature = inspect.getfullargspec(func) if signature.annotations['return'] != OPFDialogAnswer: raise WrongOPFFunctionSignature(f"Function MUST declare {OPFDialogAnswer} " f"as return value") self.__callback_before_handling[opf_id] = func def assign_function_after_handling_opf_id(self, opf_id: int, func: Callable): signature = inspect.getfullargspec(func) if signature.annotations['return'] != OPFDialogAnswer: raise WrongOPFFunctionSignature(f"Function MUST declare {OPFDialogAnswer} " f"as return value") self.__callback_after_handling[opf_id] = func