// Path: src/main/java/smile/wide/hadoop/io/StrucLogLikeWritable.java
// public class StrucLogLikeWritable implements Writable {
// 	/** Contains SMILE XML description of network*/
// 	Text nw = new Text("");
// 	/** Writable Double that contains the network structure's score*/
// 	DoubleWritable loglik = new DoubleWritable(Double.NEGATIVE_INFINITY);
// 
// 	public StrucLogLikeWritable() {
// 		nw = new Text("");
// 		loglik = new DoubleWritable(Double.NEGATIVE_INFINITY);;
// 	}
// 
// 	public StrucLogLikeWritable(String net, double ll) {
// 		nw = new Text(net);
// 		loglik = new DoubleWritable(ll);
// 	}
// 	
// 	@Override
// 	public void readFields(DataInput in) throws IOException {
// 		nw.readFields(in);
// 		loglik.readFields(in);
// 	}
// 
// 	@Override
// 	public void write(DataOutput out) throws IOException {
// 		nw.write(out);
// 		loglik.write(out);
// 	}
// 
// 	public void setNW(Text net) {
// 		nw = net;
// 	}
// 
// 	public void setNW(String net) {
// 		nw = new Text(net);
// 	}
// 
// 	public void setLogLike(DoubleWritable ll) {
// 		loglik = ll;
// 	}
// 
// 	public void setLogLike(double ll) {
// 		loglik = new DoubleWritable(ll);
// 	}
// 	
// 	public String getNw() {
// 		return nw.toString();
// 	}
// 
// 	public double getLogLike() {
// 		return loglik.get();
// 	}
// 
// 	@Override
// 	public String toString() {
// 		return nw + "\t" + loglik;
// 	}
// }


import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import smile.Network;
import smile.learning.BayesianSearch;
import smile.learning.DataSet;
import smile.wide.hadoop.io.StrucLogLikeWritable;
/*
             Licensed to the DARPA XDATA project.
       DARPA XDATA licenses this file to you 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 smile.wide.algorithms;




/**
 * Mapper class used for running SMILE's Bayesian Search algorithm.
 * Reads datafile from distributed cache, runs BayesianSearch class
 * Result is Bayesian network in String form and Structure score
 * @author m.a.dejongh@gmail.com
 *
 */
public class SMILEBSMapper extends Mapper<LongWritable, Void, Text, StrucLogLikeWritable> {
	
	/** contains name of data file used for training*/
	private String trainfile;
	/** SMILE dataset format*/
	private DataSet ds = new DataSet();
	/** key for mapper, is current a constant*/
	private static final Text mkey = new Text("smile-bayesian-search");
	/** variable to temporarily store result of mapper */
	private StrucLogLikeWritable result = new StrucLogLikeWritable();

	//Bayesian Search parameters
	/** Number of iterations the algorithm is run, after an iteration is finished a random restart occurs*/
	private int iterationCount;
	/** Probability of generating an arc between two nodes when performing the random restart*/
	private double linkProbability;
	/** Limits the number of parents a node can have*/
	private int maxParents;
	/** Limits the amount of time the algorithm can spend on searching for a network structure (0 is unlimited)*/
	private int maxSearchTime;
	/** Influences prior probability of network structures during Bayesian Search*/
	private double priorLinkProbability;
	/** Equivalent sample size for Bayesian Search, larger values increase parameter resistance to data*/
	private int priorSampleSize;
	/** Random seed used for the random restart procedure*/
	private int randSeed;

	/** Initializes class parameters*/
	@Override
	protected void setup(Context context) {
		Configuration conf = context.getConfiguration();
		trainfile = conf.get("trainfile");
		iterationCount = conf.getInt("iterationCount",20);
		linkProbability = conf.getFloat("linkProbability", 0.01f);
		maxParents = conf.getInt("maxParents",8);
		maxSearchTime = conf.getInt("maxSearchTime",0);
		priorLinkProbability = conf.getFloat("priorLinkProbability", 0.001f);
		priorSampleSize = conf.getInt("priorSampleSize", 50);
		randSeed = 0;
		ds.readFile(trainfile);
	}
	
	/**Mapper: extracts random seed from input, initializes SMILE BayesianSearch algorithm with
	 * seed, runs BayesianSearch, stores network and Bayesian score to be sent to reducer.
	 */
	@Override
	protected void map(LongWritable key, Void value, Context context) throws IOException, InterruptedException {
		System.err.println("Running map code");
		//get seeds
		long k = key.get();
		randSeed = (int)(k & 0xffffffffL);
		double score = Double.NEGATIVE_INFINITY;

// Path: src/main/java/smile/wide/hadoop/io/StrucLogLikeWritable.java
// public class StrucLogLikeWritable implements Writable {
// 	/** Contains SMILE XML description of network*/
// 	Text nw = new Text("");
// 	/** Writable Double that contains the network structure's score*/
// 	DoubleWritable loglik = new DoubleWritable(Double.NEGATIVE_INFINITY);
// 
// 	public StrucLogLikeWritable() {
// 		nw = new Text("");
// 		loglik = new DoubleWritable(Double.NEGATIVE_INFINITY);;
// 	}
// 
// 	public StrucLogLikeWritable(String net, double ll) {
// 		nw = new Text(net);
// 		loglik = new DoubleWritable(ll);
// 	}
// 	
// 	@Override
// 	public void readFields(DataInput in) throws IOException {
// 		nw.readFields(in);
// 		loglik.readFields(in);
// 	}
// 
// 	@Override
// 	public void write(DataOutput out) throws IOException {
// 		nw.write(out);
// 		loglik.write(out);
// 	}
// 
// 	public void setNW(Text net) {
// 		nw = net;
// 	}
// 
// 	public void setNW(String net) {
// 		nw = new Text(net);
// 	}
// 
// 	public void setLogLike(DoubleWritable ll) {
// 		loglik = ll;
// 	}
// 
// 	public void setLogLike(double ll) {
// 		loglik = new DoubleWritable(ll);
// 	}
// 	
// 	public String getNw() {
// 		return nw.toString();
// 	}
// 
// 	public double getLogLike() {
// 		return loglik.get();
// 	}
// 
// 	@Override
// 	public String toString() {
// 		return nw + "\t" + loglik;
// 	}
// }


// Path: src/main/java/smile/wide/algorithms/SMILEBSMapper.java
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import smile.Network;
import smile.learning.BayesianSearch;
import smile.learning.DataSet;
import smile.wide.hadoop.io.StrucLogLikeWritable;

/*
             Licensed to the DARPA XDATA project.
       DARPA XDATA licenses this file to you 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 smile.wide.algorithms;




/**
 * Mapper class used for running SMILE's Bayesian Search algorithm.
 * Reads datafile from distributed cache, runs BayesianSearch class
 * Result is Bayesian network in String form and Structure score
 * @author m.a.dejongh@gmail.com
 *
 */
public class SMILEBSMapper extends Mapper<LongWritable, Void, Text, StrucLogLikeWritable> {
	
	/** contains name of data file used for training*/
	private String trainfile;
	/** SMILE dataset format*/
	private DataSet ds = new DataSet();
	/** key for mapper, is current a constant*/
	private static final Text mkey = new Text("smile-bayesian-search");
	/** variable to temporarily store result of mapper */
	private StrucLogLikeWritable result = new StrucLogLikeWritable();

	//Bayesian Search parameters
	/** Number of iterations the algorithm is run, after an iteration is finished a random restart occurs*/
	private int iterationCount;
	/** Probability of generating an arc between two nodes when performing the random restart*/
	private double linkProbability;
	/** Limits the number of parents a node can have*/
	private int maxParents;
	/** Limits the amount of time the algorithm can spend on searching for a network structure (0 is unlimited)*/
	private int maxSearchTime;
	/** Influences prior probability of network structures during Bayesian Search*/
	private double priorLinkProbability;
	/** Equivalent sample size for Bayesian Search, larger values increase parameter resistance to data*/
	private int priorSampleSize;
	/** Random seed used for the random restart procedure*/
	private int randSeed;

	/** Initializes class parameters*/
	@Override
	protected void setup(Context context) {
		Configuration conf = context.getConfiguration();
		trainfile = conf.get("trainfile");
		iterationCount = conf.getInt("iterationCount",20);
		linkProbability = conf.getFloat("linkProbability", 0.01f);
		maxParents = conf.getInt("maxParents",8);
		maxSearchTime = conf.getInt("maxSearchTime",0);
		priorLinkProbability = conf.getFloat("priorLinkProbability", 0.001f);
		priorSampleSize = conf.getInt("priorSampleSize", 50);
		randSeed = 0;
		ds.readFile(trainfile);
	}
	
	/**Mapper: extracts random seed from input, initializes SMILE BayesianSearch algorithm with
	 * seed, runs BayesianSearch, stores network and Bayesian score to be sent to reducer.
	 */
	@Override
	protected void map(LongWritable key, Void value, Context context) throws IOException, InterruptedException {
		System.err.println("Running map code");
		//get seeds
		long k = key.get();
		randSeed = (int)(k & 0xffffffffL);
		double score = Double.NEGATIVE_INFINITY;

		BayesianSearch bs = new BayesianSearch();