importSolarFarm.php

#!/usr/bin/php
<?php
namespace SmartData;
require_once( __DIR__ . '/../Backend.php');

/*
HT_MA2_DES_DIA (nao entendi na planilha)
HT_MA2_DES_MIN (nao entendi na planilha)
HT_MA2_ESS_EXT
HT_MA2_ESS_MIN
HT_MA2_ESS_SEG
HT_MA2_FLG_MIN
HT_MA2_INV_MIN
HT_MA2_INV_SEG
HT_MA2_MME_MIN
HT_MA2_MME_SEG
HT_MA2_SBX_MIN
HT_MA2_SBX_SEG
HT_MA2_SBX_SEG_MIN
*/

use \mysqli as mysqli;

abstract class Importer// extends Backend
{
    // mysql -h 150.162.142.228 -u Pablo -pfotovoltaica stagging_ma -A

    private static $_SERVERNAME = '150.162.142.228';
    private static $_SERVERPORT = 3306;
    private static $_DBNAME     = 'stagging_ma';
    private static $_USERNAME   = 'Pablo';
    private static $_PASSWORD   = 'fotovoltaica';

    private static $BATCH_SIZE  = PHP_INT_MAX;
    private static $DOMAIN      = 'solar_farm';

    // Checks the health of databases
    public static function health_check() : bool {
        $sucess = true;
        try {
            $conn = self::_mysqlConnect();
        } catch (\Exception $e) {
            Logger::exception($e);
            $sucess = false;
        } finally {
            $conn = null;
        }
        return $sucess;
    }

    public static function get_time(){
        list($usec, $sec) = explode(' ', microtime());
        return (float) $sec + (float) $usec;
    }

    public static function insert($backend, $series, $smartdata_batch, $j, $i, $start){
        $insertBatch_start = self::get_time();
        echo "Inserting batch...\n";
        if(!$backend->insertBatch($series, ...$smartdata_batch)){
            die ("Insertion Failed!\n");
        } else {
            echo "Inserted...\n";
        }
        $insertBatch_end = self::get_time();
        $insertBatch_elapsed_time = round($insertBatch_end - $insertBatch_start, 5);
        $total_elapsed_time = round($insertBatch_end - $start, 5);
        echo $j,'/',$i,' datapoints imported in ',$insertBatch_elapsed_time,' secs / ',$total_elapsed_time,' secs', PHP_EOL;
    }

    public static function import($table, $time, $column, $unit, $error, $confidence, $x, $y, $z, $d=0, $callback=NULL) {

        $time_start = self::get_time();
        $batch_size = self::$BATCH_SIZE;
        $count = 0;

        $smartdata = new StaticSmartData (
                $unit,
                0,      // value
                $error,
                $confidence,
                $x,
                $y,
                $z,
                0,      // timestamp
                $d      // d
        );

        $series = new Series (
                    SmartData::STATIC_VERSION,
                    $unit,    // unit
                    $x,       // x
                    $y,       // y
                    $z,       // z
                    0,        // r
                    0, //1449123538000000, // t0 (epoch in microseconds)
                    PHP_INT_MAX,  // t1 (epoch in microseconds)
                    $d      // d
        );

        global $__FORCE_VERSION;
        $__FORCE_VERSION = Config_Common::VERSION_1_1;
        $cred = new Credentials(self::$DOMAIN,
                        Config::config()::CASSANDRA_SUPERUSER,
                        Config::config()::CASSANDRA_SUPERPASS);
        $backend = new Backend_V1_1($cred, true);

        echo "Creating data series...\n";
        if(!$backend->create($series))
            die ("Create error");

        $i = 0;
        echo "Pushing data... $series ({$table}.{$column})\n";

        $servername = Importer::$_SERVERNAME;
        $port       = Importer::$_SERVERPORT;
        $dbname     = Importer::$_DBNAME;
        $username   = Importer::$_USERNAME;
        $password   = Importer::$_PASSWORD;

        $mysqli = new mysqli($servername, $username, $password, $dbname);

        if ($mysqli->connect_errno) {
            echo 'Connect failed: ', $mysqli->connect_error, PHP_EOL;
            exit();
        }

        while(true){
            $query = "SELECT UNIX_TIMESTAMP(t.{$time})*1000000 as timestamp, t.{$column} as value
                      FROM {$table} t
                      LIMIT {$count},{$batch_size};";

            echo 'new query', PHP_EOL;
            $result = $mysqli->query($query, MYSQLI_USE_RESULT);
            if(!$result){
                echo 'breaking', PHP_EOL;
                break;
            }

            if(isset($smartdata_batch))
                array_splice($smartdata_batch, 0);

            $smartdata_batch = array();
            $j = 0;
            while ($row = $result->fetch_assoc()) {
                $timestamp  = $row['timestamp'];
                $value = $row['value'];

                $smartdata->time = $timestamp;

                if (is_callable($callback))
                    $smartdata->value = $callback($value);
                else
                    $smartdata->value = $value;

                $smartdata_temp = clone $smartdata;
                array_push($smartdata_batch, $smartdata_temp);

                $i = $i + 1;
                $j = $j + 1;
                if($j >= 20000){
                    self::insert($backend, $series, $smartdata_batch, $j, $i, $time_start);
                    array_splice($smartdata_batch, 0);
                    $j = 0;
                }
            }
            if($j > 0){
                self::insert($backend, $series, $smartdata_batch, $j, $i, $time_start);
                array_splice($smartdata_batch, 0);
            }


            $count+=$batch_size;

            $result->close();
        }

        $mysqli->close();

        $time_end = self::get_time();
        $elapsed_time = round($time_end - $time_start, 5);

        echo "Import finished: {$table}.{$column}  Total points: {$i}\n";
        echo 'Elapsed time: ', $elapsed_time, ' secs. Memory usage: ', round(((memory_get_peak_usage(true) / 1024) / 1024), 2), 'Mb';
        echo "________________________________________________\n";

    }

    private static function _mysqlConnect() : \PDO {
        $servername = Importer::$_SERVERNAME;
        $port       = Importer::$_SERVERPORT;
        $dbname     = Importer::$_DBNAME;
        $username   = Importer::$_USERNAME;
        $password   = Importer::$_PASSWORD;
        $conn = new \PDO("mysql:host=$servername;port=$port;dbname=$dbname", $username, $password);
        $conn->setAttribute(\PDO::ATTR_ERRMODE, \PDO::ERRMODE_EXCEPTION);
        $conn->setAttribute(\PDO::ATTR_PERSISTENT, true);
        $conn->setAttribute(\PDO::MYSQL_ATTR_USE_BUFFERED_QUERY, false);
        return $conn;
    }
}