#!/usr/bin/perl
#### #### #### #### #### #### #### #### #### #### #### #### #### #### #### #### 
#  
# License: 
#		GNU Affero General Public License, version 3.
#		See <http://www.gnu.org/licenses/>
#
# Version history:
#     	2015-02-17  first version by Monk (only InfluxDB 0.9)
#
#
# Usage:
# - Modify configuration to match your install
#
# - This script saves the last mysql timestamp that has been send to InfluxDB
#   in a json file. To work with this script this json file needs to be generated 
#   once. To do so set "$post_data = 0", "$loop = 0", "$use_limit = 2" and "$max_results = 1"
#   With "$ts_start" you can define a ms epoch timestamp from which the import process 
#   shall start.
#
# - For batch importing mysql data into an InfluxDB, generate the json file with the desired 
#   starting timestamp then set "$post_data = 1", "$loop = 1", "$use_limit = 1", 
#   "$max_results = 100" and my "$interval = 1". Then run the script with nohup.
#   I have not tested the limit for one single batch but probably $max_results can be set
#   a lot higher.
#   You can monitor the import process in the InfluxDB web interface by querying the DB, e.g.
#   with "select * from wattage.s0.consume.eg limit 1". This will return the most current 
#   timestamp that has been imported. The import process will not stop on its own but nothing bad
#   will happen when no points are available for import
#
# - To test your setup set "$debug = 1", "$post_data = 0" and "$loop = 0" 
#   data will only be logged to logfile but not send to InfluxDB
#
# - For keeping the InfluxDB updated I suggest a setting like:
#   "$post_data = 1", "$loop = 1", "$use_limit = 0", "$interval = 20", "$debug = 0" 
#   This will import all new data in 20s intervals. 
#   Note that your InfluxDB should be already pretty much updated or a huge
#   number of data points will be processed in one shot potentially freezing
#   your system
#
# - Run script with "nohup ./vzInflow.pl &" to detach from terminal
#
#### #### #### #### #### #### #### #### #### #### #### #### #### #### #### ####


use strict;
use warnings;
use LWP::UserAgent;
use JSON::XS;
use DBI;
use File::Slurp;

# flush after every write
$| = 1;

################################# CONFIGURATION START #################################

# Influx http peer IP
my $influx_peer = '192.168.178.8';
# Influx http port 
my $influx_port = '8086';
# Influx DB name
my $influx_db = 'vz2';
# Influx DB user name
my $influx_user = 'vz2';
# Influx DB password
my $influx_pwd = 'vz12';
# Default retention policy
my $influx_retention = 'mytest';
# Really post data or just do a dry run
my $post_data = 0;

# Interval to query data from mysql and push to InfluxDB
my $interval = '2';
# Set to '1' for some additional debug output to log
my $debug = 1;
# Log file where the debug output or error messages go to
my $log = '/var/log/inflow2.log';
# To run only once (for debugging) set loop to 0
my $loop = 0;
# Save max timestamp to json file
my $file_maxts = '/root/max_ts.json';

# MYSQL
# mysql host IP
my $mysql_host = '192.168.178.8';
# mysql port
my $mysql_port = 3306;
# mysql VZ user
my $mysql_user = 'vz';
# mysql VZ password
my $mysql_pass = 'demo';
# mysql VZ database name
my $mysql_db = 'volkszaehler';
# mysql use limiting and if yes specify max results for each query
# $use_limit = 0 no limit
# $use_limit = 1 limit to $max_results
# $use_limit = 2 and $ts_start to create a new starting json. Set "$loop = 0"!
my $use_limit = 1;
my $max_results = 3;
my $ts_start = '1424299242215';

# Testcase column is used for trying to calculate things directly in the
# InfluxDB. When set to "ts" it uses timestamp as value for S0 meters.
# Idea is to use the derivative function to calculate watts by the change rate
# instead of counting how many "1" are in a given time period. Usually testcase is
# "value"
my %sensors = (
	#Senor_nr	 Ch ID	Series	Tag1  Tag2		 Tag3	Testcase		  
    sensor1 => [ '10', 'power', 's0', 'consume', 'eg', 'value' ],
    sensor2 => [ '11', 'power', 's0', 'generate', 'pv', 'value' ],
    sensor3 => [ '12', 'power', 's0', 'consume', 'og', 'value'  ],
    sensor4 => [ '13', 'power', 's0', 'consume', 'ug', 'value' ],
    sensor5 => [ '17', 'power', 'ehz', 'consume', 'total', 'value' ],
    sensor6 => [ '18', 'power', 'ehz', 'generate', 'total', 'value' ],
    sensor7 => [ '19', 'power', 'ehz', 'delta', 'total', 'value' ],
    sensor8 => [ '10', 'power', 's1', 'consume', 'eg', 'ts' ],
    sensor9 => [ '11', 'power', 's1', 'generate', 'pv', 'ts' ],
    sensor10 => [ '12', 'power', 's1', 'consume', 'og', 'ts'  ],
    sensor11 => [ '13', 'power', 's1', 'consume', 'ug', 'ts' ],
);
				  
################################# CONFIGURATION END #################################

sub logit {
    my $s = shift;
    my ($logsec,$logmin,$loghour,$logmday,$logmon,$logyear,$logwday,$logyday,$logisdst)=localtime(time);
    my $logtimestamp = sprintf("%4d-%02d-%02d %02d:%02d:%02d",$logyear+1900,$logmon+1,$logmday,$loghour,$logmin,$logsec);
    $logmon++;
    my $logfile = $log;
    my $fh;
    open($fh, '>>', "$logfile") or die "$logfile: $!";
    print $fh "$logtimestamp $s\n";
    close($fh);
}


while (1) {
	
	my ($dbh, $cur_time, %to_file, @influx_points, $influx_json);
	
	# Connect to the VZ mysql DB
	$dbh = DBI->connect("DBI:mysql:database=$mysql_db;host=$mysql_host;port=$mysql_port", 
						"$mysql_user", 
						"$mysql_pass",
						{ RaiseError=>0,PrintError=>1 }
	) or logit("MYSQL ERROR: $DBI::errstr");
	
	# Read latest mysql time stamp for each channel from file
	my $json = read_file($file_maxts, { binmode => ':raw' });
    my %last_mysql = %{ decode_json $json };
	
	foreach my $sensor_details (sort keys %sensors) {
		my $channel_id = @{$sensors{$sensor_details}}[0];
		my $series_name = @{$sensors{$sensor_details}}[1];
		my $metric_source = @{$sensors{$sensor_details}}[2];
		my $metric_type = @{$sensors{$sensor_details}}[3];
		my $metric_location = @{$sensors{$sensor_details}}[4];
		my $value_type = @{$sensors{$sensor_details}}[5];
		
		my %json_body = (
						'database' => $influx_db,
						'retentionPolicy' => $influx_retention,
						'tags' => {
							'source' => $metric_source,
							'type' => $metric_type,
							'location' => $metric_location
						}
						);
		
		my ($query, $timestamp, $value, @all_values, @mysql_ts);		
		logit("INFO: Last recorded Timestamp for Channel $channel_id: $last_mysql{$channel_id}") if $debug;
		# Create mysql query and execute it
		if ( $use_limit == 1 ) {
			$query = $dbh->prepare("SELECT `timestamp` , `value` FROM `data` WHERE (`channel_id` =$channel_id AND `timestamp` >$last_mysql{$channel_id}) ORDER BY `timestamp` LIMIT $max_results");
		} elsif ( $use_limit == 0 ) {
			$query = $dbh->prepare("SELECT `timestamp` , `value` FROM `data` WHERE (`channel_id` =$channel_id AND `timestamp` >$last_mysql{$channel_id}) ORDER BY `timestamp`");
		} elsif ( $use_limit == 2 ) {
			$query = $dbh->prepare("SELECT `timestamp` , `value` FROM `data` WHERE (`channel_id` =$channel_id AND `timestamp` >$ts_start) ORDER BY `timestamp` LIMIT $max_results");
		}
		$query->execute() or logit("MYSQL ERROR: $query->err_str");
		
		# loop through mysql result
		while ( ($timestamp, $value) = $query->fetchrow_array() ) {
			my %json_point2;
			my %json_point1 = (
								'name' => $series_name,
								'timestamp' => ($timestamp * 1),
								'precision' => "ms"
							  );
			if ( $value_type eq 'value' ) {
				%json_point2 = ('value' => $value);
			} elsif ( $value_type eq 'ts' ) {
				%json_point2 = ('value' => $timestamp);
			}			
			my %json_values = (%json_point1, 'values' => \%json_point2); 
			push(@all_values, \%json_values);
			logit("INFO MYSQL: Channel_ID $channel_id Timestamp: $timestamp Value: $value") if $debug;
			push (@mysql_ts, $timestamp); 		
		}
		$query->finish();
		
		# If mysql query returned results
		if ( @mysql_ts ) {
			# Find max mysql timestamp and store it for writing to file
			@mysql_ts = sort { $a <=> $b } @mysql_ts;
			$to_file{$channel_id} = $mysql_ts[-1];
			logit("INFO MYSQL: Channel_ID $channel_id Last Timestamp: $mysql_ts[-1]") if $debug;
			@mysql_ts = ();
			
			# Encode InfluxDB json
			my %json_data = (%json_body, "points" => \@all_values);
			$influx_json = encode_json(\%json_data);
			logit("INFO JSON: Channel_ID $channel_id: $influx_json") if $debug;
			
			if ( $post_data ) {
				my $influx_url = "http://$influx_peer:$influx_port/write";
				#logit("Influx Data (POST): $influx_url") if $debug;
				my $ua = LWP::UserAgent->new();
				my $req = HTTP::Request->new(POST => $influx_url);
				$req->header('content-type' => 'application/json');
				$req->content($influx_json);
				my $resp = $ua->request($req);
				my $status_line = $resp->code . " " . $resp->message;
				logit("ERROR Influx POST: $status_line") if ($resp->is_error);
			}
		} else {
			# keep the previous timestamp for the channel
			$to_file{$channel_id} = $last_mysql{$channel_id};
		}
		
		@influx_points = ();
	}
	$dbh->disconnect();
	
	# Save latest timestamp to file
	$json = encode_json(\%to_file);
	write_file($file_maxts, { binmode => ':raw' }, $json);
	%to_file = ();
	
	# Only sleep if $loop is true
	sleep $interval if $loop;
	
	# Break "while loop" if $loop is false (run only once for debug purposes)
	last if !$loop;
 }
