view release on metacpan or  search on metacpan

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

	my $line  = $1;
	my $reply = eval { $JSON->decode($line) };
	die( 'daemon sent an unparseable reply: ' . $@ ) if $@;
	return { raw => $line, reply => $reply };
} ## end sub _read_reply

#-------------------------------------------------------------------------------
# command mode
#-------------------------------------------------------------------------------

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

	my ($which) = grep { $opt->{$_} } qw(ping stats save relearn_threshold);
	( my $cmd = $which ) =~ tr/_/-/;

	my $got   = _request( { cmd => $cmd } );
	my $reply = $got->{reply};
	die( 'daemon error: ' . $reply->{error} . "\n" ) if defined $reply->{error};

	if ( $opt->{'json'} ) {
		print $got->{raw} . "\n";
		return 1;
	}

	my $ok = $reply->{ok};
	if ( ref $ok eq 'HASH' ) {
		for my $k ( sort keys %$ok ) {
			printf "  %-20s  %s\n", $k, ( defined $ok->{$k} ? $ok->{$k} : '(unset)' );
		}
	} else {
		print( ( defined $ok ? $ok : 'ok' ) . "\n" );
	}
	return 1;
} ## end sub _command

#-------------------------------------------------------------------------------
# stream mode
#-------------------------------------------------------------------------------

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

	my $in_fh;
	if ( $opt->{'i'} eq '-' ) {
		$in_fh = \*STDIN;
	} else {
		open( $in_fh, '<', $opt->{'i'} ) or die( 'failed to open -i, "' . $opt->{'i'} . '": ' . $! . "\n" );
	}

	# -o accumulates and writes atomically at the end (matching `iforest
	# stream`), so it is unsuitable for an endless stdin; without it,
	# results print as replies arrive.
	my $results = '';
	my $emit    = sub {
		if ( defined $opt->{'o'} ) { $results .= $_[0] . "\n" }
		else                       { print $_[0] . "\n" }
	};

	my $expected_cols;
	my @rows;               # decoded rows for the pending request
	my @raw;                # matching raw input lines, for -d
	my $batch_start = 1;    # input line number of $rows[0]
	my $line_int    = 0;

	my $flush = sub {
		return unless @rows;
		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 '