Loading test.pydeleted 100644 → 0 +0 −291 Original line number Diff line number Diff line import concurrent.futures, json, logging, operator import sys from typing import Any, Dict, List, Optional, Tuple, Union TypeResults = List[Tuple[str, str, int, List[str]]] # entity_name, action, num_ok, list[error] class DescriptorLoader: def __init__( self, descriptors : Optional[Union[str, Dict]] = None, descriptors_file : Optional[str] = None, num_workers : int = 1, ) -> None: print('holaa') descriptors = { "ietf-network-slice-service:network-slice-services": { "slo-sle-templates": { "slo-sle-template": [ { "id": "high-BW-template", "description": "take the highest BW forwarding path" }, { "id": "low-latency-template", "description": "lowest possible latency forwarding behavior" } ] }, "slice-service": [ { "id": "slice_d1_d2", "description": "example slice p2p between two devices", "slo-sle-template": "low-latency-template", "status": {}, "sdps": { "sdp": [ { "id": "1", "node-id": "4.4.4.4", "service-match-criteria": { "match-criterion": [ { "index": 1, "match-type": "ietf-nss:service-any-match", "target-connection-group-id": "matrix1" } ] }, "attachment-circuits": { "attachment-circuit": [ { "id": "ac1", "description": "AC1 connected to device 1", "ac-node-id": "4.4.4.4", "ac-tp-id": "to_HL3-1-2", "ac-tags": { "ac-tags": [ { "tag-type": "ietf-nss:vlan-id", "value": [ "100" ] } ] }, "status": {} } ] }, "status": {} }, { "id": "2", "node-id": "5.5.5.5", "service-match-criteria": { "match-criterion": [ { "index": 1, "match-type": "ietf-nss:service-any-match", "target-connection-group-id": "matrix1" } ] }, "attachment-circuits": { "attachment-circuit": [ { "id": "ac2", "description": "AC2 connected to device 2", "ac-node-id": "5.5.5.5", "ac-tp-id": "eth-1/0/22.111", "ac-tags": { "ac-tags": [ { "tag-type": "ietf-nss:vlan-id", "value": [ "111" ] } ] }, "status": {} } ] }, "status": {} } ] }, "connection-groups": { "connection-group": [ { "id": "matrix1", "connectivity-type": "ietf-nss:point-to-point", "service-slo-sle-policy": { "slo-policy": { "metric-bound": [ { "metric-type": "ietf-nss:one-way-delay-maximum", "metric-unit": "milliseconds", "bound": "10" } ] } }, "connectivity-construct": [ { "id": 1, "p2p-sender-sdp": "1", "p2p-receiver-sdp": "2", "status": {} }, { "id": 2, "p2p-sender-sdp": "2", "p2p-receiver-sdp": "1", "status": {} } ] } ] } } ] } } self.__descriptors = json.loads(descriptors) if isinstance(descriptors, str) else descriptors self.__slices = self.__descriptors.get('slices' , []) #data = self.__descriptors.get('data' , {}) #Coge de la file el campo slices #ns_slice_service = data.get('ietf-network-slice-service:network-slice-services', {}) #self.__slices= ns_slice_service.get('slice-service', []) #nuevas slices self.__slices = self.__descriptors.get('ietf-network-slice-service:network-slice-services', {}) #hasta aqui bien print(type(self.__slices)) print(type(self.__slices["slice-service"])) json_out = {"slices": [ { "slice_id": { "context_id": {"context_uuid": {"uuid": "admin"}}, "slice_uuid": {} }, "name": {}, "slice_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "/settings", "resource_value": { "address_families": ["IPV4"], "bgp_as": 65000, "bgp_route_target": "65000:333", "mtu": 1512 }}} ]}, "slice_constraints": [ {"sla_capacity": {"capacity_gbps": 20.0}}, {"sla_availability": {"availability": 20.0, "num_disjoint_paths": 1, "all_active": True}}, {"sla_isolation": {"isolation_level": [0]}} ], "slice_endpoint_ids": [ ], "slice_status": {"slice_status": 1} } ]} for slice_service in self.__slices["slice-service"]: for slice in json_out["slices"]: slice["slice_id"]["slice_uuid"] = { "uuid": slice_service["id"]} slice["name"] = slice_service["description"] sdp = slice_service["sdps"]["sdp"] print(sdp) for elemento in sdp: slice["slice_config"]["config_rules"].append( {"action": 1, "custom": {"resource_key": "/device[R1]/endpoint[1/2]/settings", "resource_value": { "router_id": elemento.get("node-id",[]), "sub_interface_index": 0, "vlan_id": 111 }}}) slice["slice_endpoint_ids"].append({ "device_id": {"device_uuid": {"uuid": elemento["id"]}}, "endpoint_uuid": {"uuid": "1/2"}, "topology_id": {"context_id": {"context_uuid": {"uuid": "admin"}}, "topology_uuid": {"uuid": "admin"}} }) attcircuits = elemento["attachment-circuits"]["attachment-circuit"] for attcircuit in attcircuits: slice["slice_constraints"].append({"endpoint_location": { "endpoint_id": {"device_id": {"device_uuid": {"uuid": attcircuit["id"]}}, "endpoint_uuid": {"uuid": attcircuit["ac-tp-id"]}}, "location": {"region": "4"} }}) # Convertir a JSON de salida #json_output = json.dumps(json_out, indent=2) self.__slices = json_out.get('slices' , []) print(self.__slices) self.__results : TypeResults = list() @property def slices(self) -> Dict[str, List[Dict]]: _slices = {} for slice_ in self.__slices: context_uuid = "admin" _slices.setdefault(context_uuid, []).append(slice_) #no tenemos context_uuid en este formato, lo meto a mano? self.__slices = [format_slice_custom_config_rules (slice_ ) for slice_ in self.__slices ] return _slices @property def num_slices(self) -> Dict[str, int]: _num_slices = {} for slice_ in self.__slices: context_uuid = slice_['slice_id']['context_id']['context_uuid']['uuid'] _num_slices[context_uuid] = _num_slices.get(context_uuid, 0) + 1 return _num_slices # Format CustomConfigRules in Devices, Services and Slices provided in JSON format def process(self) -> TypeResults: # Format CustomConfigRules in Devices, Services and Slices provided in JSON format self.__slices = [format_slice_custom_config_rules (slice_ ) for slice_ in self.__slices ] # Context and Topology require to create the entity first, and add devices, links, services, # slices, etc. in a second stage. print(self.__results) return self.__results #UTILIZA LA FUNCION FORMAT_CUSTOM_CONFIG_RULES #cambio TypeResourceValue = Union[str, int, bool, float, dict, list] ''' def format_custom_config_rules(config_rules : List[Dict]) -> List[Dict]: for config_rule in config_rules: # if 'custom' not in config_rule: continue #suponemos que siempre son custom, quitamos esta linea custom_resource_value : TypeResourceValue = config_rule['attachment-circuits']['attachment-circuit'] if isinstance(custom_resource_value, (dict, list)): custom_resource_value = json.dumps(custom_resource_value, sort_keys=True, indent=0) config_rule['attachment-circuits']['attachment-circuit'] = custom_resource_value elif not isinstance(custom_resource_value, str): config_rule['attachment-circuits']['attachment-circuit'] = str(custom_resource_value) return config_rules def format_slice_custom_config_rules(slice_ : Dict) -> Dict: #donde cojo los config_rules #las config_rules parecen estar en ACs? sdps = slice_.get('sdps', {}) config_rules = sdps.get('sdp', {}) print("Tipo de sdp:", type(config_rules)) print("sdps:", config_rules) #Despues de mucho revisar, el campo sdp es List[Dict] asique voy a pasar directamente sdp a la funcion format_custom_config_rules(config_rules) config_rules = format_custom_config_rules(config_rules) slice_['sdps']['sdp'] = config_rules return slice_ ''' def format_custom_config_rules(config_rules : List[Dict]) -> List[Dict]: for config_rule in config_rules: if 'custom' not in config_rule: continue custom_resource_value : TypeResourceValue = config_rule['custom']['resource_value'] if isinstance(custom_resource_value, (dict, list)): custom_resource_value = json.dumps(custom_resource_value, sort_keys=True, indent=0) config_rule['custom']['resource_value'] = custom_resource_value return config_rules def format_slice_custom_config_rules(slice_ : Dict) -> Dict: config_rules = slice_.get('slice_config', {}).get('config_rules', []) config_rules = format_custom_config_rules(config_rules) slice_['slice_config']['config_rules'] = config_rules return slice_ # Crear una instancia de la clase mi_cargador = DescriptorLoader() results = mi_cargador.process() Loading
test.pydeleted 100644 → 0 +0 −291 Original line number Diff line number Diff line import concurrent.futures, json, logging, operator import sys from typing import Any, Dict, List, Optional, Tuple, Union TypeResults = List[Tuple[str, str, int, List[str]]] # entity_name, action, num_ok, list[error] class DescriptorLoader: def __init__( self, descriptors : Optional[Union[str, Dict]] = None, descriptors_file : Optional[str] = None, num_workers : int = 1, ) -> None: print('holaa') descriptors = { "ietf-network-slice-service:network-slice-services": { "slo-sle-templates": { "slo-sle-template": [ { "id": "high-BW-template", "description": "take the highest BW forwarding path" }, { "id": "low-latency-template", "description": "lowest possible latency forwarding behavior" } ] }, "slice-service": [ { "id": "slice_d1_d2", "description": "example slice p2p between two devices", "slo-sle-template": "low-latency-template", "status": {}, "sdps": { "sdp": [ { "id": "1", "node-id": "4.4.4.4", "service-match-criteria": { "match-criterion": [ { "index": 1, "match-type": "ietf-nss:service-any-match", "target-connection-group-id": "matrix1" } ] }, "attachment-circuits": { "attachment-circuit": [ { "id": "ac1", "description": "AC1 connected to device 1", "ac-node-id": "4.4.4.4", "ac-tp-id": "to_HL3-1-2", "ac-tags": { "ac-tags": [ { "tag-type": "ietf-nss:vlan-id", "value": [ "100" ] } ] }, "status": {} } ] }, "status": {} }, { "id": "2", "node-id": "5.5.5.5", "service-match-criteria": { "match-criterion": [ { "index": 1, "match-type": "ietf-nss:service-any-match", "target-connection-group-id": "matrix1" } ] }, "attachment-circuits": { "attachment-circuit": [ { "id": "ac2", "description": "AC2 connected to device 2", "ac-node-id": "5.5.5.5", "ac-tp-id": "eth-1/0/22.111", "ac-tags": { "ac-tags": [ { "tag-type": "ietf-nss:vlan-id", "value": [ "111" ] } ] }, "status": {} } ] }, "status": {} } ] }, "connection-groups": { "connection-group": [ { "id": "matrix1", "connectivity-type": "ietf-nss:point-to-point", "service-slo-sle-policy": { "slo-policy": { "metric-bound": [ { "metric-type": "ietf-nss:one-way-delay-maximum", "metric-unit": "milliseconds", "bound": "10" } ] } }, "connectivity-construct": [ { "id": 1, "p2p-sender-sdp": "1", "p2p-receiver-sdp": "2", "status": {} }, { "id": 2, "p2p-sender-sdp": "2", "p2p-receiver-sdp": "1", "status": {} } ] } ] } } ] } } self.__descriptors = json.loads(descriptors) if isinstance(descriptors, str) else descriptors self.__slices = self.__descriptors.get('slices' , []) #data = self.__descriptors.get('data' , {}) #Coge de la file el campo slices #ns_slice_service = data.get('ietf-network-slice-service:network-slice-services', {}) #self.__slices= ns_slice_service.get('slice-service', []) #nuevas slices self.__slices = self.__descriptors.get('ietf-network-slice-service:network-slice-services', {}) #hasta aqui bien print(type(self.__slices)) print(type(self.__slices["slice-service"])) json_out = {"slices": [ { "slice_id": { "context_id": {"context_uuid": {"uuid": "admin"}}, "slice_uuid": {} }, "name": {}, "slice_config": {"config_rules": [ {"action": 1, "custom": {"resource_key": "/settings", "resource_value": { "address_families": ["IPV4"], "bgp_as": 65000, "bgp_route_target": "65000:333", "mtu": 1512 }}} ]}, "slice_constraints": [ {"sla_capacity": {"capacity_gbps": 20.0}}, {"sla_availability": {"availability": 20.0, "num_disjoint_paths": 1, "all_active": True}}, {"sla_isolation": {"isolation_level": [0]}} ], "slice_endpoint_ids": [ ], "slice_status": {"slice_status": 1} } ]} for slice_service in self.__slices["slice-service"]: for slice in json_out["slices"]: slice["slice_id"]["slice_uuid"] = { "uuid": slice_service["id"]} slice["name"] = slice_service["description"] sdp = slice_service["sdps"]["sdp"] print(sdp) for elemento in sdp: slice["slice_config"]["config_rules"].append( {"action": 1, "custom": {"resource_key": "/device[R1]/endpoint[1/2]/settings", "resource_value": { "router_id": elemento.get("node-id",[]), "sub_interface_index": 0, "vlan_id": 111 }}}) slice["slice_endpoint_ids"].append({ "device_id": {"device_uuid": {"uuid": elemento["id"]}}, "endpoint_uuid": {"uuid": "1/2"}, "topology_id": {"context_id": {"context_uuid": {"uuid": "admin"}}, "topology_uuid": {"uuid": "admin"}} }) attcircuits = elemento["attachment-circuits"]["attachment-circuit"] for attcircuit in attcircuits: slice["slice_constraints"].append({"endpoint_location": { "endpoint_id": {"device_id": {"device_uuid": {"uuid": attcircuit["id"]}}, "endpoint_uuid": {"uuid": attcircuit["ac-tp-id"]}}, "location": {"region": "4"} }}) # Convertir a JSON de salida #json_output = json.dumps(json_out, indent=2) self.__slices = json_out.get('slices' , []) print(self.__slices) self.__results : TypeResults = list() @property def slices(self) -> Dict[str, List[Dict]]: _slices = {} for slice_ in self.__slices: context_uuid = "admin" _slices.setdefault(context_uuid, []).append(slice_) #no tenemos context_uuid en este formato, lo meto a mano? self.__slices = [format_slice_custom_config_rules (slice_ ) for slice_ in self.__slices ] return _slices @property def num_slices(self) -> Dict[str, int]: _num_slices = {} for slice_ in self.__slices: context_uuid = slice_['slice_id']['context_id']['context_uuid']['uuid'] _num_slices[context_uuid] = _num_slices.get(context_uuid, 0) + 1 return _num_slices # Format CustomConfigRules in Devices, Services and Slices provided in JSON format def process(self) -> TypeResults: # Format CustomConfigRules in Devices, Services and Slices provided in JSON format self.__slices = [format_slice_custom_config_rules (slice_ ) for slice_ in self.__slices ] # Context and Topology require to create the entity first, and add devices, links, services, # slices, etc. in a second stage. print(self.__results) return self.__results #UTILIZA LA FUNCION FORMAT_CUSTOM_CONFIG_RULES #cambio TypeResourceValue = Union[str, int, bool, float, dict, list] ''' def format_custom_config_rules(config_rules : List[Dict]) -> List[Dict]: for config_rule in config_rules: # if 'custom' not in config_rule: continue #suponemos que siempre son custom, quitamos esta linea custom_resource_value : TypeResourceValue = config_rule['attachment-circuits']['attachment-circuit'] if isinstance(custom_resource_value, (dict, list)): custom_resource_value = json.dumps(custom_resource_value, sort_keys=True, indent=0) config_rule['attachment-circuits']['attachment-circuit'] = custom_resource_value elif not isinstance(custom_resource_value, str): config_rule['attachment-circuits']['attachment-circuit'] = str(custom_resource_value) return config_rules def format_slice_custom_config_rules(slice_ : Dict) -> Dict: #donde cojo los config_rules #las config_rules parecen estar en ACs? sdps = slice_.get('sdps', {}) config_rules = sdps.get('sdp', {}) print("Tipo de sdp:", type(config_rules)) print("sdps:", config_rules) #Despues de mucho revisar, el campo sdp es List[Dict] asique voy a pasar directamente sdp a la funcion format_custom_config_rules(config_rules) config_rules = format_custom_config_rules(config_rules) slice_['sdps']['sdp'] = config_rules return slice_ ''' def format_custom_config_rules(config_rules : List[Dict]) -> List[Dict]: for config_rule in config_rules: if 'custom' not in config_rule: continue custom_resource_value : TypeResourceValue = config_rule['custom']['resource_value'] if isinstance(custom_resource_value, (dict, list)): custom_resource_value = json.dumps(custom_resource_value, sort_keys=True, indent=0) config_rule['custom']['resource_value'] = custom_resource_value return config_rules def format_slice_custom_config_rules(slice_ : Dict) -> Dict: config_rules = slice_.get('slice_config', {}).get('config_rules', []) config_rules = format_custom_config_rules(config_rules) slice_['slice_config']['config_rules'] = config_rules return slice_ # Crear una instancia de la clase mi_cargador = DescriptorLoader() results = mi_cargador.process()
