solar-index

#!/usr/bin/env perl
use strict;
use warnings;
use JSON;
use Search::Elasticsearch;
use File::Slurper qw( read_text read_binary );
use Try::Tiny;
use Path::Tiny;

sub read_json {
    my ($filename) = @_;
    from_json( read_binary($filename) );
}

my $e = Search::Elasticsearch->new();

if ( $ENV{CREATE} ) {
    try { $e->indices->delete( index => 'solar' ); };
    $e->indices->create( index => 'solar' );
    $e->indices->put_mapping(
        index => 'solar',

        # type              => 'device',
        #include_type_name => \1,
        body => {
            properties => {
                p_3phsum_kw     => { type => 'float' },    # AC power kWh
                p_mpptsum_kw    => { type => 'float' },    # AC power kWh
                ltea_3phsum_kwh => { type => 'float' },    # total energy kWh
                vln_3phavg_v    => { type => 'float' },    # AC voltage
                t_htsnk_degc    => { type => 'float' },    # heatsink temp
            }
        }
    );
    exit;
}

my @files = @ARGV ? @ARGV : <STDIN>;

my $PROCESSED = 0;

for my $file_or_dir (@files) {
    next unless -s $file_or_dir;

    #warn "processing $file_or_dir\n";

    if ( -d $file_or_dir ) {
        my $dir = path($file_or_dir);
        for my $f ( $dir->children ) {
            # warn "processing $f\n";
            process_file($f);
        }
    }
    else {
        process_file($file_or_dir);
    }
}

# printf("Processed %s files\n", $PROCESSED);

sub process_file {
    my $json_file = shift;

    my $buf;
    try { $buf = read_json($json_file) };

    return unless $buf;

    my ($id) = ( $json_file =~ m/([\d\-]+)\.json/ );

    #warn $id;

    # each device is a "document"
    my $bulk = $e->bulk_helper( index => 'solar', );
    for my $device ( @{ $buf->{devices} } ) {
        my $id = join( "-", $id, $device->{SERIAL} );
        my @dtime = split( ',', $device->{CURTIME} );
        $device->{tstamp} = sprintf( "%s-%s-%sT%s:%s:%s", @dtime );
        $bulk->index( { id => $id, source => $device } );
    }
    $bulk->flush;

    $PROCESSED++;

    if ( !$PROCESSED % 1000 ) {
        warn "processed $PROCESSED\n";
    }
}