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
+}