DataVideoSink.java

/*
 * Created on May 07, 2004, Terry E. Weymouth
 * ***********************************************************************************
 * CVS Header: $Header: /disks/cvs/neesgrid/turbine/src/org/nees/rbnb/DataVideoSink.java,v 1.9 2004/07/22 18:50:27 hubbard Exp $
 * ***********************************************************************************
 */
package org.nees.rbnb;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import javax.media.protocol.PullBufferDataSource;
import javax.media.protocol.PullBufferStream;

import java.util.Date;
import java.util.TimeZone;
import java.text.SimpleDateFormat;

import java.awt.Dimension;
import java.awt.image.BufferedImage;
import java.awt.Image;

import javax.media.MediaLocator;
import javax.media.Time;
import javax.media.*;
import javax.media.control.*;
import javax.media.protocol.*;
import javax.media.protocol.DataSource;
import javax.media.datasink.*;
import javax.media.format.VideoFormat;
import javax.media.format.AudioFormat;
import javax.imageio.ImageIO;
import com.sun.image.codec.jpeg.*;

import com.sun.media.multiplexer.*;

import com.sun.image.codec.jpeg.ImageFormatException;
import com.sun.image.codec.jpeg.JPEGCodec;
import com.sun.image.codec.jpeg.JPEGImageDecoder;

import com.rbnb.sapi.*;

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

public class DataVideoSink {
    
    double[] dataArray = new double[0];
    double[] timeArray = new double[0];
    
    Sink dataSink;
    Sink videoSink;

    boolean connected = false;
    
    int videoWidth = 480;
    int videoHeight = 320;
    
    byte [] prevImage = (byte[]) null;
    
    /* ***********           for testing               **************** */
    public static boolean DEBUG = false;

    private static final String SERVER_NAME = "neestpm.mcs.anl.gov";
    private static final String SERVER_PORT = "3333";
    private static final String SINK_NAME = "DataVideoSink";
    
    private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("MMM d, yyyy h:mm:ss.SSS 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 static double USER_GIVEN_FRAME_RATE = 1.0;
    private static int ITEMS_TO_SKIP = 0;

    private static String t_sinkName = SINK_NAME;
    private static String t_dataSourcePath = null;
    private static String t_videoSourcePath = null;
    private static double t_startTime = -1;
    private static double t_duration = -1;
    private static double t_userGivenFrameRate = USER_GIVEN_FRAME_RATE;
    private static int t_itemsToSkip = ITEMS_TO_SKIP;
     
    private static String t_outputURL;

    private static String t_serverName = SERVER_NAME;
    private static String t_serverPort = SERVER_PORT;
    private static String t_server = t_serverName + ":" + t_serverPort;

    // this problem should be fixed
//  private static double uglyOffset = 5.0;
//  
//  public static void setUglyOffest(double value)
//  {
//      uglyOffset = value;
//  }
//  
//  public static double getUglyOffest()
//  {
//      return uglyOffset;
//  }

    public static void main(String[] args) {

        try {
            if (setArgs(args))
            {
                DataVideoSink s = new DataVideoSink();
                if (s.connect(t_server,t_sinkName))
                {
                    System.out.println("DataVideoSink: Connection made to server = "
                        + t_server + " as " + t_sinkName + " for " + t_dataSourcePath
                        + " and as " + t_sinkName + "_video" + " for " + t_videoSourcePath + ".");
                    System.out.println("Set to request a fetch of time and duration = " 
                        + t_startTime + "," + t_duration);

                    s.fetchData(t_dataSourcePath, t_startTime, t_duration);

                    int items = s.getNumberOfItems();
                    System.out.println("Number of Items is " + items);                  

                    s.makeMovie((float)t_userGivenFrameRate, 
                        t_outputURL, t_videoSourcePath, t_itemsToSkip);

                    double [] times = s.getTimeArray();
                    double [] data = s.getDataArray();
                    
                    DEBUG = true;
                    
                    System.out.println("Number of Items is " + items);                  
                    for (int i = 0; i < items; i++)
                    {
                        long unixTime = (long)(times[i] * 1000.0); // convert sec to millisec
                        String time = DATE_FORMAT.format(new Date(unixTime));

                        System.out.println("  " + i 
                            + " ("+ times[i] + " - " + time + "): " 
                            + data[i]); 
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            printUsage();
        }
        System.exit(0);
    }
    
    private static void printUsage() {
        System.out.println("DataVideoSink: usage is... (for testing only)");        
        System.out.println("DataVideoSink ");
        System.out.println("[-s Server Hostname *" + SERVER_NAME + "] ");
        System.out.println("[-p Server Port Number *" + SERVER_PORT + "] ");
        System.out.println("[-k Sink Name *" + SINK_NAME + " ]");
        System.out.println("[-f Frame Rate in final movie *" + USER_GIVEN_FRAME_RATE + "]");
        System.out.println("[-i Items to skip in data for each movie frame *" + 
            ITEMS_TO_SKIP +"]");
        System.out.println("-d Data Source Path - required");
        System.out.println("-v Video Source path - required");
        System.out.println("-a Start Time - required");
        System.out.println("-z End Time - required");
        System.out.println("-m Movie output file URL - required");
//      System.out.println("[-u magic offset *"+ uglyOffset + "]");
        System.out.println("Note: times can either be yyyy-mm-dd:hh:mm:ss.nnn or");
        System.out.println("an arbitraty floating point number");
    }
    
    public static boolean setArgs(String[] args)
        throws Exception
    {
        double t_endTime = -1;
        //parse args
        try {
            ArgHandler ah=new ArgHandler(args);
            if (ah.checkFlag('h')) {
                printUsage();
                return false;
            }
            if (ah.checkFlag('s')) {
                String a=ah.getOption('s');
                if (a!=null) t_serverName=a;
            }
            if (ah.checkFlag('p')) {
                String a=ah.getOption('p');
                if (a!=null) t_serverPort=a;
            }
            if (ah.checkFlag('k')) {
                String a=ah.getOption('k');
                if (a!=null) t_sinkName=a;
            }
            if (ah.checkFlag('d')) {
                String a=ah.getOption('d');
                if (a!=null) t_dataSourcePath=a;
            }
            if (ah.checkFlag('v')) {
                String a=ah.getOption('v');
                if (a!=null) t_videoSourcePath=a;
            }
            if (ah.checkFlag('m')) {
                String a=ah.getOption('m');
                if (a!=null) t_outputURL=a;
            }
            if (ah.checkFlag('a')) {
                String a=ah.getOption('a');
                if (a!=null)
                {
                    try
                    {
                        double value = getTimeOrDouble(a);
                        t_startTime = value;
                    }
                    catch (Exception ex)
                    {
                        System.out.println("Failed to parse start time (" + a + "): " + ex);
                    }
                }
            }
            if (ah.checkFlag('z')) {
                String a=ah.getOption('z');
                if (a!=null)
                {
                    try
                    {
                        double value = getTimeOrDouble(a);
                        t_endTime = value;
                    }
                    catch (Exception ex)
                    {
                        System.out.println("Failed to parse end time (" + a + "): " + ex);
                    }
                }
            }
            if (ah.checkFlag('f')) {
                String a=ah.getOption('f');
                if (a!=null)
                {
                    try
                    {
                        double value = Double.parseDouble(a);
                        t_userGivenFrameRate = value;
                    }
                    catch (Exception ex)
                    {
                        System.out.println("Failed to parse Frame Rate (" + a + "): " + ex);
                    }
                }
            }
//          if (ah.checkFlag('u')) {
//              String a=ah.getOption('u');
//              if (a!=null)
//              {
//                  try
//                  {
//                      double value = Double.parseDouble(a);
//                      uglyOffset = value;
//                  }
//                  catch (Exception ex)
//                  {
//                      System.out.println("Failed to parse magic offset (" + a + "): " + ex);
//                  }
//              }
//          }
            if (ah.checkFlag('i')) {
                String a=ah.getOption('i');
                if (a!=null)
                {
                    try
                    {
                        int value = Integer.parseInt(a);
                        t_itemsToSkip = value;
                    }
                    catch (Exception ex)
                    {
                        System.out.println("Failed to parse Frame Rate (" + a + "): " + ex);
                    }
                }
            }
        } catch (Exception e) {
            throw new Exception("Arg setting failed: " + e);
        }

        if ((t_startTime > 0) && (t_endTime > 0))
        {
            t_duration = t_endTime - t_startTime;
            if (t_duration < 0)
            {
                System.out.println("End Time must be after Start Time.");
                printUsage();
                return false;
            }
        }

        t_server = t_serverName + ":" + t_serverPort;
        
        if (t_startTime < 0)
        {
            System.out.println("Start Time is required.");
            printUsage();
            return false;
        }

        if (t_duration < 0 ) 
        {
            System.out.println("End Time is required.");
            printUsage();
            return false;
        }

        if (t_dataSourcePath == null)
        {
            System.out.println("Data Source Path is required.");
            printUsage();
            return false;
        }

        if (t_videoSourcePath == null)
        {
            System.out.println("Video Source Path is required.");
            printUsage();
            return false;
        }
        
        if (t_outputURL == null)
        {
            System.out.println("Output File URL is required.");
            printUsage();
            return false;
        }

        return true;
    } // setArgs

    public static double getTimeOrDouble(String arg) throws Exception
    {
        double value = 0.0;
        boolean gotit = false;
        String reason = null;
                
        try{
            Date d = INPUT_FORMAT.parse(arg);
            long t = d.getTime();
            value = ((double)t)/1000.0;
            gotit = true;
        } catch (Exception e1)
        {
            reason = e1.toString();
            gotit = false;
        }

        if (!gotit)
        try {
            value = Double.parseDouble(arg);
            gotit = true;
        } catch (Exception e2)
        {
            reason = reason + "; " + e2.toString();
            gotit = false;
        }

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

    } // getTimeOrDouble

/* ***********       end of "for testing"             **************** */

    public boolean connect(String server, String sinkName)
    {
        ChannelMap sMap;
        try {
            // Create a sinks and connect:
            dataSink = new Sink();
            dataSink.OpenRBNBConnection(server,sinkName);
            
            videoSink = new Sink();
            videoSink.OpenRBNBConnection(server,sinkName + "_video");
            
            connected = true;
        } catch (SAPIException se) { se.printStackTrace(); }
        return connected;
    } // connect()

    public void disconnect()
    {
        dataSink.CloseRBNBConnection();
        videoSink.CloseRBNBConnection();
        connected = false;
    }
    public void fetchData(String dataSourcePath, double startTime, double duration)
    {
        fetchData(dataSourcePath, startTime, duration, "absolute");
    }
    
    public void fetchData(String dataSourcePath,
        double startTime, double duration, String requestType)
    {
        dataArray = new double[0];
        timeArray = new double[0];
        
        if (!connected) return;

        if (DEBUG) System.out.println("Attempting fetch (at "
            + dataSourcePath + ") with: "
            + "start time = " + startTime + ", "
            + "duration = " + duration + ", "
            + "request type = " + requestType + ". "
            );
        try {
            ChannelMap dataMap = new ChannelMap();
            int dataIndex = dataMap.Add(dataSourcePath);
            dataSink.Request(dataMap, startTime, duration, requestType);
            ChannelMap res = dataSink.Fetch(-1,dataMap);
            if (res != null)
            {
                if (res.NumberOfChannels() < 1)
                {
                    if (DEBUG) System.out.println("no channels returned in fetch");
                    return;
                }
                dataArray = res.GetDataAsFloat64(dataIndex);
                timeArray = res.GetTimes(dataIndex);
            }
        } catch (SAPIException se) {
            se.printStackTrace();
            return;
        } catch (IllegalStateException se) {
            // not connected ot RBNB server
            se.printStackTrace(); 
            return;
        } catch (IllegalArgumentException se) {
            // bad parameters on Request
            se.printStackTrace(); 
            return;
        }
        
    } // fatchData
    public void makeMovie(float frameRate,
        String oURL, String videoSourcePath,
        int dataItemsToSkip) throws Exception
    {

        setWidthHeight(videoSourcePath);
        
        PullBufferDataSource ds = (PullBufferDataSource)
            new RBNBImageData(
                getWidth(), getHeight(), frameRate,
                videoSourcePath,dataItemsToSkip);

        // Generate the output media locators.
        JpegImagesToMovie imageToMovie = new JpegImagesToMovie();
        MediaLocator oml;
        
        if ((oml = imageToMovie.createMediaLocator(oURL)) == null) {
            throw new Exception("Cannot build media locator from: " + oURL);
        }
        
        imageToMovie.makeMovieFromPullBufferDataSource(
            getWidth(), getHeight(), frameRate, ds, oml);
                        
    }

    public int getNumberOfItems() {
        return timeArray.length;
    } // getNumberOfItems

    public double[] getTimeArray() {
        return timeArray;
    } // getTimeArray

    public double[] getDataArray() {
        return dataArray;
    } // getDataArray
    
    
    private int getWidth() {
        return videoWidth;
    }

    private int getHeight() {
        return videoHeight;
    }
    
    private void setWidth(int val)
    {
        videoWidth = val;
    }

    private void setHeight(int val)
    {
        videoHeight = val;
    }

    private void setWidthHeight(String videoSourcePath)
    {
        try
        {
            if (timeArray.length > 0)
            {
                // ugly correction -- this should be fixed!
//              double UGLY_CORRECTION = uglyOffset * 60.0 * 60.0; // uglyOffset hrs in seconds
//  
//              System.out.println("Ugly correction = " + UGLY_CORRECTION + "(" +uglyOffset + ")");
                
//              byte[] image = getRBNBImage(videoSourcePath, timeArray[0]+UGLY_CORRECTION);
                byte[] image = getRBNBImage(videoSourcePath, timeArray[0]);
                JPEGImageDecoder decoder = JPEGCodec.createJPEGDecoder(new ByteArrayInputStream(image));
                BufferedImage bi;
                bi = decoder.decodeAsBufferedImage();
                videoWidth = bi.getWidth();
                videoHeight = bi.getHeight();
            }
            else
            {
                videoWidth = 480;
                videoHeight = 320;
            }
        }
        catch (Exception e)
        {
            System.out.println("Possible problem? (setWidthHeight) = " + e);
            videoWidth = 480;
            videoHeight = 320;          
        }
    }
    
    public byte[] getRBNBImage(String videoSourcePath, double videoTime)
        throws SAPIException
    {
        if (DEBUG)
        {
            long unixTime = (long)(videoTime * 1000.0); // convert sec to millisec
            String time = DATE_FORMAT.format(new Date(unixTime));
            System.out.println("Video Source Path = " + videoSourcePath);
            System.out.println("Request time stamp = " + time);
        }
        ChannelMap videoMap = new ChannelMap();
        int videoIndex = videoMap.Add(videoSourcePath);
        videoSink.Request(videoMap,videoTime-1.0,3.0,"Absolute");
        videoMap = videoSink.Fetch(5000, videoMap);
        if (videoMap.GetIfFetchTimedOut())
            throw new SAPIException("Request on channel (" + videoSourcePath + ") timed out. No Data.");
        if ((videoMap.NumberOfChannels() < 1)
                || (!videoMap.GetName(0).equals(videoSourcePath)))
            throw new SAPIException("Requested channel (" + videoSourcePath + ") no data");
        if (DEBUG)
        {
            double[] realTimes = videoMap.GetTimes(0);
            double realtime = realTimes[0];
            double diff = realtime - videoTime;
            long realUnixTime = (long)(realtime * 1000.0); // convert sec to millisec
            String realTimeString = DATE_FORMAT.format(new Date(realUnixTime));
            System.out.println("The numbers " + videoTime + "," + realtime + "," + diff);
            System.out.println("Actual time stamp = " + realTimeString);
        }
        return videoMap.GetDataAsByteArray(0)[0];
    }
    
    private class RBNBImageData extends PullBufferDataSource
    {

        RBNBImageDataStream streams[];

        RBNBImageData(int width, int height, float frameRate,
            String videoSourcePath ,int dataItemsToSkip)
        {
            streams = new RBNBImageDataStream[1];
            streams[0] = new RBNBImageDataStream(width, height, frameRate,
                    videoSourcePath, dataItemsToSkip);
        }

        public void setLocator(MediaLocator source) {
        }

        public MediaLocator getLocator() {
            return null;
        }

        public String getContentType() {
            return ContentDescriptor.RAW;
        }

        public void connect() {
        }

        public void disconnect() {
        }

        public void start() {
        }

        public void stop() {
        }

        public PullBufferStream[] getStreams() {
            return streams;
        }

        public Time getDuration() {
            return DURATION_UNKNOWN;
        }

        public Object[] getControls() {
            return new Object[0];
        }

        public Object getControl(String type) {
            return null;
        }
    }
    
    class RBNBImageDataStream implements PullBufferStream
    {

        String videoSourcePath;
        int dataItemsToSkip;
        VideoFormat format;

        int width;
        int height;

        int nextImage = 0;  // index of the next image to be read.
        boolean ended = false;

        public RBNBImageDataStream(int width, int height, float frameRate,
            String videoSourcePath ,int dataItemsToSkip)
        {

            format = new VideoFormat(VideoFormat.JPEG,
                    new Dimension(getWidth(), getHeight()),
                    Format.NOT_SPECIFIED,
                    Format.byteArray,
                    frameRate);

            this.videoSourcePath = videoSourcePath;
            this.dataItemsToSkip = dataItemsToSkip;
            this.width = width;
            this.height = height;       
        }

        public boolean willReadBlock() {
            return false;
        }

        public void read(Buffer buf) throws IOException {

            if (DEBUG) System.out.println("processing image " + nextImage);
            // Check if we've finished all the frames.
            if (nextImage >= timeArray.length) {
                // We are done.  Set EndOfMedia.
                System.err.println("Array: Done processing all images.");
                buf.setEOM(true);
                buf.setOffset(0);
                buf.setLength(0);
                ended = true;
                return;
            }
            
            // ugly correction -- this should be fixed
//          double UGLY_CORRECTION = uglyOffset * 60.0 * 60.0; // uglyOffset hrs in seconds
//
//          System.out.println("Ugly correction = " + UGLY_CORRECTION + "(" +uglyOffset + ")");
            
            byte[] data;
            try {
//              data = getRBNBImage(videoSourcePath, timeArray[nextImage] + UGLY_CORRECTION);
                data = getRBNBImage(videoSourcePath, timeArray[nextImage]);
                prevImage = data;
            } catch (SAPIException e) {
                if (prevImage == null)
                {
                    System.out.println("Image fatch failed, skipping: " + e);
                    data = new byte[0];
                    buf.setData(data);
                    buf.setOffset(0);
                    buf.setLength(0);
                    nextImage += dataItemsToSkip + 1;
                    return;
                }
                data = prevImage;
            }
            nextImage += dataItemsToSkip + 1;

            buf.setOffset(0);
            buf.setLength(data.length);
            buf.setFormat(format);
            buf.setFlags(buf.getFlags() | buf.FLAG_KEY_FRAME);
            buf.setData(data);

            //     Check the height and width of the read image to make sure that it is what 
            //     is expected - if not resize and update buffer

            JPEGImageDecoder decoder = JPEGCodec.createJPEGDecoder(new ByteArrayInputStream(data));

            //     decode JPEG image to raw image
            BufferedImage bi;
            bi = decoder.decodeAsBufferedImage();
            if ( height != bi.getHeight() || width != bi.getWidth() ) {
                System.out.println("Resizing... to "+width+" x "+height);
        
                Image newimg = bi.getScaledInstance(width, height, 0);
    
                System.out.println("Converting back to buffered image...");
    
                BufferedImage dest = null;
                dest = new BufferedImage(width, height, dest.TYPE_INT_RGB);
                dest.getGraphics().drawImage((Image) newimg, 0, 0, null);
    
                System.out.println("resized height="+dest.getHeight()+" w="+dest.getWidth());
                
                setHeight(dest.getHeight());
                setWidth(dest.getWidth());
                
                // encode scaled image as JPEG image                            
                ByteArrayOutputStream out = new ByteArrayOutputStream();
                JPEGImageEncoder encoder = JPEGCodec.createJPEGEncoder(out);
                System.out.println("Encoding resized image");
                encoder.encode(dest);
                System.out.println("Setting image as buffer");
                // get JPEG image data as byte array
                byte[] newdata = out.toByteArray();
                buf.setData(newdata);
                System.out.println("Length="+newdata.length);
                buf.setLength(newdata.length);
            } // End if must resize

        } // read

        public Format getFormat() {
            return format;
        }

        public ContentDescriptor getContentDescriptor() {
            return new ContentDescriptor(ContentDescriptor.RAW);
        }

        public long getContentLength() {
            return 0;
        }

        public boolean endOfStream() {
            return ended;
        }

        public Object[] getControls() {
            return new Object[0];
        }

        public Object getControl(String type) {
            return null;
        }
    } // RBNBImageDataStream
}

---