From 5186b9790e5a388369a82adc2b1c5cee28ff9167 Mon Sep 17 00:00:00 2001
From: Alessandro Genova <alessandro.genova@kitware.com>
Date: Tue, 3 Dec 2019 16:58:03 -0500
Subject: [PATCH] Add multiple ways to aggregate an image being streamed

These changes are needed to support both the annular mask and maximum
diffraction pipelines
---
 src/components/pipeline/index.tsx |  5 +++++
 src/stem-image/aggregation.ts     | 26 ++++++++++++++++++++++++++
 src/stem-image/data.ts            | 11 +++++++----
 src/stem-image/pipelines.ts       | 13 +++++++++++--
 4 files changed, 49 insertions(+), 6 deletions(-)
 create mode 100644 src/stem-image/aggregation.ts

diff --git a/src/components/pipeline/index.tsx b/src/components/pipeline/index.tsx
index fefec7e..9da2ee9 100644
--- a/src/components/pipeline/index.tsx
+++ b/src/components/pipeline/index.tsx
@@ -10,6 +10,7 @@ import StatusBar from './status';
 import Dialog from './dialog';
 import AddWorker from './add-worker';
 import SplitButton from '../split-button';
+import { getAggregationFunction } from '../../stem-image/aggregation';
 
 export enum ButtonOptions {
   GenerateImage = "generate-image",
@@ -250,6 +251,10 @@ class PipelineWrapper extends Component<Props, State> {
   onPipelineCreated(pipeline: PipelineCreationData) {
     const { extraValues } = this.props;
     const { pipelineId, workerId, name, info } = pipeline;
+
+    const aggregationFn = getAggregationFunction(info.aggregation);
+    this.source.setAggregationFunction(aggregationFn);
+
     const executeParams = {
       pipelineId,
       workerId,
diff --git a/src/stem-image/aggregation.ts b/src/stem-image/aggregation.ts
new file mode 100644
index 0000000..9c66e1a
--- /dev/null
+++ b/src/stem-image/aggregation.ts
@@ -0,0 +1,26 @@
+import {
+  PipelineAggregation
+} from './pipelines';
+
+export type AggregationFunction = (aggregated: number, current: number) => number;
+
+export const SumAggregation: AggregationFunction = (aggregated, current) => aggregated + current;
+
+export const MaxAggregation: AggregationFunction = (aggregated, current) => Math.max(aggregated, current);
+
+export const MinAggregation: AggregationFunction = (aggregated, current) => Math.min(aggregated, current);
+
+export function getAggregationFunction(aggregation: PipelineAggregation) : AggregationFunction {
+  switch (aggregation) {
+    case PipelineAggregation.Max: {
+      return SumAggregation;
+    }
+    case PipelineAggregation.Min: {
+      return MinAggregation;
+    }
+    case PipelineAggregation.Sum:
+    default: {
+      return SumAggregation;
+    }
+  }
+}
diff --git a/src/stem-image/data.ts b/src/stem-image/data.ts
index a517629..4bfb3fc 100644
--- a/src/stem-image/data.ts
+++ b/src/stem-image/data.ts
@@ -5,6 +5,7 @@ import {
 import { StreamConnection } from './connection';
 import { MultiSubjectProducer, IObserver } from './subject';
 import { PipelineExecutionData } from './pipelines';
+import { AggregationFunction, SumAggregation } from './aggregation';
 import { decode } from '@msgpack/msgpack';
 
 export interface ImageDataSource {
@@ -98,6 +99,7 @@ export class StreamImageDataSource extends BaseImageDataSource implements ImageD
   private connection: StreamConnection | null = null;
   private sizeEvent: string = "";
   private dataEvent: string = "";
+  private aggregationFn: AggregationFunction = SumAggregation;
 
   constructor() {
     super();
@@ -114,6 +116,10 @@ export class StreamImageDataSource extends BaseImageDataSource implements ImageD
     this.connection.subscribe(this.dataEvent, this.dataObserver);
   }
 
+  setAggregationFunction(fn: AggregationFunction) {
+    this.aggregationFn = fn;
+  }
+
   resetConnection() {
     if (this.connection) {
       this.connection.unsubscribe(this.sizeEvent, this.sizeObserver);
@@ -153,11 +159,8 @@ export class StreamImageDataSource extends BaseImageDataSource implements ImageD
   }
 
   private updateImageChunk(values: Float64Array) {
-    const {width, height} = this.size;
-    const n = width * height;
-
     for(let i = 0; i < values.length; ++i) {
-        this.data[i] += values[i];
+        this.data[i] = this.aggregationFn(this.data[i], values[i]);
     }
 
     this.updateRange();
diff --git a/src/stem-image/pipelines.ts b/src/stem-image/pipelines.ts
index feddd14..74276ea 100644
--- a/src/stem-image/pipelines.ts
+++ b/src/stem-image/pipelines.ts
@@ -5,7 +5,9 @@ export interface PipelineInfo {
   description: string;
   displayName: string;
   parameters: {[name: string]: ServerField};
-  output: PipelineIO
+  input: PipelineIO;
+  output: PipelineIO;
+  aggregation: PipelineAggregation;
 }
 
 export interface PipelineCreationData {
@@ -25,10 +27,17 @@ export interface PipelineExecutionData {
 }
 
 export enum PipelineName {
-  AnnularMask = 'annular'
+  AnnularMask = 'annular',
+  MaximumDiffraction = 'maximum_diffraction'
 }
 
 export enum PipelineIO {
   Image = 'image',
   Frame = 'frame'
 }
+
+export enum PipelineAggregation {
+  Sum = 'sum',
+  Max = 'max',
+  Min = 'min'
+}