import copy from typing import Dict, List, Tuple from common.DeviceTypes import DeviceTypeEnum from common.tools.object_factory.ConfigRule import json_config_rule_set from context.proto.context_pb2 import DeviceDriverEnum, DeviceOperationalStatusEnum DEVICE_DISABLED = DeviceOperationalStatusEnum.DEVICEOPERATIONALSTATUS_DISABLED DEVICE_EMUPR_TYPE = DeviceTypeEnum.EMULATED_PACKET_ROUTER.value DEVICE_EMUOLS_TYPE = DeviceTypeEnum.EMULATED_OPTICAL_LINE_SYSTEM.value DEVICE_EMU_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_UNDEFINED] DEVICE_EMU_ADDRESS = '127.0.0.1' DEVICE_EMU_PORT = '0' DEVICE_PR_TYPE = DeviceTypeEnum.PACKET_ROUTER.value DEVICE_PR_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_OPENCONFIG] DEVICE_TAPI_TYPE = DeviceTypeEnum.OPTICAL_LINE_SYSTEM.value DEVICE_TAPI_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_TRANSPORT_API] DEVICE_P4_TYPE = DeviceTypeEnum.P4_SWITCH.value DEVICE_P4_DRIVERS = [DeviceDriverEnum.DEVICEDRIVER_P4] def json_device_id(device_uuid : str): return {'device_uuid': {'uuid': device_uuid}} def json_device( device_uuid : str, device_type : str, status : DeviceOperationalStatusEnum, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = [] ): return { 'device_id' : json_device_id(device_uuid), 'device_type' : device_type, 'device_config' : {'config_rules': copy.deepcopy(config_rules)}, 'device_operational_status': status, 'device_drivers' : copy.deepcopy(drivers), 'device_endpoints' : copy.deepcopy(endpoints), } def json_device_emulated_packet_router_disabled( device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_EMU_DRIVERS ): return json_device( device_uuid, DEVICE_EMUPR_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers) def json_device_emulated_tapi_disabled( device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_EMU_DRIVERS ): return json_device( device_uuid, DEVICE_EMUOLS_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers) def json_device_packetrouter_disabled( device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_PR_DRIVERS ): return json_device( device_uuid, DEVICE_PR_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers) def json_device_tapi_disabled( device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_TAPI_DRIVERS ): return json_device( device_uuid, DEVICE_TAPI_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers) def json_device_p4_disabled( device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = DEVICE_P4_DRIVERS ): return json_device( device_uuid, DEVICE_P4_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers) def json_device_connect_rules(address : str, port : int, settings : Dict = {}): return [ json_config_rule_set('_connect/address', address), json_config_rule_set('_connect/port', port), json_config_rule_set('_connect/settings', settings), ] def json_device_emulated_connect_rules( endpoint_descriptors : List[Tuple[str, str, List[int]]], address : str = DEVICE_EMU_ADDRESS, port : int = DEVICE_EMU_PORT ): settings = {'endpoints': [ {'uuid': endpoint_uuid, 'type': endpoint_type, 'sample_types': sample_types} for endpoint_uuid,endpoint_type,sample_types in endpoint_descriptors ]} return json_device_connect_rules(address, port, settings=settings)