DaqToRbnb.java

package org.nees.rbnb;

import org.nees.daq.*;

import java.io.IOException;
import java.net.UnknownHostException;
import java.net.Socket;
import java.util.Hashtable;
import java.util.Calendar;
import java.util.GregorianCalendar;
import com.rbnb.sapi.*;

import com.rbnb.utility.ArgHandler; //for argument parsing
//import COM.Creare.Utility.ArgHandler; //for argument parsing

public class DaqToRbnb extends RBNBBase
{

    private ControlPort controlPort = null;
    private DataThread dataThread = null;
    private ChannelEntry[] channels = new ChannelEntry[0];
    
    private static final String DEFAULT_DAQ_SERVER_NAME = "DAQServer";
    private static final String DEFAULT_DAQ_SERVER = "localhost";
    private static final String DEFAULT_DAQ_CONTROL_PORT_NAME = "DAQControlPort";   
    private static final int DEFAULT_DAQ_CONTROL_PORT = 55055;
    private static final String DEFAULT_DAQ_DATA_PORT_NAME = "DAQDataPort";
    private static final int DEFAULT_DAQ_DATA_PORT = 55056;
    private static final String DEFAULT_RBNB_SOURCE_NAME_NAME = "SourceName";   
    private static final String DEFAULT_RBNB_SOURCE_NAME = "FromDAQ";
    private static final String CACHE_SIZE_NAME = "CacheSize";  
    private static final int DEFAULT_CACHE_SIZE=100;
    
    private String daqServerName = DEFAULT_DAQ_SERVER;
    private int daqControlPort = DEFAULT_DAQ_CONTROL_PORT;
    private int daqDataPort = DEFAULT_DAQ_DATA_PORT;

    private String rbnbSourceName = DEFAULT_RBNB_SOURCE_NAME;
    
    private int cacheSize = DEFAULT_CACHE_SIZE;

    private Socket controlSocket = null;
    private Socket dataSocket = null;
    
    private Source source;
    boolean connected = false;

    Thread mainThread;

    private String[] parameterNameArray;
    private String[] parameterTypeArray;
    private Object[] parameterDefaultArray;

    private double timeOffset = 0.0; // in seconds

    public static void main(String[] args) {
        // start from command line
        DaqToRbnb control = new DaqToRbnb();
        if (control.setArgs(args))
        {
            control.startThread();
        }
    }

    public void printUsage() {
        System.out.println("DaqToRbnb: usage is...");       
        System.out.println("DaqToRbnb ");
        super.printUsage();  // uses h, s, p
        System.out.println("[-q DAQ Server *" + DEFAULT_DAQ_SERVER + "] ");
        System.out.println("[-c DAQ Control Port *" + DEFAULT_DAQ_CONTROL_PORT + "] ");
        System.out.println("[-d DAQ Data Port *" + DEFAULT_DAQ_DATA_PORT + "] ");
        System.out.println("[-n RBNB Source Name *" + DEFAULT_RBNB_SOURCE_NAME + "] ");
        System.out.println("[-z cache size *" + DEFAULT_CACHE_SIZE + "]");
        double hours = timeOffset/(60.0*60.0);
        System.out.println("[-o time offset, floating point, hours to GMT *"+ hours +"]");
        System.out.println("[-D flag (t/f) to debug print data time stamps *f]");
        System.out.println();
    }
    
    protected boolean setInstanceArgs(String[] args) throws IllegalArgumentException
    {
        String a;
        //parse args
        try {
            ArgHandler ah=new ArgHandler(args);
            if (ah.checkFlag('q')) {
                a=ah.getOption('q');
                if (a!=null) daqServerName=a;
            }
            if (ah.checkFlag('c')) {
                a=ah.getOption('c');
                if (a!=null) daqControlPort=Integer.parseInt(a);
            }
            if (ah.checkFlag('d')) {
                a=ah.getOption('d');
                if (a!=null) daqDataPort=Integer.parseInt(a);
            }
            if (ah.checkFlag('n')) {
                a=ah.getOption('n');
                if (a!=null) rbnbSourceName=a;
            }
            if (ah.checkFlag('z')) {
                a=ah.getOption('z');
                if (a!=null)
                try
                {
                    Integer i =  new Integer(a);
                    int value = i.intValue();
                    cacheSize = value;
                }
                catch (Exception ignore) {} 
            }
            if (ah.checkFlag('o')) {
                a=ah.getOption('o');
                if (a!=null)
                try
                {
                    double value =  Double.parseDouble(a); // in hours
                    timeOffset = (long)(value*60.0*60.0); // in seconds
                }
                catch (Exception ignore) {} 
            }
            if (ah.checkFlag('D')){
                a=ah.getOption('D');
                if (a!=null)
                {
                    if (a.equals("t")) ISOtoRbnbTime.DEBUG = true;
                }
            }
        } catch (Exception e) {
            throw new IllegalArgumentException("DaqToRbnb argument exception "+e.getMessage());
        }
        System.out.println("Starting DaqToRbnb with");
        System.out.println("  DAQ: server = " + daqServerName + 
            "; control port = "+ daqControlPort + "; data port = " + daqDataPort );
        System.out.println("  RBNB: server = " + getServer()
             + "; source name = " + rbnbSourceName );
        System.out.println("  Time offset (in seconds) = " + timeOffset
            + ", which is " + timeOffset/(60.0*60.0) + " hours");
        System.out.println("  Use DaqToRbnb -h to see optional parameters");
        return true;
    }

    public DaqToRbnb() {
        computeDefaultTimeOffset();
        setParamterArrays();
        initialize();
    }

    protected void setParamterArrays() {
        String [] pNameArray = super.getBaseParameterNameArray();
        String [] pTypeArray = super.getBaseParameterTypeArray();
        Object [] pDefaultArray = super.getBaseParameterDefaultArray();
        
        int base = pNameArray.length;
        int numberOfparameters = 5;
        parameterNameArray = new String[base + numberOfparameters];
        parameterTypeArray = new String[base + numberOfparameters];
        parameterDefaultArray = new Object[base + numberOfparameters];
        
        for (int i = 0; i < base; i++)
        {
            parameterNameArray[i] = pNameArray[i];
            parameterTypeArray[i] = pTypeArray[i];
            parameterDefaultArray[i] = pDefaultArray[i];
        }
        
        parameterNameArray[base + 0] = DEFAULT_DAQ_SERVER_NAME;
        parameterTypeArray[base + 0] = RBNBBaseParameterHolder.STRING_TYPE;
        parameterDefaultArray[base + 0] = new String(DEFAULT_DAQ_SERVER);       

        parameterNameArray[base + 1] = DEFAULT_DAQ_CONTROL_PORT_NAME;
        parameterTypeArray[base + 1] = RBNBBaseParameterHolder.INTEGER_TYPE;
        parameterDefaultArray[base + 1] = new Integer(DEFAULT_DAQ_CONTROL_PORT);        

        parameterNameArray[base + 2] = DEFAULT_DAQ_DATA_PORT_NAME;
        parameterTypeArray[base + 2] = RBNBBaseParameterHolder.INTEGER_TYPE;
        parameterDefaultArray[base + 2] = new Integer(DEFAULT_DAQ_DATA_PORT);       

        parameterNameArray[base + 3] = DEFAULT_RBNB_SOURCE_NAME_NAME;
        parameterTypeArray[base + 3] = RBNBBaseParameterHolder.STRING_TYPE;
        parameterDefaultArray[base + 3] = new String(DEFAULT_RBNB_SOURCE_NAME);     

        parameterNameArray[base + 4] = CACHE_SIZE_NAME;
        parameterTypeArray[base + 4] = RBNBBaseParameterHolder.INTEGER_TYPE;
        parameterDefaultArray[base + 4] = new Integer(DEFAULT_CACHE_SIZE);      

    }

    /* (non-Javadoc)
     * @see org.nees.rbnb.RBNBBase#setVariablesFromParameters()
     */
    public void setVariablesFromParameters() {
        super.setBaseVarialbesFromParametes();

        daqServerName = DEFAULT_DAQ_SERVER;
        try{
            Object obj = getValueWithDefault(
                DEFAULT_DAQ_SERVER_NAME, DEFAULT_DAQ_SERVER);
            daqServerName = (String)obj;
        } catch (Throwable ignore){}

        daqControlPort = DEFAULT_DAQ_CONTROL_PORT;
        try{
            Integer i = new Integer(DEFAULT_DAQ_CONTROL_PORT);
            Object obj = getValueWithDefault(DEFAULT_DAQ_CONTROL_PORT_NAME, i);
            daqControlPort = ((Integer)obj).intValue();
        } catch (Throwable ignore){}

        daqDataPort = DEFAULT_DAQ_DATA_PORT;
        try{
            Integer i = new Integer(DEFAULT_DAQ_DATA_PORT);
            Object obj = getValueWithDefault(DEFAULT_DAQ_DATA_PORT_NAME, i);
            daqDataPort = ((Integer)obj).intValue();
        } catch (Throwable ignore){}

        rbnbSourceName = DEFAULT_RBNB_SOURCE_NAME;
        try{
            Object obj = getValueWithDefault(
                DEFAULT_RBNB_SOURCE_NAME_NAME, DEFAULT_RBNB_SOURCE_NAME);
            rbnbSourceName = (String)obj;
        } catch (Throwable ignore){}

        cacheSize = DEFAULT_CACHE_SIZE;
        try{
            Integer i = new Integer(DEFAULT_CACHE_SIZE);
            Object obj = getValueWithDefault(CACHE_SIZE_NAME, i);
            cacheSize = ((Integer)obj).intValue();
        } catch (Throwable ignore){}

    }

    private void computeDefaultTimeOffset() {
        Calendar calendar = new GregorianCalendar();
        long tz = calendar.get(Calendar.ZONE_OFFSET);
        long dt = calendar.get(Calendar.DST_OFFSET);
        System.out.println("Default time: Time Zone offset: "
               + (-((double)(tz/1000))/(60.0*60.0))); // in hours
        System.out.println("Defalut time: Daylight Savings Time offest (in hours): "
               + (-((double)(dt/1000))/(60.0*60.0))); // in hours
        // Time zone offset
        timeOffset = - (double)((tz + dt)/1000); // in seconds 
    }

    private void connect()
    {
        // Note: do this inside the main thread because of long delays
        // in the connection times!
        try
        {
            startDaqConnections(); // note both connects must be made first
            connectToDaqControl();
            connected = true;
        }
        catch (Throwable t) { t.printStackTrace(); }
    }
    
    private void startDaqConnections()
        throws UnknownHostException, IOException
    {
        controlSocket = new Socket(daqServerName,daqControlPort);

        try
        {
            Thread.sleep(1000);
        } catch (Exception ignore) {}
        System.out.println("...Continued");
        
        dataSocket = new Socket(daqServerName,daqDataPort);
    }

    private void connectToDaqControl()
        throws UnknownHostException, IOException
    {
        controlPort = null;
        
        ControlPort port = null;

        // use a local vaiable in case there is an exception
        port = new ControlPort(controlSocket);

        System.out.println("Pause for one sec after ControlPort...");
        try
        {
            Thread.sleep(1000);
        } catch (Exception ignore) {}
        System.out.println("...Continued");

        controlPort = port;
    }

    private void disconnect()
    {
        try {
            // unsubscribe to all channels
            for (int i = 0; i < channels.length; i ++)
            {
                channels[i].unsubscribe();
            }
            controlSocket.close();
            dataSocket.close();
            source.CloseRBNBConnection();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    private void buildChannels()
        throws IOException, SAPIException
    {
        if (controlPort == null) return;

        dataThread = new DataThread(dataSocket);
    
        source=new Source(cacheSize, "none", 0);
        source.OpenRBNBConnection(getServer(), rbnbSourceName);
    
        String[] channelList = controlPort.getChannels();
        
        channels = new ChannelEntry[channelList.length];

        for (int i = 0; i < channelList.length; i++)
        {
            String name = channelList[i];
            channels[i] = new ChannelEntry(name);
            dataThread.addListener(name, channels[i]);
        }
    }

    private void exec() throws IOException
    {
        if (controlPort == null) return;

        if (dataThread == null) return;

        for (int i = 0; i < channels.length; i++)
        {
            channels[i].subscribe();
        }

        dataThread.start();
        
    }
    
    private class ChannelEntry implements DaqListener
    {
            
        String name;
        ChannelMap map;
        boolean subscribed = false;
        int index;

        ChannelEntry(String name) throws SAPIException
        {
            this.name = name;
            map = new ChannelMap();
            index = map.Add(name);
            source.Register(map);
            System.out.println("RBNB: Connection made to server = "
                + getServer() + 
                " as source = " + rbnbSourceName + 
                " with channel = " + name + ".");
        }

        void subscribe() throws IOException
        {
            controlPort.subscribe(name);
            subscribed = true;
        }
        
        void unsubscribe() throws IOException
        {
            controlPort.unsubscribe(name);
            subscribed = false;
        }
    
        /*
            @brief Put DAQ data into turbine
            
            Implements the data listener; data arrives from DAQ and is put on RBNB.
            
            @param name Channel name
            @param time timestamp, RBNB format, in seconds
            @param data DAQ datum
            @see DaqListener#postData
            @bug the timeOffset was added when it was discovered that the DAQ controller
            did not know about time zones and was using the system time from the machine
            of origin as THE TIME.
        */
        public void postData(String name, double  time, double data)
            throws SAPIException
        {
            if (!name.equals(this.name)) return;
            
            map.PutTime(time + timeOffset, 0.0); // in seconds
            
            //map.PutTimeAuto("timeofday");
            double dataArray[] = new double[1];
            dataArray[0] = data;
            map.PutDataAsFloat64(index,dataArray);
            // System.out.println("Put Data, " + name + ": " + data);   
            source.Flush(map);
        }
    } // ChannelListener

    public void startThread()
    {       
        // Use this inner class to hide the public run method
        Runnable r = new Runnable() {
            public void run() {
              runWork();
            }
        };
        mainThread = new Thread(r, "DaqToRbnb");
        mainThread.start();
        System.out.println("DaqToRbnb: Started main thread.");
    }

    public void stopThread()
    {
        if (!connected) return;     
        mainThread.interrupt();
        System.out.println("DaqToRbnb: Stopped thread.");
    }

    private void runWork()
    {
        while(true)
        {
            connect();
            if (connected)
            {
                try
                {
                    buildChannels();
                    exec();
                    while (isRunning())
                    {   // check every 5 seconds
                        try {Thread.sleep(5*1000);} catch (Exception ignore){}
                    }
                }
                catch (Throwable t)
                {
                }
                disconnect();
                if (dataThread.isRunning())
                {
                    try{
                        dataThread.stop();
                    }
                    catch (Throwable ignore){}
                }
                dataThread = null;
            }
            // retry every 30 seconds
            try {Thread.sleep(30*1000);} catch (Exception ignore){}
        } // retry
    }

    public boolean isRunning()
    {
        return (connected && dataThread.isRunning());
    }

    /* (non-Javadoc)
     * @see org.nees.rbnb.RBNBBase#getParameterNameArray()
     */
    public String[] getParameterNameArray() {
        return parameterNameArray;
    }

    /* (non-Javadoc)
     * @see org.nees.rbnb.RBNBBase#getParameterTypeArray()
     */
    public String[] getParameterTypeArray() {
        return parameterTypeArray;
    }

    /* (non-Javadoc)
     * @see org.nees.rbnb.RBNBBase#getParameterDefaultsArray()
     */
    public Object[] getParameterDefaultArray() {
        return parameterDefaultArray;
    }

    /* (non-Javadoc)
     * @see org.nees.rbnb.RBNBBase#start()
     */
    public boolean start() {
        if (isRunning()) return false;
        if (connected) disconnect();
        setVariablesFromParameters();
        startThread();
        return true;
    }

    /* (non-Javadoc)
     * @see org.nees.rbnb.RBNBBase#stop()
     */
    public boolean stop() {
        if (!isRunning()) return false;
        stopThread();
        disconnect();
        return true;
    }
}

---