Loading src/service/service/tools/OpticalTools.py +6 −2 Original line number Diff line number Diff line Loading @@ -18,8 +18,8 @@ from typing import Dict, List, Tuple from common.method_wrappers.ServiceExceptions import NotFoundException from common.proto.context_pb2 import( ConfigActionEnum, ConfigRule, ConfigRule_Custom, Connection, ContextId, Device, DeviceId, Empty, EndPointId, OpticalBand, OpticalBandId, Service, TopologyId, Uuid Device, DeviceId, Empty, EndPointId, OpticalBand, OpticalBandId, OpticalBandList, Service, TopologyId, Uuid ) from common.proto.pathcomp_pb2 import PathCompReply from common.tools.context_queries.OpticalConfig import find_optical_band Loading Loading @@ -588,6 +588,10 @@ def extend_optical_band(reply, optical_band_text) -> Service: MSG = 'optical_band_id not found in reply({:s})/optical_band_text({:s})' raise KeyError(MSG.format(str(reply), str(optical_band_text))) context_client = ContextClient() optical_bands : OpticalBandList = context_client.GetOpticalBand(Empty()) LOGGER.warning('GetOpticalBand result: {:s}'.format(grpc_message_to_json_string(optical_bands))) ob_index = optical_band_res['optical_band_id'] optical_band = find_optical_band(ob_index=ob_index) if optical_band is None: Loading Loading
src/service/service/tools/OpticalTools.py +6 −2 Original line number Diff line number Diff line Loading @@ -18,8 +18,8 @@ from typing import Dict, List, Tuple from common.method_wrappers.ServiceExceptions import NotFoundException from common.proto.context_pb2 import( ConfigActionEnum, ConfigRule, ConfigRule_Custom, Connection, ContextId, Device, DeviceId, Empty, EndPointId, OpticalBand, OpticalBandId, Service, TopologyId, Uuid Device, DeviceId, Empty, EndPointId, OpticalBand, OpticalBandId, OpticalBandList, Service, TopologyId, Uuid ) from common.proto.pathcomp_pb2 import PathCompReply from common.tools.context_queries.OpticalConfig import find_optical_band Loading Loading @@ -588,6 +588,10 @@ def extend_optical_band(reply, optical_band_text) -> Service: MSG = 'optical_band_id not found in reply({:s})/optical_band_text({:s})' raise KeyError(MSG.format(str(reply), str(optical_band_text))) context_client = ContextClient() optical_bands : OpticalBandList = context_client.GetOpticalBand(Empty()) LOGGER.warning('GetOpticalBand result: {:s}'.format(grpc_message_to_json_string(optical_bands))) ob_index = optical_band_res['optical_band_id'] optical_band = find_optical_band(ob_index=ob_index) if optical_band is None: Loading