Read Zarr Connectome Files (#8717)

- Support reading connectome files registered as attachment
- read the new zarr3-based format
- for hdf5 connectome files, always assume static list of synapse type
names (there only ever was this static list, except for very few
exceptions, compare
https://scm.slack.com/archives/C5AKLAV0B/p1750852209211939?thread_ts=1705502230.128199&cid=C5AKLAV0B

### Steps to test:
- In connectome tab, load synapses both from hdf5 and zarr connectome
file, should work as expected.

### TODOs:
- [x] list + look up connectome file keys
- [x] delegate to correct service
- [x] re-connnect hdf5 connectome file service
- [x] How to deal with synapse type names for legacy files? Compare
[slack
thread](https://scm.slack.com/archives/C5AKLAV0B/p1750851969007069?thread_ts=1705502230.128199&cid=C5AKLAV0B)
- [x] implement zarr connectome file service
- [x] enable cache clear
- [x] test
- [x] extract string constants like for the other attachments

### Issues:

- contributes to #8618
- contributes to #8567

------
- [x] Added changelog entry (create a `$PR_NUMBER.md` file in
`unreleased_changes` or use `./tools/create-changelog-entry.py`)
- [x] Removed dev-only changes like prints and application.conf edits
- [x] Considered [common edge
cases](../blob/master/.github/common_edge_cases.md)
- [x] Needs datastore update after deployment

---------

Co-authored-by: MichaelBuessemeyer <[email protected]>

Author: fm3

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)