path: root/gi/posterior-regularisation/prjava/src/phrase/PhraseCluster.java

package phrase;

import gnu.trove.TIntArrayList;
import org.apache.commons.math.special.Gamma;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.PrintStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Pattern;

import phrase.Corpus.Edge;


public class PhraseCluster {
	
	public int K;
	private int n_phrases, n_words, n_contexts, n_positions;
	public Corpus c;
	public ExecutorService pool; 

	double[] lambdaPTCT;
	double[][] lambdaPT;
	boolean cacheLambda = true;

	// emit[tag][position][word] = p(word | tag, position in context)
	double emit[][][];
	// pi[phrase][tag] = p(tag | phrase)
	double pi[][];
	
	public PhraseCluster(int numCluster, Corpus corpus)
	{
		K=numCluster;
		c=corpus;
		n_words=c.getNumWords();
		n_phrases=c.getNumPhrases();
		n_contexts=c.getNumContexts();
		n_positions=c.getNumContextPositions();

		emit=new double [K][n_positions][n_words];
		pi=new double[n_phrases][K];
		
		for(double [][]i:emit)
			for(double []j:i)
				arr.F.randomise(j, true);

		for(double []j:pi)
			arr.F.randomise(j, true);
	}
	
	void useThreadPool(ExecutorService pool)
	{
		this.pool = pool;
	}

	public double EM(int phraseSizeLimit)
	{
		double [][][]exp_emit=new double [K][n_positions][n_words];
		double []exp_pi=new double[K];
		
		for(double [][]i:exp_emit)
			for(double []j:i)
				Arrays.fill(j, 1e-10);
		
		double loglikelihood=0;
		
		//E
		for(int phrase=0; phrase < n_phrases; phrase++)
		{
			if (phraseSizeLimit >= 1 && c.getPhrase(phrase).size() > phraseSizeLimit)
				continue;

			Arrays.fill(exp_pi, 1e-10);
			
			List<Edge> contexts = c.getEdgesForPhrase(phrase);

			for (int ctx=0; ctx<contexts.size(); ctx++)
			{
				Edge edge = contexts.get(ctx);
				
				double p[]=posterior(edge);
				double z = arr.F.l1norm(p);
				assert z > 0;
				loglikelihood += edge.getCount() * Math.log(z);
				arr.F.l1normalize(p);
				
				double count = edge.getCount();
				//increment expected count
				TIntArrayList context = edge.getContext();
				for(int tag=0;tag<K;tag++)
				{
					for(int pos=0;pos<n_positions;pos++){
						exp_emit[tag][pos][context.get(pos)]+=p[tag]*count;
					}
					exp_pi[tag]+=p[tag]*count;
				}
			}
			arr.F.l1normalize(exp_pi);
			System.arraycopy(exp_pi, 0, pi[phrase], 0, K);
		}

		//M
		for(double [][]i:exp_emit)
			for(double []j:i)
				arr.F.l1normalize(j);
			
		emit=exp_emit;

		return loglikelihood;
	}
	
	public double PREM(double scalePT, double scaleCT, int phraseSizeLimit)
	{
		if (scaleCT == 0)
		{
			if (pool != null)
				return PREM_phrase_constraints_parallel(scalePT, phraseSizeLimit);
			else
				return PREM_phrase_constraints(scalePT, phraseSizeLimit);
		}
		else // FIXME: ignores phraseSizeLimit
			return this.PREM_phrase_context_constraints(scalePT, scaleCT);
	}

	
	public double PREM_phrase_constraints(double scalePT, int phraseSizeLimit)
	{
		double [][][]exp_emit=new double[K][n_positions][n_words];
		double []exp_pi=new double[K];
		
		for(double [][]i:exp_emit)
			for(double []j:i)
				Arrays.fill(j, 1e-10);
		
		if (lambdaPT == null && cacheLambda)
			lambdaPT = new double[n_phrases][];
		
		double loglikelihood=0, kl=0, l1lmax=0, primal=0;
		int failures=0, iterations=0;
		long start = System.currentTimeMillis();
		//E
		for(int phrase=0; phrase<n_phrases; phrase++)
		{
			if (phraseSizeLimit >= 1 && c.getPhrase(phrase).size() > phraseSizeLimit)
			{
				//System.arraycopy(pi[phrase], 0, exp_pi[phrase], 0, K);
				continue;
			}
			
			Arrays.fill(exp_pi, 1e-10);
			
			// FIXME: add rare edge check to phrase objective & posterior processing
			PhraseObjective po = new PhraseObjective(this, phrase, scalePT, (cacheLambda) ? lambdaPT[phrase] : null);
			boolean ok = po.optimizeWithProjectedGradientDescent();
			if (!ok) ++failures;
			if (cacheLambda) lambdaPT[phrase] = po.getParameters();
			iterations += po.getNumberUpdateCalls();
			double [][] q=po.posterior();
			loglikelihood += po.loglikelihood();
			kl += po.KL_divergence();
			l1lmax += po.l1lmax();
			primal += po.primal(scalePT);
			List<Edge> edges = c.getEdgesForPhrase(phrase);

			for(int edge=0;edge<q.length;edge++){
				Edge e = edges.get(edge);
				TIntArrayList context = e.getContext();
				double contextCnt = e.getCount();
				//increment expected count
				for(int tag=0;tag<K;tag++){
					for(int pos=0;pos<n_positions;pos++){
						exp_emit[tag][pos][context.get(pos)]+=q[edge][tag]*contextCnt;
					}
					
					exp_pi[tag]+=q[edge][tag]*contextCnt;
					
				}
			}
			arr.F.l1normalize(exp_pi);
			System.arraycopy(exp_pi, 0, pi[phrase], 0, K);
		}
		
		long end = System.currentTimeMillis();
		if (failures > 0)
			System.out.println("WARNING: failed to converge in " + failures + "/" + n_phrases + " cases");
		System.out.println("\tmean iters:     " + iterations/(double)n_phrases + " elapsed time " + (end - start) / 1000.0);
		System.out.println("\tllh:            " + loglikelihood);
		System.out.println("\tKL:             " + kl);
		System.out.println("\tphrase l1lmax:  " + l1lmax);
		
		//M
		for(double [][]i:exp_emit)
			for(double []j:i)
				arr.F.l1normalize(j);
		emit=exp_emit;
		
		return primal;
	}

	public double PREM_phrase_constraints_parallel(final double scalePT, int phraseSizeLimit)
	{
		assert(pool != null);
		
		final LinkedBlockingQueue<PhraseObjective> expectations 
			= new LinkedBlockingQueue<PhraseObjective>();
		
		double [][][]exp_emit=new double [K][n_positions][n_words];
		double [][]exp_pi=new double[n_phrases][K];
		
		for(double [][]i:exp_emit)
			for(double []j:i)
				Arrays.fill(j, 1e-10);
		for(double []j:exp_pi)
			Arrays.fill(j, 1e-10);
		
		double loglikelihood=0, kl=0, l1lmax=0, primal=0;
		final AtomicInteger failures = new AtomicInteger(0);
		final AtomicLong elapsed = new AtomicLong(0l);
		int iterations=0;
		long start = System.currentTimeMillis();
		List<Future<PhraseObjective>> results = new ArrayList<Future<PhraseObjective>>();
		
		if (lambdaPT == null && cacheLambda)
			lambdaPT = new double[n_phrases][];

		//E
		for(int phrase=0;phrase<n_phrases;phrase++) {
			if (phraseSizeLimit >= 1 && c.getPhrase(phrase).size() > phraseSizeLimit) {
				System.arraycopy(pi[phrase], 0, exp_pi[phrase], 0, K);
				continue;
			}

			final int p=phrase;
			results.add(pool.submit(new Callable<PhraseObjective>() {
				public PhraseObjective call() {
					//System.out.println("" + Thread.currentThread().getId() + " optimising lambda for " + p);
					long start = System.currentTimeMillis();
					PhraseObjective po = new PhraseObjective(PhraseCluster.this, p, scalePT, (cacheLambda) ? lambdaPT[p] : null);
					boolean ok = po.optimizeWithProjectedGradientDescent();
					if (!ok) failures.incrementAndGet();
					long end = System.currentTimeMillis();
					elapsed.addAndGet(end - start);
					//System.out.println("" + Thread.currentThread().getId() + " done optimising lambda for " + p);
					return po;
				}
			}));
		}
		
		// aggregate the expectations as they become available
		for (Future<PhraseObjective> fpo : results)
		{
			try {
				//System.out.println("" + Thread.currentThread().getId() + " reading queue #" + count);

				// wait (blocking) until something is ready
				PhraseObjective po = fpo.get();
				// process
				int phrase = po.phrase;
				if (cacheLambda) lambdaPT[phrase] = po.getParameters();
				//System.out.println("" + Thread.currentThread().getId() + " taken phrase " + phrase);
				double [][] q=po.posterior();
				loglikelihood += po.loglikelihood();
				kl += po.KL_divergence();
				l1lmax += po.l1lmax();
				primal += po.primal(scalePT);
				iterations += po.getNumberUpdateCalls();

				List<Edge> edges = c.getEdgesForPhrase(phrase);
				for(int edge=0;edge<q.length;edge++){
					Edge e = edges.get(edge);
					TIntArrayList context = e.getContext();
					double contextCnt = e.getCount();
					//increment expected count
					for(int tag=0;tag<K;tag++){
						for(int pos=0;pos<n_positions;pos++){
							exp_emit[tag][pos][context.get(pos)]+=q[edge][tag]*contextCnt;
						}
						exp_pi[phrase][tag]+=q[edge][tag]*contextCnt;
					}
				}
			} catch (InterruptedException e) {
				System.err.println("M-step thread interrupted. Probably fatal!");
				throw new RuntimeException(e);
			} catch (ExecutionException e) {
				System.err.println("M-step thread execution died. Probably fatal!");
				throw new RuntimeException(e);
			}
		}
		
		long end = System.currentTimeMillis();
		
		if (failures.get() > 0)
			System.out.println("WARNING: failed to converge in " + failures.get() + "/" + n_phrases + " cases");
		System.out.println("\tmean iters:     " + iterations/(double)n_phrases + " walltime " + (end-start)/1000.0 + " threads " + elapsed.get() / 1000.0);
		System.out.println("\tllh:            " + loglikelihood);
		System.out.println("\tKL:             " + kl);
		System.out.println("\tphrase l1lmax:  " + l1lmax);
		
		//M
		for(double [][]i:exp_emit)
			for(double []j:i)
				arr.F.l1normalize(j);
		emit=exp_emit;
		
		for(double []j:exp_pi)
			arr.F.l1normalize(j);
		pi=exp_pi;
		
		return primal;
	}
	
	public double PREM_phrase_context_constraints(double scalePT, double scaleCT)
	{	
		double[][][] exp_emit = new double [K][n_positions][n_words];
		double[][] exp_pi = new double[n_phrases][K];

		//E step
		PhraseContextObjective pco = new PhraseContextObjective(this, lambdaPTCT, pool, scalePT, scaleCT);
		boolean ok = pco.optimizeWithProjectedGradientDescent();
		if (cacheLambda) lambdaPTCT = pco.getParameters();

		//now extract expectations
		List<Corpus.Edge> edges = c.getEdges();
		for(int e = 0; e < edges.size(); ++e)
		{
			double [] q = pco.posterior(e);
			Corpus.Edge edge = edges.get(e);

			TIntArrayList context = edge.getContext();
			double contextCnt = edge.getCount();
			//increment expected count
			for(int tag=0;tag<K;tag++)
			{
				for(int pos=0;pos<n_positions;pos++)
					exp_emit[tag][pos][context.get(pos)]+=q[tag]*contextCnt;
				exp_pi[edge.getPhraseId()][tag]+=q[tag]*contextCnt;
			}
		}
		
		System.out.println("\tllh:            " + pco.loglikelihood());
		System.out.println("\tKL:             " + pco.KL_divergence());
		System.out.println("\tphrase l1lmax:  " + pco.phrase_l1lmax());
		System.out.println("\tcontext l1lmax: " + pco.context_l1lmax());
		
		//M step
		for(double [][]i:exp_emit)
			for(double []j:i)
				arr.F.l1normalize(j);
		emit=exp_emit;
		
		for(double []j:exp_pi)
			arr.F.l1normalize(j);
		pi=exp_pi;
		
		return pco.primal();
	}	
		
	/**
	 * @param phrase index of phrase
	 * @param ctx array of context
	 * @return unnormalized posterior
	 */
	public double[] posterior(Corpus.Edge edge) 
	{
		double[] prob;
		
		if(edge.getTag()>=0){
			prob=new double[K];
			prob[edge.getTag()]=1;
			return prob;
		}
		
		if (edge.getPhraseId() < n_phrases)
			prob = Arrays.copyOf(pi[edge.getPhraseId()], K);
		else
		{
			prob = new double[K];
			Arrays.fill(prob, 1.0);
		}
		
		TIntArrayList ctx = edge.getContext();
		for(int tag=0;tag<K;tag++)
		{
			for(int c=0;c<n_positions;c++)
			{
				int word = ctx.get(c);
				if (!this.c.isSentinel(word) && word < n_words)
					prob[tag]*=emit[tag][c][word];
			}
		}
		
		return prob;
	}
	
	public void displayPosterior(PrintStream ps, List<Edge> testing)
	{	
		for (Edge edge : testing)
		{
			double probs[] = posterior(edge);
			arr.F.l1normalize(probs);

			// emit phrase
			ps.print(edge.getPhraseString());
			ps.print("\t");
			ps.print(edge.getContextString(true));
			int t=arr.F.argmax(probs);
			ps.println(" ||| C=" + t + " T=" + edge.getCount() + " P=" + probs[t]);
			//ps.println("# probs " + Arrays.toString(probs));
		}
	}
	
	public void displayModelParam(PrintStream ps)
	{
		final double EPS = 1e-6;
		ps.println("phrases " + n_phrases + " tags " + K + " positions " + n_positions);
		
		for (int i = 0; i < n_phrases; ++i)
			for(int j=0;j<pi[i].length;j++)
				if (pi[i][j] > EPS)
					ps.println(i + " " + j + " " + pi[i][j]);

		ps.println();
		for (int i = 0; i < K; ++i)
		{
			for(int position=0;position<n_positions;position++)
			{
				for(int word=0;word<emit[i][position].length;word++)
				{
					if (emit[i][position][word] > EPS)
						ps.println(i + " " + position + " " + word + " " + emit[i][position][word]);
				}
			}
		}
	}
	
	double phrase_l1lmax()
	{
		double sum=0;
		for(int phrase=0; phrase<n_phrases; phrase++)
		{
			double [] maxes = new double[K];
			for (Edge edge : c.getEdgesForPhrase(phrase))
			{
				double p[] = posterior(edge);
				arr.F.l1normalize(p);
				for(int tag=0;tag<K;tag++)
					maxes[tag] = Math.max(maxes[tag], p[tag]);
			}
			for(int tag=0;tag<K;tag++)
				sum += maxes[tag];
		}
		return sum;
	}

	double context_l1lmax()
	{
		double sum=0;
		for(int context=0; context<n_contexts; context++)
		{
			double [] maxes = new double[K];
			for (Edge edge : c.getEdgesForContext(context))
			{
				double p[] = posterior(edge);
				arr.F.l1normalize(p);
				for(int tag=0;tag<K;tag++)
					maxes[tag] = Math.max(maxes[tag], p[tag]);
			}
			for(int tag=0;tag<K;tag++)
				sum += maxes[tag];
		}
		return sum;
	}

	public void loadParameters(BufferedReader input) throws IOException
	{	
		final double EPS = 1e-50;
		
		// overwrite pi, emit with ~zeros
		for(double [][]i:emit)
			for(double []j:i)
				Arrays.fill(j, EPS);

		for(double []j:pi)
			Arrays.fill(j, EPS);

		String line = input.readLine();
		assert line != null;

		Pattern space = Pattern.compile(" +");
		String[] parts = space.split(line);
		assert parts.length == 6;

		assert parts[0].equals("phrases");
		int phrases = Integer.parseInt(parts[1]);
		int tags = Integer.parseInt(parts[3]);
		int positions = Integer.parseInt(parts[5]);
		
		assert phrases == n_phrases;
		assert tags == K;
		assert positions == n_positions;

		// read in pi
		while ((line = input.readLine()) != null)
		{
			line = line.trim();
			if (line.isEmpty()) break;
			
			String[] tokens = space.split(line);
			assert tokens.length == 3;
			int p = Integer.parseInt(tokens[0]);
			int t = Integer.parseInt(tokens[1]);
			double v = Double.parseDouble(tokens[2]);

			pi[p][t] = v;
		}
		
		// read in emissions
		while ((line = input.readLine()) != null)
		{
			String[] tokens = space.split(line);
			assert tokens.length == 4;
			int t = Integer.parseInt(tokens[0]);
			int p = Integer.parseInt(tokens[1]);
			int w = Integer.parseInt(tokens[2]);
			double v = Double.parseDouble(tokens[3]);

			emit[t][p][w] = v;
		}
	}
}