<commit_before>package com.freevariable.surlaplaque;

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

import org.apache.spark.mllib.clustering._

import com.freevariable.surlaplaque.importer._
import com.freevariable.surlaplaque.data._
// import com.freevariable.surlaplaque.power._
import com.freevariable.surlaplaque.app._

object ReplHarness extends Common {
    
    def setup(args: Array[String], providedApp: Option[SLP] = None) = {
        // XXX: add optional parameters here to support cluster execution
        val app = providedApp.getOrElse(new SLP(new SparkContext(master, appName)))
        
        app.processFiles(SLP.expandArgs(args))
    }
    
    def mapActivities(args: Array[String]) = {
        val data = setup(args)
        
        val apairs = data.map((tp:Trackpoint) => Pair(tp.activity.getOrElse("UNKNOWN"), Pair(tp.timestamp, tp)))
        
        val activities = apairs.keys.distinct.collect
        
        val pairs = for (activity <- activities) yield {
            val filtered = apairs.filter((tup) => {val (a,_) = tup ; a == activity});
            val timestampedTrackpoints = filtered.map((tup) => {val (_,(t,tp)) = tup; (t, tp.watts)}).sortByKey();
            val samples = timestampedTrackpoints.collect;
            Pair(activity, samples)
        }
        
        pairs.toMap
    }
}

object BucketApp extends Common {
    
    def main(args: Array[String]) {
        // XXX: add optional parameters here to support cluster execution
        val app = new SLP(new SparkContext(master, appName))
        
        val data = app.processFiles(SLP.expandArgs(args))
        
        val emptyBuckets = ZoneHistogram.make(ftp)
        
        val counts = data.map((tp:Trackpoint) => (tp.watts)).countByValue
        
        val buckets = counts.foldLeft(emptyBuckets)((b, tup:Pair[Double,Long]) => {
            val (w,ct) = tup
            b.record(w,ct)
            })
        
        Console.println(buckets)
    }
}

object WaveletClusterApp extends Common {
    import scala.compat.Platform.currentTime
    import org.apache.spark.rdd.RDD
    import com.freevariable.surlaplaque.wavelets._
    import com.freevariable.surlaplaque.power.NP
    import org.apache.spark.SparkConf
    
    val OFFSET = 30
    val KEEP = 0.15
    
    def processActivities(args: Array[String]) = {
        val conf = new SparkConf()
                     .setMaster(master)
                     .setAppName(appName)
                     .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val app = new SLP(new SparkContext(conf))
        
        val data = ReplHarness.setup(args, Some(app))
        
        val apairs = data.map((tp:Trackpoint) => Pair(tp.activity.getOrElse("UNKNOWN"), Pair(tp.timestamp, tp)))
        
        val activities = apairs.keys.distinct.collect
        
        // XXX: this is somewhat bogus
        val pairs = for (activity <- activities) yield {
            val filtered = apairs.filter((tup) => {val (a,_) = tup ; a == activity})
            val timestampedTrackpoints = filtered.map((tup) => {val (_,(t,tp)) = tup; (t, tp.watts)}).sortByKey()
            val samples = timestampedTrackpoints.collect
            Pair(activity, samples.map((ttp) => ttp._2))
        }
        
        app.context.parallelize(pairs)
    }
    
    def transformWavelets(rdd: RDD[(String, Array[Double])], keep: Double) = {
        val we = new WaveletExtractor(1024, OFFSET, keep)
        rdd.flatMap((pair) => {
            val (activity, samples) = pair
            for (((wavelet, sampleWindow),idx) <- (we.transformAndAbstract(samples).zipWithIndex)) yield ((activity,idx), wavelet, sampleWindow)
        })
    }
    
    def findClusters(awpairs: RDD[((String, Int), Array[Double])]) = {
        val numClusters = getEnvValue("SLP_WAVELET_CLUSTERS", "24").toInt
        val numIterations = getEnvValue("SLP_ITERATIONS", "50").toInt

        val km = new KMeans()
        km.setK(numClusters)
        km.setMaxIterations(numIterations)
        km.run(awpairs.map((tup) => {val (_,darr) = tup ; darr}).cache)
    }
    
    def runClustering(args: Array[String]) = {
        val beforeWavelets = currentTime
        val aspairs = processActivities(args).cache
        val keep = getEnvValue("SLP_COEFFICIENT_KEEP_RATIO", KEEP.toString).toDouble
        
        val awpairs = transformWavelets(aspairs, keep)
        val afterWavelets = currentTime

        val waveletTime = afterWavelets - beforeWavelets
        Console.println(s"Wavelet transformation took $waveletTime ms")

        val beforeClustering = currentTime
        val model = findClusters(awpairs.map({case(ao,w,smps) => (ao,w)}).cache)
        val afterClustering = currentTime
        
        val clusterTime = afterClustering - beforeClustering
        Console.println(s"Cluster centroid optimization took $clusterTime ms")
        
        (aspairs, awpairs, model)
    }
            
     def main(args: Array[String]) = {
         val (aspairs, awpairs, model) = runClustering(args)
         
         val predictions = awpairs.map({case (activityAndOffset, coeffs, _s) => (model.predict(coeffs), activityAndOffset)}).sortByKey()
         val np_pairs = awpairs.map({case (activityAndOffset, _c, samples) => (activityAndOffset, NP.calculate(samples))}).collectAsMap()
         
         for (tup <- predictions.collect) {
             tup match {
                 case (ctr,ao @ (a,o)) => {
                     val hours = (o * OFFSET) / (60*60)
                     val minutes = ((o * OFFSET) / 60) - (hours*60)
                     val seconds = (o * OFFSET) % 60
                     val np = np_pairs.getOrElse(ao, -1.0)

                     Console.println("%s at offset %d:%02d:%02d (NP %f) is in cluster %d".format(a, hours, minutes, seconds, np, ctr))
                 }
             }
         }
         
         for ((activity, samples) <- aspairs) {
             val np = NP.calculate(samples)
             Console.println(s"NP for $activity is $np")
         }
     }<commit_msg>added kj calculations<commit_after>package com.freevariable.surlaplaque;

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._

import org.apache.spark.mllib.clustering._

import com.freevariable.surlaplaque.importer._
import com.freevariable.surlaplaque.data._
// import com.freevariable.surlaplaque.power._
import com.freevariable.surlaplaque.app._

object ReplHarness extends Common {
    
    def setup(args: Array[String], providedApp: Option[SLP] = None) = {
        // XXX: add optional parameters here to support cluster execution
        val app = providedApp.getOrElse(new SLP(new SparkContext(master, appName)))
        
        app.processFiles(SLP.expandArgs(args))
    }
    
    def mapActivities(args: Array[String]) = {
        val data = setup(args)
        
        val apairs = data.map((tp:Trackpoint) => Pair(tp.activity.getOrElse("UNKNOWN"), Pair(tp.timestamp, tp)))
        
        val activities = apairs.keys.distinct.collect
        
        val pairs = for (activity <- activities) yield {
            val filtered = apairs.filter((tup) => {val (a,_) = tup ; a == activity});
            val timestampedTrackpoints = filtered.map((tup) => {val (_,(t,tp)) = tup; (t, tp.watts)}).sortByKey();
            val samples = timestampedTrackpoints.collect;
            Pair(activity, samples)
        }
        
        pairs.toMap
    }
}

object BucketApp extends Common {
    
    def main(args: Array[String]) {
        // XXX: add optional parameters here to support cluster execution
        val app = new SLP(new SparkContext(master, appName))
        
        val data = app.processFiles(SLP.expandArgs(args))
        
        val emptyBuckets = ZoneHistogram.make(ftp)
        
        val counts = data.map((tp:Trackpoint) => (tp.watts)).countByValue
        
        val buckets = counts.foldLeft(emptyBuckets)((b, tup:Pair[Double,Long]) => {
            val (w,ct) = tup
            b.record(w,ct)
            })
        
        Console.println(buckets)
    }
}

object WaveletClusterApp extends Common {
    import scala.compat.Platform.currentTime
    import org.apache.spark.rdd.RDD
    import com.freevariable.surlaplaque.wavelets._
    import com.freevariable.surlaplaque.power.NP
    import org.apache.spark.SparkConf
    
    val OFFSET = 30
    val KEEP = 0.15
    
    def processActivities(args: Array[String]) = {
        val conf = new SparkConf()
                     .setMaster(master)
                     .setAppName(appName)
                     .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val app = new SLP(new SparkContext(conf))
        
        val data = ReplHarness.setup(args, Some(app))
        
        val apairs = data.map((tp:Trackpoint) => Pair(tp.activity.getOrElse("UNKNOWN"), Pair(tp.timestamp, tp)))
        
        val activities = apairs.keys.distinct.collect
        
        // XXX: this is somewhat bogus
        val pairs = for (activity <- activities) yield {
            val filtered = apairs.filter((tup) => {val (a,_) = tup ; a == activity})
            val timestampedTrackpoints = filtered.map((tup) => {val (_,(t,tp)) = tup; (t, tp.watts)}).sortByKey()
            val samples = timestampedTrackpoints.collect
            Pair(activity, samples.map((ttp) => ttp._2))
        }
        
        app.context.parallelize(pairs)
    }
    
    def transformWavelets(rdd: RDD[(String, Array[Double])], keep: Double) = {
        val we = new WaveletExtractor(1024, OFFSET, keep)
        rdd.flatMap((pair) => {
            val (activity, samples) = pair
            for (((wavelet, sampleWindow),idx) <- (we.transformAndAbstract(samples).zipWithIndex)) yield ((activity,idx), wavelet, sampleWindow)
        })
    }
    
    def findClusters(awpairs: RDD[((String, Int), Array[Double])]) = {
        val numClusters = getEnvValue("SLP_WAVELET_CLUSTERS", "24").toInt
        val numIterations = getEnvValue("SLP_ITERATIONS", "50").toInt

        val km = new KMeans()
        km.setK(numClusters)
        km.setMaxIterations(numIterations)
        km.run(awpairs.map((tup) => {val (_,darr) = tup ; darr}).cache)
    }
    
    def runClustering(args: Array[String]) = {
        val beforeWavelets = currentTime
        val aspairs = processActivities(args).cache
        val keep = getEnvValue("SLP_COEFFICIENT_KEEP_RATIO", KEEP.toString).toDouble
        
        val awpairs = transformWavelets(aspairs, keep)
        val afterWavelets = currentTime

        val waveletTime = afterWavelets - beforeWavelets
        Console.println(s"Wavelet transformation took $waveletTime ms")

        val beforeClustering = currentTime
        val model = findClusters(awpairs.map({case(ao,w,smps) => (ao,w)}).cache)
        val afterClustering = currentTime
        
        val clusterTime = afterClustering - beforeClustering
        Console.println(s"Cluster centroid optimization took $clusterTime ms")
        
        (aspairs, awpairs, model)
    }
    
     def main(args: Array[String]) = {
         val (aspairs, awpairs, model) = runClustering(args)
         
         val calculate_np = NP.calculate _
         
         val predictions = awpairs.map({case (activityAndOffset, coeffs, _s) => (model.predict(coeffs), activityAndOffset)}).sortByKey()
         val np_pairs = awpairs.map({case (activityAndOffset, _c, samples) => (activityAndOffset, calculate_np(samples))}).collectAsMap()
         val kj_rdd = awpairs.map({case (aao,c,s) => (aao,s.reduce(_ + _) / 1000.0)}).cache
         val kj_pairs = kj_rdd.collectAsMap()
         val top_kjs = kj_rdd.map({case (aao,kj) => (kj, aao)}).sortByKey().collect()
         
         for (tup <- predictions.collect) {
             tup match {
                 case (ctr,ao @ (a,o)) => {
                     val hours = (o * OFFSET) / (60*60)
                     val minutes = ((o * OFFSET) / 60) - (hours*60)
                     val seconds = (o * OFFSET) % 60
                     val np = np_pairs.getOrElse(ao, -1.0)
                     val kj = kj_pairs.getOrElse(ao, -1.0)
                     Console.println("%s at offset %d:%02d:%02d (NP %f, kj %f) is in cluster %d".format(a, hours, minutes, seconds, np, kj, ctr))
                 }
             }
         }
         
         for (tup <- top_kjs) {
           val (kj, aao) = tup
           val (a,o) = aao
           val hours = (o * OFFSET) / (60*60)
           val minutes = ((o * OFFSET) / 60) - (hours*60)
           val seconds = (o * OFFSET) % 60

           Console.println("%s at offset %d:%02d:%02d was %f kj".format(a, hours, minutes, seconds, kj))
         }
         
         for ((activity, samples) <- aspairs) {
             val np = NP.calculate(samples)
             val kj = samples.reduce(_ + _) / 1000
             Console.println(s"NP for $activity is $np ($kj kj)")
         }
     }
{'fraction_non_alphanumeric': 0.12183250927070458, 'fraction_numerical': 0.006566749072929543, 'mean_word_length': 2.2113619449268174, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2009912', 'n_tokens_mistral': 4133, 'n_tokens_neox': 3946, 'n_words': 848}