Pre-merge code cleanup

    from webui.service.optical_link.routes import optical_link      # pylint: disable=import-outside-toplevel

    app.register_blueprint(optical_link)

    from webui.service.tapi.routes import tapi                      # pylint: disable=import-outside-toplevel

    app.register_blueprint(tapi)

    from webui.service.service.routes import service                # pylint: disable=import-outside-toplevel

    app.register_blueprint(service)

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

# Copyright 2022-2024 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.

from flask_wtf import FlaskForm

from flask_wtf.file import FileAllowed

from wtforms import StringField , IntegerField ,SubmitField , FileField

class SocketForm(FlaskForm):

   ip            = StringField('Ip')

   port        = IntegerField("Port")

   submit = SubmitField('Connect')

class UploadService(FlaskForm):

   descriptors = FileField(

        'Descriptors',

        validators=[

            FileAllowed(['json'], 'JSON Descriptors only!')

        ])

   submit = SubmitField('Submit')

 No newline at end of file

# Copyright 2022-2024 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.

import json, logging , time

from flask import (

    request, redirect, render_template, Blueprint, flash, session, url_for,

    current_app, make_response

)

from common.proto.context_pb2 import (

    Empty, ServiceList, OpticalLink , TopologyId , OpticalConfigId

)

from common.tools.object_factory.OpticalLink import extract_endpoint_names

from common.tools.context_queries.OpticalConfig import opticalconfig_uuid_get_duuid

from context.client.ContextClient import ContextClient

from device.client.DeviceClient import DeviceClient

from service.client.ServiceClient import ServiceClient

from slice.client.SliceClient import SliceClient

from tapi.client.TapiClient import TapiClient

from .forms import SocketForm , UploadService

from google.protobuf.json_format import Parse

tapi = Blueprint('tapi', __name__, url_prefix="/tapi")

context_client = ContextClient()

device_client = DeviceClient()

service_client = ServiceClient()

slice_client = SliceClient()

LOGGER = logging.getLogger(__name__)

DESCRIPTOR_LOADER_NUM_WORKERS = 10

@tapi.route('/tapi_main',methods=["GET"])

def tapi_main () : 

    session.pop('tapi_endpoint',None)

    return redirect(url_for('tapi.home'))

@tapi.route("/" , methods=['GET', 'POST'])

def home() :

    topo_lsts = None

    socket_form = SocketForm ()

    ip = None 

    port = None

    tapi_client=None

    if request.method == 'GET' : 

        if 'tapi_endpoint' in session  : 

            ip,port = session['tapi_endpoint'].split(':')

    if socket_form.validate_on_submit():       

        ip = socket_form.ip.data

        port = socket_form.port.data 

    if ip is not None and port is not None :   

        try: 

                tapi_client = TapiClient(ip , port)

                LOGGER.info(f'tapi started {ip  }: {port} ')

                if tapi_client is not None : 

                    LOGGER.info(f'checking existing ')

                    topo_lsts=tapi_client.GetListTopologies(Empty())

                    #topo_lsts = tapi_client.CheckConnectivity()

                    LOGGER.info(f'topo existed {topo_lsts}')

                if  topo_lsts   is not None  : 

                    session['tapi_endpoint'] = f'{ip}:{port}'

                    return  render_template( 'tapi/topo.html',topo_lsts=topo_lsts)

        except Exception as  err : 

            LOGGER.info(f" error from connecting tapi {err}") 

            flash(f'Connection is failed: `{str(err)}`', 'danger')

        finally:

            if tapi_client is not None : 

                tapi_client.close()

    return render_template('tapi/home.html', socket_form=socket_form)

@tapi.route("/retmote_topo/<path:topo_uuid>/<path:context_uuid>" , methods=['GET', 'POST'])

def get_retmote_topo (topo_uuid :str ,context_uuid :str) : 

    topo=None

    endpoints_names={}

    if 'tapi_endpoint' in session : 

        try: 

            ip ,port = session['tapi_endpoint'].split(':')

            tapi_client = TapiClient(ip , port)

            LOGGER.info(f'fetching details { topo_uuid} ')

            if tapi_client is not None : 

                topo_id = TopologyId()

                topo_id.topology_uuid.uuid = topo_uuid

                topo_id.context_id.context_uuid.uuid=context_uuid

                topo=tapi_client.GetTopology(topo_id)

                #topo_lsts = tapi_client.CheckConnectivity()

                endpoints_names = extract_endpoint_names(topo)

                LOGGER.info(f'topo existed {topo}')

        except Exception as  err : 

            LOGGER.info(f" error from connecting tapi {err}") 

            flash(f'Connection is failed: `{str(err)}`', 'danger')

        finally: 

            tapi_client.close()

    else : 

        flash("Please Connect To Desired Machine First!", "warning")

        return redirect(url_for("tapi.home"))         

    logging.info(f"endpoinst_names { endpoints_names}")            

    return  render_template( 'tapi/topo_detail.html',topo=topo,endpoints_names=endpoints_names)

@tapi.route("/set_retmote_service/<path:topo_uuid>/<path:context_uuid>" , methods=['GET', 'POST'])

def add_service (topo_uuid :str ,context_uuid :str) :        

        ip = None

        port = None

        service_form = UploadService()

        if 'tapi_endpoint' in session  : 

            ip,port = session['tapi_endpoint'].split(':')

        else : 

            flash("Please Connect To Desired Machine First!", "warning")

            return redirect(url_for("tapi.home"))    

        if service_form.validate_on_submit():   

            descriptors_file = request.files[service_form.descriptors.name]

            descriptors_data = descriptors_file.read()

            descriptors = json.loads(descriptors_data)   

            request_message = Parse(json.dumps(descriptors), ServiceList()) 

            tapi_client = TapiClient(ip , port)

            if tapi_client is not None : 

                try: 

                  tapi_client.SetService(request_message)

                  flash(f'Service is deployed successfully to {ip}:{port}', 'success')

                  return redirect(url_for('tapi.tapi_main'))

                except Exception as  err : 

                    LOGGER.info(f" error from connecting tapi {err}") 

                    flash(f'Connection is failed: `{str(err)}`', 'danger')

                finally: 

                  tapi_client.close() 

            else : 

                    flash(f'Error in Connction ', 'danger')

        return render_template('tapi/add_service.html', service_form=service_form)            

@tapi.route('<path:device_uuid>/detail',methods=['GET'])    

def show_details(device_uuid):

    ip=''

    port =''

    if 'tapi_endpoint' in session  : 

            ip,port = session['tapi_endpoint'].split(':')

    else : 

            flash("Please Connect To Desired Machine First!", "warning")

            return redirect(url_for("tapi.home"))   

    # config_uuid = opticalconfig_uuid_get_duuid (device_uuid)

    # opticalconfigId = OpticalConfigId()

    # opticalconfigId.opticalconfig_uuid = config_uuid

    device_details = []

    device_name=''

    try : 

        tapi_client = TapiClient(ip , port)

        resposne = tapi_client.GetOpticalConfig(device_uuid)

        if (response and response.opticalconfig_id.opticalconfig_uuid !=''):

            opticalConfig = OpticalConfig()

            opticalConfig.CopyFrom(response)

            device_name = ""

            config = json.loads(opticalConfig.config)

            logging.info(f"config {config}")

            if "device_name" in config:

                device_name = config["device_name"]

                config_type = config["type"]

                if config_type == DeviceTypeEnum.OPTICAL_TRANSPONDER._value_:

                    LOGGER.info("config details from show detail %s",config)

                    if 'channels' in config:

                        for channel in config['channels'] :

                            new_config={}

                            new_config["name"]=channel['name']

                            new_config['operationalMode']=channel['operational-mode'] if 'operational-mode' in channel else ''

                            new_config['targetOutputPower']=channel['target-output-power'] if 'target-output-power' in channel else ''

                            new_config["frequency"]=channel['frequency'] if 'frequency' in channel else ''

                            new_config['line_port']=channel["line-port"] if 'line-port' in channel else ''

                            new_config["status"] = channel['status'] if 'status' in channel else ""

                            device_details.append(new_config)

                if config_type == DeviceTypeEnum.OPTICAL_ROADM._value_:

                    LOGGER.info("config details from show detail %s",config)

                    if 'channels' in config:

                        for channel in config['channels'] :

                            new_config={}

                            new_config["band_name"]=channel['band_name'] if 'band_name' in channel else None

                            new_config['type']=channel['type'] if 'type' in channel else ''

                            new_config['src_port']=channel['src_port'] if 'src_port' in channel else ''

                            new_config['dest_port']=channel['dest_port'] if 'dest_port' in channel else ''

                            new_config["lower_frequency"]=channel['lower_frequency'] if 'lower_frequency' in channel else ''

                            new_config["upper_frequency"]=channel['upper_frequency'] if 'upper_frequency' in channel else ''

                            new_config["status"] = channel['status'] if 'status' in channel else ""

                            new_config['optical_band_parent']= channel['optical_band_parent'] if 'optical_band_parent' in channel else ''

                            new_config['channel_index']= channel['channel_index'] if 'channel_index' in channel else ''

                            device_details.append(new_config)

                if config_type == DeviceTypeEnum.OPEN_ROADM._value_:

                    if 'interfaces' in config :

                        for interface in config["interfaces"]:

                            if interface["type"] != "Null":

                                new_config={}

                                new_config["name"]=interface["name"] if "name" in interface else '' 

                                new_config["administrative_state"]=interface[ "administrative_state"]

                                new_config["circuit_pack_name"]=interface["circuit_pack_name"]

                                new_config["port"]=interface["port"]

                                new_config["interface_list"]=interface["interface_list"]

                                new_config["frequency"]=interface["frequency"]

                                new_config["width"]=interface[ "width"]

                                new_config["type"]=interface["type"]

                                device_details.append(new_config)

            LOGGER.info("device details  %s",device_details)

       #return render_template( 'tapi/device_details.html', device=device_details,config_id=config_uuid,device_name=device_name,type=config_type)

    except Exception as  err : 

                    LOGGER.info(f" error from showing oc details {err}") 

                    flash(f'Error : `{str(err)}`', 'danger')

    return render_template(

        'tapi/device_details.html', device=device_details,device_uuid=device_uuid,device_name=device_name

    )

 No newline at end of file