view release on metacpan or  search on metacpan

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

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

sub abstract { 'Client for iforest streamd: stream rows through it or send it commands' }

sub description {
	'Talks to a running `iforest streamd` daemon over its Unix socket,
speaking the same one-JSON-document-per-line protocol.

Stream mode (-i) feeds rows through the daemon and prints one result
per row, in order.  Input is CSV by default (positional rows, matching
`iforest stream`; fields are sent as numbers when they look like
numbers and as raw strings otherwise, so munged columns pass through
untouched) and the output is `$score,$label` lines, with -d prepending
the input columns.  With --jsonl each input line is instead a JSON row
-- an array for positional data, an object for a tagged row through
the full munger plan -- and the output is the daemon\'s reply JSON
lines verbatim.  Rows are sent in --batch sized messages, lockstep;
each request is tagged with its starting input line number, so a bad
row dies naming the input line (rows earlier in that message were
already applied by the daemon -- prequential learning is not
transactional).  All row validation is the daemon\'s: only it knows
whether the model munges.

Command mode sends exactly one of --ping, --stats, --save, or
--relearn-threshold and renders the reply as text (--json for the raw
reply).  The exit code is 0 on ok and non-zero on error, connect
failure, or timeout, so `iforest streamc --set web --ping` works
directly in health checks.

--set/--socket resolve the socket path exactly as streamd does, so the
same flags reach the same daemon.
';
} ## end sub description

sub validate {
	my ( $self, $opt, $args ) = @_;

	if ( defined( $opt->{'set'} ) && $opt->{'set'} !~ /\A[A-Za-z0-9+\-@_]+\z/ ) {
		$self->usage_error( '--set, "'
				. $opt->{'set'}
				. '", must match /\A[A-Za-z0-9+\-@_]+\z/ (letters, digits, and + - @ _ only)' );
	}

	my @cmds = grep { $opt->{$_} } qw(ping stats save relearn_threshold);
	if ( defined( $opt->{'i'} ) ) {
		if ( scalar @cmds ) {
			$self->usage_error('-i may not be combined with --ping/--stats/--save/--relearn-threshold');
		}
	} elsif ( scalar @cmds != 1 ) {
		$self->usage_error(
			'need either -i (stream mode) or exactly one of --ping, --stats, --save, --relearn-threshold');
	}

	if ( defined( $opt->{'i'} ) && $opt->{'i'} ne '-' ) {
		if ( !-f $opt->{'i'} ) {