package org.mwg.experiments.smartgridprofiling.gmm;

import org.mwg.*;
import org.mwg.core.scheduler.ExecutorScheduler;
import org.mwg.importer.ImporterPlugin;
import org.mwg.ml.MLPlugin;
import org.mwg.ml.algorithm.profiling.GaussianMixtureNode;
import org.mwg.ml.algorithm.profiling.GaussianTreeNode;
import org.mwg.task.Action;
import org.mwg.task.Task;
import org.mwg.task.TaskContext;
import org.mwg.task.TaskFunctionConditional;
import org.omg.PortableInterceptor.SYSTEM_EXCEPTION;

import java.util.concurrent.atomic.AtomicLong;

import static org.mwg.importer.ImporterActions.readFiles;
import static org.mwg.task.Actions.*;

/**
 * Created by assaad on 06/07/16.
 */
public class ParralelTest {
    public static void main(String[] arg) {
        final String csvdir = "/Users/assaad/work/github/data/consumption/londonpower/";
        final double[] err = new double[]{0.5 * 0.5, 0.25 * 0.25, 0.9 * 0.9, 10 * 10};

        final AtomicLong counter = new AtomicLong(0);
        final AtomicLong starttime = new AtomicLong(0);

        final Graph g = new org.mwg.GraphBuilder()
                .withPlugin(new ImporterPlugin())
                .withMemorySize(3000000)
                .saveEvery(20000)
                // .withOffHeapMemory()
                .withStorage(new LevelDBStorage(csvdir + "leveldb/").useNative(false))
                .withPlugin(new MLPlugin())

                // .withScheduler(new ExecutorScheduler())
                .build();
        g.connect(new Callback<Boolean>() {
            @Override
            public void on(Boolean result) {
                starttime.set(System.nanoTime());
                final Task t = readFiles(csvdir + "NDsim/allusers/")
                        .foreachPar(
                                ifThen(new TaskFunctionConditional() {
                                           @Override
                                           public boolean eval(TaskContext context) {
                                               return context.resultAsString().contains("csv");
                                           }
                                       }, then(new Action() {
                                            @Override
                                            public void eval(TaskContext context) {
                                                //create node and set as var

                                                GaussianTreeNode profiler = (GaussianTreeNode) context.graph().newTypedNode(0, 0, GaussianTreeNode.NAME);
                                                profiler.set(GaussianMixtureNode.PRECISION, err); //Minimum covariance in both axis
                                                context.setVariable("profiler", profiler);
                                                profiler.free();
                                                context.setUnsafeResult(context.result());

                                            }
                                        }).action("readLines", "{{result}}")
                                                .foreach(
                                                        split(",").then(new Action() {
                                                            @Override
                                                            public void eval(TaskContext context) {
                                                                String[] values = context.resultAsStringArray();
                                                                double[] vector = new double[4];
                                                                double[] features = new double[3];


                                                                vector[0] = Double.parseDouble(values[1]);
                                                                vector[1] = Double.parseDouble(values[2]);
                                                                vector[2] = Double.parseDouble(values[3]);
                                                                vector[3] = Double.parseDouble(values[4]);

                                                                System.arraycopy(vector, 0, features, 0, vector.length - 1);


                                                                GaussianTreeNode profiler = (GaussianTreeNode) context.variable("profiler");
                                                                profiler.internalLearn(vector, features, new Callback<Boolean>() {
                                                                    @Override
                                                                    public void on(Boolean result) {
                                                                        long c = counter.addAndGet(1);
                                                                        if (c % 20000 == 0) {
                                                                            long end = System.nanoTime();
                                                                            double time = end - starttime.get();
                                                                            time = time / 1000000000;
                                                                            time = c / time;
                                                                            double d = c;
                                                                            d = d / 1000000;
                                                                            System.out.println(d + " m " + time + " values/sec");
                                                                        }
                                                                    }
                                                                });


                                                                //TODO lookup and train

                                                                //end
                                                                context.setUnsafeResult(null);


                                                            }
                                                        })
                                                )
                                ));
                t.executeWith(g, null, null, false, new Callback<Object>() {
                    @Override
                    public void on(Object result) {
                        System.out.println("end!");
                    }
                });
            }
        });

        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

@@ -431,61 +431,8 @@
 nal;
-%0Aimport org.omg.PortableInterceptor.SYSTEM_EXCEPTION;
 %0A%0Aim
@@ -972,24 +972,25 @@
 icLong(0);%0A%0A
+%0A
         fina
@@ -1074,24 +1074,68 @@
 erPlugin())%0A
+                .withPlugin(new MLPlugin())%0A
             
@@ -1201,48 +1201,8 @@
 00)%0A
-                // .withOffHeapMemory()%0A
     
@@ -1280,61 +1280,15 @@
 ive(
-false))%0A                .withPlugin(new MLPlugin(
+true
 ))%0A
-%0A
     
@@ -1298,19 +1298,16 @@
         
- //
  .withSc
@@ -1337,16 +1337,28 @@
 eduler()
+.workers(30)
 )%0A      
@@ -1373,24 +1373,80 @@
   .build();%0A
+%0A        DeferCounterSync waiter= g.newSyncCounter(1);%0A%0A
         g.co
@@ -4802,17 +4802,17 @@
 if (c %25 
-2
+1
 0000 == 
@@ -5515,16 +5515,18 @@
 (d + %22 m
+ @
  %22 + tim
@@ -5541,17 +5541,32 @@
 lues/sec
-%22
+, counter: %22 + c
 );%0A     
@@ -5706,78 +5706,1104 @@
     
-%7D%0A                                                                %7D);%0A
+    context.setUnsafeResult(null);%0A                                                                    %7D%0A                                                                %7D);%0A%0A                                                              /*  long c = counter.addAndGet(1);%0A                                                                if (c %25 1000 == 0) %7B%0A                                                                    long end = System.nanoTime();%0A                                                                    double time = end - starttime.get();%0A                                                                    time = time / 1000000000;%0A                                                                    time = c / time;%0A                                                                    double d = c;%0A                                                                    d = d / 1000000;%0A                                                                    System.out.println(d + %22 m @ %22 + time + %22 values/sec%22);%0A                                                                %7D*/
 %0A%0A  
@@ -6963,103 +6963,8 @@
 end%0A
-                                                                context.setUnsafeResult(null);%0A
 %0A%0A  
@@ -7375,24 +7375,64 @@
 ln(%22end!%22);%0A
+                        waiter.count();%0A
             
@@ -7518,25 +7518,26 @@
     
-Thread.sleep(1000
+waiter.waitResult(
 );%0A 
@@ -7556,19 +7556,8 @@
 ch (
-Interrupted
 Exce

update test