openfl/transport/grpc/director_server.py
async def SetNewExperiment(self, stream, context):  # NOQA:N802
        """Request to set new experiment."""
        logger.info(f'SetNewExperiment request has got {stream}')
        # TODO: add streaming reader
        data_file_path = self.root_dir / str(uuid.uuid4())
        with open(data_file_path, 'wb') as data_file:
            async for request in stream:
                if request.experiment_data.size == len(request.experiment_data.npbytes):
                    data_file.write(request.experiment_data.npbytes)
                else:
                    raise Exception('Bad request')

        tensor_dict = None
        if request.model_proto:
            tensor_dict, _ = deconstruct_model_proto(request.model_proto, NoCompressionPipeline())

        caller = self.get_caller(context)

        is_accepted = await self.director.set_new_experiment(
            experiment_name=request.name,
            sender_name=caller,
            tensor_dict=tensor_dict,
            collaborator_names=request.collaborator_names,
            experiment_archive_path=data_file_path
        )

        logger.info('Send response')
        return director_pb2.SetNewExperimentResponse(accepted=is_accepted)
Request to set new experiment.
Request to set new experiment.
Request to set new experiment .
async def SetNewExperiment(self, stream, context):  
        logger.info(f'SetNewExperiment request has got {stream}')
        data_file_path = self.root_dir / str(uuid.uuid4())
        with open(data_file_path, 'wb') as data_file:
            async for request in stream:
                if request.experiment_data.size == len(request.experiment_data.npbytes):
                    data_file.write(request.experiment_data.npbytes)
                else:
                    raise Exception('Bad request')
        tensor_dict = None
        if request.model_proto:
            tensor_dict, _ = deconstruct_model_proto(request.model_proto, NoCompressionPipeline())
        caller = self.get_caller(context)
        is_accepted = await self.director.set_new_experiment(
            experiment_name=request.name,
            sender_name=caller,
            tensor_dict=tensor_dict,
            collaborator_names=request.collaborator_names,
            experiment_archive_path=data_file_path
        )
        logger.info('Send response')
        return director_pb2.SetNewExperimentResponse(accepted=is_accepted)
async def SetNewExperiment ( self , stream , context ) : logger . info ( f'SetNewExperiment request has got {stream}' ) data_file_path = self . root_dir / str ( uuid . uuid4 ( ) ) with open ( data_file_path , 'wb' ) as data_file : async for request in stream : if request . experiment_data . size == len ( request . experiment_data . npbytes ) : data_file . write ( request . experiment_data . npbytes ) else : raise Exception ( 'Bad request' ) tensor_dict = None if request . model_proto : tensor_dict , _ = deconstruct_model_proto ( request . model_proto , NoCompressionPipeline ( ) ) caller = self . get_caller ( context ) is_accepted = await self . director . set_new_experiment ( experiment_name = request . name , sender_name = caller , tensor_dict = tensor_dict , collaborator_names = request . collaborator_names , experiment_archive_path = data_file_path ) logger . info ( 'Send response' ) return director_pb2 . SetNewExperimentResponse ( accepted = is_accepted )
Request to set new experiment.
Request to set new experiment .
# NOQA:N802 """Request to set new experiment.""" # TODO: add streaming reader
self
stream
context