Remove unnecessary docstrings and clean up comments in cisco_connector and dump_templates modules (d49046bd) · Commits · TFS / nsc

src/realizer/tfs/helpers/cisco_connector.py

+1 −206

Original line number	Diff line number	Diff line
		@@ -11,7 +11,6 @@
		# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
		# See the License for the specific language governing permissions and
		# limitations under the License.
		""" This file is an original contribution from Telefonica Innovación Digital S.L. """

		# This file is an original contribution from Telefonica Innovación Digital S.L.

		@@ -24,10 +23,6 @@ class cisco_connector():
		self.configs=configs

		def execute_commands(self, commands):
		"""Execute a list of commands on the Cisco device via SSH.
		Connects to the device and sends configuration commands.
		This method is used to send configuration commands to the Cisco device."""

		try:
		# Configuración del dispositivo
		device = {
		@@ -51,10 +46,6 @@ class cisco_connector():
		logging.error("Failed to execute commands on %s: %s",self.address, str(e))

		def create_command_template(self, config):
		"""Create a command template for configuring L2VPN on a Cisco device.
		This method generates the necessary commands to configure L2VPN
		based on the provided configuration parameters."""

		commands = [
		"l2vpn",
		f"pw-class l2vpn_vpws_profile_example_{config['number']}",
		@@ -84,10 +75,6 @@ class cisco_connector():
		return commands

		def full_create_command_template(self):
		"""Create a full command template for configuring L2VPN on a Cisco device.
		This method generates all necessary commands to configure L2VPN based on
		the provided configurations."""

		commands =[]
		for config in self.configs:
		commands_temp = self.create_command_template(config)
		@@ -97,202 +84,10 @@ class cisco_connector():
		return commands

		def create_command_template_delete(self):
		"""Create a command template for deleting L2VPN configuration on a Cisco device.
		This method generates the necessary commands to remove L2VPN configurations."""

		commands = [
		"no l2vpn",
		]

		commands.append("commit")
		commands.append("end")
		return commands

		def send_network_slice_request(data: str, action: str = "create") -> dict:

		url = 'http://192.168.1.143:9090/api/resource-allocation/transport-network-slice-l3'
		headers = {'Content-Type': 'application/json'}
		try:
		if action == "delete":
		data = {
		"ietf-network-slice-service:network-slice-services": {
		"slice-service": [
		{
		"id": data
		}
		]
		}
		}
		response = requests.delete(url, headers=headers, json=data, timeout=15)
		elif action == "create":
		response = requests.post(url, headers=headers, json=data, timeout=15)
		else:
		raise ValueError("Invalid action. Use 'create' or 'delete'.")
		except requests.exceptions.RequestException as e:
		logging.error(f"HTTP request failed: {e}")
		return {}

		if response.ok:
		return response.json()
		else:
		print(f"Request failed with status code {response.status_code}: {response.text}")
		response.raise_for_status()

		def group_block(group, action, group_id_override=None, node = None):
		active = "true" if action == 'create' else "false"
		group_id = group_id_override if group_id_override is not None else group["digital_sub_carriers_group_id"]
		if node == "leaf":
		return {
		"digital_sub_carriers_group_id": group_id,
		"digital_sub_carrier_id": [
		{'sub_carrier_id': 1, 'active': active},
		{'sub_carrier_id': 2, 'active': active},
		{'sub_carrier_id': 3, 'active': active},
		{'sub_carrier_id': 4, 'active': active}
		]
		}
		else:
		return {
		"digital_sub_carriers_group_id": group_id,
		"digital_sub_carrier_id": [
		{
		"sub_carrier_id": sid,
		"active": active,
		}
		for sid in group["subcarrier-id"]
		]
		}

		def generate_rules(connectivity_service, intent, action):
		src_name = connectivity_service.get("source", "FALTA VALOR")
		dest_list = connectivity_service.get("destination", ["FALTA VALOR"])
		dest_str = ",".join(dest_list)
		config_rules = []

		network_slice_uuid_str = f"{src_name}_to_{dest_str}"
		tunnel_uuid = str(uuid.uuid5(uuid.NAMESPACE_DNS, network_slice_uuid_str))
		provisionamiento = {
		"network-slice-uuid": network_slice_uuid_str,
		"viability": True,
		"actions": []
		}

		attributes = connectivity_service["connectivity-service"]["tapi-connectivity:connectivity-service"]["connection"][0]["optical-connection-attributes"]
		groups = attributes["subcarrier-attributes"]["digital-subcarrier-group"]
		# central_freq = (attributes["central-frequency"])
		# tx_power = (attributes["Tx-power"])
		# spacing = attributes["digital-subcarrier-spacing"]
		operational_mode = attributes["modulation"]["operational-mode"]
		# port = attributes["modulation"]["port"]
		hub_groups = [
		group_block(group, action, group_id_override=index + 1)
		for index, group in enumerate(groups)
		]
		hub = {
		"name": "channel-1",
		"frequency": 195000000,
		"target_output_power": 0,
		"operational_mode": operational_mode,
		"operation" : "merge",
		"digital_sub_carriers_group": hub_groups
		}

		leaves = []
		for dest, group in zip(connectivity_service["destination"], groups):
		if dest == "T1.1":
		name = "channel-1"
		freq = 195006250
		if dest == "T1.2":
		name = "channel-3"
		freq = 195018750
		if dest == "T1.3":
		name = "channel-5"
		freq = 195031250
		leaf = {
		"name": name,
		"frequency": freq,
		"target_output_power": group["Tx-power"],
		"operational_mode": int(group["operational-mode"]),
		"operation" : "merge",
		"digital_sub_carriers_group": [group_block(group, action, group_id_override=1, node = "leaf")]
		}

		leaves.append(leaf)

		final_json = {"components": [hub] + leaves}
		if action == 'create':
		provisionamiento["actions"].append({
		"type": "XR_AGENT_ACTIVATE_TRANSCEIVER",
		"layer": "OPTICAL",
		"content": final_json,
		"controller-uuid": "IPoWDM Controller"
		})

		nodes = {}
		sdp_list = intent['ietf-network-slice-service:network-slice-services']['slice-service'][0]['sdps']['sdp']

		for sdp in sdp_list:
		node = sdp['node-id']
		attachments = sdp['attachment-circuits']['attachment-circuit']
		for ac in attachments:
		ip = ac.get('ac-ipv4-address', None)
		prefix = ac.get('ac-ipv4-prefix-length', None)
		vlan = 500
		nodes[node] = {
		"ip-address": ip,
		"ip-mask": prefix,
		"vlan-id": vlan
		}

		provisionamiento["actions"].append({
		"type": "CONFIG_VPNL3",
		"layer": "IP",
		"content": {
		"tunnel-uuid": tunnel_uuid,
		"src-node-uuid": src_name,
		"src-ip-address": nodes[src_name]["ip-address"],
		"src-ip-mask": str(nodes[src_name]["ip-mask"]),
		"src-vlan-id": nodes[src_name]["vlan-id"],
		"dest1-node-uuid": dest_list[0],
		"dest1-ip-address": nodes[dest_list[0]]["ip-address"],
		"dest1-ip-mask": str(nodes[dest_list[0]]["ip-mask"]),
		"dest1-vlan-id": nodes[dest_list[0]]["vlan-id"],
		"dest2-node-uuid": dest_list[1],
		"dest2-ip-address": nodes[dest_list[1]]["ip-address"],
		"dest2-ip-mask": str(nodes[dest_list[1]]["ip-mask"]),
		"dest2-vlan-id": nodes[dest_list[1]]["vlan-id"],
		"dest3-node-uuid": dest_list[2],
		"dest3-ip-address": nodes[dest_list[2]]["ip-address"],
		"dest3-ip-mask": str(nodes[dest_list[2]]["ip-mask"]),
		"dest3-vlan-id": nodes[dest_list[2]]["vlan-id"]
		},
		"controller-uuid": "IP Controller"
		})

		config_rules.append(provisionamiento)
		else:
		nodes = []
		nodes.append(src_name)
		for dst in dest_list: nodes.append(dst)
		aux = tunnel_uuid + '-' + src_name + '-' + '-'.join(dest_list)
		provisionamiento["actions"].append({
		"type": "DEACTIVATE_XR_AGENT_TRANSCEIVER",
		"layer": "OPTICAL",
		"content": final_json,
		"controller-uuid": "IPoWDM Controller",
		"uuid" : aux,
		"nodes": nodes
		})
		config_rules.append(provisionamiento)

		return config_rules

		def get_sip_from_name(context, node_name):

		context = context.json()
		topologies = context.get("tapi-topology:topology-context", {}).get("topology", [])
		for topology in topologies:
		nodes = topology.get("nodes", [])
		for node in nodes:
		if node.get("name") == node_name:
		return node.get("uuid")
		return None

src/utils/dump_templates.py

+0 −2

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

		# This file is an original contribution from Telefonica Innovación Digital S.L.
		""" This file contains constants used throughout the NSC application. """

		import json, os
		from src.config.constants import TEMPLATES_PATH
		@@ -26,7 +25,6 @@ def dump_templates(nbi_file, ietf_file, realizer_file):
		if not current_app.config["DUMP_TEMPLATES"]:
		return


		templates = {
		"nbi_template.json": nbi_file,
		"ietf_template.json": ietf_file,