Merge branch 'dev/code-restruture' into 'develop'

# Copyright 2022-2025 ETSI SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# 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.

import os

import logging

from flask import Flask

from flask_restx import Api

from flask_cors import CORS

from swagger.tfs_namespace import tfs_ns

from swagger.ixia_namespace import ixia_ns

from src.Constants import NSC_PORT, WEBUI_DEPLOY

from src.config.constants import NSC_PORT

from src.webui.gui import gui_bp

from src.config.config import create_config

def create_app():

    """Factory para crear la app Flask con la configuración cargada"""

    app = Flask(__name__)

    app = create_config(app)

    CORS(app)

    # Configure logging to provide clear and informative log messages

    logging.basicConfig(

        level=app.config["LOGGING_LEVEL"],

        format="%(levelname)s - %(message)s"

    )

    # Create API instance

    api = Api(

        app,

    # Register namespaces

    api.add_namespace(tfs_ns, path="/tfs")

    api.add_namespace(ixia_ns, path="/ixia")

#gui_bp = Blueprint('gui', __name__, template_folder='templates')

if WEBUI_DEPLOY:

    app.secret_key = 'clave-secreta-dev' 

    if app.config["WEBUI_DEPLOY"]:

        app.secret_key = "clave-secreta-dev"

        app.register_blueprint(gui_bp)

    return app

# Solo arrancamos el servidor si ejecutamos el script directamente

if __name__ == "__main__":

    app = create_app()

    app.run(host="0.0.0.0", port=NSC_PORT, debug=True)

from src.config.constants import DATABASE_PATH

from src.utils.send_response import send_response

import os, json, logging

from flask import current_app

class Api:

    def __init__(self, slice_service):

        self.slice_service = slice_service

    def add_flow(self, intent):

        """

        Create a new transport network slice.

        Args:

            intent (dict): Network slice intent in 3GPP or IETF format

        Returns:

            Result of the Network Slice Controller (NSC) operation

        API Endpoint:

            POST /slice

        Raises:

            ValueError: If no transport network slices are found

            Exception: For unexpected errors during slice creation process

        """

        try:

            result = self.slice_service.nsc(intent)

            if not result:

                return send_response(False, code=404, message="No intents found")

            return send_response(

                True,

                code=201,

                data=result 

            )

        except Exception as e:

            # Handle unexpected errors

            return send_response(False, code=500, message=str(e))

    def get_flows(self,slice_id=None):

        """

        Retrieve transport network slice information.

        This method allows retrieving:

        - All transport network slices

        - A specific slice by its ID

        Args:

            slice_id (str, optional): Unique identifier of a specific slice. 

                                      Defaults to None.

        Returns:

            dict or list: 

            - If slice_id is provided: Returns the specific slice details

            - If slice_id is None: Returns a list of all slices

            - Returns an error response if no slices are found

        API Endpoint:

            GET /slice/{id}

        Raises:

            ValueError: If no transport network slices are found

            Exception: For unexpected errors during file processing

        """

        try:

            # Read slice database from JSON file

            with open(os.path.join(DATABASE_PATH, "slice_ddbb.json"), 'r') as file:

                content = json.load(file)

            # If specific slice ID is provided, find and return matching slice

            if slice_id:

                for slice in content:

                    if slice["slice_id"] == slice_id:

                        return slice, 200

                raise ValueError("Transport network slices not found")

            # If no slices exist, raise an error

            if len(content) == 0:

                raise ValueError("Transport network slices not found")

            # Return all slices if no specific ID is given

            return [slice for slice in content if slice.get("controller") == self.slice_service.controller_type], 200

        except ValueError as e:

            # Handle case where no slices are found

            return send_response(False, code=404, message=str(e))

        except Exception as e:

            # Handle unexpected errors

            return send_response(False, code=500, message=str(e))

    def modify_flow(self,slice_id, intent):

        """

        Modify an existing transport network slice.

        Args:

            slice_id (str): Unique identifier of the slice to modify

            intent (dict): New intent configuration for the slice

        Returns:

            Result of the Network Slice Controller (NSC) operation

        API Endpoint:

            PUT /slice/{id}

        """

        try:

            result = self.slice_service.nsc(intent, slice_id)

            if not result:

                return send_response(False, code=404, message="Slice not found")

            return send_response(

                True,

                code=200,

                message="Slice modified successfully",

                data=result

            )

        except Exception as e:

            # Handle unexpected errors

            return send_response(False, code=500, message=str(e))

    def delete_flows(self, slice_id=None):

        """

        Delete transport network slice(s).

        This method supports:

        - Deleting a specific slice by ID

        - Deleting all slices

        - Optional cleanup of L2VPN configurations

        Args:

            slice_id (str, optional): Unique identifier of slice to delete. 

                                      Defaults to None.

        Returns:

            dict: Response indicating successful deletion or error details

        API Endpoint:

            DELETE /slice/{id}

        Raises:

            ValueError: If no slices are found to delete

            Exception: For unexpected errors during deletion process

        Notes:

            - If controller_type is TFS, attempts to delete from Teraflow

            - If need_l2vpn_support is True, performs additional L2VPN cleanup

        """

        try:

            # Read current slice database

            with open(os.path.join(DATABASE_PATH, "slice_ddbb.json"), 'r') as file:

                content = json.load(file)

            id = None

            # Delete specific slice if slice_id is provided

            if slice_id:

                for i, slice in enumerate(content):

                    if slice["slice_id"] == slice_id and slice.get("controller") == self.slice_service.controller_type:

                        del content[i]

                        id = i

                        break

                # Raise error if slice not found

                if id is None:

                    raise ValueError("Transport network slice not found")

                # Update slice database

                with open(os.path.join(DATABASE_PATH, "slice_ddbb.json"), 'w') as file:

                    json.dump(content, file, indent=4)

                logging.info(f"Slice {slice_id} removed successfully")

                return {}, 204

            # Delete all slices

            else:

                # Optional: Delete in Teraflow if configured

                if self.slice_service.controller_type == "TFS":

                    # TODO: should send a delete request to Teraflow

                    if current_app.config["TFS_L2VPN_SUPPORT"]:

                        self.slice_service.tfs_l2vpn_delete()

                data_removed = [slice for slice in content if slice.get("controller") == self.slice_service.controller_type] 

                # Verify slices exist before deletion

                if len(data_removed) == 0:

                    raise ValueError("Transport network slices not found")

                filtered_data = [slice for slice in content if slice.get("controller") != self.slice_service.controller_type]    

                # Clear slice database

                with open(os.path.join(DATABASE_PATH, "slice_ddbb.json"), 'w') as file:

                    json.dump(filtered_data, file, indent=4)

                logging.info("All slices removed successfully")

                return {}, 204

        except ValueError as e:

            return send_response(False, code=404, message=str(e))

        except Exception as e:

            return send_response(False, code=500, message=str(e))

 No newline at end of file

# -------------------------

# General

# -------------------------

LOGGING_LEVEL=INFO  # Options: CRITICAL, ERROR, WARNING, INFO, DEBUG, NOTSET

DUMP_TEMPLATES=false

# -------------------------

# Mapper

# -------------------------

# Flag to determine if the NSC performs NRPs

NRP_ENABLED=false

# Planner Flags

PLANNER_ENABLED=true

# Flag to determine if external PCE is used

PCE_EXTERNAL=false

# -------------------------

# Realizer

# -------------------------

# If true, no config sent to controllers

DUMMY_MODE=true

# -------------------------

# Teraflow

# -------------------------

TFS_IP=127.0.0.1

UPLOAD_TYPE=WEBUI   # Options: WEBUI o NBI

# Flag to determine if additional L2VPN configuration support is required for deploying L2VPNs with path selection

TFS_L2VPN_SUPPORT=false

# -------------------------

# IXIA

# -------------------------

IXIA_IP=127.0.0.1

# -------------------------

# WebUI

# -------------------------

WEBUI_DEPLOY=true

import os

from dotenv import load_dotenv

from flask import Flask

import logging

# Load .env file if present

load_dotenv()

LOG_LEVELS = {

    "CRITICAL": logging.CRITICAL,

    "ERROR": logging.ERROR,

    "WARNING": logging.WARNING,

    "INFO": logging.INFO,

    "DEBUG": logging.DEBUG,

    "NOTSET": logging.NOTSET,

}

def create_config(app: Flask):

    """Load flags into Flask app.config"""

    # Default logging level

    app.config["LOGGING_LEVEL"] = LOG_LEVELS.get(os.getenv("LOGGING_LEVEL", "INFO").upper(),logging.INFO)

    # Dump templates

    app.config["DUMP_TEMPLATES"] = os.getenv("DUMP_TEMPLATES", "false").lower() == "true"

    # Mapper

    app.config["NRP_ENABLED"] = os.getenv("NRP_ENABLED", "false").lower() == "true"

    app.config["PLANNER_ENABLED"] = os.getenv("PLANNER_ENABLED", "false").lower() == "true"

    app.config["PCE_EXTERNAL"] = os.getenv("PCE_EXTERNAL", "false").lower() == "true"

    # Realizer

    app.config["DUMMY_MODE"] = os.getenv("DUMMY_MODE", "true").lower() == "true"

    # Teraflow

    app.config["TFS_IP"] = os.getenv("TFS_IP", "127.0.0.1")

    app.config["UPLOAD_TYPE"] = os.getenv("UPLOAD_TYPE", "WEBUI")

    app.config["TFS_L2VPN_SUPPORT"] = os.getenv("TFS_L2VPN_SUPPORT", "false").lower() == "true"

    # IXIA

    app.config["IXIA_IP"] = os.getenv("IXIA_IP", "127.0.0.1")

    # WebUI

    app.config["WEBUI_DEPLOY"] = os.getenv("WEBUI_DEPLOY", "false").lower() == "true"

    return app