view release on metacpan or  search on metacpan

lib/Algorithm/Classifier/IsolationForest/App/Command/streamc.pm  view on Meta::CPAN

package Algorithm::Classifier::IsolationForest::App::Command::streamc;

use strict;
use warnings;
use Algorithm::Classifier::IsolationForest::App -command;
use File::Slurp      qw(write_file);
use File::Spec       ();
use Scalar::Util     qw(looks_like_number);
use IO::Socket::UNIX ();
use IO::Select       ();

# JSON::MaybeXS codec and the connected socket, set up in execute.
my $JSON;
my $SOCK;
my $TIMEOUT;
my $READ_BUF = '';

sub opt_spec {
	return (
		[
			'set=s',
			'Named streamd instance to talk to; the socket becomes <set>.sock under the run dir, '
				. 'exactly as streamd resolves it. Must match /\A[A-Za-z0-9+\-@_]+\z/.'
		],
		[
			'socket=s',
			'Unix domain socket streamd listens on; default /var/run/iforest_streamd/streamd.sock. With '
				. '--set this is instead the base run dir (default /var/run/iforest_streamd) holding <set>.sock.',
			{ 'completion' => 'files' }
		],
		[ 'timeout=i', 'Seconds to wait for each reply from the daemon.', { 'default' => 30 } ],

		# stream mode
		[
			'i=s',
			'Input to stream through the daemon, one row per line; - reads stdin.',
			{ 'completion' => 'files' }
		],
		[ 'o=s', 'Output the results to this file instead of printing.', { 'completion' => 'files' } ],
		[ 'w',   'If the file specified via -o exists, over write it.' ],
		[ 'd',   'Include the input data in the output (CSV input only).' ],
		[
			'mode=s',
			"What each row does: 'prequential' (score against the model as it stood, then learn -- the "
				. "default), 'learn' (learn only, no output), or 'score' (score only, nothing learned).",
			{ 'default' => 'prequential' }
		],
		[
			'jsonl',
			'Input lines are JSON rows instead of CSV: an array is positional, an object is a tagged row '
				. '(full munger plan, raw values may contain anything JSON can). Output is the daemon\'s '
				. 'reply JSON lines verbatim, one per request (--batch 1 for one per row).'
		],
		[
			'batch=i',
			'Rows per request message. Bigger amortises round trips; 1 gives per-row latency for '
				. 'tail -F style pipelines.',
			{ 'default' => 256 }
		],

		# command mode
		[ 'ping',              'Check the daemon is alive; exits 0 on pong.' ],
		[ 'stats',             'Print the daemon stats (seen, window, threshold, connections, set, ...).' ],
		[ 'save',              'Ask the daemon to save the model now; prints the file name.' ],
		[ 'relearn-threshold', 'Ask the daemon to relearn the contamination decision threshold.' ],
		[ 'json',              'Command mode: print the raw JSON reply instead of the text rendering.' ],
	);
} ## end sub opt_spec

lib/Algorithm/Classifier/IsolationForest/App/Command/streamc.pm  view on Meta::CPAN

		my $got   = _request( { rows => [@rows], mode => $opt->{'mode'}, tag => $batch_start } );
		my $reply = $got->{reply};
		if ( defined $reply->{error} ) {
			my $line = $batch_start;
			my $err  = $reply->{error};
			# Batch errors come back as "row N: ..." with N relative to
			# the message; map it back to the input line.
			if ( $err =~ s/\Arow (\d+): // ) {
				$line = $batch_start + $1;
			}
			die( 'line ' . $line . ' of input: ' . $err . "\n" );
		} ## end if ( defined $reply->{error} )
		if ( $opt->{'mode'} ne 'learn' ) {
			if ( $opt->{'jsonl'} ) {
				$emit->( $got->{raw} );
			} else {
				my $pairs = $reply->{scores};
				for my $i ( 0 .. $#$pairs ) {
					my $prefix = $opt->{'d'} ? $raw[$i] . ',' : '';
					$emit->( $prefix . $pairs->[$i][0] . ',' . $pairs->[$i][1] );
				}
			}
		} ## end if ( $opt->{'mode'} ne 'learn' )
		@rows        = ();
		@raw         = ();
		$batch_start = $line_int + 1;
	}; ## end $flush = sub

	while ( my $line = <$in_fh> ) {
		$line_int++;
		chomp $line;
		if ( $line =~ /^\s*$/ ) {
			$flush->();    # keep line-number accounting exact across blanks
			$batch_start = $line_int + 1;
			next;
		}

		if ( $opt->{'jsonl'} ) {
			my $row = eval { $JSON->decode($line) };
			die( 'line ' . $line_int . ' of -i did not parse as JSON: ' . $@ ) if $@;
			die( 'line ' . $line_int . ' of -i must be a JSON array (positional) or object (tagged)' . "\n" )
				unless ref $row eq 'ARRAY' || ref $row eq 'HASH';
			push @rows, $row;
		} else {
			my @fields = split( /,/, $line, -1 );
			if ( !defined $expected_cols ) {
				$expected_cols = scalar @fields;
				die( 'Line ' . $line_int . ' of input has no columns' ) if $expected_cols < 1;
			} elsif ( scalar @fields != $expected_cols ) {
				die(      'Line '
						. $line_int
						. ' of input has '
						. scalar(@fields)
						. ' columns but expected '
						. $expected_cols );
			}

			# Numeric-looking fields travel as JSON numbers, everything
			# else as strings for the daemon's munger plan to handle; the
			# daemon owns validation either way.
			push @rows, [ map { looks_like_number($_) ? 0 + $_ : $_ } @fields ];
			push @raw,  $line;
		} ## end else [ if ( $opt->{'jsonl'} ) ]

		$flush->() if scalar @rows >= $opt->{'batch'};
	} ## end while ( my $line = <$in_fh> )
	$flush->();

	if ( defined $opt->{'o'} ) {
		write_file( $opt->{'o'}, { 'atomic' => 1 }, $results );
	}
	return 1;
} ## end sub _stream

return 1;