openfl/transport/grpc/director_server.py
async def GetTrainedModel(self, request, context):  # NOQA:N802
        """RPC for retrieving trained models."""
        logger.info('Request GetTrainedModel has got!')

        if request.model_type == director_pb2.GetTrainedModelRequest.BEST_MODEL:
            model_type = 'best'
        elif request.model_type == director_pb2.GetTrainedModelRequest.LAST_MODEL:
            model_type = 'last'
        else:
            logger.error('Incorrect model type')
            return director_pb2.TrainedModelResponse()

        caller = self.get_caller(context)

        trained_model_dict = self.director.get_trained_model(
            experiment_name=request.experiment_name,
            caller=caller,
            model_type=model_type
        )

        if trained_model_dict is None:
            return director_pb2.TrainedModelResponse()

        model_proto = construct_model_proto(trained_model_dict, 0, NoCompressionPipeline())

        return director_pb2.TrainedModelResponse(model_proto=model_proto)
RPC for retrieving trained models.
RPC for retrieving trained models.
RPC for retrieving trained models .
async def GetTrainedModel(self, request, context):  
        logger.info('Request GetTrainedModel has got!')
        if request.model_type == director_pb2.GetTrainedModelRequest.BEST_MODEL:
            model_type = 'best'
        elif request.model_type == director_pb2.GetTrainedModelRequest.LAST_MODEL:
            model_type = 'last'
        else:
            logger.error('Incorrect model type')
            return director_pb2.TrainedModelResponse()
        caller = self.get_caller(context)
        trained_model_dict = self.director.get_trained_model(
            experiment_name=request.experiment_name,
            caller=caller,
            model_type=model_type
        )
        if trained_model_dict is None:
            return director_pb2.TrainedModelResponse()
        model_proto = construct_model_proto(trained_model_dict, 0, NoCompressionPipeline())
        return director_pb2.TrainedModelResponse(model_proto=model_proto)
async def GetTrainedModel ( self , request , context ) : logger . info ( 'Request GetTrainedModel has got!' ) if request . model_type == director_pb2 . GetTrainedModelRequest . BEST_MODEL : model_type = 'best' elif request . model_type == director_pb2 . GetTrainedModelRequest . LAST_MODEL : model_type = 'last' else : logger . error ( 'Incorrect model type' ) return director_pb2 . TrainedModelResponse ( ) caller = self . get_caller ( context ) trained_model_dict = self . director . get_trained_model ( experiment_name = request . experiment_name , caller = caller , model_type = model_type ) if trained_model_dict is None : return director_pb2 . TrainedModelResponse ( ) model_proto = construct_model_proto ( trained_model_dict , 0 , NoCompressionPipeline ( ) ) return director_pb2 . TrainedModelResponse ( model_proto = model_proto )
RPC for retrieving trained models.
RPC for retrieving trained models .
# NOQA:N802 """RPC for retrieving trained models."""
self
request
context