diff --git a/ApplicationLibCode/Application/Tools/KeyValueStore/RiaKeyValueStoreUtil.cpp b/ApplicationLibCode/Application/Tools/KeyValueStore/RiaKeyValueStoreUtil.cpp index 23dad220ba..8bcdfd0187 100644 --- a/ApplicationLibCode/Application/Tools/KeyValueStore/RiaKeyValueStoreUtil.cpp +++ b/ApplicationLibCode/Application/Tools/KeyValueStore/RiaKeyValueStoreUtil.cpp @@ -23,3 +23,22 @@ std::vector RiaKeyValueStoreUtil::convertToFloatVector( const std::option return floatVector; } + +//-------------------------------------------------------------------------------------------------- +/// +//-------------------------------------------------------------------------------------------------- +std::vector RiaKeyValueStoreUtil::convertToByteVector( const std::vector& floatVec ) +{ + if ( floatVec.empty() ) return {}; + + // Calculate the total size needed for the byte array + size_t sizeInBytes = floatVec.size() * sizeof( float ); + + // Create a vector of bytes with the appropriate size + std::vector byteVec( sizeInBytes ); + + // Copy the binary data from the float vector to the byte vector + std::memcpy( byteVec.data(), floatVec.data(), sizeInBytes ); + + return byteVec; +} diff --git a/ApplicationLibCode/Application/Tools/KeyValueStore/RiaKeyValueStoreUtil.h b/ApplicationLibCode/Application/Tools/KeyValueStore/RiaKeyValueStoreUtil.h index b64a50afeb..32ae9355eb 100644 --- a/ApplicationLibCode/Application/Tools/KeyValueStore/RiaKeyValueStoreUtil.h +++ b/ApplicationLibCode/Application/Tools/KeyValueStore/RiaKeyValueStoreUtil.h @@ -28,4 +28,5 @@ class RiaKeyValueStoreUtil { public: static std::vector convertToFloatVector( const std::optional>& input ); + static std::vector convertToByteVector( const std::vector& floatVec ); }; diff --git a/GrpcInterface/GrpcProtos/KeyValueStore.proto b/GrpcInterface/GrpcProtos/KeyValueStore.proto index 013b68e667..eb1d161caf 100644 --- a/GrpcInterface/GrpcProtos/KeyValueStore.proto +++ b/GrpcInterface/GrpcProtos/KeyValueStore.proto @@ -7,6 +7,8 @@ import "Definitions.proto"; service KeyValueStore { rpc SetValue(stream KeyValueStoreInputChunk) returns (ClientToServerStreamReply) {} + rpc GetValue(KeyValueStoreOutputRequest) + returns (stream KeyValueStoreOutputChunk) {} } message KeyValueStoreInputChunk { @@ -23,3 +25,7 @@ message KeyValueInputParameters { } message KeyValueStoreChunk { repeated float values = 1; } + +message KeyValueStoreOutputRequest { string name = 1; } + +message KeyValueStoreOutputChunk { repeated float values = 1; } diff --git a/GrpcInterface/Python/rips/project.py b/GrpcInterface/Python/rips/project.py index 12d43434f1..2e0a836b65 100644 --- a/GrpcInterface/Python/rips/project.py +++ b/GrpcInterface/Python/rips/project.py @@ -518,3 +518,40 @@ def __generate_key_value_store_input_chunks(self, array, name): # Final empty message to signal completion chunk = KeyValueStore_pb2.KeyValueStoreInputChunk() yield chunk + + +@add_method(Project) +def key_values_async( + self, + key, +): + """Get float values for a given key. Async: returns an iterator. + + Arguments: + key(str): The key to get. + + Returns: + An iterator to a chunk object containing an array of float values + Loop through the chunks and then the values within the chunk to get all values. + """ + request = KeyValueStore_pb2.KeyValueStoreOutputRequest(name=key) + for chunk in self.__key_value_store_stub.GetValue(request): + yield chunk + + +@add_method(Project) +def key_values(self, key): + """Get float values for a given key. Synchronous: returns a list. ` + + Arguments: + key(str): The key to get. + + Returns: + A list of float values + """ + all_values = [] + generator = self.key_values_async(key) + for chunk in generator: + for value in chunk.values: + all_values.append(value) + return all_values diff --git a/GrpcInterface/Python/rips/tests/test_key_value_store.py b/GrpcInterface/Python/rips/tests/test_key_value_store.py new file mode 100644 index 0000000000..b9a397fbd0 --- /dev/null +++ b/GrpcInterface/Python/rips/tests/test_key_value_store.py @@ -0,0 +1,26 @@ +import sys +import os +import pytest + +sys.path.insert(1, os.path.join(sys.path[0], "../../")) +import rips + + +def test_set_and_get_values(rips_instance, initialize_test): + project = rips_instance.project + + key = "my-key" + input_values = [float(x) for x in range(2000)] + project.set_key_values(key, input_values) + + output_values = project.key_values(key) + assert input_values == output_values + + +def test_get_non_existing_key(rips_instance, initialize_test): + project = rips_instance.project + + key = "this-key-does-not-exist" + + with pytest.raises(rips.RipsError, match="No matching key found."): + project.key_values(key) diff --git a/GrpcInterface/RiaGrpcKeyValueStoreService.cpp b/GrpcInterface/RiaGrpcKeyValueStoreService.cpp index 276f88a397..382697adb3 100644 --- a/GrpcInterface/RiaGrpcKeyValueStoreService.cpp +++ b/GrpcInterface/RiaGrpcKeyValueStoreService.cpp @@ -17,9 +17,12 @@ ////////////////////////////////////////////////////////////////////////////////// #include "RiaGrpcKeyValueStoreService.h" +#include "KeyValueStore.pb.h" #include "RiaApplication.h" #include "RiaGrpcCallbacks.h" #include "RiaGrpcCaseService.h" +#include "RiaGrpcHelper.h" +#include "RiaKeyValueStoreUtil.h" #include "RiaLogging.h" #include "cafAssert.h" @@ -79,6 +82,55 @@ public: return grpc::Status( grpc::INVALID_ARGUMENT, "Need to have parameters in first message" ); } + //-------------------------------------------------------------------------------------------------- + /// + //-------------------------------------------------------------------------------------------------- + Status init( const KeyValueStoreOutputRequest* request ) + { + std::string name = request->name(); + + RiaLogging::debug( QString( "Output stream request. Name='%1'" ).arg( QString::fromStdString( name ) ) ); + + auto result = RiaApplication::instance()->keyValueStore()->get( name ); + if ( result.has_value() ) + { + m_data = RiaKeyValueStoreUtil::convertToFloatVector( result ); + m_name = name; + m_cellCount = m_data.size(); + return grpc::Status::OK; + } + else + { + return grpc::Status( grpc::NOT_FOUND, "No matching key found." ); + } + } + + //-------------------------------------------------------------------------------------------------- + /// + //-------------------------------------------------------------------------------------------------- + Status assignStreamReply( KeyValueStoreOutputChunk* reply ) + { + // How many data units will fit into one stream package? + const size_t packageSize = RiaGrpcHelper::numberOfDataUnitsInPackage( sizeof( float ) ); + const size_t remainingData = m_cellCount - m_streamedValueCount; + const size_t chunkSize = std::min( packageSize, remainingData ); + + if ( chunkSize == 0 ) + { + return grpc::Status( grpc::OUT_OF_RANGE, + "We've reached the end. This is not an error but means transmission is finished" ); + } + + reply->mutable_values()->Reserve( static_cast( chunkSize ) ); + for ( size_t i = 0; i < chunkSize; ++i ) + { + reply->add_values( m_data[m_streamedValueCount + i] ); + } + + m_streamedValueCount += chunkSize; + return grpc::Status::OK; + } + //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- @@ -124,26 +176,7 @@ public: QString( "Stream finished: name=%1 size=%2" ).arg( QString::fromStdString( m_name ) ).arg( m_data.size() ) ); if ( !m_name.empty() && !m_data.empty() ) { - auto convertFromFloatVectorToBytes = []( const std::vector& float_vec ) -> std::vector - { - if ( float_vec.empty() ) - { - return {}; - } - - // Calculate the total size needed for the byte array - size_t size_in_bytes = float_vec.size() * sizeof( float ); - - // Create a vector of bytes with the appropriate size - std::vector byte_vec( size_in_bytes ); - - // Copy the binary data from the float vector to the byte vector - std::memcpy( byte_vec.data(), float_vec.data(), size_in_bytes ); - - return byte_vec; - }; - - RiaApplication::instance()->keyValueStore()->set( m_name, convertFromFloatVectorToBytes( m_data ) ); + RiaApplication::instance()->keyValueStore()->set( m_name, RiaKeyValueStoreUtil::convertToByteVector( m_data ) ); } } @@ -164,6 +197,15 @@ grpc::Status RiaGrpcKeyValueStoreService::SetValue( grpc::ServerContext* { return stateHandler->receiveStreamRequest( request, reply ); } + +grpc::Status RiaGrpcKeyValueStoreService::GetValue( grpc::ServerContext* context, + const rips::KeyValueStoreOutputRequest* request, + rips::KeyValueStoreOutputChunk* reply, + RiaKeyValueStoreStateHandler* stateHandler ) +{ + return stateHandler->assignStreamReply( reply ); +} + //-------------------------------------------------------------------------------------------------- /// //-------------------------------------------------------------------------------------------------- @@ -180,6 +222,17 @@ std::vector RiaGrpcKeyValueStoreService::createCallba &Self::SetValue, &Self::RequestSetValue, new RiaKeyValueStoreStateHandler( true ) ) }; + for ( int i = 0; i < NUM_CONCURRENT_CLIENT_TO_SERVER_STREAMS; ++i ) + { + callbacks.push_back( + new RiaGrpcServerToClientStreamCallback( this, + &Self::GetValue, + &Self::RequestGetValue, + new RiaKeyValueStoreStateHandler ) ); + } return callbacks; } diff --git a/GrpcInterface/RiaGrpcKeyValueStoreService.h b/GrpcInterface/RiaGrpcKeyValueStoreService.h index aa89eb5fe8..22aa376823 100644 --- a/GrpcInterface/RiaGrpcKeyValueStoreService.h +++ b/GrpcInterface/RiaGrpcKeyValueStoreService.h @@ -38,5 +38,10 @@ public: rips::ClientToServerStreamReply* reply, RiaKeyValueStoreStateHandler* stateHandler ); + grpc::Status GetValue( grpc::ServerContext* context, + const rips::KeyValueStoreOutputRequest* request, + rips::KeyValueStoreOutputChunk* reply, + RiaKeyValueStoreStateHandler* stateHandler ); + std::vector createCallbacks() override; };