Read Zarr Connectome Files

unreleased_changes/8717.md

@@ -0,0 +1,2 @@
+### Added
+- Connectomes can now also be read from the new zarr3-based format, and from remote object storage.

util/src/main/scala/com/scalableminds/util/collections/SequenceUtils.scala

@@ -1,5 +1,7 @@
 package com.scalableminds.util.collections
 
+import scala.collection.Searching.{Found, InsertionPoint}
+
 object SequenceUtils {
   def findUniqueElement[T](list: Seq[T]): Option[T] = {
     val uniqueElements = list.distinct
@@ -51,4 +53,11 @@ object SequenceUtils {
       val batchTo = Math.min(to, (batchIndex + 1) * batchSize + from - 1)
       (batchFrom, batchTo)
     }
+
+  // Search in a sorted array, returns Box of index where element is found or, if missing, where element would be inserted
+  def searchSorted(haystack: Array[Long], needle: Long): Int =
+    haystack.search(needle) match {
+      case Found(i)          => i
+      case InsertionPoint(i) => i
+    }
 }

webknossos-datastore/app/com/scalableminds/webknossos/datastore/DataStoreModule.scala

@@ -4,6 +4,11 @@ import org.apache.pekko.actor.ActorSystem
 import com.google.inject.AbstractModule
 import com.google.inject.name.Names
 import com.scalableminds.webknossos.datastore.services._
+import com.scalableminds.webknossos.datastore.services.connectome.{
+  ConnectomeFileService,
+  Hdf5ConnectomeFileService,
+  ZarrConnectomeFileService
+}
 import com.scalableminds.webknossos.datastore.services.mapping.{
   AgglomerateService,
   Hdf5AgglomerateService,
@@ -52,6 +57,9 @@ class DataStoreModule extends AbstractModule {
     bind(classOf[SegmentIndexFileService]).asEagerSingleton()
     bind(classOf[ZarrSegmentIndexFileService]).asEagerSingleton()
     bind(classOf[Hdf5SegmentIndexFileService]).asEagerSingleton()
+    bind(classOf[ConnectomeFileService]).asEagerSingleton()
+    bind(classOf[ZarrConnectomeFileService]).asEagerSingleton()
+    bind(classOf[Hdf5ConnectomeFileService]).asEagerSingleton()
     bind(classOf[NeuroglancerPrecomputedMeshFileService]).asEagerSingleton()
     bind(classOf[RemoteSourceDescriptorService]).asEagerSingleton()
     bind(classOf[ChunkCacheService]).asEagerSingleton()

webknossos-datastore/app/com/scalableminds/webknossos/datastore/controllers/DataSourceController.scala

@@ -25,6 +25,12 @@ import com.scalableminds.webknossos.datastore.services.uploading._
 import com.scalableminds.webknossos.datastore.storage.DataVaultService
 import com.scalableminds.util.tools.Box.tryo
 import com.scalableminds.util.tools.{Box, Empty, Failure, Full}
+import com.scalableminds.webknossos.datastore.services.connectome.{
+  ByAgglomerateIdsRequest,
+  BySynapseIdsRequest,
+  ConnectomeFileService,
+  SynapticPartnerDirection
+}
 import com.scalableminds.webknossos.datastore.services.mapping.AgglomerateService
 import play.api.data.Form
 import play.api.data.Forms.{longNumber, nonEmptyText, number, tuple}
@@ -452,14 +458,16 @@ class DataSourceController @Inject()(
       meshFileService.clearCache(dataSourceId, layerName)
     val closedSegmentIndexFileHandleCount =
       segmentIndexFileService.clearCache(dataSourceId, layerName)
+    val closedConnectomeFileHandleCount =
+      connectomeFileService.clearCache(dataSourceId, layerName)
     val reloadedDataSource: InboxDataSource = dataSourceService.dataSourceFromDir(
       dataSourceService.dataBaseDir.resolve(organizationId).resolve(datasetDirectoryName),
       organizationId)
     datasetErrorLoggingService.clearForDataset(organizationId, datasetDirectoryName)
     val clearedVaultCacheEntriesOpt = dataSourceService.invalidateVaultCache(reloadedDataSource, layerName)
     clearedVaultCacheEntriesOpt.foreach { clearedVaultCacheEntries =>
       logger.info(
-        s"Cleared caches for ${layerName.map(l => s"layer '$l' of ").getOrElse("")}dataset $organizationId/$datasetDirectoryName: closed $closedAgglomerateFileHandleCount agglomerate file handles, $closedMeshFileHandleCount mesh file handles, $closedSegmentIndexFileHandleCount segment index file handles, removed $clearedBucketProviderCount bucketProviders, $clearedVaultCacheEntries vault cache entries and $removedChunksCount image chunk cache entries.")
+        s"Cleared caches for ${layerName.map(l => s"layer '$l' of ").getOrElse("")}dataset $organizationId/$datasetDirectoryName: closed $closedAgglomerateFileHandleCount agglomerate file handles, $closedMeshFileHandleCount mesh file handles, $closedSegmentIndexFileHandleCount segment index file handles, $closedConnectomeFileHandleCount connectome file handles, removed $clearedBucketProviderCount bucketProviders, $clearedVaultCacheEntries vault cache entries and $removedChunksCount image chunk cache entries.")
     }
     reloadedDataSource
   }
@@ -510,21 +518,12 @@ class DataSourceController @Inject()(
     Action.async { implicit request =>
       accessTokenService.validateAccessFromTokenContext(
         UserAccessRequest.readDataSources(DataSourceId(datasetDirectoryName, organizationId))) {
-        val connectomeFileNames =
-          connectomeFileService.exploreConnectomeFiles(organizationId, datasetDirectoryName, dataLayerName)
         for {
-          mappingNames <- Fox.serialCombined(connectomeFileNames.toList) { connectomeFileName =>
-            val path =
-              connectomeFileService.connectomeFilePath(organizationId,
-                                                       datasetDirectoryName,
-                                                       dataLayerName,
-                                                       connectomeFileName)
-            connectomeFileService.mappingNameForConnectomeFile(path)
-          }
-          connectomesWithMappings = connectomeFileNames
-            .zip(mappingNames)
-            .map(tuple => ConnectomeFileNameWithMappingName(tuple._1, tuple._2))
-        } yield Ok(Json.toJson(connectomesWithMappings))
+          (dataSource, dataLayer) <- dataSourceRepository.getDataSourceAndDataLayer(organizationId,
+                                                                                    datasetDirectoryName,
+                                                                                    dataLayerName)
+          connectomeFileInfos <- connectomeFileService.listConnectomeFiles(dataSource.id, dataLayer)
+        } yield Ok(Json.toJson(connectomeFileInfos))
       }
     }
 
@@ -535,10 +534,13 @@ class DataSourceController @Inject()(
       accessTokenService.validateAccessFromTokenContext(
         UserAccessRequest.readDataSources(DataSourceId(datasetDirectoryName, organizationId))) {
         for {
-          meshFilePath <- Fox.successful(
-            connectomeFileService
-              .connectomeFilePath(organizationId, datasetDirectoryName, dataLayerName, request.body.connectomeFile))
-          synapses <- connectomeFileService.synapsesForAgglomerates(meshFilePath, request.body.agglomerateIds)
+          (dataSource, dataLayer) <- dataSourceRepository.getDataSourceAndDataLayer(organizationId,
+                                                                                    datasetDirectoryName,
+                                                                                    dataLayerName)
+          meshFileKey <- connectomeFileService.lookUpConnectomeFileKey(dataSource.id,
+                                                                       dataLayer,
+                                                                       request.body.connectomeFile)
+          synapses <- connectomeFileService.synapsesForAgglomerates(meshFileKey, request.body.agglomerateIds)
         } yield Ok(Json.toJson(synapses))
       }
     }
@@ -551,12 +553,18 @@ class DataSourceController @Inject()(
       accessTokenService.validateAccessFromTokenContext(
         UserAccessRequest.readDataSources(DataSourceId(datasetDirectoryName, organizationId))) {
         for {
-          meshFilePath <- Fox.successful(
-            connectomeFileService
-              .connectomeFilePath(organizationId, datasetDirectoryName, dataLayerName, request.body.connectomeFile))
-          agglomerateIds <- connectomeFileService.synapticPartnerForSynapses(meshFilePath,
+          directionValidated <- SynapticPartnerDirection
+            .fromString(direction)
+            .toFox ?~> "could not parse synaptic partner direction"
+          (dataSource, dataLayer) <- dataSourceRepository.getDataSourceAndDataLayer(organizationId,
+                                                                                    datasetDirectoryName,
+                                                                                    dataLayerName)
+          meshFileKey <- connectomeFileService.lookUpConnectomeFileKey(dataSource.id,
+                                                                       dataLayer,
+                                                                       request.body.connectomeFile)
+          agglomerateIds <- connectomeFileService.synapticPartnerForSynapses(meshFileKey,
                                                                              request.body.synapseIds,
-                                                                             direction)
+                                                                             directionValidated)
         } yield Ok(Json.toJson(agglomerateIds))
       }
     }
@@ -568,10 +576,13 @@ class DataSourceController @Inject()(
       accessTokenService.validateAccessFromTokenContext(
         UserAccessRequest.readDataSources(DataSourceId(datasetDirectoryName, organizationId))) {
         for {
-          meshFilePath <- Fox.successful(
-            connectomeFileService
-              .connectomeFilePath(organizationId, datasetDirectoryName, dataLayerName, request.body.connectomeFile))
-          synapsePositions <- connectomeFileService.positionsForSynapses(meshFilePath, request.body.synapseIds)
+          (dataSource, dataLayer) <- dataSourceRepository.getDataSourceAndDataLayer(organizationId,
+                                                                                    datasetDirectoryName,
+                                                                                    dataLayerName)
+          meshFileKey <- connectomeFileService.lookUpConnectomeFileKey(dataSource.id,
+                                                                       dataLayer,
+                                                                       request.body.connectomeFile)
+          synapsePositions <- connectomeFileService.positionsForSynapses(meshFileKey, request.body.synapseIds)
         } yield Ok(Json.toJson(synapsePositions))
       }
     }
@@ -583,10 +594,13 @@ class DataSourceController @Inject()(
       accessTokenService.validateAccessFromTokenContext(
         UserAccessRequest.readDataSources(DataSourceId(datasetDirectoryName, organizationId))) {
         for {
-          meshFilePath <- Fox.successful(
-            connectomeFileService
-              .connectomeFilePath(organizationId, datasetDirectoryName, dataLayerName, request.body.connectomeFile))
-          synapseTypes <- connectomeFileService.typesForSynapses(meshFilePath, request.body.synapseIds)
+          (dataSource, dataLayer) <- dataSourceRepository.getDataSourceAndDataLayer(organizationId,
+                                                                                    datasetDirectoryName,
+                                                                                    dataLayerName)
+          meshFileKey <- connectomeFileService.lookUpConnectomeFileKey(dataSource.id,
+                                                                       dataLayer,
+                                                                       request.body.connectomeFile)
+          synapseTypes <- connectomeFileService.typesForSynapses(meshFileKey, request.body.synapseIds)
         } yield Ok(Json.toJson(synapseTypes))
       }
     }

webknossos-datastore/app/com/scalableminds/webknossos/datastore/datareaders/DatasetArray.scala

@@ -193,6 +193,8 @@ class DatasetArray(vaultPath: VaultPath,
                                                                tc: TokenContext): Fox[MultiArray] =
     if (shape.contains(0)) {
       Fox.successful(MultiArrayUtils.createEmpty(header.resolvedDataType, rank))
+    } else if (shape.exists(_ < 0)) {
+      Fox.failure(s"Trying to read negative shape from DatasetArray: ${shape.mkString(",")}")
     } else {
       val totalOffset: Array[Long] = offset.zip(header.voxelOffset).map { case (o, v) => o - v }.padTo(offset.length, 0)
       val chunkIndices = ChunkUtils.computeChunkIndices(datasetShape, chunkShape, shape, totalOffset)