Common - Object Factory: (7271b2be) · Commits · TFS / controller

src/common/tools/object_factory/Device.py

+29 −23

Original line number	Diff line number	Diff line
		@@ -13,7 +13,7 @@
		# limitations under the License.

		import copy
		from typing import Dict, List, Tuple
		from typing import Dict, List, Optional, Tuple
		from common.DeviceTypes import DeviceTypeEnum
		from common.proto.context_pb2 import DeviceDriverEnum, DeviceOperationalStatusEnum
		from common.tools.object_factory.ConfigRule import json_config_rule_set
		@@ -50,10 +50,10 @@ 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] = []
		device_uuid : str, device_type : str, status : DeviceOperationalStatusEnum, name : Optional[str] = None,
		endpoints : List[Dict] = [], config_rules : List[Dict] = [], drivers : List[Dict] = []
		):
		return {
		result = {
		'device_id' : json_device_id(device_uuid),
		'device_type' : device_type,
		'device_config' : {'config_rules': copy.deepcopy(config_rules)},
		@@ -61,74 +61,80 @@ def json_device(
		'device_drivers' : copy.deepcopy(drivers),
		'device_endpoints' : copy.deepcopy(endpoints),
		}
		if name is not None: result['name'] = name
		return result

		def json_device_emulated_packet_router_disabled(
		device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		device_uuid : str, name : Optional[str] = None, 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,
		device_uuid, DEVICE_EMUPR_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
		drivers=drivers)

		def json_device_emulated_tapi_disabled(
		device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		device_uuid : str, name : Optional[str] = None, 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,
		device_uuid, DEVICE_EMUOLS_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
		drivers=drivers)

		def json_device_emulated_datacenter_disabled(
		device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		drivers : List[Dict] = DEVICE_EMU_DRIVERS
		):
		return json_device(
		device_uuid, DEVICE_EMUDC_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules,
		device_uuid, DEVICE_EMUDC_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
		drivers=drivers)

		def json_device_packetrouter_disabled(
		device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		device_uuid : str, name : Optional[str] = None, 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)
		device_uuid, DEVICE_PR_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
		drivers=drivers)

		def json_device_tapi_disabled(
		device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		device_uuid : str, name : Optional[str] = None, 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)
		device_uuid, DEVICE_TAPI_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
		drivers=drivers)

		def json_device_xr_constellation_disabled(
		device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		drivers : List[Dict] = DEVICE_XR_CONSTELLATION_DRIVERS
		):
		return json_device(
		device_uuid, DEVICE_XR_CONSTELLATION_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules,
		drivers=drivers)
		device_uuid, DEVICE_XR_CONSTELLATION_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints,
		config_rules=config_rules, drivers=drivers)

		def json_device_microwave_disabled(
		device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		drivers : List[Dict] = DEVICE_MICROWAVE_DRIVERS
		):
		return json_device(
		device_uuid, DEVICE_MICROWAVE_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules,
		device_uuid, DEVICE_MICROWAVE_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
		drivers=drivers)

		def json_device_p4_disabled(
		device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		device_uuid : str, name : Optional[str] = None, 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)
		device_uuid, DEVICE_P4_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
		drivers=drivers)

		def json_device_tfs_disabled(
		device_uuid : str, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		device_uuid : str, name : Optional[str] = None, endpoints : List[Dict] = [], config_rules : List[Dict] = [],
		drivers : List[Dict] = DEVICE_TFS_DRIVERS
		):
		return json_device(
		device_uuid, DEVICE_TFS_TYPE, DEVICE_DISABLED, endpoints=endpoints, config_rules=config_rules, drivers=drivers)
		device_uuid, DEVICE_TFS_TYPE, DEVICE_DISABLED, name=name, endpoints=endpoints, config_rules=config_rules,
		drivers=drivers)

		def json_device_connect_rules(address : str, port : int, settings : Dict = {}):
		return [

src/common/tools/object_factory/Link.py

+5 −3

Original line number	Diff line number	Diff line
		@@ -13,7 +13,7 @@
		# limitations under the License.

		import copy
		from typing import Dict, List, Tuple
		from typing import Dict, List, Optional, Tuple

		def get_link_uuid(a_endpoint_id : Dict, z_endpoint_id : Dict) -> str:
		return '{:s}/{:s}=={:s}/{:s}'.format(
		@@ -23,8 +23,10 @@ def get_link_uuid(a_endpoint_id : Dict, z_endpoint_id : Dict) -> str:
		def json_link_id(link_uuid : str) -> Dict:
		return {'link_uuid': {'uuid': link_uuid}}

		def json_link(link_uuid : str, endpoint_ids : List[Dict]) -> Dict:
		return {'link_id': json_link_id(link_uuid), 'link_endpoint_ids': copy.deepcopy(endpoint_ids)}
		def json_link(link_uuid : str, endpoint_ids : List[Dict], name : Optional[str] = None) -> Dict:
		result = {'link_id': json_link_id(link_uuid), 'link_endpoint_ids': copy.deepcopy(endpoint_ids)}
		if name is not None: result['name'] = name
		return result

		def compose_link(endpoint_a, endpoint_z) -> Tuple[Dict, Dict]:
		link_uuid = get_link_uuid(endpoint_a['endpoint_id'], endpoint_z['endpoint_id'])

Admin message