/*
 * Copyright (c) 2016-2017 The Regents of the University of California.
 * All rights reserved.
 *
 * '$Author: crawl $'
 * '$Date: 2018-02-07 18:53:35 +0000 (Wed, 07 Feb 2018) $'
 * '$Revision: 34661 $'
 *
 * Permission is hereby granted, without written agreement and without
 * license or royalty fees, to use, copy, modify, and distribute this
 * software and its documentation for any purpose, provided that the above
 * copyright notice and the following two paragraphs appear in all copies
 * of this software.
 *
 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY
 * FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
 * ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF
 * THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE
 * PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF
 * CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,
 * ENHANCEMENTS, OR MODIFICATIONS.
 *
 */

package org.kepler.spark.mllib;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.kepler.spark.actor.SparkSQLActor;
import org.kepler.webview.server.WebViewServer; // use webview's vertx instance

import io.vertx.core.Vertx;
import io.vertx.core.http.HttpClient;
import io.vertx.core.http.HttpClientOptions;
import io.vertx.core.http.WebSocket;
import io.vertx.core.json.JsonObject;
import ptolemy.actor.Actor;
import ptolemy.actor.TypedIOPort;
import ptolemy.actor.parameters.PortParameter;
import ptolemy.data.IntToken;
import ptolemy.data.LongToken;
import ptolemy.data.ObjectToken;
import ptolemy.data.StringToken;
import ptolemy.data.Token;
import ptolemy.data.expr.Parameter;
import ptolemy.data.expr.StringParameter;
import ptolemy.data.type.BaseType;
import ptolemy.data.type.ObjectType;
import ptolemy.kernel.CompositeEntity;
import ptolemy.kernel.util.Attribute;
import ptolemy.kernel.util.ChangeRequest;
import ptolemy.kernel.util.IllegalActionException;
import ptolemy.kernel.util.NameDuplicationException;
import ptolemy.kernel.util.NamedObj;
import ptolemy.kernel.util.SingletonAttribute;

/**
 * Actor for retrieving weather station data from the Firemap web socket server.
 */
public class GetLiveMeasurements extends SparkSQLActor
{
    protected Vertx vertx;

    public StringParameter websocketHost;    // firemap.sdsc.edu
    public StringParameter websocketPort;    // 8443;
    public StringParameter websocketUri;     // /livewx/websocket
    public StringParameter messageType;
    public StringParameter dropCorruptedData;
    public Parameter minFiringInterval;
    public Parameter firingQueueThreshold;
    private static String WEBSOCKET_HOST;
    private static int WEBSOCKET_PORT;
    private static String WEBSOCKET_URI;

    private final long DEFAULT_FIRING_INTERVAL = 60000;
    private final int DEFAULT_QUEUE_THRS = 1;

    public PortParameter observables;
    public TypedIOPort output;
    public TypedIOPort count;

    private WebSocket _websocket = null;
    public String[] _observables = null;

    private String _messageTypeValue;
    private StructType _schema;

    // A queue that holds data samples yet to be output by the actor.
    // _handleMessage() enqueues incoming websocket data,
    // fire() dequeues and outputs a dataset
    private Queue<Message> _queue;

    // long for storing timestamp of last actor firing
    // Used for timing GetLiveMeasurement's firing.
    private long timeOfLastFire;

    // if true, sets any incoming measurement value containing "#" to null.
    private boolean dropCorrupted;

    /**
     * Initializes the actor's parameters and Vertex property.
     *
     * @param container
     * @param name
     * @throws IllegalActionException
     * @throws NameDuplicationException
     */
    public GetLiveMeasurements(CompositeEntity container, String name)
        throws IllegalActionException, NameDuplicationException
    {
        super(container, name);

        messageType = new StringParameter(this, "messageType");
        messageType.setExpression("R0");
        messageType.addChoice("R0");
        messageType.addChoice("R1");
        messageType.addChoice("R2");
        messageType.addChoice("R3");
        messageType.addChoice("R5");

        observables = new PortParameter(this, "observables");
        observables.setStringMode(true);
        observables.setTypeEquals(BaseType.STRING);
        observables.getPort().setTypeEquals(BaseType.STRING);
        new Attribute(observables.getPort(), "_showName");

        websocketHost = new StringParameter(this, "websocketHost");
        websocketHost.setExpression("firemap.sdsc.edu");

        websocketPort = new StringParameter(this, "websocketPort");
        websocketPort.setExpression("5446");

        websocketUri = new StringParameter(this, "websocketUri");
        websocketUri.setExpression("");

        dropCorruptedData = new StringParameter(this, "dropCorruptedData");
        dropCorruptedData.setExpression("true");

        minFiringInterval = new Parameter(this, "minFiringInterval");
        minFiringInterval.setTypeEquals(BaseType.LONG);
        minFiringInterval.setToken(new LongToken(DEFAULT_FIRING_INTERVAL));

        firingQueueThreshold = new Parameter(this, "firingQueueThreshold");
        firingQueueThreshold.setTypeEquals(BaseType.INT);
        firingQueueThreshold.setToken(new IntToken(DEFAULT_QUEUE_THRS));

        output = new TypedIOPort(this, "output", false, true);
        output.setTypeEquals(new ObjectType(Dataset.class));
        new SingletonAttribute(output, "_showName");

        count = new TypedIOPort(this, "count", false, true);
        count.setTypeEquals(BaseType.LONG);
        new SingletonAttribute(count, "_showName");
    }

    /**
     * Handles changing the message type parameter, and validates it.
     *
     * @param attribute The attribute that changed.
     * @throws IllegalActionException
     */
    @Override
    public void attributeChanged(Attribute attribute)
        throws IllegalActionException
    {
        if(attribute == messageType) {
            Token token = messageType.getToken();

            if(token != null) {
                String value = ((StringToken) token).stringValue();

                if(value != null && !value.trim().isEmpty()) {
                    String[] choices = messageType.getChoices();

                    if(!Arrays.asList(choices).contains(value))  {
                        throw new IllegalActionException(this,
                                         "Unsupported messageType: " + value);
                    }

                    _messageTypeValue = value;
                }
            }
        }
        else {
            super.attributeChanged(attribute);
        }
    }


    /**
     * Call this to tell DE director to fire the actor.
     *
     * @throws IllegalActionException
     */
    private void _fireActor() throws IllegalActionException
    {
        final Actor actor = this;
        final NamedObj namedObj = output;
        ChangeRequest request = new ChangeRequest(this, "fire " + getName())
            {
                protected void _execute() throws Exception {
                    getDirector().fireAtCurrentTime(actor);
                }

                @Override
                public NamedObj getLocality() {
                    return namedObj;
                }

            };
        request.setPersistent(false);
        getContainer().requestChange(request);
    }


    /**
     * Sets up the websocket and other member variables once per workflow
     * execution.
     *
     * @throws IllegalActionException
     */
    @Override
    public void initialize()
        throws IllegalActionException
    {
        super.initialize();

        // Get the websocket parameters, store them in member variables
        WEBSOCKET_HOST = websocketHost.getValueAsString();
        WEBSOCKET_PORT = Integer.parseInt(websocketPort.getValueAsString());
        WEBSOCKET_URI = websocketUri.getValueAsString();

        // Set options for websocket
        //vertx = Vertx.vertx();
        vertx = WebViewServer.vertx();
        HttpClientOptions options = new HttpClientOptions();
        options.setDefaultHost(WEBSOCKET_HOST);
        options.setDefaultPort(WEBSOCKET_PORT);
        options.setSsl(true);
        options.setTrustAll(true);

        // Create the websocket, define its message handler
        HttpClient client = vertx.createHttpClient(options);
        client.websocket(WEBSOCKET_URI, websocket -> {

                _websocket = websocket;

                System.out.println("Web socket connected");

                websocket.frameHandler(frame -> {
                            String data = frame.textData();
                            JsonObject json = new JsonObject(data);
                            //JsonObject message = json.getJsonObject("message");
                            _handleMessage(json);  // websocket updated 2017-07-19
                        });

                websocket.closeHandler(x -> {
                        System.out.println("Web socket closed");
                    });

                websocket.writeFinalTextFrame("start");
                });


        // Get the set observables and format them into a string array
        observables.update();

        StringToken observablesToken = (StringToken) observables.getToken();
        String observablesValue = (observablesToken == null ?
                                   "" : observablesToken.stringValue().trim());

        if(observablesValue.isEmpty()) {
            throw new IllegalActionException(this, "Must specify observables.");
        }

        _observables = observablesValue.split("\\s*,\\s*");

        List<StructField> fields = new ArrayList<>();
        StructField field;
        // add 7 observables as first 7 columns
        for(String observable : _observables) {
            field = DataTypes.createStructField(observable,
                                                DataTypes.DoubleType, true);
            fields.add(field);
        }

        // add name and timestampMillis columns
        field = DataTypes.createStructField("name",DataTypes.StringType, true);
        fields.add(field);
        field = DataTypes.createStructField("timestampMillis",
                                            DataTypes.LongType, true);
        fields.add(field);

        // create schema
        _schema = DataTypes.createStructType(fields);

        // create message queue
        _queue = new ConcurrentLinkedQueue<Message>();

        // initialize time counter and dropCorrupted
        timeOfLastFire = System.currentTimeMillis();

        // get dropCorrupted
        dropCorrupted = Boolean.parseBoolean(dropCorruptedData.getValueAsString());

    }

    /**
     * Waits for weather station data then packages it into a Spark data frame.
     *
     * @throws IllegalActionException
     */
    @Override
    public void fire()
        throws IllegalActionException
    {
        super.fire();

        synchronized(_queue) {
            List<Row> rows = new ArrayList<>();

            while(_queue.size() > 0) {
                // get message and extract name
                Message msg  = _queue.poll();
                String name = msg.name;
                name = name.replaceAll("\\s+","");  // clean up name string
                Double[] vector = msg.getVector(_observables);

                // TODO: find a more dynamic way of creating new row with
                // different data types as its values(double,...,double,String)
                //Row row = new GenericRow(vector);
                Row row = RowFactory.create(vector[0], vector[1], vector[2],
                                            vector[3], vector[4], vector[5],
                                            vector[6], name,
                                            msg.timestampMillis);
                rows.add(row);
            }

            Dataset<Row> df = _sqlContext.createDataFrame(rows, _schema);

            output.broadcast(new ObjectToken(df, Dataset.class));

            if (count.numberOfSinks() > 0)
                count.broadcast(new LongToken(df.count()));
        }

    }

    /**
     * Close the websocket connection when the workflow ends.
     *
     * @throws IllegalActionException
     */
    @Override
    public void wrapup()
        throws IllegalActionException
    {
        super.wrapup();

        if(_websocket != null)  {
            _websocket.close();
        }
    }


    /**
     * Converts the websocket message JSON into a message instance, if it's the
     * correct message type.
     * @param message
     */
    private void _handleMessage(JsonObject message)
    {
        String mType = "0" + _messageTypeValue; // TODO  What is this "0"?

        if(mType.equals(message.getString("mtype")))  {
            //System.out.println("Accepted message: " + message.toString());

            Message currentMessage = new Message(message);

            try {
                // attempt to trigger an exception
                Double[] vector = currentMessage.getVector(_observables);

                _queue.add(currentMessage);

                // After waiting for the specified interval,
                //   if msg queue not empty, notify DE director to fire actor
                if (System.currentTimeMillis() - timeOfLastFire >
                    ((LongToken)minFiringInterval.getToken()).longValue()
                    && _queue.size() >=
                    ((IntToken)firingQueueThreshold.getToken()).intValue() ){

                    // reset timestamp variable
                    timeOfLastFire = System.currentTimeMillis();

                    // attempt to signal DE director to call fire()
                    try {
                        //System.out.println("Firing actor");
                        _fireActor();
                    } catch (IllegalActionException e) {
                        System.out.println("Error firing GetLiveMeasurements." + e);
                    }
                }

            } catch (Exception e) {
                System.out.println("Invalid observable. " + e);
            }
        }
    }


    /**
     * Class for storing web socket responses
     */
    private class Message
    {
        public String name = "";
        public Long timestampMillis = null;
        public Double windSpeedMin = null;
        public Double windSpeedAvg = null;
        public Double windSpeedMax = null;
        public Double windDirectionMin = null;
        public Double windDirectionAvg = null;
        public Double windDirectionMax = null;
        public Double airPressure = null;
        public Double airTemperature = null;
        public Double internalTemperature = null;
        public Double relativeHumidity = null;
        public Double rainAccumulation = null;
        public Double rainDuration = null;
        public Double rainIntensity = null;
        public Double rainPeakIntentsity = null;
        public Double hailAccumulation = null;
        public Double hailDuration = null;
        public Double hailIntensity = null;
        public Double hailPeakIntentsity = null;
        public Double heatingTemperature = null;
        public Double heatingVoltage = null;
        public Double supplyVoltage = null;
        public Double ref35Voltage = null;

        /**
         * Empty constructor for simple initialization.
         */
        public Message() {
            timestampMillis = System.currentTimeMillis();
        }

        /**
         * Initialize with a message response JSON object.
         *
         * @param json
         */
        public Message(JsonObject json)
        {
            setValuesFromJson(json);
            timestampMillis = System.currentTimeMillis();
        }

        /**
         * Sets the values from a message response JSON object.
         *
         * @param json
         */
        public void setValuesFromJson(JsonObject json)
        {
            Iterator it = json.iterator();

            while(it.hasNext()) {
                Map.Entry pair = (Map.Entry) it.next();
                String key = (String) pair.getKey();
                Object value = pair.getValue();

                if(key.equals("Name")) {
                    name = (String) value;

                } else if(!key.equals("mtype")) {
                    try {
                        setValue(key, value);
                    } catch(Exception e) {
                        System.out.println("WARNING: " + e.toString());
                    }
                }
            }
        }

        /**
         * Gets a measurement value.
         *
         * @param key
         * @return
         * @throws Exception
         */
        public Double getValue(String key)
            throws IllegalActionException
        {
            switch(key) {
            case "Sn": return windSpeedMin;
            case "Sm": return windSpeedAvg;
            case "Sx": return windSpeedMax;
            case "Dn": return windDirectionMin;
            case "Dm": return windDirectionAvg;
            case "Dx": return windDirectionMax;
            case "Pa": return airPressure;
            case "Ta": return airTemperature;
            case "Tp": return internalTemperature;
            case "Ua": return relativeHumidity;
            case "Rc": return rainAccumulation;
            case "Rd": return rainDuration;
            case "Ri": return rainIntensity;
            case "Rp": return rainPeakIntentsity;
            case "Hc": return hailAccumulation;
            case "Hd": return hailDuration;
            case "Hi": return hailIntensity;
            case "Hp": return hailPeakIntentsity;
            case "Th": return heatingTemperature;
            case "Vh": return heatingVoltage;
            case "Vs": return supplyVoltage;
            case "Vr": return ref35Voltage;
            }

            throw new IllegalActionException("\"" + key +
                                             "\" is not a valid property.");
        }

        /**
         * Sets a measurement value.
         * Value can be a number, or a string. If it's a string, it must match the
         * type of a number followed by a unit.
         * @param key
         * @param value
         * @throws Exception
         */
        public void setValue(String key, Object value)
            throws Exception
        {
            Double doubleValue = null;

            if(value instanceof Number) {
                doubleValue = (Double) value;

            } else if(value instanceof String) {
                String stringValue = (String) value;
                Pattern pattern = Pattern.compile("([0-9\\-.]+)(.+)");
                Matcher matcher = pattern.matcher(stringValue);

                // catch numeric values containing "#" (corrupted data)
                if (dropCorrupted && stringValue.contains("#"))
                    throw new Exception("Corrupted data for " + name
                                        + " - " + key + ":" + stringValue);

                if(!matcher.matches()) {
                    throw new Exception("\"" + stringValue + "\" is not a valid value.");
                }

                String number = matcher.group(1);
                String unit = matcher.group(2);

                //doubleValue = _normalizeUnit(key, Double.parseDouble(number), unit);
                doubleValue = Double.parseDouble(number);

            } else if(value != null) {
                throw new Exception("Value passed was of an invalid type.");
            }

            switch(key) {
            case "Sn": windSpeedMin = doubleValue; break;
            case "Sm": windSpeedAvg = doubleValue; break;
            case "Sx": windSpeedMax = doubleValue; break;
            case "Dn": windDirectionAvg = doubleValue; break;
            case "Dm": windDirectionAvg = doubleValue; break;
            case "Dx": windDirectionMax = doubleValue; break;
            case "Pa": airPressure = doubleValue; break;
            case "Ta": airTemperature = doubleValue; break;
            case "Tp": internalTemperature = doubleValue; break;
            case "Ua": relativeHumidity = doubleValue; break;
            case "Rc": rainAccumulation = doubleValue; break;
            case "Rd": rainDuration = doubleValue; break;
            case "Ri": rainIntensity = doubleValue; break;
            case "Rp": rainPeakIntentsity = doubleValue; break;
            case "Hc": hailAccumulation = doubleValue; break;
            case "Hd": hailDuration = doubleValue; break;
            case "Hi": hailIntensity = doubleValue; break;
            case "Hp": hailPeakIntentsity = doubleValue; break;
            case "Th": heatingTemperature = doubleValue; break;
            case "Vh": heatingVoltage = doubleValue; break;
            case "Vs": supplyVoltage = doubleValue; break;
            case "Vr": ref35Voltage = doubleValue; break;
            default:
                throw new Exception("\"" + key + "\" is not a valid property.");

            }
        }

        /**
         * Gets an array of measurement values from an array of measurement names.
         *
         * @param observables
         * @return
         * @throws Exception
         */
        public Double[] getVector(String[] observables)
            throws IllegalActionException
        {
            Double[] vector = new Double[observables.length];

            for(int i = 0; i < observables.length; i++) {
                vector[i] = getValue(observables[i]);
            }

            return vector;
        }

    }
}