Add segmentation pipeline

api/compute/filter.go

@@ -85,7 +85,7 @@ func filterData(client *compute.Client, ds *api.Dataset, filterParams *api.Filte
 
  // output the filtered results as the data in the filtered dataset
  _, outputDataFile := getPreFilteringOutputDataFile(outputFolder)
- err = util.CopyFile(filteredData, outputDataFile)
+ err = util.CopyFile(filteredData.ResultURI, outputDataFile)
  if err != nil {
  return "", nil, err
  }

api/compute/pipeline.go

@@ -115,10 +115,17 @@ type QueueItem struct {
 
 // QueueResponse represents the result from processing a queue item.
 type QueueResponse struct {
- Output interface{}
+ Output *PipelineOutput
  Error error
 }
 
+// PipelineOutput represents an output from executing a queued pipeline.
+type PipelineOutput struct {
+ SolutionID string
+ ResultURI string
+ FittedSolutionID string
+}
+
 // Queue uses a buffered channel to queue tasks and provides the result via channels.
 type Queue struct {
  mu sync.RWMutex
@@ -194,6 +201,9 @@ func (q *Queue) Done() {
 // InitializeCache sets up an empty cache or if a source file provided, reads
 // the cache from the source file.
 func InitializeCache(sourceFile string, readEnabled bool) error {
+ // register the output type for the cache!
+ gob.Register(&PipelineOutput{})
+
  var c *gc.Cache
  if util.FileExists(sourceFile) {
  b, err := ioutil.ReadFile(sourceFile)
@@ -234,7 +244,7 @@ func InitializeQueue(config *env.Config) {
 
 // SubmitPipeline executes pipelines using the client and returns the result URI.
 func SubmitPipeline(client *compute.Client, datasets []string, datasetsProduce []string, searchRequest *pipeline.SearchSolutionsRequest,
- fullySpecifiedStep *description.FullySpecifiedPipeline, allowedValueTypes []string, shouldCache bool) (string, error) {
+ fullySpecifiedStep *description.FullySpecifiedPipeline, allowedValueTypes []string, shouldCache bool) (*PipelineOutput, error) {
 
  request := compute.NewExecPipelineRequest(datasets, datasetsProduce, fullySpecifiedStep.Pipeline)
 
@@ -254,12 +264,12 @@ func SubmitPipeline(client *compute.Client, datasets []string, datasetsProduce [
  if cache.readEnabled {
  if shouldCache {
  if err != nil {
- return "", err
+ return nil, err
  }
  entry, found := cache.cache.Get(hashedPipelineUniqueKey)
  if found {
  log.Infof("returning cached entry for pipeline")
- return entry.(string), nil
+ return entry.(*PipelineOutput), nil
  }
  }
  } else {
@@ -268,25 +278,24 @@ func SubmitPipeline(client *compute.Client, datasets []string, datasetsProduce [
  // get equivalency key for enqueuing
  hashedPipelineEquivKey, err := queueTask.hashEquivalent()
  if err != nil {
- return "", err
+ return nil, err
  }
 
  resultChan := queue.Enqueue(hashedPipelineEquivKey, queueTask)
  defer queue.Done()
 
  result := <-resultChan
  if result.Error != nil {
- return "", result.Error
+ return nil, result.Error
  }
 
- datasetURI := result.Output.(string)
- cache.cache.Set(hashedPipelineUniqueKey, datasetURI, gc.DefaultExpiration)
+ cache.cache.Set(hashedPipelineUniqueKey, result.Output, gc.DefaultExpiration)
  err = cache.PersistCache()
  if err != nil {
  log.Warnf("error persisting cache: %v", err)
  }
 
- return datasetURI, nil
+ return result.Output, nil
 }
 
 func runPipelineQueue(queue *Queue) {
@@ -316,6 +325,8 @@ func runPipelineQueue(queue *Queue) {
  // listen for completion
  var errPipeline error
  var datasetURI string
+ var fittedSolutionID string
+ var solutionID string
  err = pipelineTask.request.Listen(func(status compute.ExecPipelineStatus) {
  // check for error
  if status.Error != nil {
@@ -324,6 +335,8 @@ func runPipelineQueue(queue *Queue) {
 
  if status.Progress == compute.RequestCompletedStatus {
  datasetURI = status.ResultURI
+ fittedSolutionID = status.FittedSolutionID
+ solutionID = status.SolutionID
  }
  })
  if err != nil {
@@ -342,7 +355,11 @@ func runPipelineQueue(queue *Queue) {
 
  datasetURI = strings.Replace(datasetURI, "file://", "", -1)
 
- queueTask.returnResult(&QueueResponse{Output: datasetURI})
+ queueTask.returnResult(&QueueResponse{&PipelineOutput{
+ ResultURI: datasetURI,
+ FittedSolutionID: fittedSolutionID,
+ SolutionID: solutionID,
+ }, nil})
  }
 
  log.Infof("ending queue processing")

api/compute/search.go

@@ -39,6 +39,7 @@ type searchResult struct {
 type pipelineSearchContext struct {
  searchID string
  dataset string
+ task []string
  storageName string
  sourceDatasetURI string
  trainDatasetURI string
@@ -278,7 +279,7 @@ func (s *SolutionRequest) dispatchSolutionSearchPipeline(statusChan chan Solutio
  if ok {
  // reformat result to have one row per d3m index since confidences
  // can produce one row / class
- resultURI, err = reformatResult(resultURI)
+ resultURI, err = reformatResult(resultURI, s.TargetFeature.HeaderName)
  if err != nil {
  return nil, err
  }

api/compute/segment.go

@@ -0,0 +1,58 @@
+//
+// Copyright © 2021 Uncharted Software Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package compute
+
+import (
+ "strconv"
+
+ "github.com/pkg/errors"
+
+ "github.com/uncharted-distil/distil/api/util/imagery"
+)
+
+// BuildSegmentationImage uses the raw segmentation output to build an image layer.
+func BuildSegmentationImage(rawSegmentation [][]interface{}) (map[string][]byte, error) {
+ // output is mapping of d3m index to new segmentation layer
+ output := map[string][]byte{}
+ // need to output all the masks as images
+ for _, r := range rawSegmentation[1:] {
+ // create the image that captures the mask
+ d3mIndex := r[0].(string)
+ rawMask := r[1].([]interface{})
+ rawFloats := make([][]float64, len(rawMask))
+ for i, f := range rawMask {
+ dataF := f.([]interface{})
+ nestedFloats := make([]float64, len(dataF))
+ for j, nf := range dataF {
+ fp, err := strconv.ParseFloat(nf.(string), 64)
+ if err != nil {
+ return nil, errors.Wrapf(err, "unable to parse mask")
+ }
+ nestedFloats[j] = fp
+ }
+ rawFloats[i] = nestedFloats
+ }
+
+ filter := imagery.ConfidenceMatrixToImage(rawFloats, imagery.MagmaColorScale, uint8(255))
+ imageBytes, err := imagery.ImageToPNG(filter)
+ if err != nil {
+ return nil, err
+ }
+ output[d3mIndex] = imageBytes
+ }
+
+ return output, nil
+}