Service component:

# limitations under the License.

# 

import json

import requests

import uuid

from common.Constants import *

import functools, json, logging, requests, uuid

from typing import List

from common.Constants import ServiceNameEnum

from common.proto.context_pb2 import(

    Device, DeviceId, Service, Connection, EndPointId, TopologyId, ContextId, Uuid,

    ConfigRule, ConfigActionEnum, ConfigRule_Custom

)

from common.proto.pathcomp_pb2 import PathCompReply

from common.Settings import (

    ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, find_environment_variables, get_env_var_name

    ENVVAR_SUFIX_SERVICE_BASEURL_HTTP, ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC,

    find_environment_variables, get_env_var_name

)

from service.service.tools.replies import (

    reply_uni_txt, optical_band_uni_txt, reply_bid_txt, optical_band_bid_txt

)

from service.service.tools.replies import reply_uni_txt, optical_band_uni_txt, reply_bid_txt, optical_band_bid_txt

log = logging.getLogger(__name__)

testing = False

TESTING = False

get_optical_controller_setting = functools.partial(get_env_var_name, ServiceNameEnum.OPTICALCONTROLLER)

VAR_NAME_OPTICAL_CTRL_BASEURL_HTTP = get_optical_controller_setting(ENVVAR_SUFIX_SERVICE_BASEURL_HTTP)

VAR_NAME_OPTICAL_CTRL_SCHEMA       = get_optical_controller_setting('SCHEMA')

VAR_NAME_OPTICAL_CTRL_HOST         = get_optical_controller_setting(ENVVAR_SUFIX_SERVICE_HOST)

VAR_NAME_OPTICAL_CTRL_PORT         = get_optical_controller_setting(ENVVAR_SUFIX_SERVICE_PORT_GRPC)

VAR_NAME_OPTICAL_CONTROLLER_HOST = get_env_var_name(ServiceNameEnum.OPTICALCONTROLLER, ENVVAR_SUFIX_SERVICE_HOST)

VAR_NAME_OPTICAL_CONTROLLER_PORT = get_env_var_name(ServiceNameEnum.OPTICALCONTROLLER, ENVVAR_SUFIX_SERVICE_PORT_GRPC)

OPTICAL_CTRL_BASE_URL = '{:s}://{:s}:{:s}/OpticalTFS'

opticalcontrollers_url = find_environment_variables([

    VAR_NAME_OPTICAL_CONTROLLER_HOST,

    VAR_NAME_OPTICAL_CONTROLLER_PORT,

def get_optical_controller_base_url() -> str:

    settings = find_environment_variables([

        VAR_NAME_OPTICAL_CTRL_BASEURL_HTTP,

        VAR_NAME_OPTICAL_CTRL_SCHEMA,

        VAR_NAME_OPTICAL_CTRL_HOST,

        VAR_NAME_OPTICAL_CTRL_PORT,

    ])

OPTICAL_IP   = opticalcontrollers_url.get(VAR_NAME_OPTICAL_CONTROLLER_HOST)

OPTICAL_PORT = opticalcontrollers_url.get(VAR_NAME_OPTICAL_CONTROLLER_PORT)

log.info(str(OPTICAL_IP), str(OPTICAL_PORT))

    base_url = settings.get(VAR_NAME_OPTICAL_CTRL_BASEURL_HTTP)

    if base_url is not None:

        log.debug('Optical Controller: base_url={:s}'.format(str(base_url)))

        return base_url

    host = settings.get(VAR_NAME_OPTICAL_CTRL_HOST)

    port = int(settings.get(VAR_NAME_OPTICAL_CTRL_PORT, 80))

    MSG = 'Optical Controller not found: settings={:s}'

    if host is None: raise Exception(MSG.format(str(settings)))

    if port is None: raise Exception(MSG.format(str(settings)))

    schema = settings.get(VAR_NAME_OPTICAL_CTRL_SCHEMA, 'http')

    base_url = OPTICAL_CTRL_BASE_URL.format(schema, host, port)

    log.debug('Optical Controller: base_url={:s}'.format(str(base_url)))

    return base_url

def get_uuids_from_names(devices: List[Device], device_name: str, port_name: str):

    device_uuid = ""

def add_lightpath(src, dst, bitrate, bidir, ob_band) -> str:

    if not testing:

    if not TESTING:

        urlx = ""

        headers = {"Content-Type": "application/json"}

        base_url = get_optical_controller_base_url()

        if ob_band is None:

            if bidir is None:

                bidir = 1

            urlx = "http://{}:{}/OpticalTFS/AddFlexLightpath/{}/{}/{}/{}".format(OPTICAL_IP, OPTICAL_PORT, src, dst, bitrate, bidir)

            urlx = "{:s}/AddFlexLightpath/{:s}/{:s}/{:s}/{:s}".format(base_url, src, dst, bitrate, bidir)

        else:

            if bidir is None:

                bidir = 1

            urlx = "http://{}:{}/OpticalTFS/AddFlexLightpath/{}/{}/{}/{}/{}".format(OPTICAL_IP, OPTICAL_PORT, src, dst, bitrate, bidir, ob_band)

            urlx = "{:s}/AddFlexLightpath/{:s}/{:s}/{:s}/{:s}/{:s}".format(base_url, src, dst, bitrate, bidir, ob_band)

        r = requests.put(urlx, headers=headers)

        reply = r.text 

        return reply

def get_optical_band(idx) -> str:

    if not testing:

        urlx = "http://{}:{}/OpticalTFS/GetOpticalBand/{}".format(OPTICAL_IP, OPTICAL_PORT, idx)

    if not TESTING:

        base_url = get_optical_controller_base_url()

        urlx = "{:s}/GetOpticalBand/{:s}".format(base_url, idx)

        headers = {"Content-Type": "application/json"}

        r = requests.get(urlx, headers=headers)

        reply = r.text 

def delete_lightpath(flow_id, src, dst, bitrate) -> str:

    reply = "200"

    if not testing:

        urlx = "http://{}:{}/OpticalTFS/DelLightpath/{}/{}/{}/{}".format(OPTICAL_IP, OPTICAL_PORT, flow_id, src, dst, bitrate)

    if not TESTING:

        base_url = get_optical_controller_base_url()

        urlx = "{:s}/DelLightpath/{:s}/{:s}/{:s}/{:s}".format(base_url, flow_id, src, dst, bitrate)

        headers = {"Content-Type": "application/json"}

        r = requests.delete(urlx, headers=headers)

def get_lightpaths() -> str:

    urlx = "http://{}:{}/OpticalTFS/GetLightpaths".format(OPTICAL_IP, OPTICAL_PORT)

    base_url = get_optical_controller_base_url()

    urlx = "{:s}/GetLightpaths".format(base_url)

    headers = {"Content-Type": "application/json"}

    r = requests.get(urlx, headers=headers)