/*
 *    This program is free software; you can redistribute it and/or modify
 *    it under the terms of the GNU General Public License as published by
 *    the Free Software Foundation; either version 2 of the License, or
 *    (at your option) any later version.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU General Public License for more details.
 *
 *    You should have received a copy of the GNU General Public License
 *    along with this program; if not, write to the Free Software
 *    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
 */

/*
 *   BoundaryPanelDistrubuted.java
 *   Copyright (C) 2003 University of Waikato, Hamilton, New Zealand
 *
 */

package weka.gui.boundaryvisualizer;

import weka.classifiers.Classifier;
import weka.classifiers.AbstractClassifier;
import weka.core.FastVector;
import weka.core.Instances;
import weka.core.Utils;
import weka.experiment.Compute;
import weka.experiment.RemoteExperimentEvent;
import weka.experiment.RemoteExperimentListener;
import weka.experiment.TaskStatusInfo;

import java.awt.BorderLayout;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileReader;
import java.io.ObjectInputStream;
import java.rmi.Naming;
import java.util.Vector;

/**
 * This class extends BoundaryPanel with code for distributing the
 * processing necessary to create a visualization among a list of
 * remote machines. Specifically, a visualization is broken down and
 * processed row by row using the available remote computers.
 *
 * @author <a href="mailto:mhall@cs.waikato.ac.nz">Mark Hall</a>
 * @version $Revision: 5928 $
 * @since 1.0
 * @see BoundaryPanel
 */
public class BoundaryPanelDistributed
  extends BoundaryPanel {

  /** for serialization */
  private static final long serialVersionUID = -1743284397893937776L;

  /** a list of RemoteExperimentListeners */
  protected Vector m_listeners = new Vector();

  /** Holds the names of machines with remoteEngine servers running */
  protected Vector m_remoteHosts = new Vector();
  
  /** The queue of available hosts */
  private weka.core.Queue m_remoteHostsQueue = new weka.core.Queue();

  /** The status of each of the remote hosts */
  private int [] m_remoteHostsStatus;

  /** The number of times tasks have failed on each remote host */
  private int [] m_remoteHostFailureCounts;

  protected static final int AVAILABLE=0;
  protected static final int IN_USE=1;
  protected static final int CONNECTION_FAILED=2;
  protected static final int SOME_OTHER_FAILURE=3;

  protected static final int MAX_FAILURES=3;

  /** Set to true if MAX_FAILURES exceeded on all hosts or connections fail 
      on all hosts or user aborts plotting */
  private boolean m_plottingAborted = false;

  /** The number of hosts removed due to exceeding max failures */
  private int m_removedHosts;

  /** The count of failed sub-tasks */
  private int m_failedCount;

  /** The count of successfully completed sub-tasks */
  private int m_finishedCount;

  /** The queue of sub-tasks waiting to be processed */
  private weka.core.Queue m_subExpQueue = new weka.core.Queue();

  /** number of seconds between polling server */
  private int m_minTaskPollTime = 1000;

  private int [] m_hostPollingTime;

  /**
   * Creates a new <code>BoundaryPanelDistributed</code> instance.
   *
   * @param panelWidth width of the display
   * @param panelHeight height of the display
   */
  public BoundaryPanelDistributed(int panelWidth, int panelHeight) {
    super(panelWidth, panelHeight);
  }

  /**
   * Set a list of host names of machines to distribute processing to
   *
   * @param remHosts a Vector of host names (Strings)
   */
  public void setRemoteHosts(Vector remHosts) {
    m_remoteHosts = remHosts;
  }

  /**
   * Add an object to the list of those interested in recieving update
   * information from the RemoteExperiment
   * @param r a listener
   */
  public void addRemoteExperimentListener(RemoteExperimentListener r) {
    m_listeners.addElement(r);
  }

  protected void initialize() {
    super.initialize();

    m_plottingAborted = false;
    m_finishedCount = 0;
    m_failedCount = 0;

    // initialize all remote hosts to available
    m_remoteHostsStatus = new int [m_remoteHosts.size()];    
    m_remoteHostFailureCounts = new int [m_remoteHosts.size()];

    m_remoteHostsQueue = new weka.core.Queue();

    if (m_remoteHosts.size() == 0) {
      System.err.println("No hosts specified!");
      System.exit(1);
    }

    // prime the hosts queue
    m_hostPollingTime = new int [m_remoteHosts.size()];
    for (int i=0;i<m_remoteHosts.size();i++) {
      m_remoteHostsQueue.push(new Integer(i));
      m_hostPollingTime[i] = m_minTaskPollTime;
    }

    // set up sub taskss (just holds the row numbers to be processed
    m_subExpQueue = new weka.core.Queue();
    for (int i = 0; i < m_panelHeight; i++) {
      m_subExpQueue.push(new Integer(i));
    }
    
    try {
      // need to build classifier and data generator
      m_classifier.buildClassifier(m_trainingData);
    } catch (Exception ex) {
      ex.printStackTrace();
      System.exit(1);
    }
    
    boolean [] attsToWeightOn;
    // build DataGenerator
    attsToWeightOn = new boolean[m_trainingData.numAttributes()];
    attsToWeightOn[m_xAttribute] = true;
    attsToWeightOn[m_yAttribute] = true;
    
    m_dataGenerator.setWeightingDimensions(attsToWeightOn);    
    try {
      m_dataGenerator.buildGenerator(m_trainingData);
    } catch (Exception ex) {
      ex.printStackTrace();
      System.exit(1);
    }
  }

  /**
   * Start processing
   *
   * @exception Exception if an error occurs
   */
  public void start() throws Exception {
    // done in the sub task
    /*     m_numOfSamplesPerGenerator = 
	   (int)Math.pow(m_samplesBase, m_trainingData.numAttributes()-3); */

    m_stopReplotting = true;
    if (m_trainingData == null) {
      throw new Exception("No training data set (BoundaryPanel)");
    }
    if (m_classifier == null) {
      throw new Exception("No classifier set (BoundaryPanel)");
    }
    if (m_dataGenerator == null) {
      throw new Exception("No data generator set (BoundaryPanel)");
    }
    if (m_trainingData.attribute(m_xAttribute).isNominal() || 
	m_trainingData.attribute(m_yAttribute).isNominal()) {
      throw new Exception("Visualization dimensions must be numeric "
			  +"(BoundaryPanel)");
    }
    
    computeMinMaxAtts();
    initialize();

    // launch tasks on all available hosts
    int totalHosts = m_remoteHostsQueue.size();
    for (int i = 0; i < totalHosts; i++) {
      availableHost(-1);
      Thread.sleep(70);
    }
  }

  /**
   * Push a host back onto the list of available hosts and launch a waiting
   * Task (if any).
   *
   * @param hostNum the number of the host to return to the queue. -1
   * if no host to return.
   */
  protected synchronized void availableHost(int hostNum) {
    if (hostNum >= 0) { 
      if (m_remoteHostFailureCounts[hostNum] < MAX_FAILURES) {
	m_remoteHostsQueue.push(new Integer(hostNum));
      } else {
	notifyListeners(false,true,false,"Max failures exceeded for host "
			+((String)m_remoteHosts.elementAt(hostNum))
			+". Removed from host list.");
	m_removedHosts++;
      }
    }

    // check for all sub exp complete or all hosts failed or failed count
    // exceeded
    if (m_failedCount == (MAX_FAILURES * m_remoteHosts.size())) {
      m_plottingAborted = true;
      notifyListeners(false,true,true,"Plotting aborted! Max failures "
		      +"exceeded on all remote hosts.");
      return;
    }

    /*    System.err.println("--------------");
    System.err.println("exp q :"+m_subExpQueue.size());
    System.err.println("host list size "+m_remoteHosts.size());
    System.err.println("actual host list size "+m_remoteHostsQueue.size());
    System.err.println("removed hosts "+m_removedHosts); */
    if (m_subExpQueue.size() == 0 && 
	(m_remoteHosts.size() == 
	 (m_remoteHostsQueue.size() + m_removedHosts))) {
      if (m_plotTrainingData) {
	plotTrainingData();
      }
      notifyListeners(false,true,true,"Plotting completed successfully.");

      return;
    }


    if (checkForAllFailedHosts()) {
      return;
    }

    if (m_plottingAborted && 
	(m_remoteHostsQueue.size() + m_removedHosts) == 
	m_remoteHosts.size()) {
      notifyListeners(false,true,true,"Plotting aborted. All remote tasks "
		      +"finished.");
    }

    if (!m_subExpQueue.empty() && !m_plottingAborted) {
      if (!m_remoteHostsQueue.empty()) {
	int availHost, waitingTask;
	try {
	  availHost = ((Integer)m_remoteHostsQueue.pop()).intValue();
	  waitingTask = ((Integer)m_subExpQueue.pop()).intValue();
	  launchNext(waitingTask, availHost);
	} catch (Exception ex) {
	  ex.printStackTrace();
	}
      }
    }    
  }

  /**
   * Inform all listeners of progress
   * @param status true if this is a status type of message
   * @param log true if this is a log type of message
   * @param finished true if the remote task has finished
   * @param message the message.
   */
  private synchronized void notifyListeners(boolean status, 
					    boolean log, 
					    boolean finished,
					    String message) {
    if (m_listeners.size() > 0) {
      for (int i=0;i<m_listeners.size();i++) {
	RemoteExperimentListener r = 
	  (RemoteExperimentListener)(m_listeners.elementAt(i));
	r.remoteExperimentStatus(new RemoteExperimentEvent(status,
							   log,
							   finished,
							   message));
      }
    } else {
      System.err.println(message);
    }
  }

  /**
   * Check to see if we have failed to connect to all hosts
   */
  private boolean checkForAllFailedHosts() {
    boolean allbad = true;
    for (int i = 0; i < m_remoteHostsStatus.length; i++) {
      if (m_remoteHostsStatus[i] != CONNECTION_FAILED) {
	allbad = false;
	break;
      }
    }
    if (allbad) {
      m_plottingAborted = true;
      notifyListeners(false,true,true,"Plotting aborted! All connections "
		      +"to remote hosts failed.");
    }
    return allbad;
  }

  /**
   * Increment the number of successfully completed sub experiments
   */
  protected synchronized void incrementFinished() {
    m_finishedCount++;
  }

  /**
   * Increment the overall number of failures and the number of failures for
   * a particular host
   * @param hostNum the index of the host to increment failure count
   */
  protected synchronized void incrementFailed(int hostNum) {
    m_failedCount++;
    m_remoteHostFailureCounts[hostNum]++;
  }

  /**
   * Push an experiment back on the queue of waiting experiments
   * @param expNum the index of the experiment to push onto the queue
   */
  protected synchronized void waitingTask(int expNum) {
    m_subExpQueue.push(new Integer(expNum));
  }

  protected void launchNext(final int wtask, final int ah) {
    Thread subTaskThread;
    subTaskThread = new Thread() {
	public void run() {
	  m_remoteHostsStatus[ah] = IN_USE;
	  //	  m_subExpComplete[wtask] = TaskStatusInfo.PROCESSING;
	  RemoteBoundaryVisualizerSubTask vSubTask = 
	    new RemoteBoundaryVisualizerSubTask();
	  vSubTask.setXAttribute(m_xAttribute);
	  vSubTask.setYAttribute(m_yAttribute);
	  vSubTask.setRowNumber(wtask);
	  vSubTask.setPanelWidth(m_panelWidth);
	  vSubTask.setPanelHeight(m_panelHeight);
	  vSubTask.setPixHeight(m_pixHeight);
	  vSubTask.setPixWidth(m_pixWidth);
	  vSubTask.setClassifier(m_classifier);
	  vSubTask.setDataGenerator(m_dataGenerator);
	  vSubTask.setInstances(m_trainingData);
	  vSubTask.setMinMaxX(m_minX, m_maxX);
	  vSubTask.setMinMaxY(m_minY, m_maxY);
	  vSubTask.setNumSamplesPerRegion(m_numOfSamplesPerRegion);
	  vSubTask.setGeneratorSamplesBase(m_samplesBase);
	  try {
	    String name = "//"
	      +((String)m_remoteHosts.elementAt(ah))
	      +"/RemoteEngine";
	    Compute comp = (Compute) Naming.lookup(name);
	    // assess the status of the sub-exp
	    notifyListeners(false,true,false,"Starting row "
			    +wtask
			    +" on host "
			    +((String)m_remoteHosts.elementAt(ah)));
	    Object subTaskId = comp.executeTask(vSubTask);
	    boolean finished = false;
	    TaskStatusInfo is = null;
	    long startTime = System.currentTimeMillis();
	    while (!finished) {
	      try {
		Thread.sleep(Math.max(m_minTaskPollTime, 
				      m_hostPollingTime[ah]));
		
		TaskStatusInfo cs = (TaskStatusInfo)comp.
		  checkStatus(subTaskId);
		if (cs.getExecutionStatus() == TaskStatusInfo.FINISHED) {
		  // push host back onto queue and try launching any waiting 
		  // sub-experiments
		  long runTime = System.currentTimeMillis() - startTime;
		  runTime /= 4;
		  if (runTime < 1000) {
		    runTime = 1000;
		  }
		  m_hostPollingTime[ah] = (int)runTime;

		  // Extract the row from the result
		  RemoteResult rr =  (RemoteResult)cs.getTaskResult();
		  double [][] probs = rr.getProbabilities();
		  
		  for (int i = 0; i < m_panelWidth; i++) {
		    m_probabilityCache[wtask][i] = probs[i];
		    if (i < m_panelWidth-1) {
		      plotPoint(i, wtask, probs[i], false);
		    } else {
		      plotPoint(i, wtask, probs[i], true);
		    }
		  }
		  notifyListeners(false, true, false,  cs.getStatusMessage());
		  m_remoteHostsStatus[ah] = AVAILABLE;
		  incrementFinished();
		  availableHost(ah);
		  finished = true;
		} else if (cs.getExecutionStatus() == 
			   TaskStatusInfo.FAILED) {
		  // a non connection related error---possibly host doesn't have
		  // access to data sets or security policy is not set up
		  // correctly or classifier(s) failed for some reason
		  notifyListeners(false, true, false,  
				  cs.getStatusMessage());
		  m_remoteHostsStatus[ah] = SOME_OTHER_FAILURE;
		  //		  m_subExpComplete[wexp] = TaskStatusInfo.FAILED;
		  notifyListeners(false,true,false,"Row "+wtask
				  +" "+cs.getStatusMessage()
				  +". Scheduling for execution on another host.");
		  incrementFailed(ah);
		  // push experiment back onto queue
		  waitingTask(wtask);	
		  // push host back onto queue and try launching any waiting 
		  // Tasks. Host is pushed back on the queue as the
		  // failure may be temporary.
		  availableHost(ah);
		  finished = true;
		} else {
		  if (is == null) {
		    is = cs;
		    notifyListeners(false, true, false, cs.getStatusMessage());
		  } else {
		    RemoteResult rr = (RemoteResult)cs.getTaskResult();
		    if (rr != null) {
		      int percentComplete = rr.getPercentCompleted();
		      String timeRemaining = "";
		      if (percentComplete > 0 && percentComplete < 100) {
			double timeSoFar = (double)System.currentTimeMillis() -
			  (double)startTime;
			double timeToGo = 
			  ((100.0 - percentComplete) 
			   / (double)percentComplete) * timeSoFar;
			if (timeToGo < m_hostPollingTime[ah]) {
			  m_hostPollingTime[ah] = (int)timeToGo;
			}
			String units = "seconds";
			timeToGo /= 1000.0;
			if (timeToGo > 60) {
			  units = "minutes";
			  timeToGo /= 60.0;
			}
			if (timeToGo > 60) {
			  units = "hours";
			  timeToGo /= 60.0;
			}
			timeRemaining = " (approx. time remaining "
			  +Utils.doubleToString(timeToGo, 1)+" "+units+")";
		      }
		      if (percentComplete < 25 
			  /*&& minTaskPollTime < 30000*/) {		
			if (percentComplete > 0) {
			  m_hostPollingTime[ah] = 
			    (int)((25.0 / (double)percentComplete) * 
				  m_hostPollingTime[ah]);
			} else {
			  m_hostPollingTime[ah] *= 2;
			}
			if (m_hostPollingTime[ah] > 60000) {
			  m_hostPollingTime[ah] = 60000;
			}
		      }
		      notifyListeners(false, true, false,
				      "Row "+wtask+" "+percentComplete
				      +"% complete"+timeRemaining+".");
		    } else {
		      notifyListeners(false, true, false,
				      "Row "+wtask+" queued on "
				      +((String)m_remoteHosts.
					elementAt(ah)));
		      if (m_hostPollingTime[ah] < 60000) {
			m_hostPollingTime[ah] *= 2;
		      }
		    }

		    is = cs;
		  }
		}
	      } catch (InterruptedException ie) {
		ie.printStackTrace();
	      }
	    }
	  } catch (Exception ce) {
	    m_remoteHostsStatus[ah] = CONNECTION_FAILED;
	    m_removedHosts++;
	    System.err.println(ce);
	    ce.printStackTrace();
	    notifyListeners(false,true,false,"Connection to "
			    +((String)m_remoteHosts.elementAt(ah))
			    +" failed. Scheduling row "
			    +wtask
			    +" for execution on another host.");
	    checkForAllFailedHosts();
	    waitingTask(wtask);
	  } finally {
	    if (isInterrupted()) {
	      System.err.println("Sub exp Interupted!");
	    }
	  }
	}
      };
    subTaskThread.setPriority(Thread.MIN_PRIORITY);
    subTaskThread.start();
  }

  /**
   * Main method for testing this class
   *
   * @param args a <code>String[]</code> value
   */
  public static void main (String [] args) {
    try {
      if (args.length < 8) {
	System.err.println("Usage : BoundaryPanelDistributed <dataset> "
			   +"<class col> <xAtt> <yAtt> "
			   +"<base> <# loc/pixel> <kernel bandwidth> "
			   +"<display width> "
			   +"<display height> <classifier "
			   +"[classifier options]>");
	System.exit(1);
      }
      
      Vector hostNames = new Vector();
      // try loading hosts file
      try {
	BufferedReader br = new BufferedReader(new FileReader("hosts.vis"));
	String hostName = br.readLine();
	while (hostName != null) {
	  System.out.println("Adding host "+hostName);
	  hostNames.add(hostName);
	  hostName = br.readLine();
	}
	br.close();
      } catch (Exception ex) {
	System.err.println("No hosts.vis file - create this file in "
			   +"the current directory with one host name "
			   +"per line, or use BoundaryPanel instead.");
	System.exit(1);
      }

      final javax.swing.JFrame jf = 
	new javax.swing.JFrame("Weka classification boundary visualizer");
      jf.getContentPane().setLayout(new BorderLayout());

      System.err.println("Loading instances from : "+args[0]);
      java.io.Reader r = new java.io.BufferedReader(
			 new java.io.FileReader(args[0]));
      final Instances i = new Instances(r);
      i.setClassIndex(Integer.parseInt(args[1]));

      //      bv.setClassifier(new Logistic());
      final int xatt = Integer.parseInt(args[2]);
      final int yatt = Integer.parseInt(args[3]);
      int base = Integer.parseInt(args[4]);
      int loc = Integer.parseInt(args[5]);

      int bandWidth = Integer.parseInt(args[6]);
      int panelWidth = Integer.parseInt(args[7]);
      int panelHeight = Integer.parseInt(args[8]);

      final String classifierName = args[9];
      final BoundaryPanelDistributed bv = 
	new BoundaryPanelDistributed(panelWidth,panelHeight);
      bv.addRemoteExperimentListener(new RemoteExperimentListener() {
	  public void remoteExperimentStatus(RemoteExperimentEvent e) {
	    if (e.m_experimentFinished) {
	      String classifierNameNew = 
		classifierName.substring(classifierName.lastIndexOf('.')+1, 
					 classifierName.length());
	      bv.saveImage(classifierNameNew+"_"+i.relationName()
			   +"_X"+xatt+"_Y"+yatt+".jpg");
	    } else {
	      System.err.println(e.m_messageString);
	    }
	  }
	});
      bv.setRemoteHosts(hostNames);

      jf.getContentPane().add(bv, BorderLayout.CENTER);
      jf.setSize(bv.getMinimumSize());
      //      jf.setSize(200,200);
      jf.addWindowListener(new java.awt.event.WindowAdapter() {
	  public void windowClosing(java.awt.event.WindowEvent e) {
	    jf.dispose();
	    System.exit(0);
	  }
	});

      jf.pack();
      jf.setVisible(true);
      //      bv.initialize();
      bv.repaint();
      

      String [] argsR = null;
      if (args.length > 10) {
	argsR = new String [args.length-10];
	for (int j = 10; j < args.length; j++) {
	  argsR[j-10] = args[j];
	}
      }
      Classifier c = AbstractClassifier.forName(args[9], argsR);
      KDDataGenerator dataGen = new KDDataGenerator();
      dataGen.setKernelBandwidth(bandWidth);
      bv.setDataGenerator(dataGen);
      bv.setNumSamplesPerRegion(loc);
      bv.setGeneratorSamplesBase(base);
      bv.setClassifier(c);
      bv.setTrainingData(i);
      bv.setXAttribute(xatt);
      bv.setYAttribute(yatt);

      try {
	// try and load a color map if one exists
	FileInputStream fis = new FileInputStream("colors.ser");
	ObjectInputStream ois = new ObjectInputStream(fis);
	FastVector colors = (FastVector)ois.readObject();
	bv.setColors(colors);	
      } catch (Exception ex) {
	System.err.println("No color map file");
      }
      bv.start();
    } catch (Exception ex) {
      ex.printStackTrace();
    }
  }
}