GrabDataMultipleSink.java

/*
 * Created on April 16, 2004
 */
package org.nees.rbnb;

import java.io.IOException;
import java.io.PrintWriter;
import java.io.FileWriter;
import java.io.File;
import java.util.Vector;
import java.util.Iterator;
import java.util.Stack;
import java.util.Date;
import java.util.TimeZone;
import java.text.SimpleDateFormat;

import java.util.regex.*;

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

public class GrabDataMultipleSink {
    
    private static final String SERVER_NAME = "localhost";
    private static final String SERVER_PORT = "3333";
    private static final String SINK_NAME = "GrabDataMutiple";
    private static final String SEARCH_CHANNEL_PATH = ".*";
    private static final String ARCHIVE_DIRECTORY = ".";
    
    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("MMM d, yyyy h:mm aa");
    private static final TimeZone TZ = TimeZone.getTimeZone("GMT");

    private static final SimpleDateFormat INPUT_FORMAT = new SimpleDateFormat("yyyy-MM-dd;hh:mm:ss.SSS");

    static
    {
        DATE_FORMAT.setTimeZone(TZ);
        INPUT_FORMAT.setTimeZone(TZ);
    }

    private String serverName = SERVER_NAME;
    private String serverPort = SERVER_PORT;
    private String server = serverName + ":" + serverPort;
    private String sinkName = SINK_NAME;
    private String searchChannelPath = SEARCH_CHANNEL_PATH;
    private String archiveDirectory = ARCHIVE_DIRECTORY;
    private boolean includeHidden = false;
    
    private Sink sink = null;
    private ChannelMap sMap;
    private int index;
    
    private boolean useStart = false;
    private double startTime;
    private boolean useDuration = false;
    private double endTime;
    private double duration;
    
    public static void main(String[] args) {
        GrabDataMultipleSink s = new GrabDataMultipleSink(args);
        s.exec();
    } // main
    
    private void printUsage() {
        System.out.println("GrabDataMultipleSink: usage is...");        
        System.out.println("GrabDataMultipleSink ");
        System.out.println("[-s Server Hostname *" + SERVER_NAME + "] ");
        System.out.println("[-o Server Port Number *" + SERVER_PORT + "] ");
        System.out.println("[-k Sink Name *" + SINK_NAME + " ]");
        System.out.println("[-c Channel Path Pattern *" + SEARCH_CHANNEL_PATH + "] ");
        System.out.println("[-d Archive directory root *" + ARCHIVE_DIRECTORY + "]");
        System.out.println("[-x Include Hidden (true|false) *false]");
        System.out.println("[-a Start Time *now]");
        System.out.println("[-z End Time *<unspecified>]");
        System.out.println("Note: times can either be yyyy-mm-dd;hh:mm:ss.nnn or");
        System.out.println("an arbitraty floating point number");
    } // printUsage
    
    public GrabDataMultipleSink(String[] args) {
        //parse args
        try {
            ArgHandler ah=new ArgHandler(args);
            if (ah.checkFlag('h')) {
                printUsage();
                System.exit(0);             
            }
            if (ah.checkFlag('s')) {
                String a=ah.getOption('s');
                if (a!=null) serverName=a;
            }
            if (ah.checkFlag('p')) {
                String a=ah.getOption('p');
                if (a!=null) serverPort=a;
            }
            if (ah.checkFlag('c')) {
                String a=ah.getOption('c');
                if (a!=null) searchChannelPath=a;
            }
            if (ah.checkFlag('k')) {
                String a=ah.getOption('k');
                if (a!=null) sinkName=a;
            }
            if (ah.checkFlag('d')) {
                String a=ah.getOption('d');
                if (a!=null) archiveDirectory=a;
            }
            if (ah.checkFlag('x')) {
                String a=ah.getOption('x');
                if (a!=null)
                {
                    if (a.equals("true")) includeHidden = true;
                    if (a.equals("false")) includeHidden = false;
                } 
            }
            if (ah.checkFlag('a')) {
                String a=ah.getOption('a');
                if (a!=null)
                {
                    try
                    {
                        double value = getTimeOrDouble(a);
                        startTime = value;
                        useStart = true;
                    }
                    catch (Exception ingore) {}
                }
            }
            if (ah.checkFlag('z')) {
                String a=ah.getOption('z');
                if (a!=null)
                {
                    try
                    {
                        double value = getTimeOrDouble(a);
                        endTime = value;
                        useDuration = true;
                    }
                    catch (Exception ingore) {}
                }
            }
        } catch (Exception e) {
            System.err.println("GrabDataMultipleSink argument exception "+e.getMessage());
            e.printStackTrace();
            System.exit(0);
        }
        
        if (useDuration && !useStart) // gat a endTime without a startTime
        {
            useDuration = false;
            System.out.println("Got End Time without a Start Time; ignored End Time.");
        }
        if (useStart && useDuration && (endTime < startTime))
        {
            useStart = false;
            useDuration = false;
            System.out.println("End Time is less then Start Time; ignoring both.");
        }
        
        if (useDuration)
        {
            duration = endTime - startTime;
        }

        server = serverName + ":" + serverPort;
        
        System.out.println("Starting GrabDataMultipleSink on " + server + " as " + sinkName);
        System.out.println("  Requesting " + searchChannelPath);
        System.out.println("  Use GrabDataMultipleSink -h to see optional parameters");
    } // GrabDataMultipleSink
    
    public void exec()
    {
        try {
            // Create a sink and connect:
            sink=new Sink();
            sink.OpenRBNBConnection(server,sinkName);
            
            // get all the channel paths that match the pattern
            Vector collection = new Vector();
            sMap = new ChannelMap();
            sink.RequestRegistration();     
            sMap = sink.Fetch(-1,sMap);
            ChannelTree tree = ChannelTree.createFromChannelMap(sMap);

            Pattern p = Pattern.compile(searchChannelPath);
            // for each channel path, check match, collect matches...
            
            Iterator nodes = tree.iterator();
            while (nodes.hasNext())
            {
                ChannelTree.Node n = (ChannelTree.Node)nodes.next();
System.out.println("Checking " + n.getFullName() + ";" + n.getName());
                if (!includeHidden && n.getFullName().startsWith("_")) continue;
                if (n.getType() != ChannelTree.CHANNEL) continue;
                String name = n.getFullName();
                Matcher m = p.matcher(name);
                if (m.matches())
                {
System.out.println("Matches");
                    boolean isSource = false;
                    ChannelTree.Node upNode = n.getParent();
                    while ((!isSource) || (upNode != null))
                    {
                        if (upNode.getType() == ChannelTree.SOURCE) isSource = true;
                        upNode = upNode.getParent();
                    }
                    if (isSource)
                    {
System.out.println("... and is a source.");
                        collection.add(n);
                    }
                    else
System.out.println("... and is NOT a source.");
                }                   
            }
            
            Iterator channels = collection.iterator();

            if (channels.hasNext())
            {
                System.out.println("GrabDataMultipleSink: Intending connectsions to server = "
                    + server + " as " + sinkName + "monotering channels: ");
                while (channels.hasNext())
                {
                    ChannelTree.Node candidate = (ChannelTree.Node)channels.next();
                    System.out.println(candidate.getFullName());
                    new GrabDataSink(candidate);
                }
            }
            else
                System.out.println("GrabDataMultipleSink: No channels to monitor.");
            sink.CloseRBNBConnection();
            
        } catch (SAPIException se) { se.printStackTrace(); }
    } // exec

    private double getTimeOrDouble(String arg) throws Exception
    {
        double value = 0.0;
        boolean gotit = false;
                
        try{
            Date d = INPUT_FORMAT.parse(arg);
            long t = d.getTime();
            value = ((double)t)/1000.0;
            gotit = true;
        } catch (Exception ignore)
        {
            gotit = false;
        }

        if (!gotit)
        try {
            value = Double.parseDouble(arg);
            gotit = true;
        } catch (Exception ignore)
        {
            gotit = false;
        }

        if (!gotit) throw(new Exception("Failed to parse time " + arg));        
        
        return value;

    } // getTimeOrDouble
    
    private class GrabDataSink
    {   
        Sink sink = null;
        ChannelMap sMap;
        int index;
        
        String name;
        PrintWriter out = null;
        
        boolean connected = false;
    
        Thread stringDataThread;
        boolean runit = false;

        ChannelTree.Node itsNode;

        public GrabDataSink(ChannelTree.Node node)
        {
            itsNode = node;
            name = itsNode.getFullName().replace('/','_');
            if (connect(name))
            {
                connected = true;
                if (createFile())
                    writeFileHeader();
                    startThread();
            }
        } // GrabDataSink
        
        private void writeFileHeader() {
            // TODO Auto-generated method stub
            
        } // writeFileHeader()

        private boolean connect(String name)
        {
            try
            {
                sink=new Sink();
                sink.OpenRBNBConnection(server,sinkName + "_" + name);
                sMap = new ChannelMap();
                index = sMap.Add(itsNode.getFullName());
                sink.Subscribe(sMap,"newest");
            }
            catch (Throwable th)
            {
                System.out.println("Failed connection for " + name
                    + "; reason: " + th);
                return false;           
            }
            System.out.println("Sucessful connection for " + name);
            return true;
        } // connect
    
        private boolean createFile()
        {
            ChannelTree.Node p = itsNode.getParent();
            Stack st = new Stack();
            
            // get the path for the file
            while (p != null)
            {
                st.push(p);
                p = p.getParent();
            }
            
            // make sure that the directories on the path exist
            String path = archiveDirectory;
            while (!st.empty())
            {
                p = (ChannelTree.Node)st.pop();
                path = path + File.separator + p.getName();
            }
            File f = new File(path);
System.out.println("Appempting to create the directory path " + path);
            f.mkdirs();
            
            // Open file on path
            path = path + File.separator + itsNode.getName();
System.out.println("Trying to create file for path "+ path);
            try {
                out = new PrintWriter(new FileWriter(path));
            } catch (IOException e) {
                System.out.println("Can not create file " + e);
                return false;
            }
            return true;
        } // createFile
        
        private void startThread()
        {
            
            if (!connected) return;
            
            // Use this inner class to hide the public run method
            Runnable r = new Runnable() {
                public void run() {
                  runWork();
                }
            };
            runit = true;
            stringDataThread = new Thread(r, itsNode.getName());
            stringDataThread.start();
            System.out.println("GrabDataSink: Started thread for " + itsNode.getFullName());
        } // startThread
    
        public void stopThread()
        {
            runit = false;
            stringDataThread.interrupt();
            System.out.println("GrabDataSink: Stopped thread for " + itsNode.getFullName());
        } // stopThread
        
        private void runWork ()
        {
            try {
                while(isRunning())
                {
                    ChannelMap m = sink.Fetch(-1);
                    if (m == null)
                    {
                        System.out.println("Data fetch failed for " + itsNode.getFullName());
                        continue;
                    }
                    double startTime = m.GetTimeStart(index);
                    double duration = m.GetTimeDuration(index);
                    double[] data = m.GetDataAsFloat64(index);
                    long unixTime = (long)(startTime * 1000.0); // convert sec to millisec
                    String time1 = DATE_FORMAT.format(new Date(unixTime));
                    unixTime = (long)((startTime + duration) * 1000.0);
                    String time2 = DATE_FORMAT.format(new Date(unixTime));
                    System.out.println("Data Received (for " 
                        + time1 + " to " + time2
                        + "GMT -- " + itsNode.getFullName() + "): ");
                    out.print(startTime + " " + duration);
                    for (int i = 0; i < data.length; i++)
                    {
                        System.out.println("  " + i + ": " + data[i]);
                        out.print(" " + data[i]);
                    }
                    out.println();
                    out.flush();
                }
            } catch (SAPIException se) {
                se.printStackTrace(); 
            }
            stringDataThread = null;
        } // runWork
        
        private boolean isRunning()
        {
            return (connected && runit);
        } // isRunning
    
    } // class GrabDataSink

}

---