X-Git-Url: http://dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FRBN.pm;h=6ec93569b01c0bd837217803c2abf553b2d0fa29;hb=bd8b8aa6d37a7dde70d093c5552c29f519ac9613;hp=91d1a9fd2a18c224df5b81dcba7fd7e0c1e2090c;hpb=841b61045ab2b8fa959eed04909c17e092b7a16c;p=spider.git diff --git a/perl/RBN.pm b/perl/RBN.pm index 91d1a9fd..6ec93569 100644 --- a/perl/RBN.pm +++ b/perl/RBN.pm @@ -11,15 +11,63 @@ package RBN; use 5.10.1; -use DXUtil; +use lib qw {.}; + use DXDebug; +use DXUtil; use DXLog; use DXUser; use DXChannel; -use Math::Round qw(nearest); +use Math::Round qw(nearest nearest_floor); use Date::Parse; -use Time::HiRes qw(clock_gettime CLOCK_REALTIME); +use Time::HiRes qw(gettimeofday); use Spot; +use DXJSON; +use IO::File; + +use constant { + ROrigin => 0, + RQrg => 1, + RCall => 2, + RMode => 3, + RStrength => 4, + RTime => 5, + RUtz => 6, + Respot => 7, + RQra => 8, + RSpotData => 9, + }; + +use constant { + SQrg => 0, + SCall => 1, + STime => 2, + SComment => 3, + SOrigin => 4, + SZone => 11, + }; +use constant { + OQual => 0, + OAvediff => 1, + OSpare => 2, + ODiff => 3, + }; +use constant { + CTime => 0, + CQual => 1, + CData => 2, + }; + +use constant { + DScore => 0, + DGood => 1, + DBad => 2, + DLastin => 3, + DEviants => 4, + }; + + +our $DATA_VERSION = 1; our @ISA = qw(DXChannel); @@ -27,17 +75,46 @@ our $startup_delay = 5*60; # don't send anything out until this timer has expir # this is to allow the feed to "warm up" with duplicates # so that the "big rush" doesn't happen. -our $minspottime = 60*60; # the time between respots of a callsign - if a call is +our $minspottime = 30*60; # the time between respots of a callsign - if a call is # still being spotted (on the same freq) and it has been # spotted before, it's spotted again after this time # until the next minspottime has passed. our $beacontime = 5*60; # same as minspottime, but for beacons (and shorter) -our $dwelltime = 6; # the amount of time to wait for duplicates before issuing +our $dwelltime = 10; # the amount of time to wait for duplicates before issuing # a spot to the user (no doubt waiting with bated breath). -our $filterdef = $Spot::filterdef; # we use the same filter as the Spot system. Can't think why. +our $filterdef = $Spot::filterdef; # we use the same filter as the Spot system. Can't think why :-). + +my $spots; # the GLOBAL spot cache + +my %runtime; # how long each channel has been running + +our $cachefn = localdata('rbn_cache'); +our $cache_valid = 4*60; # The cache file is considered valid if it is not more than this old + +our $maxqrgdiff = 10; # the maximum +our $minqual = 2; # the minimum quality we will accept for output + +my $json; +my $noinrush = 0; # override the inrushpreventor if set +our $maxdeviants = 5; # the number of deviant QRGs to record for skimmer records + +sub init +{ + $json = DXJSON->new; + if (check_cache()) { + $noinrush = 1; + } else { + $spots = {VERSION=>$DATA_VERSION}; + } + if (defined $DB::VERSION) { + $noinrush = 1; + $json->indent(1); + } + +} sub new { @@ -47,17 +124,27 @@ sub new my $pkg = shift; my $call = shift; -# DXProt::_add_thingy($main::routeroot, [$call, 0, 0, 1, undef, undef, $self->hostname], ); - $self->{spot} = {}; $self->{last} = 0; $self->{noraw} = 0; $self->{nospot} = 0; + $self->{nouser} = {}; $self->{norbn} = 0; + $self->{noraw10} = 0; + $self->{nospot10} = 0; + $self->{nouser10} = {}; + $self->{norbn10} = 0; + $self->{nospothour} = 0; + $self->{nouserhour} = {}; + $self->{norbnhour} = 0; + $self->{norawhour} = 0; $self->{sort} = 'N'; $self->{lasttime} = $main::systime; $self->{minspottime} = $minspottime; $self->{beacontime} = $beacontime; $self->{showstats} = 0; + $self->{pingint} = 0; + $self->{nopings} = 0; + $self->{queue} = {}; return $self; } @@ -84,7 +171,10 @@ sub start ($h) = $line =~ /host=([\da..fA..F:]+)/; $line =~ s/\s*host=[\da..fA..F:]+// if $h; } - $self->{hostname} = $h if $h; + if ($h) { + $h =~ s/^::ffff://; + $self->{hostname} = $h; + } } $self->{width} = 80 unless $self->{width} && $self->{width} > 80; $self->{consort} = $line; # save the connection type @@ -112,8 +202,10 @@ sub start $user->qra(DXBearing::lltoqra($lat, $long)) if (defined $lat && defined $long); } - # start inrush timer - $self->{inrushpreventor} = $main::systime + $startup_delay; + # if we have been running and stopped for a while + # if the cache is warm enough don't operate the inrush preventor + $self->{inrushpreventor} = exists $runtime{$call} && $runtime{$call} > $startup_delay || $noinrush ? 0 : $main::systime + $startup_delay; + dbg("RBN: noinrush: $noinrush, setting inrushpreventor on $self->{call} to $self->{inrushpreventor}"); } my @queue; # the queue of spots ready to send @@ -123,10 +215,7 @@ sub normal my $self = shift; my $line = shift; my @ans; - my $spots = $self->{spot}; - - # save this for them's that need it - my $rawline = $line; + my $dbgrbn = isdbg('rbn'); # remove leading and trailing spaces chomp $line; @@ -135,7 +224,7 @@ sub normal # add base RBN - my $tim = $main::systime; + my $now = $main::systime; # parse line dbg "RBN:RAW,$line" if isdbg('rbnraw'); @@ -149,14 +238,23 @@ sub normal my $qra = $spd, $spd = '' if is_qra($spd); $u = $qra if $qra; + # is this anything like a callsign? + unless (is_callsign($call)) { + dbg("RBN: ERROR $call from $origin on $qrg is invalid, dumped"); + return; + } + $origin =~ s/\-(?:\d{1,2}\-)?\#$//; # get rid of all the crap we aren't interested in $sort ||= ''; $tx ||= ''; $qra ||= ''; - dbg qq{or:$origin qr:$qrg ca:$call mo:$mode s:$s m:$m sp:$spd u:$u sort:$sort t:$t tx:$tx qra:$qra} if isdbg('rbn'); + dbg qq{RBN:input decode or:$origin qr:$qrg ca:$call mo:$mode s:$s m:$m sp:$spd u:$u sort:$sort t:$t tx:$tx qra:$qra} if $dbgrbn; + ++$self->{noraw}; + ++$self->{noraw10}; + ++$self->{norawhour}; my $b; @@ -205,145 +303,111 @@ sub normal # process to just the standard "message passing" which has been shown to be able to sustain over 5000 # per second (limited by the test program's output and network speed, rather than DXSpider's handling). - my $nqrg = nearest(1, $qrg); # normalised to nearest Khz + my $search = 5; + my $nqrg = nearest(1, $qrg * 10); # normalised to nearest Khz my $sp = "$call|$nqrg"; # hopefully the skimmers will be calibrated at least this well! - my $spp = sprintf("$call|%d", $nqrg+1); # but, clearly, my hopes are rudely dashed - my $spm = sprintf("$call|%d", $nqrg-1); # in BOTH directions! - # do we have it? - my $spot = $spots->{$sp}; - $spot = $spots->{$spp}, $sp = $spp, dbg(qq{RBN: SPP using $spp for $sp}) if !$spot && exists $spots->{$spp}; - $spot = $spots->{$spm}, $sp = $spm, dbg(qq{RBN: SPM using $spm for $sp}) if !$spot && exists $spots->{$spm}; + # find it? + my $cand = $spots->{$sp}; + unless ($cand) { + my ($i, $new); + for ($i = $nqrg; !$cand && $i <= $nqrg+$search; $i += 1) { + $new = "$call|$i"; + $cand = $spots->{$new}, last if exists $spots->{$new}; + } + if ($cand) { + my $diff = $i - $nqrg; + dbg(qq{RBN: QRG Diff using $new (+$diff) for $sp for qrg $qrg}) if (isdbg('rbnqrg') || $dbgrbn); + $sp = $new; + } + } + unless ($cand) { + my ($i, $new); + for ($i = $nqrg; !$cand && $i >= $nqrg-$search; $i -= 1) { + $new = "$call|$i"; + $cand = $spots->{$new}, last if exists $spots->{$new}; + } + if ($cand) { + my $diff = $nqrg - $i; + dbg(qq{RBN: QRG Diff using $new (-$diff) for $sp for qrg $qrg}) if (isdbg('rbnqrg') || $dbgrbn); + $sp = $new; + } + } - # if we have one and there is only one slot and that slot's time isn't expired for respot then return my $respot = 0; - if ($spot && ref $spot) { - if (@$spot == 1) { - unless ($self->{minspottime} > 0 && $tim - $spot->[0] >= $self->{minspottime}) { - dbg("RBN: key: '$sp' call: $call qrg: $qrg DUPE \@ ". atime(int $spot->[0])) if isdbg('rbn'); + if ($cand && ref $cand) { + if (@$cand <= CData) { + unless ($self->{minspottime} > 0 && $now - $cand->[CTime] >= $self->{minspottime}) { + dbg("RBN: key: '$sp' call: $call qrg: $qrg DUPE \@ ". atime(int $cand->[CTime])) if $dbgrbn; return; } - dbg("RBN: key: '$sp' RESPOTTING call: $call qrg: $qrg last seen \@ ". atime(int $spot->[0])) if isdbg('rbn'); - undef $spot; # it's about to be recreated (in one place) + dbg("RBN: key: '$sp' RESPOTTING call: $call qrg: $qrg last seen \@ ". atime(int $cand->[CTime])) if $dbgrbn; + $cand->[CTime] = $now; ++$respot; } # otherwise we have a spot being built up at the moment - } elsif ($spot) { - dbg("RBN: key '$sp' = '$spot' not ref"); + } elsif ($cand) { + dbg("RBN: key '$sp' = '$cand' not ref"); return; } # here we either have an existing spot record buildup on the go, or we need to create the first one - unless ($spot) { - $spots->{$sp} = $spot = [clock_gettime(CLOCK_REALTIME)];; - dbg("RBN: key: '$sp' call: $call qrg: $qrg NEW" . $respot ? ' RESPOT' : '') if isdbg('rbn'); + unless ($cand) { + $spots->{$sp} = $cand = [$now, 0]; + dbg("RBN: key: '$sp' call: $call qrg: $qrg NEW" . ($respot ? ' RESPOT' : '')) if $dbgrbn; } # add me to the display queue unless we are waiting for initial in rush to finish - return unless $self->{inrushpreventor} < $main::systime; - push @{$self->{queue}}, $sp if @$spot == 1; # queue the KEY (not the record) + return unless $noinrush || $self->{inrushpreventor} < $main::systime; # build up a new record and store it in the buildup # deal with the unix time my ($hh,$mm) = $t =~ /(\d\d)(\d\d)Z$/; my $utz = $hh*3600 + $mm*60 + $main::systime_daystart; # possible issue with late spot from previous day - $utz -= 86400 if $utz > $tim+3600; # too far ahead, drag it back one day + $utz -= 86400 if $utz > $now+3600; # too far ahead, drag it back one day # create record and add into the buildup my $r = [$origin, nearest(.1, $qrg), $call, $mode, $s, $t, $utz, $respot, $u]; - dbg("RBN: key: '$sp' ADD RECORD call: $call qrg: $qrg origin: $origin") if isdbg('rbn'); - my @s = Spot::prepare($r->[1], $r->[2], $r->[6], '', $r->[0]); + my @s = Spot::prepare($r->[RQrg], $r->[RCall], $r->[RUtz], '', $r->[ROrigin]); + if ($s[5] == 666) { + dbg("RBN: ERROR invalid prefix/callsign $call from $origin-# on $qrg, dumped"); + return; + } + if ($self->{inrbnfilter}) { my ($want, undef) = $self->{inrbnfilter}->it($s); - next unless $want; + return unless $want; } - $r->[9] = \@s; + $r->[RSpotData] = \@s; - push @$spot, $r; + ++$self->{queue}->{$sp};# unless @$cand>= CData; # queue the KEY (not the record) - # At this point we run the queue to see if anything can be sent onwards to the punter - my $now = clock_gettime(CLOCK_REALTIME); + dbg("RBN: key: '$sp' ADD RECORD call: $call qrg: $qrg origin: $origin") if $dbgrbn; - # now run the waiting queue which just contains KEYS ($call|$qrg) - foreach $sp (@{$self->{queue}}) { - my $cand = $spots->{$sp}; - unless ($cand && $cand->[0]) { - dbg "RBN Cand " . ($cand ? 'def' : 'undef') . " [0] " . ($cand->[0] ? 'def' : 'undef') . " dwell $dwelltime"; - next; - } - if ($now >= $cand->[0] + $dwelltime ) { - # we have a candidate, create qualitee value(s); - unless (@$cand > 1) { - dbg "RBN: QUEUE key '$sp' MISSING RECORDS " . dd($cand) if isdbg 'rbn'; - shift @{$self->{queue}}; - next; - } - my $savedtime = shift @$cand; # save the start time - my $r = $cand->[0]; - my $quality = @$cand; - $quality = 9 if $quality > 9; - $quality = "Q:$quality"; - if (isdbg('progress')) { - my $s = "RBN: SPOT key: '$sp' = $r->[2] on $r->[1] \@ $r->[5] $quality"; - $s .= " route: $self->{call}"; - dbg($s); - } - - send_dx_spot($self, $quality, $cand); - - # clear out the data and make this now just "spotted", but no further action required until respot time - dbg "RBN: QUEUE key '$sp' cleared" if isdbg 'rbn'; - - $spots->{$sp} = [$savedtime]; - shift @{$self->{queue}}; - } else { - dbg sprintf("RBN: QUEUE key: '$sp' SEND time not yet reached %.1f secs left", $spot->[0] + $dwelltime - $now) if isdbg 'rbnqueue'; - } - } - + push @$cand, $r; } else { - dbg "RBN:DATA,$line" if isdbg('rbn'); - } - - # # periodic clearing out of the two caches - if (($tim % 60 == 0 && $tim > $self->{last}) || ($self->{last} && $tim >= $self->{last} + 60)) { - my $count = 0; - my $removed = 0; - while (my ($k,$v) = each %{$spots}) { - if ($tim - $v->[0] > $self->{minspottime}*2) { - delete $spots->{$k}; - ++$removed; - } - else { - ++$count; - } - } - dbg "RBN:ADMIN,$self->{call},spot cache remain: $count removed: $removed"; # if isdbg('rbn'); - dbg "RBN:" . join(',', "STAT", $self->{noraw}, $self->{norbn}, $self->{nospot}) if $self->{showstats}; - $self->{noraw} = $self->{norbn} = $self->{nospot} = 0; - $self->{last} = int($tim / 60) * 60; + dbg "RBN:DATA,$line" if $dbgrbn; } } - - -# } -# } - # we should get the spot record minus the time, so just an array of record (arrays) sub send_dx_spot { my $self = shift; my $quality = shift; - my $spot = shift; + my $cand = shift; + ++$self->{norbn}; + ++$self->{norbn10}; + ++$self->{norbnhour}; + # $r = [$origin, $qrg, $call, $mode, $s, $utz, $respot]; - my $mode = $spot->[0]->[3]; # as all the modes will be the same; + my $mode = $cand->[CData]->[RMode]; # as all the modes will be the same; my @dxchan = DXChannel::get_all(); @@ -357,15 +421,20 @@ sub send_dx_spot ++$want if $user->wantbeacon && $mode =~ /^BCN|DXF/; ++$want if $user->wantcw && $mode =~ /^CW/; ++$want if $user->wantrtty && $mode =~ /^RTT/; - ++$want if $user->wantpsk && $mode =~ /^PSK/; - ++$want if $user->wantcw && $mode =~ /^CW/; + ++$want if $user->wantpsk && $mode =~ /^PSK|FSK|MSK/; ++$want if $user->wantft && $mode =~ /^FT/; - ++$want unless $want; # send everything if nothing is selected. - next unless $want; + dbg(sprintf("RBN: spot selection for $dxchan->{call} mode: '$mode' want: $want flags rbn:%d ft:%d bcn:%d cw:%d psk:%d rtty:%d", + $user->wantrbn, + $user->wantft, + $user->wantbeacon, + $user->wantcw, + $user->wantpsk, + $user->wantrtty, + )) if isdbg('rbnll'); # send one spot to one user out of the ones that we have - $self->dx_spot($dxchan, $quality, $spot) if $want; + $self->dx_spot($dxchan, $quality, $cand) if $want; } } @@ -374,90 +443,417 @@ sub dx_spot my $self = shift; my $dxchan = shift; my $quality = shift; - my $spot = shift; - + my $cand = shift; + my $call = $dxchan->{call}; + my $seeme = $dxchan->user->rbnseeme(); my $strength = 100; # because it could if we talk about FTx my $saver; - my %zone; - my %qrg; my $respot; my $qra; - - foreach my $r (@$spot) { + + ++$self->{nousers}->{$call}; + ++$self->{nousers10}->{$call}; + ++$self->{nousershour}->{$call}; + + my $filtered; + my $rf = $dxchan->{rbnfilter} || $dxchan->{spotsfilter}; + my $comment; + + foreach my $r (@$cand) { # $r = [$origin, $qrg, $call, $mode, $s, $t, $utz, $respot, $qra]; # Spot::prepare($qrg, $call, $utz, $comment, $origin); + next unless $r && ref $r; - my $comment = sprintf "%-3s %2ddB $quality", $r->[3], $r->[4]; - $respot = 1 if $r->[7]; - $qra = $r->[8] if !$qra && $r->[8] && is_qra($r->[8]); + $qra = $r->[RQra] if !$qra && $r->[RQra] && is_qra($r->[RQra]); - my $s = $r->[9]; # the prepared spot - $s->[3] = $comment; # apply new generated comment - - - ++$zone{$s->[11]}; # save the spotter's zone - ++$qrg{$s->[0]}; # and the qrg + $comment = sprintf "%-3s %2ddB $quality", $r->[RMode], $r->[RStrength]; + my $s = $r->[RSpotData]; # the prepared spot + $s->[SComment] = $comment; # apply new generated comment - - my $want = 0; - my $rf = $dxchan->{rbnfilter} || $dxchan->{spotsfilter}; - if ($rf) { - ($want, undef) = $rf->it($s); - next unless $want; - $saver = $s; - dbg("RBN: FILTERED call: $s->[1] qrg: $s->[0] origin: $s->[4] dB: $r->[4]") if isdbg 'rbn'; - last; + ++$zone{$s->[SZone]}; # save the spotter's zone + + # if the 'see me' flag is set, then show all the spots without further adornment (see set/rbnseeme for more info) + if ($seeme) { + send_final($dxchan, $s); + next; } # save the lowest strength one - if ($r->[4] < $strength) { - $strength = $r->[4]; + if ($r->[RStrength] < $strength) { + $strength = $r->[RStrength]; $saver = $s; - dbg("RBN: STRENGTH call: $s->[1] qrg: $s->[0] origin: $s->[4] dB: $r->[4]") if isdbg 'rbn'; + dbg("RBN: STRENGTH spot: $s->[SCall] qrg: $s->[SQrg] origin: $s->[SOrigin] dB: $r->[RStrength] < $strength") if isdbg 'rbnll'; + } + + if ($rf) { + my ($want, undef) = $rf->it($s); + dbg("RBN: FILTERING for $call spot: $s->[SCall] qrg: $s->[SQrg] origin: $s->[SOrigin] dB: $r->[RStrength] com: '$s->[SComment]' want: " . ($want ? 'YES':'NO')) if isdbg 'rbnll'; + next unless $want; + $filtered = $s; } } + if ($rf) { + $saver = $filtered; # if nothing passed the filter's lips then $saver == $filtered == undef ! + } + if ($saver) { my $buf; # create a zone list of spotters - delete $zone{$saver->[11]}; # remove this spotter's zone (leaving all the other zones) + delete $zone{$saver->[SZone]}; # remove this spotter's zone (leaving all the other zones) my $z = join ',', sort {$a <=> $b} keys %zone; - # determine the most likely qrg and then set it - my $mv = 0; - my $fk; - my $c = 0; - while (my ($k, $v) = each %qrg) { - $fk = $k, $mv = $v if $v > $mv; - ++$c; - } - $saver->[0] = $fk; - $saver->[3] .= '*' if $c > 1; - $saver->[3] .= '+' if $respot; - $saver->[3] .= " Z:$z" if $z; + # alter spot data accordingly + $saver->[SComment] .= " Z:$z" if $z; + + send_final($dxchan, $saver); + + ++$self->{nospot}; + ++$self->{nospot10}; + ++$self->{nospothour}; - dbg("RBN: SENDING call: $saver->[1] qrg: $saver->[0] origin: $saver->[4] $saver->[3]") if isdbg 'rbn'; - if ($dxchan->{ve7cc}) { - my $call = $saver->[4]; - $saver->[4] .= '-#'; - $buf = VE7CC::dx_spot($dxchan, @$saver); - $saver->[4] = $call; - } else { - $buf = $dxchan->format_dx_spot(@$saver); - } - $buf =~ s/^DX/RB/; - $dxchan->local_send('N', $buf); - if ($qra) { - my $user = DXUser::get_current($saver->[1]) || DXUser->new($saver->[1]); + my $user = DXUser::get_current($saver->[SCall]) || DXUser->new($saver->[SCall]); unless ($user->qra && is_qra($user->qra)) { $user->qra($qra); - dbg("RBN: update qra on $saver->[1] to $qra"); + dbg("RBN: update qra on $saver->[SCall] to $qra"); $user->put; } } } } +sub send_final +{ + my $dxchan = shift; + my $saver = shift; + my $call = $dxchan->{call}; + my $buf; + + dbg("RBN: SENDING to $call spot: $saver->[SCall] qrg: $saver->[SQrg] origin: $saver->[SOrigin] $saver->[SComment]") if isdbg 'rbnll'; + if ($dxchan->{ve7cc}) { + my $call = $saver->[SOrigin]; + $saver->[SOrigin] .= '-#'; + $buf = VE7CC::dx_spot($dxchan, @$saver); + $saver->[SOrigin] = $call; + } else { + my $call = $saver->[SOrigin]; + $saver->[SOrigin] = substr($call, 0, 6); + $saver->[SOrigin] .= '-#'; + $buf = $dxchan->format_dx_spot(@$saver); + $saver->[SOrigin] = $call; + } + $dxchan->local_send('N', $buf); +} + +# per second +sub process +{ + my $rbnskim = isdbg('rbnskim'); + + foreach my $dxchan (DXChannel::get_all()) { + next unless $dxchan->is_rbn; + + # At this point we run the queue to see if anything can be sent onwards to the punter + my $now = $main::systime; + my $ta = [gettimeofday]; + my $items = 0; + + # now run the waiting queue which just contains KEYS ($call|$qrg) + foreach my $sp (keys %{$dxchan->{queue}}) { + my $cand = $spots->{$sp}; + ++$items; + unless ($cand && $cand->[CTime]) { + dbg "RBN Cand $sp " . ($cand ? 'def' : 'undef') . " [CTime] " . ($cand->[CTime] ? 'def' : 'undef') . " dwell $dwelltime"; + next; + } + if ($now >= $cand->[CTime] + $dwelltime ) { + # we have a candidate, create qualitee value(s); + unless (@$cand > CData) { + dbg "RBN: QUEUE key '$sp' MISSING RECORDS, IGNORED" . dd($cand) if isdbg 'rbnqueue'; + next; + } + dbg "RBN: QUEUE PROCESSING key: '$sp' $now >= $cand->[CTime]" if isdbg 'rbnqueue'; + my $quality = @$cand - CData; + my $spotters = $quality; + + # dump it and remove it from the queue if it is of unadequate quality + if ($quality < $minqual) { + if ($rbnskim) { + my $r = $cand->[CData]; + if ($r) { + my $s = "RBN:SKIM Ignored (Q:$quality < Q:$minqual) key: '$sp' = $r->[RCall] on $r->[RQrg] by $r->[ROrigin] \@ $r->[RTime] route: $dxchan->{call}"; + dbg($s); + } + } + delete $spots->{$sp}; # don't remember it either - this means that a spot HAS to come in with sufficient spotters to be processed. + delete $dxchan->{queue}->{$sp}; + next; + } + + $quality = 9 if $quality > 9; + $cand->[CQual] = $quality if $quality > $cand->[CQual]; + + my $r; + + # this scores each candidate according to its skimmer's QRG score (i.e. how often it agrees with its peers) + # what happens is hash of all QRGs in candidates are incremented by that skimmer's reputation for "accuracy" + # or, more exactly, past agreement with the consensus. This score can be from -5 -> +5. + my %qrg = (); + my $skimmer; + my $sk; + my $band; + my %seen = (); + foreach $r (@$cand) { + next unless ref $r; + if (exists $seen{$r->[ROrigin]}) { + undef $r; + next; + } + $seen{$r->[ROrigin]} = 1; + $band ||= int $r->[RQrg] / 1000; + $sk = "SKIM|$r->[ROrigin]|$band"; # thus only once per set of candidates + $skimmer = $spots->{$sk}; + unless ($skimmer) { + $skimmer = $spots->{$sk} = [1, 0, 0, $now, []]; # this first time, this new skimmer gets the benefit of the doubt on frequency. + dbg("RBN:SKIM new slot $sk " . $json->encode($skimmer)) if $rbnskim; + } + $qrg{$r->[RQrg]} += ($skimmer->[DScore] || 1); + } + + # determine the most likely qrg and then set it - NOTE (-)ve votes, generated by the skimmer scoring system above, are ignored + my @deviant; + my $c = 0; + my $mv = 0; + my $qrg = 0; + while (my ($k, $votes) = each %qrg) { + if ($votes >= $mv) { + $qrg = $k; + $mv = $votes; + } + ++$c; + } + + # Ignore possible spots with 0 QRG score - as determined by the skimmer scoring system above - as they are likely to be wrong + unless ($qrg > 0) { + if ($rbnskim) { + my $keys; + while (my ($k, $v) = (each %qrg)) { + $keys .= "$k=>$v, "; + } + $keys =~ /,\s*$/; + my $i = 0; + foreach $r (@$cand) { + next unless $r && ref $r; + dbg "RBN:SKIM cand $i QRG likely wrong from '$sp' = $r->[RCall] on $r->[RQrg] by $r->[ROrigin] \@ $r->[RTime] (qrgs: $keys c: $c) route: $dxchan->{call}, ignored"; + ++$i; + } + } + delete $spots->{$sp}; # get rid + delete $dxchan->{queue}->{$sp}; + next; + } + + # detemine and spit out the deviants. Then adjust the scores according to whether it is a deviant or good + # NOTE: deviant nodes can become good (or less bad), and good nodes bad (or less good) on each spot that + # they generate. This is based solely on each skimmer's agreement (or not) with the "consensus" score generated + # above ($qrg). The resultant score + good + bad is stored per band and will be used the next time a spot + # appears on this band from each skimmer. + foreach $r (@$cand) { + next unless $r && ref $r; + my $diff = $c > 1 ? nearest(.1, $r->[RQrg] - $qrg) : 0; + $sk = "SKIM|$r->[ROrigin]|$band"; + $skimmer = $spots->{$sk}; + if ($diff) { + ++$skimmer->[DBad] if $skimmer->[DBad] < $maxdeviants; + --$skimmer->[DGood] if $skimmer->[DGood] > 0; + push @deviant, sprintf("$r->[ROrigin]:%+.1f", $diff); + push @{$skimmer->[DEviants]}, $diff; + shift @{$skimmer->[DEviants]} while @{$skimmer->[DEviants]} > $maxdeviants; + } else { + ++$skimmer->[DGood] if $skimmer->[DGood] < $maxdeviants; + --$skimmer->[DBad] if $skimmer->[DBad] > 0; + shift @{$skimmer->[DEviants]}; + } + $skimmer->[DScore] = $skimmer->[DGood] - $skimmer->[DBad]; + my $lastin = difft($skimmer->[DLastin], $now, 2); + my $difflist = join(', ', @{$skimmer->[DEviants]}); + $difflist = " ($difflist)" if $difflist; + dbg("RBN:SKIM key $sp slot $sk $r->[RQrg] - $qrg = $diff Skimmer score: $skimmer->[DGood] - $skimmer->[DBad] = $skimmer->[DScore] lastseen:$lastin ago$difflist") if $rbnskim; + $skimmer->[DLastin] = $now; + $r->[RSpotData]->[SQrg] = $qrg if $qrg && $c > 1; # set all the QRGs to the agreed value + } + + $qrg = (sprintf "%.1f", $qrg)+0; + $r = $cand->[CData]; + $r->[RQrg] = $qrg; + my $squality = "Q:$cand->[CQual]"; + $squality .= '*' if $c > 1; + $squality .= '+' if $r->[Respot]; + + if (isdbg('progress')) { + my $s = "RBN: SPOT key: '$sp' = $r->[RCall] on $r->[RQrg] by $r->[ROrigin] \@ $r->[RTime] $squality route: $dxchan->{call}"; + my $td = @deviant; + $s .= " QRGScore $mv Deviants ($td/$spotters): "; + $s .= join(', ', sort @deviant) if $td; + dbg($s); + } + + # finally send it out to any waiting public + send_dx_spot($dxchan, $squality, $cand); + + # clear out the data and make this now just "spotted", but no further action required until respot time + dbg "RBN: QUEUE key '$sp' cleared" if isdbg 'rbn'; + + delete $spots->{$sp}; + delete $dxchan->{queue}->{$sp}; + + # calculate new sp (which will be 70% likely the same as the old one) + # we do this to cope with the fact that the first spotter may well be "wrongly calibrated" giving a qrg that disagrees with the majority. + # and we want to store the key that corresponds to majority opinion. + my $nqrg = nearest(1, $qrg * 10); # normalised to nearest Khz + my $nsp = "$r->[RCall]|$nqrg"; + if ($sp ne $nsp) { + dbg("RBN:SKIM CHANGE KEY sp '$sp' -> '$nsp' for storage") if $rbnskim; + $spots->{$nsp} = [$now, $cand->[CQual]]; + } + } + else { + dbg sprintf("RBN: QUEUE key: '$sp' SEND time not yet reached %.1f secs left", $cand->[CTime] + $dwelltime - $now) if isdbg 'rbnqueue'; + } + } + if (isdbg('rbntimer')) { + my $diff = _diffus($ta); + dbg "RBN: TIMER process queue for call: $dxchan->{call} $items spots $diff uS"; + } + } +} + +sub per_minute +{ + foreach my $dxchan (DXChannel::get_all()) { + next unless $dxchan->is_rbn; + dbg "RBN:STATS minute $dxchan->{call} raw: $dxchan->{noraw} retrieved spots: $dxchan->{norbn} delivered: $dxchan->{nospot} after filtering to users: " . scalar keys %{$dxchan->{nousers}} if isdbg('rbnstats'); + if ($dxchan->{noraw} == 0 && $dxchan->{lasttime} > 60) { + LogDbg('RBN', "RBN: no input from $dxchan->{call}, disconnecting"); + $dxchan->disconnect; + } + $dxchan->{noraw} = $dxchan->{norbn} = $dxchan->{nospot} = 0; $dxchan->{nousers} = {}; + $runtime{$dxchan->{call}} += 60; + } + + # save the spot cache + write_cache() unless $main::systime + $startup_delay < $main::systime;; +} + +sub per_10_minute +{ + my $count = 0; + my $removed = 0; + while (my ($k,$cand) = each %{$spots}) { + next if $k eq 'VERSION'; + next if $k =~ /^O\|/; + next if $k =~ /^SKIM\|/; + + if ($main::systime - $cand->[CTime] > $minspottime*2) { + delete $spots->{$k}; + ++$removed; + } + else { + ++$count; + } + } + dbg "RBN:STATS spot cache remain: $count removed: $removed"; # if isdbg('rbn'); + foreach my $dxchan (DXChannel::get_all()) { + next unless $dxchan->is_rbn; + my $nq = keys %{$dxchan->{queue}}; + my $pc = $dxchan->{noraw10} ? sprintf("%.1f%%",$dxchan->{norbn10}*100/$dxchan->{noraw10}) : '0.0%'; + dbg "RBN:STATS 10-minute $dxchan->{call} queue: $nq raw: $dxchan->{noraw10} retrieved spots: $dxchan->{norbn10} ($pc) delivered: $dxchan->{nospot10} after filtering to users: " . scalar keys %{$dxchan->{nousers10}}; + $dxchan->{noraw10} = $dxchan->{norbn10} = $dxchan->{nospot10} = 0; $dxchan->{nousers10} = {}; + } +} + +sub per_hour +{ + foreach my $dxchan (DXChannel::get_all()) { + next unless $dxchan->is_rbn; + my $nq = keys %{$dxchan->{queue}}; + my $pc = $dxchan->{norawhour} ? sprintf("%.1f%%",$dxchan->{norbnhour}*100/$dxchan->{norawhour}) : '0.0%'; + dbg "RBN:STATS hour $dxchan->{call} queue: $nq raw: $dxchan->{norawhour} retrieved spots: $dxchan->{norbnhour} ($pc) delivered: $dxchan->{nospothour} after filtering to users: " . scalar keys %{$dxchan->{nousershour}}; + $dxchan->{norawhour} = $dxchan->{norbnhour} = $dxchan->{nospothour} = 0; $dxchan->{nousershour} = {}; + } +} + +sub finish +{ + write_cache(); +} + +sub write_cache +{ + my $ta = [ gettimeofday ]; + $json->indent(1) if isdbg 'rbncache'; + my $s = eval {$json->encode($spots)}; + if ($s) { + my $fh = IO::File->new(">$cachefn") or confess("writing $cachefn $!"); + $fh->print($s); + $fh->close; + } else { + dbg("RBN:Write_cache error '$@'"); + } + $json->indent(0); + my $diff = _diffms($ta); + dbg("RBN:WRITE_CACHE time to write: $diff mS"); +} + +sub check_cache +{ + if (-e $cachefn) { + my $mt = (stat($cachefn))[9]; + my $t = $main::systime - $mt || 1; + my $p = difft($mt, 2); + if ($t < $cache_valid) { + dbg("RBN:check_cache '$cachefn' spot cache exists, created $p ago and not too old"); + my $fh = IO::File->new($cachefn); + my $s; + if ($fh) { + local $/ = undef; + $s = <$fh>; + dbg("RBN:check_cache cache read size " . length $s); + $fh->close; + } else { + dbg("RBN:check_cache file read error $!"); + return undef; + } + if ($s) { + eval {$spots = $json->decode($s)}; + if ($spots && ref $spots) { + if (exists $spots->{VERSION} && $spots->{VERSION} == $DATA_VERSION) { + # now clean out anything that has spot build ups in progress + while (my ($k, $cand) = each %$spots) { + next if $k eq 'VERSION'; + next if $k =~ /^O\|/; + next if $k =~ /^SKIM\|/; + if (@$cand > CData) { + $spots->{$k} = [$cand->[CTime], $cand->[CQual]]; + } + } + dbg("RBN:check_cache spot cache restored"); + return 1; + } + } + dbg("RBN::checkcache error decoding $@"); + } + } else { + my $d = difft($main::systime-$cache_valid); + dbg("RBN::checkcache '$cachefn' created $p ago is too old (> $d), ignored"); + } + } else { + dbg("RBN:check_cache '$cachefn' spot cache not present"); + } + + return undef; +} + 1;