boundarypaneldistributed.java

一个数据挖掘软件ALPHAMINERR的整个过程的JAVA版源代码

  }

  /**
   * 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 = Classifier.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();
    }
  }
}