FloatBackOfffice/TEST/test_gps_partition.php

<?php
require '../backend.php';

// We will not limit the SQL we will exhaust the allowed memory
// we will process data in ${hard_limit} batches and hopefully catch up
$hard_limit = 500000;

//PostgreSQL
$gps_db_host = $savvyext->cfgReadChar('gpsdatabase.host');
$gps_db_port = $savvyext->cfgReadLong('gpsdatabase.port');
$gps_db_name = $savvyext->cfgReadChar('gpsdatabase.name');
$gps_db_user = $savvyext->cfgReadChar('gpsdatabase.user');
$gps_db_pass = $savvyext->cfgReadChar('gpsdatabase.pass');
$conn_string_gps = "host=${gps_db_host} port=${gps_db_port} dbname=${gps_db_name} user=${gps_db_user} password=${gps_db_pass}";
$pgconn_gps = pg_connect($conn_string_gps);

$q = "select member_id,count(*) from members_tracking group by member_id";
$r = pg_query($pgconn_gps, $q);
if ($r && pg_num_rows($r)) {
    while ($f=pg_fetch_row($r)) {
        echo "[".date("Y-m-d H:i:s")."] Processing member_id=".$f[0]." (total member records: ".$f[1].")\n";
        //if ($f[0]==22) continue; // Crash!
        $p = processMemberGPSDataPartition($f[0]);
    }
}

pg_close($pgconn_gps);

function processMemberGPSDataPartition($member_id) {
    global $pgconn_gps, $hard_limit;
    $min_id = 0;
    $max_id = 0;
    // Min id (the biggest id we have stopped processing last time)
    $q = "SELECT max(last_id) FROM members_tracking_index WHERE member_id=${member_id}";
    $r = pg_query($pgconn_gps, $q);
    if ($f=pg_fetch_row($r)) {
        $min_id = (int)$f[0];
    }
    
    // Max id (we will limit the processing pool to get consistent results)
    $q = "SELECT max(id) FROM members_tracking WHERE member_id=${member_id}";
    $r = pg_query($pgconn_gps, $q);
    if ($f=pg_fetch_row($r)) {
        $max_id = $f[0];
    }
    echo "[".date("Y-m-d H:i:s")."] Process from ${min_id} to ${max_id} (member_id=${member_id})\n";
    
    // Select id's for processing, we will have a separate threads for each device id
    $pids = [];
    $q = "select id,device_id,ttime from members_tracking ";
    $q.= "where member_id=${member_id} and id>${min_id} and id<${max_id} ";
    $q.= "order by device_id,ttime limit ${hard_limit}";
    $r = pg_query($pgconn_gps, $q);
    $pid = 0;
    $total = 0;
    while ($f=pg_fetch_row($r)) {
        if ($f[1]==NULL || $f[1]<1) {
            $f[1] = "NULL";
        }
        if (!array_key_exists($f[1],$pids)) {
            $pids[$f[1]] = [];
            // Try to get the pid
            $ttime = $f[2];
            $q1 = "SELECT id FROM members_tracking WHERE member_id=${member_id} AND ";
            $q1.= ($f[1]==NULL ? " device_id IS NULL" : " device_id=".$f[1]);
            $q1.= " AND ttime<'${ttime}'::timestamp ORDER BY ttime DESC LIMIT 1";
            $r1 =pg_query($pgconn_gps, $q1);
            if ($r1 && pg_num_rows($r1) && $f1=pg_fetch_row($r1)) {
                $pid = $f1[0];
            }
            pg_free_result($r1);
        }
        if ($pid>0) {
            $pids[$f[1]][] = [$pid,$f[0]];
        }
        $pid = $f[0];
        $total++;
    }
    pg_free_result($r);
    
    $processed = 0;
    foreach ($pids as $device_id=>$ids) {
        echo "[".date("Y-m-d H:i:s")."] Processing device_id=${device_id} (member_id=${member_id})\n";
        $last_id = 0;
        foreach ($ids as list($pid,$nid)) { // prev,next
            echo ".";
            $q = "SELECT a.id AS nid, b.id AS pid, ";
            $q.= "ST_Distance(a.gps,b.gps) AS distance, EXTRACT(epoch FROM a.ttime-b.ttime) AS duration ";
            $q.= "FROM members_tracking a LEFT JOIN members_tracking b ON (b.id=${pid}) WHERE a.id=${nid}";
            $r = pg_query($pgconn_gps, $q);
            $f = pg_fetch_assoc($r);
            pg_free_result($r);
            $q = "UPDATE members_tracking SET previous_id=".$f["pid"].",distance=".((int)$f["distance"]).",duration=".((int)(1000*$f["duration"]));
            $q.= "WHERE id=".$f["nid"]." RETURNING *";
            $r = pg_query($pgconn_gps, $q);
            pg_free_result($r);
            $last_id = $f["nid"];
        }
        echo "\n";
        echo "[".date("Y-m-d H:i:s")."] last_id=${last_id} (member_id=${member_id},device_id=${device_id})\n";
        if ($last_id>0) {
            $plast_id = 0;
            $q = "SELECT last_id FROM members_tracking_index WHERE member_id=${member_id} AND device_id=${device_id}";
            $r = pg_query($pgconn_gps, $q);
            if ($r && pg_num_rows($r) && $f=pg_fetch_row($r)) {
                $plast_id = $f[0];
            }
            pg_free_result($r);
            if ($plast_id>0 && $plast_id<$last_id) {
                $q = "UPDATE members_tracking_index SET last_id=${last_id} ";
                $q.= "WHERE member_id=${member_id} AND device_id=${device_id} ";
                $q.= "RETURNING *";
                $r = pg_query($pgconn_gps, $q);
            } else {
                $q = "INSERT INTO members_tracking_index (member_id,device_id,last_id) ";
                $q.= "VALUES(${member_id},${device_id},${last_id}) ";
                $q.= "RETURNING *";
                $r = pg_query($pgconn_gps, $q);
            }
            if ($r && pg_num_rows($r) && $f=pg_fetch_assoc($r)) {
                echo "[".date("Y-m-d H:i:s")."] New last_id=${last_id} (member_id=${member_id},device_id=${device_id})\n";
            } else {
                echo "[".date("Y-m-d H:i:s")."] ERROR: ".pg_last_error($pgconn_gps)."\n";
                echo "$q\n";
            }
            pg_free_result($r);
        } else {
            echo "[".date("Y-m-d H:i:s")."] Invalid last_id=${last_id} (member_id=${member_id},device_id=${device_id})\n";
        }
        $processed++;
    }
    echo "[".date("Y-m-d H:i:s")."] Processed: ${processed} (member_id=${member_id})\n";
    unset($pids);
    return $processed;    
}

function distanceBetweenTwoGpsCoordinates($lat1,$lon1,$lat2,$lon2,$unit) {
    //error_log("public function distanceBetweenTwoGpsCoordinates($lat1,$lon1,$lat2,$lon2,$unit)");
    if (($lat1 == $lat2) && ($lon1 == $lon2)) {
        return 0;
    }
    $theta = $lon1 - $lon2;
    $dist = sin(deg2rad($lat1)) * sin(deg2rad($lat2)) +  cos(deg2rad($lat1)) * cos(deg2rad($lat2)) * cos(deg2rad($theta));
    $dist = acos($dist);
    $dist = rad2deg($dist);
    $miles = $dist * 60 * 1.1515;
    $unit = strtoupper($unit);
    if ($unit == "K") {
        return ($miles * 1.609344);
    }
    if ($unit == "N") {
        ($miles * 0.8684);
    }
    return $miles;
}