package es.tid.cosmos.mobility.btslabelling;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;

import es.tid.cosmos.mobility.clientlabelling.VectorCreateNodeDayhourJob;
import es.tid.cosmos.mobility.clientlabelling.VectorFuseNodeDaygroupJob;
import es.tid.cosmos.mobility.clientlabelling.VectorNormalizedJob;

/**
 *
 * @author dmicol
 */
public final class BtsLabellingRunner {
    private BtsLabellingRunner() {
    }
    
    public static void run(Path btsCommsPath, Path vectorBtsClusterPath,
                           Path tmpDir, Configuration conf) throws Exception {
        Path btsCountsPath = tmpDir.suffix("bts_counts");
        {
            VectorFilterBtsJob job = new VectorFilterBtsJob(conf);
            job.configure(btsCommsPath, btsCountsPath);
            if (!job.waitForCompletion(true)) {
                throw new Exception("Failed to run " + job.getJobName());
            }
        }

        Path btsSumComsPath = tmpDir.suffix("bts_sum_coms");
        {
            VectorSumComsBtsJob job = new VectorSumComsBtsJob(conf);
            job.configure(btsCountsPath, btsSumComsPath);
            if (!job.waitForCompletion(true)) {
                throw new Exception("Failed to run " + job.getJobName());
            }
        }

        Path btsDayhourPath = tmpDir.suffix("bts_dayhour");
        {
            VectorCreateNodeDayhourJob job = new VectorCreateNodeDayhourJob(
                    conf);
            job.configure(btsSumComsPath, btsDayhourPath);
            if (!job.waitForCompletion(true)) {
                throw new Exception("Failed to run " + job.getJobName());
            }
        }

        Path vectorBtsPath = tmpDir.suffix("vector_bts");
        {
            VectorFuseNodeDaygroupJob job = new VectorFuseNodeDaygroupJob(conf);
            job.configure(btsDayhourPath, vectorBtsPath);
            if (!job.waitForCompletion(true)) {
                throw new Exception("Failed to run " + job.getJobName());
            }
        }

        Path vectorBtsNormPath = tmpDir.suffix("vector_bts_norm");
        {
            VectorNormalizedJob job = new VectorNormalizedJob(conf);
            job.configure(vectorBtsPath, vectorBtsNormPath);
            if (!job.waitForCompletion(true)) {
                throw new Exception("Failed to run " + job.getJobName());
            }
        }

        Path vectorBtsClusterSinfiltPath = tmpDir.suffix(
                "vector_bts_cluster_sinfilt");
        {
            ClusterBtsGetMinDistanceJob job = new ClusterBtsGetMinDistanceJob(
                    conf);
            job.configure(vectorBtsNormPath, vectorBtsClusterSinfiltPath);
            if (!job.waitForCompletion(true)) {
                throw new Exception("Failed to run " + job.getJobName());
            }
        }

        {
            FilterBtsVectorJob job = new FilterBtsVectorJob(conf);
            job.configure(vectorBtsClusterSinfiltPath, vectorBtsClusterPath);
            if (!job.waitForCompletion(true)) {
                throw new Exception("Failed to run " + job.getJobName());
            }
        }
    }
}

@@ -560,16 +560,20 @@
 h tmpDir
+Path
 , Config
@@ -585,16 +585,28 @@
 on conf)
+%0A           
  throws 
@@ -642,38 +642,45 @@
 ountsPath = 
-tmpDir.suffix(
+new Path(tmpDirPath, 
 %22bts_counts%22
@@ -988,38 +988,45 @@
 mComsPath = 
-tmpDir.suffix(
+new Path(tmpDirPath, 
 %22bts_sum_com
@@ -1344,30 +1344,37 @@
 rPath = 
-tmpDir.suffix(
+new Path(tmpDirPath, 
 %22bts_day
@@ -1726,38 +1726,45 @@
 orBtsPath = 
-tmpDir.suffix(
+new Path(tmpDirPath, 
 %22vector_bts%22
@@ -2095,30 +2095,37 @@
 mPath = 
-tmpDir.suffix(
+new Path(tmpDirPath, 
 %22vector_
@@ -2470,22 +2470,28 @@
 h = 
-tmpDir.suffix(
+new Path(tmpDirPath,
 %0A   

Fix paths in BTS labelling.