Merge pull request #585 from JuliaRobotics/feature/3Q20/Datablobs_as_nodes

CGDFG blobs as nodes

Author: GearsAD

src/CloudGraphsDFG/CloudGraphsDFG.jl

@@ -76,7 +76,17 @@ import ...DistributedFactorGraphs:  setSolverParams!,
                                     getEstimateFields,
                                     _getname,
                                     _getmodule,
-                                    getTypeFromSerializationModule
+                                    getTypeFromSerializationModule,
+                                    listBlobStores,
+                                    BlobStoreEntry,
+                                    AbstractDataEntry,
+                                    MongodbDataEntry,
+                                    getDataEntries,
+                                    listDataEntries,
+                                    getDataEntry,
+                                    addDataEntry!,
+                                    updateDataEntry!,
+                                    deleteDataEntry!
 
 using Neo4j
 using Base64

src/CloudGraphsDFG/entities/CloudGraphsDFG.jl

@@ -13,6 +13,7 @@ mutable struct CloudGraphsDFG{T <: AbstractParams} <: AbstractDFG{T}
     description::String #TODO Maybe remove description
     addHistory::Vector{Symbol}
     solverParams::T # Solver parameters
+    blobStores::Dict{Symbol, AbstractBlobStore}
 
     # inner constructor for all constructors in common
     function CloudGraphsDFG{T}(neo4jInstance::Neo4jInstance,
@@ -25,7 +26,8 @@ mutable struct CloudGraphsDFG{T <: AbstractParams} <: AbstractDFG{T}
                                createSessionNodes::Bool=true,
                                userData::Dict{Symbol, String} = Dict{Symbol, String}(),
                                robotData::Dict{Symbol, String} = Dict{Symbol, String}(),
-                               sessionData::Dict{Symbol, String} = Dict{Symbol, String}()) where T <: AbstractParams
+                               sessionData::Dict{Symbol, String} = Dict{Symbol, String}(),
+                               blobStores::Dict{Symbol, AbstractBlobStore} = Dict{Symbol, AbstractBlobStore}()) where T <: AbstractParams
         # Validate the userId, robotId, and sessionId
         !isValidLabel(userId) && error("'$userId' is not a valid User ID")
         !isValidLabel(robotId) && error("'$robotId' is not a valid Robot ID")
@@ -35,7 +37,7 @@ mutable struct CloudGraphsDFG{T <: AbstractParams} <: AbstractDFG{T}
         # graph = Neo4j.getgraph(neo4jConnection)
         # neo4jInstance = Neo4jInstance(neo4jConnection, graph)
 
-        dfg = new{T}(neo4jInstance, userId, robotId, sessionId, description, addHistory, solverParams)
+        dfg = new{T}(neo4jInstance, userId, robotId, sessionId, description, addHistory, solverParams, blobStores)
         # Create the session if it doesn't already exist
         if createSessionNodes
             createDfgSessionIfNotExist(dfg)
@@ -93,7 +95,6 @@ function CloudGraphsDFG(; hostname="localhost",
                           solverParams::T=NoSolverParams(),
                           kwargs...) where T <: AbstractParams
 
-    @info "Creating $sessionId"
     return CloudGraphsDFG{T}(hostname,
                              port,
                              username,
@@ -140,4 +141,5 @@ function show(io::IO, ::MIME"text/plain", c::CloudGraphsDFG)
     println(io, " - Neo4J instance: $(c.neo4jInstance.connection.host)")
     println(io, " - Session: $(c.userId):$(c.robotId):$(c.sessionId)")
     println(io, " - Description: ", c.description)
+    println(io, " - Blob Stores: ", listBlobStores(c))
 end

src/CloudGraphsDFG/services/CloudGraphsDFG.jl

@@ -175,7 +175,7 @@ function updateVariable!(dfg::CloudGraphsDFG, variable::DFGVariable; skipAddErro
     !skipAddError && !exist && @warn "Variable label '$(variable.label)' does not exist in the factor graph, adding"
 
     # Create/update the base variable
-    # NOTE: We are no merging the variable.tags into the labels anymore. We can index by that but not
+    # NOTE: We are not merging the variable.tags into the labels anymore. We can index by that but not
     # going to pollute the graph with unnecessary (and potentially dangerous) labels.
     addProps = Dict("softtype" => "\"$(string(typeof(getSofttype(variable))))\"")
     query = """
@@ -239,7 +239,10 @@ function getVariable(dfg::CloudGraphsDFG, label::Union{Symbol, String})
     for solverKey in listVariableSolverData(dfg, label)
         variable.solverDataDict[solverKey] = getVariableSolverData(dfg, label, solverKey)
     end
-    # TODO - data entries
+    dataDict = getDataEntries(dfg, label)
+    for (k,v) in dataDict
+        variable.dataDict[k] = v
+    end
 
     return variable
 end
@@ -254,6 +257,9 @@ function mergeVariableData!(dfg::CloudGraphsDFG, sourceVariable::DFGVariable; cu
     for (k,v) in sourceVariable.solverDataDict
         updateVariableSolverData!(dfg, getLabel(sourceVariable), v, currentTransaction=currentTransaction)
     end
+    for (k,v) in sourceVariable.dataDict
+        updateDatEntry!(dfg, getLabel(sourceVariable), v, currentTransaction=currentTransaction)
+    end
     return sourceVariable
 end
 
@@ -632,6 +638,83 @@ function deletePPE!(dfg::CloudGraphsDFG, variablekey::Symbol, ppekey::Symbol=:de
     return _unpackPPE(dfg, props)
 end
 
+## DataEntry CRUD
+
+function getDataEntries(dfg::CloudGraphsDFG, label::Symbol; currentTransaction::Union{Nothing, Neo4j.Transaction}=nothing)
+    entries = Dict{Symbol, BlobStoreEntry}()
+    # TODO: Optimize if necessary.
+    for key in listDataEntries(dfg, label, currentTransaction=currentTransaction)
+        entry = getDataEntry(dfg, label, key, currentTransaction=currentTransaction)
+        entries[entry.label] = entry
+    end
+    return entries
+end
+
+function listDataEntries(dfg::CloudGraphsDFG, label::Symbol; currentTransaction::Union{Nothing, Neo4j.Transaction}=nothing)
+    return _listVarSubnodesForType(
+        dfg, 
+        label, 
+        BlobStoreEntry, 
+        "label"; 
+        currentTransaction=currentTransaction)
+end
+
+function getDataEntry(dfg::CloudGraphsDFG, label::Symbol, key::Symbol; currentTransaction::Union{Nothing, Neo4j.Transaction}=nothing)
+    properties = _getVarSubnodeProperties(
+        dfg, label, 
+        BlobStoreEntry, 
+        key; 
+        currentTransaction=currentTransaction)
+    return Unmarshal.unmarshal(
+        BlobStoreEntry,
+        properties)
+end
+
+function addDataEntry!(dfg::CloudGraphsDFG, label::Symbol, bde::BlobStoreEntry; currentTransaction::Union{Nothing, Neo4j.Transaction}=nothing)
+    if bde.label in listDataEntries(dfg, label, currentTransaction=currentTransaction)
+        error("Data label '$(bde.label)' already exists")
+    end
+    packed = _matchmergeVariableSubnode!(
+        dfg,
+        label,
+        _getLabelsForInst(dfg, bde, parentKey=label),
+        bde,
+        :DATA,
+        currentTransaction=currentTransaction)
+    return Unmarshal.unmarshal(
+            BlobStoreEntry,
+            packed)
+end
+
+function updateDataEntry!(dfg::CloudGraphsDFG, label::Symbol,  bde::BlobStoreEntry; currentTransaction::Union{Nothing, Neo4j.Transaction}=nothing)
+    if !(bde.label in listDataEntries(dfg, label, currentTransaction=currentTransaction))
+        @warn "Data label '$(bde.label)' does not exist, adding"
+    end
+    packed = _matchmergeVariableSubnode!(
+        dfg,
+        label,
+        _getLabelsForInst(dfg, bde, parentKey=label),
+        bde,
+        :DATA,
+        currentTransaction=currentTransaction)
+    return Unmarshal.unmarshal(
+            BlobStoreEntry,
+            packed)
+end
+
+function deleteDataEntry!(dfg::CloudGraphsDFG, label::Symbol, key::Symbol; currentTransaction::Union{Nothing, Neo4j.Transaction}=nothing)
+    props = _deleteVarSubnode!(
+        dfg,
+        label,
+        :DATA,
+        _getLabelsForType(dfg, BlobStoreEntry, parentKey=label),
+        key,
+        currentTransaction=currentTransaction)
+    return Unmarshal.unmarshal(
+        BlobStoreEntry,
+        props)
+end
+
 ## VariableSolverData CRUD
 
 """

src/CloudGraphsDFG/services/CommonFunctions.jl

@@ -147,6 +147,9 @@ function _structToNeo4jProps(inst::Union{User, Robot, Session, PVND, N, APPE, AB
         if field isa ZonedDateTime
             val = "datetime(\"$(string(field))\")"
         end
+        if field isa UUID
+            val = "\"$(string(field))\""
+        end
         # TODO: Switch this to decorator pattern
         if typeof(inst) <: DFGNode
             # Variables
@@ -248,7 +251,7 @@ function _getLabelsForInst(dfg::CloudGraphsDFG,
     typeof(inst) <: DFGFactor && push!(labels, String(getLabel(inst)))
     typeof(inst) <: AbstractPointParametricEst && push!(labels, String(inst.solverKey))
     typeof(inst) <: VariableNodeData && push!(labels, String(inst.solverKey))
-    typeof(inst) <: AbstractDataEntry && push!(labels, String(inst.key))
+    typeof(inst) <: AbstractDataEntry && push!(labels, String(inst.label))
     return labels
 end
 

src/DataBlobs/services/BlobStores.jl

@@ -77,9 +77,13 @@ function addData!(dfg::AbstractDFG, blobstore::AbstractBlobStore, label::Symbol,
     return de=>db
 end
 
-function updateData!(dfg::AbstractDFG, blobstore::AbstractBlobStore, label::Symbol,  entry::AbstractDataEntry, blob::Vector{UInt8})
-    assertHash(entry, blob, hashfunction=hashfunction)
-    de = updateDataEntry!(dfg, label, entry)
+function updateData!(dfg::AbstractDFG, blobstore::AbstractBlobStore, label::Symbol,  entry::AbstractDataEntry, blob::Vector{UInt8}; hashfunction = sha256)
+    # Recalculate the hash - NOTE Assuming that this is going to be a BlobStoreEntry. TBD.
+    newEntry = BlobStoreEntry(entry.label, entry.id, blobstore.key, bytes2hex(hashfunction(blob)),
+        "$(dfg.userId)|$(dfg.robotId)|$(dfg.sessionId)|$(label)",
+        entry.description, entry.mimeType, entry.createdTimestamp)
+
+    de = updateDataEntry!(dfg, label, newEntry)
     db = updateDataBlob!(blobstore, de, blob)
     return de=>db
 end
@@ -156,9 +160,27 @@ function addDataBlob!(store::FolderStore{T}, entry::BlobStoreEntry, data::T) whe
     blobfilename = joinpath(store.folder,"$(entry.id).dat")
     entryfilename = joinpath(store.folder,"$(entry.id).json")
     if isfile(blobfilename)
-        error("Key '$(id)' blob already exists.")
+        error("Key '$(entry.id)' blob already exists.")
     elseif isfile(entryfilename)
-        error("Key '$(id)' entry already exists, but no blob.")
+        error("Key '$(entry.id)' entry already exists, but no blob.")
+    else
+        open(blobfilename, "w") do f
+            write(f, data)
+        end
+        open(entryfilename, "w") do f
+            JSON.print(f, entry)
+        end
+        return data
+    end
+end
+
+function updateDataBlob!(store::FolderStore{T},  entry::BlobStoreEntry, data::T) where T
+    blobfilename = joinpath(store.folder,"$(entry.id).dat")
+    entryfilename = joinpath(store.folder,"$(entry.id).json")
+    if !isfile(blobfilename)
+        @warn "Key '$(entry.id)' doesn't exist."
+    elseif !isfile(entryfilename)
+        @warn "Key '$(entry.id)' doesn't exist."
     else
         open(blobfilename, "w") do f
             write(f, data)
@@ -170,6 +192,7 @@ function addDataBlob!(store::FolderStore{T}, entry::BlobStoreEntry, data::T) whe
     end
 end
 
+
 function deleteDataBlob!(store::FolderStore{T}, entry::BlobStoreEntry) where T
     blobfilename = joinpath(store.folder,"$(entry.id).dat")
     entryfilename = joinpath(store.folder,"$(entry.id).json")

src/DataBlobs/services/InMemoryDataEntryBlob.jl

@@ -5,7 +5,7 @@ export InMemoryDataEntry
 """
     $(TYPEDEF)
 Store data temporary in memory.
-NOTE: Neigher Entry nor Blob will be persisted.
+NOTE: Neither Entry nor Blob will be persisted.
 """
 struct InMemoryDataEntry{T} <: AbstractDataEntry
     label::Symbol

src/DistributedFactorGraphs.jl

@@ -54,7 +54,8 @@ export getBlobStore,
        addBlobStore!,
        updateBlobStore!,
        deleteBlobStore!,
-       emptyBlobStore!
+       emptyBlobStore!,
+       listBlobStores
 
 # TODO Not sure these are needed or should work everywhere, implement in cloud?
 export updateUserData!, updateRobotData!, updateSessionData!, deleteUserData!, deleteRobotData!, deleteSessionData!

src/services/AbstractDFG.jl

@@ -196,6 +196,7 @@ addBlobStore!(dfg::AbstractDFG, bs::AbstractBlobStore) = push!(dfg.blobStores, g
 updateBlobStore!(dfg::AbstractDFG, bs::AbstractBlobStore) = push!(dfg.blobStores, getKey(bs)=>bs)
 deleteBlobStore!(dfg::AbstractDFG, key::Symbol) = pop!(dfg.blobStores, key)
 emptyBlobStore!(dfg::AbstractDFG) = empty!(dfg.blobStores)
+listBlobStores(dfg::AbstractDFG) = collect(keys(dfg.blobStores))
 
 ##==============================================================================
 ## CRUD Interfaces

test/interfaceTests.jl

@@ -118,6 +118,9 @@ end
     else
         @test_skip DataEntriesTestBlock!(fg1, var2)
     end
+
+    # New data blob API
+    blobsTestBlock!(fg1, var2)
 end
 
 @testset "TODO Sorteer groep" begin

test/testBlocks.jl

@@ -867,6 +867,46 @@ function  DataEntriesTestBlock!(fg, v2)
     @test listDataEntries(v1) == Symbol[]
 end
 
+function blobsTestBlock!(fg, v1)
+    @testset "Data blob tests" begin
+        # Blobstore functions
+        fs = FolderStore("/tmp/$(string(uuid4())[1:8])")
+        # Adding
+        addBlobStore!(fg, fs)
+        # Listing
+        @test listBlobStores(fg) == [fs.key]
+        # Getting
+        @test getBlobStore(fg, fs.key) == fs
+        # Deleting
+        @test deleteBlobStore!(fg, fs.key) == fs
+        # Updating
+        updateBlobStore!(fg, fs)
+        @test listBlobStores(fg) == [fs.key]
+        # Emptying
+        emptyBlobStore!(fg)
+        @test listBlobStores(fg) == []
+        # Add it back
+        addBlobStore!(fg, fs)
+
+        # Data functions
+        testData = rand(UInt8, 50)
+        # Adding 
+        newData = addData!(fg, fs.key, getLabel(v1), :testing, testData)
+        # Listing
+        @test :testing in listDataEntries(fg, getLabel(v1))
+        # Getting
+        data = getData(fg, fs, getLabel(v1), :testing)
+        @test data[1].hash == newData[1].hash
+        @test data[2] == newData[2]
+        # Updating
+        updateData = updateData!(fg, fs, getLabel(v1), newData[1], rand(UInt8, 50))
+        @test updateData[1].hash != data[1].hash
+        @test updateData[2] != data[2]
+        # Deleting
+        retData = deleteData!(fg, getLabel(v1), :testing)
+    end
+end
+
 
 function testGroup!(fg, v1, v2, f0, f1)
     # "TODO Sorteer groep"