// Path: src/main/java/acromusashi/stream/ml/clustering/kmeans/entity/KmeansPoint.java
// public class KmeansPoint implements Serializable
// {
//     /** serialVersionUID */
//     private static final long serialVersionUID = -5518876772129721631L;
// 
//     /** データ座標 */
//     private double[]          dataPoint;
// 
//     /**
//      * パラメータを指定せずにインスタンスを生成する。
//      */
//     public KmeansPoint()
//     {}
// 
//     /**
//      * @return the dataPoint
//      */
//     public double[] getDataPoint()
//     {
//         return this.dataPoint;
//     }
// 
//     /**
//      * @param dataPoint the dataPoint to set
//      */
//     public void setDataPoint(double[] dataPoint)
//     {
//         this.dataPoint = dataPoint;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public boolean equals(Object other)
//     {
//         if (this == other)
//         {
//             return true;
//         }
// 
//         if (!(other instanceof KmeansPoint))
//         {
//             return false;
//         }
// 
//         KmeansPoint otherPoint = (KmeansPoint) other;
// 
//         for (int i = 0; i < this.dataPoint.length; ++i)
//         {
//             if (this.dataPoint[i] != otherPoint.getDataPoint()[i])
//             {
//                 return false;
//             }
//         }
// 
//         return true;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public int hashCode()
//     {
//         return MathUtils.hash(this.dataPoint);
//     }
// 
//     /**
//      * 文字列表現を返す。<br>
//      * StormからDRPC応答を生成するコンポーネント<br>
//      * (ReturnResultsReducer#complete(ReturnResultsState, TridentCollector))<br>
//      * に問題があり、JSON文字列とならない状態でレスポンスが生成されるため、結果をダブルクォートで囲っている。
//      * 
//      * @return 文字列表現
//      */
//     @Override
//     public String toString()
//     {
//         // TODO Stormの応答生成部の問題が対応された際にダブルクォートを削除
//         String result = "\""
//                 + ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE) + "\"";
//         return result;
//     }
// }


import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.TridentOperationContext;
import storm.trident.tuple.TridentTuple;
import acromusashi.stream.ml.clustering.kmeans.entity.KmeansPoint;
import backtype.storm.tuple.Values;
/**
* Copyright (c) Acroquest Technology Co, Ltd. All Rights Reserved.
* Please read the associated COPYRIGHTS file for more details.
*
* THE SOFTWARE IS PROVIDED BY Acroquest Technolog Co., Ltd.,
* WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING
* BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDER BE LIABLE FOR ANY
* CLAIM, DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
* OR DISTRIBUTING THIS SOFTWARE OR ITS DERIVATIVES.
*/
package acromusashi.stream.ml.clustering.kmeans;




/**
 * KMeansの対象ポイントを生成するFunctionクラス
 * 
 * @author kimura
 */
public class KmeansCreator extends BaseFunction
{
    /** serialVersionUID */
    private static final long   serialVersionUID = 4620883521921594615L;

    /** logger */
    private static final Logger logger           = LoggerFactory.getLogger(KmeansCreator.class);

    /** 読みこんだデータを分割する文字列 */
    private String              delimeter        = ",";

    /**
     * パラメータを指定せずにインスタンスを生成する。
     */
    public KmeansCreator()
    {}

    /**
     * {@inheritDoc}
     */
    @SuppressWarnings("rawtypes")
    @Override
    public void prepare(Map conf, TridentOperationContext context)
    {
        if (logger.isDebugEnabled() == true)
        {
            logger.debug("prepared started. taskIndex=" + context.getPartitionIndex()
                    + ", taxkNum=" + context.numPartitions());
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void execute(TridentTuple tuple, TridentCollector collector)
    {
        String receivedStr = tuple.getString(0);

        String[] splitedStr = StringUtils.split(receivedStr, this.delimeter);
        int dataNum = splitedStr.length;
        double[] points = new double[splitedStr.length];

        try
        {
            for (int index = 0; index < dataNum; index++)
            {
                points[index] = Double.parseDouble(splitedStr[index].trim());
            }

// Path: src/main/java/acromusashi/stream/ml/clustering/kmeans/entity/KmeansPoint.java
// public class KmeansPoint implements Serializable
// {
//     /** serialVersionUID */
//     private static final long serialVersionUID = -5518876772129721631L;
// 
//     /** データ座標 */
//     private double[]          dataPoint;
// 
//     /**
//      * パラメータを指定せずにインスタンスを生成する。
//      */
//     public KmeansPoint()
//     {}
// 
//     /**
//      * @return the dataPoint
//      */
//     public double[] getDataPoint()
//     {
//         return this.dataPoint;
//     }
// 
//     /**
//      * @param dataPoint the dataPoint to set
//      */
//     public void setDataPoint(double[] dataPoint)
//     {
//         this.dataPoint = dataPoint;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public boolean equals(Object other)
//     {
//         if (this == other)
//         {
//             return true;
//         }
// 
//         if (!(other instanceof KmeansPoint))
//         {
//             return false;
//         }
// 
//         KmeansPoint otherPoint = (KmeansPoint) other;
// 
//         for (int i = 0; i < this.dataPoint.length; ++i)
//         {
//             if (this.dataPoint[i] != otherPoint.getDataPoint()[i])
//             {
//                 return false;
//             }
//         }
// 
//         return true;
//     }
// 
//     /**
//      * {@inheritDoc}
//      */
//     @Override
//     public int hashCode()
//     {
//         return MathUtils.hash(this.dataPoint);
//     }
// 
//     /**
//      * 文字列表現を返す。<br>
//      * StormからDRPC応答を生成するコンポーネント<br>
//      * (ReturnResultsReducer#complete(ReturnResultsState, TridentCollector))<br>
//      * に問題があり、JSON文字列とならない状態でレスポンスが生成されるため、結果をダブルクォートで囲っている。
//      * 
//      * @return 文字列表現
//      */
//     @Override
//     public String toString()
//     {
//         // TODO Stormの応答生成部の問題が対応された際にダブルクォートを削除
//         String result = "\""
//                 + ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE) + "\"";
//         return result;
//     }
// }


// Path: src/main/java/acromusashi/stream/ml/clustering/kmeans/KmeansCreator.java
import java.util.Map;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.operation.TridentOperationContext;
import storm.trident.tuple.TridentTuple;
import acromusashi.stream.ml.clustering.kmeans.entity.KmeansPoint;
import backtype.storm.tuple.Values;

/**
* Copyright (c) Acroquest Technology Co, Ltd. All Rights Reserved.
* Please read the associated COPYRIGHTS file for more details.
*
* THE SOFTWARE IS PROVIDED BY Acroquest Technolog Co., Ltd.,
* WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING
* BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
* IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDER BE LIABLE FOR ANY
* CLAIM, DAMAGES SUFFERED BY LICENSEE AS A RESULT OF USING, MODIFYING
* OR DISTRIBUTING THIS SOFTWARE OR ITS DERIVATIVES.
*/
package acromusashi.stream.ml.clustering.kmeans;




/**
 * KMeansの対象ポイントを生成するFunctionクラス
 * 
 * @author kimura
 */
public class KmeansCreator extends BaseFunction
{
    /** serialVersionUID */
    private static final long   serialVersionUID = 4620883521921594615L;

    /** logger */
    private static final Logger logger           = LoggerFactory.getLogger(KmeansCreator.class);

    /** 読みこんだデータを分割する文字列 */
    private String              delimeter        = ",";

    /**
     * パラメータを指定せずにインスタンスを生成する。
     */
    public KmeansCreator()
    {}

    /**
     * {@inheritDoc}
     */
    @SuppressWarnings("rawtypes")
    @Override
    public void prepare(Map conf, TridentOperationContext context)
    {
        if (logger.isDebugEnabled() == true)
        {
            logger.debug("prepared started. taskIndex=" + context.getPartitionIndex()
                    + ", taxkNum=" + context.numPartitions());
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void execute(TridentTuple tuple, TridentCollector collector)
    {
        String receivedStr = tuple.getString(0);

        String[] splitedStr = StringUtils.split(receivedStr, this.delimeter);
        int dataNum = splitedStr.length;
        double[] points = new double[splitedStr.length];

        try
        {
            for (int index = 0; index < dataNum; index++)
            {
                points[index] = Double.parseDouble(splitedStr[index].trim());
            }

            KmeansPoint result = new KmeansPoint();