//
//Copyright 2016  Cityzen Data
//
//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 io.warp10.script.aggregator;

import io.warp10.continuum.gts.GeoTimeSerie;
import io.warp10.continuum.gts.GeoTimeSerie.TYPE;
import io.warp10.script.NamedWarpScriptFunction;
import io.warp10.script.WarpScriptAggregatorFunction;
import io.warp10.script.WarpScriptBucketizerFunction;
import io.warp10.script.WarpScriptMapperFunction;
import io.warp10.script.WarpScriptReducerFunction;
import io.warp10.script.WarpScriptException;

import com.geoxp.GeoXPLib;

/**
 * Return the operation or of the values on the interval.
 * The returned location will be the centroid of all locations.
 * The returned elevation will be the average of all elevations.
 */
public class Or extends NamedWarpScriptFunction implements WarpScriptAggregatorFunction, WarpScriptMapperFunction, WarpScriptBucketizerFunction, WarpScriptReducerFunction {

  private final boolean ignoreNulls;

  public Or(String name, boolean ignoreNulls) {
    super(name);
    this.ignoreNulls = ignoreNulls;
  }

  @Override
  public Object apply(Object[] args) throws WarpScriptException {
    long[] ticks = (long[]) args[3];
    long[] locations = (long[]) args[4];
    long[] elevations = (long[]) args[5];
    Object[] values = (Object[]) args[6];

    if (0 == ticks.length) {
      return new Object[] { Long.MAX_VALUE, GeoTimeSerie.NO_LOCATION, GeoTimeSerie.NO_ELEVATION, null };
    }

    TYPE orType = TYPE.UNDEFINED;
    Boolean res = false;
    long tickor = 0L;
    long latitudes = 0L;
    long longitudes = 0L;
    int locationcount = 0;
    long elev = 0L;
    int elevationcount = 0;

    int nulls = 0;

    for (int i = 0; i < values.length; i++) {
      Object value = values[i];

      if (null == value) {
        nulls++;
        continue;
        //return new Object[] { Long.MAX_VALUE, GeoTimeSerie.NO_LOCATION, GeoTimeSerie.NO_ELEVATION, null };
      } else {

        tickor += ticks[i] - ticks[0];

        if (GeoTimeSerie.NO_LOCATION != locations[i]) {
          long[] xy = GeoXPLib.xyFromGeoXPPoint(locations[i]);
          latitudes += xy[0];
          longitudes += xy[1];
          locationcount++;
        }

        if (GeoTimeSerie.NO_ELEVATION != elevations[i]) {
          elev += elevations[i];
          elevationcount++;
        }

        if (TYPE.BOOLEAN == orType) {
          res = res || ((Boolean) value).booleanValue();
        } else {
          // No type detected yet,
          // check value

          if (value instanceof Boolean) {
            orType = TYPE.BOOLEAN;
            res = ((Boolean) value).booleanValue();
          } else {
            //
            // Or with an other type than Boolean has no meaning
            //
            return new Object[] { Long.MAX_VALUE, GeoTimeSerie.NO_LOCATION, GeoTimeSerie.NO_ELEVATION, null };
          }
        }
      }
    }

    long meanlocation = GeoTimeSerie.NO_LOCATION;
    long meanelevation = GeoTimeSerie.NO_ELEVATION;

    if (locationcount > 0) {
      latitudes = latitudes / locationcount;
      longitudes = longitudes / locationcount;
      meanlocation = GeoXPLib.toGeoXPPoint(latitudes, longitudes);
    }

    if (elevationcount > 0) {
      meanelevation = elev / elevationcount;
    }

    Object orvalue = null;

    //
    // If we should not ignore nulls and there were some nulls, return null
    //

    if (!ignoreNulls && nulls > 0) {
      return new Object[] { Long.MAX_VALUE, GeoTimeSerie.NO_LOCATION, GeoTimeSerie.NO_ELEVATION, null };
    }

    if (TYPE.BOOLEAN == orType) {
      orvalue = res;
    }

    return new Object[] { ticks[0] + (tickor / ticks.length), meanlocation, meanelevation, orvalue };
  }
}
@@ -1978,19 +1978,18 @@
     
-B
+b
 oolean 
+o
 r
-es
  = f
@@ -2829,548 +2829,50 @@
     
-if (TYPE.BOOLEAN == orType) %7B%0A          res = res
+or = or
  %7C%7C 
-((
 Boolean
-) value).booleanValue();%0A        %7D else %7B%0A          // No type detected yet,%0A          // check value%0A%0A          if (value instanceof Boolean) %7B%0A            orType = TYPE.BOOLEAN;%0A            res = ((Boolean) value).booleanValue();%0A          %7D else %7B%0A            //%0A            // Or with an other type than Boolean has no meaning%0A            //%0A            return new Object%5B%5D %7B Long.MAX_VALUE, GeoTimeSerie.NO_LOCATION, GeoTimeSerie.NO_ELEVATION, null %7D;%0A          %7D%0A        %7D
+.TRUE.equals(values%5Bi%5D);
 %0A   
@@ -3267,36 +3267,8 @@
  %7D%0A%0A
-    Object orvalue = null;%0A%0A
     
@@ -3507,70 +3507,8 @@
  %7D%0A%0A
-    if (TYPE.BOOLEAN == orType) %7B%0A      orvalue = res;%0A    %7D%0A%0A
     
@@ -3596,21 +3596,16 @@
 tion, or
-value
  %7D;%0A  %7D%0A

Change aggregator And and Or syntax to match the one of the OpAnd and OpOr