X-Git-Url: http://dxcluster.net/gitweb/gitweb.cgi?a=blobdiff_plain;f=perl%2FRBN.pm;h=92c66f1a51820431b1c8c3bbda63e6cdacddcafc;hb=5d10d2095199b0d798926fa9e79a3f580c296b67;hp=15b4aa479fc86b1dd2702233dda887f0d380e243;hpb=cbb522ef802d48991734a4ce803fa6ffa9774588;p=spider.git diff --git a/perl/RBN.pm b/perl/RBN.pm index 15b4aa47..92c66f1a 100644 --- a/perl/RBN.pm +++ b/perl/RBN.pm @@ -18,7 +18,7 @@ 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(gettimeofday); use Spot; @@ -58,6 +58,14 @@ use constant { CData => 2, }; +use constant { + DScore => 0, + DGood => 1, + DBad => 2, + DLastin => 3, + DEviants => 4, + }; + our $DATA_VERSION = 1; @@ -91,10 +99,12 @@ 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; + $json->canonical(0); if (check_cache()) { $noinrush = 1; } else { @@ -206,7 +216,7 @@ sub normal my $self = shift; my $line = shift; my @ans; -# my $spots = $self->{spot}; + my $dbgrbn = isdbg('rbn'); # remove leading and trailing spaces chomp $line; @@ -241,7 +251,7 @@ sub normal $sort ||= ''; $tx ||= ''; $qra ||= ''; - 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 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}; @@ -294,37 +304,33 @@ 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 $nearest = 1; my $search = 5; - my $mult = 10; - my $tqrg = $qrg * $mult; - my $nqrg = nearest($nearest, $tqrg); # normalised to nearest Khz -# my $nqrg = nearest_even($qrg); # normalised to nearest Khz + my $nqrg = nearest(1, $qrg * 10); # normalised to nearest Khz my $sp = "$call|$nqrg"; # hopefully the skimmers will be calibrated at least this well! # find it? my $cand = $spots->{$sp}; unless ($cand) { my ($i, $new); - for ($i = $tqrg; !$cand && $i <= $tqrg+$search; $i += 1) { + 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 - $tqrg; - dbg(qq{RBN: QRG Diff using $new (+$diff) for $sp for qrg $qrg}) if (isdbg('rbnqrg') || isdbg('rbn')); + 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 = $tqrg; !$cand && $i >= $tqrg-$search; $i -= 1) { + for ($i = $nqrg; !$cand && $i >= $nqrg-$search; $i -= 1) { $new = "$call|$i"; $cand = $spots->{$new}, last if exists $spots->{$new}; } if ($cand) { - my $diff = $tqrg - $i; - dbg(qq{RBN: QRG Diff using $new (-$diff) for $sp for qrg $qrg}) if (isdbg('rbnqrg') || isdbg('rbn')); + my $diff = $nqrg - $i; + dbg(qq{RBN: QRG Diff using $new (-$diff) for $sp for qrg $qrg}) if (isdbg('rbnqrg') || $dbgrbn); $sp = $new; } } @@ -334,11 +340,11 @@ sub normal 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 isdbg('rbn'); + 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 $cand->[CTime])) if isdbg('rbn'); + dbg("RBN: key: '$sp' RESPOTTING call: $call qrg: $qrg last seen \@ ". atime(int $cand->[CTime])) if $dbgrbn; $cand->[CTime] = $now; ++$respot; } @@ -352,7 +358,7 @@ sub normal # here we either have an existing spot record buildup on the go, or we need to create the first one unless ($cand) { $spots->{$sp} = $cand = [$now, 0]; - dbg("RBN: key: '$sp' call: $call qrg: $qrg NEW" . ($respot ? ' RESPOT' : '')) if isdbg('rbn'); + 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 @@ -380,12 +386,12 @@ sub normal ++$self->{queue}->{$sp};# unless @$cand>= CData; # queue the KEY (not the record) - dbg("RBN: key: '$sp' ADD RECORD call: $call qrg: $qrg origin: $origin") if isdbg('rbn'); + dbg("RBN: key: '$sp' ADD RECORD call: $call qrg: $qrg origin: $origin") if $dbgrbn; push @$cand, $r; } else { - dbg "RBN:DATA,$line" if isdbg('rbn'); + dbg "RBN:DATA,$line" if $dbgrbn; } } @@ -440,6 +446,7 @@ sub dx_spot my $quality = 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; @@ -457,16 +464,22 @@ sub dx_spot foreach my $r (@$cand) { # $r = [$origin, $qrg, $call, $mode, $s, $t, $utz, $respot, $qra]; # Spot::prepare($qrg, $call, $utz, $comment, $origin); - next unless ref $r; + next unless $r && ref $r; $qra = $r->[RQra] if !$qra && $r->[RQra] && is_qra($r->[RQra]); $comment = sprintf "%-3s %2ddB $quality", $r->[RMode], $r->[RStrength]; my $s = $r->[RSpotData]; # the prepared spot $s->[SComment] = $comment; # apply new generated comment - + ++$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->[RStrength] < $strength) { $strength = $r->[RStrength]; @@ -479,7 +492,6 @@ sub dx_spot 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; -# last; } } @@ -496,22 +508,8 @@ sub dx_spot # alter spot data accordingly $saver->[SComment] .= " Z:$z" if $z; - 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; - } -# $buf =~ s/^DX/RB/; - $dxchan->local_send('N', $buf); - + send_final($dxchan, $saver); + ++$self->{nospot}; ++$self->{nospot10}; ++$self->{nospothour}; @@ -527,18 +525,46 @@ sub dx_spot } } +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; @@ -546,70 +572,165 @@ sub process 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 'rbn'; + dbg "RBN: QUEUE key '$sp' MISSING RECORDS, IGNORED" . dd($cand) if isdbg 'rbnqueue'; + 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; } 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; - my %qrg; + + # 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; - ++$qrg{$r->[RQrg]}; + 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 + + # 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; + my $qrg = 0; while (my ($k, $votes) = each %qrg) { - $qrg = $k, $mv = $votes if $votes > $mv; + if ($votes >= $mv) { + $qrg = $k; + $mv = $votes; + } ++$c; } - # spit out the deviants - if ($c > 1) { - foreach $r (@$cand) { - next unless ref $r; - my $diff = nearest(.1, $qrg - $r->[RQrg]); - push @deviant, sprintf("$r->[ROrigin]:%+.1f", $diff) if $diff != 0; - $r->[RSpotData]->[SQrg] = $qrg; # set all the QRGs to the agreed value + + # 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; + $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 ($cand->[CQual] >= $minqual) { - if (isdbg('progress')) { - my $s = "RBN: SPOT key: '$sp' = $r->[RCall] on $r->[RQrg] by $r->[ROrigin] \@ $r->[RTime] $squality route: $dxchan->{call}"; - $s .= " Deviants: " . join(', ', sort @deviant) if @deviant; - dbg($s); - } - send_dx_spot($dxchan, $squality, $cand); - } elsif (isdbg('rbn')) { - my $s = "RBN: SPOT IGNORED(Q $cand->[CQual] < $minqual) key: '$sp' = $r->[RCall] on $r->[RQrg] by $r->[ROrigin] \@ $r->[RTime] $squality route: $dxchan->{call}"; + 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'; - - $spots->{$sp} = [$now, $cand->[CQual]]; + delete $dxchan->{queue}->{$sp}; + delete $spots->{$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 @@ -636,6 +757,7 @@ sub per_10_minute 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}; @@ -649,7 +771,8 @@ sub per_10_minute foreach my $dxchan (DXChannel::get_all()) { next unless $dxchan->is_rbn; my $nq = keys %{$dxchan->{queue}}; - dbg "RBN:STATS 10-minute $dxchan->{call} queue: $nq raw: $dxchan->{noraw10} retrieved spots: $dxchan->{norbn10} delivered: $dxchan->{nospot10} after filtering to users: " . scalar keys %{$dxchan->{nousers10}}; + 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} = {}; } } @@ -659,7 +782,8 @@ sub per_hour foreach my $dxchan (DXChannel::get_all()) { next unless $dxchan->is_rbn; my $nq = keys %{$dxchan->{queue}}; - dbg "RBN:STATS hour $dxchan->{call} queue: $nq raw: $dxchan->{norawhour} retrieved spots: $dxchan->{norbnhour} delivered: $dxchan->{nospothour} after filtering to users: " . scalar keys %{$dxchan->{nousershour}}; + 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} = {}; } } @@ -671,10 +795,21 @@ sub finish sub write_cache { - my $fh = IO::File->new(">$cachefn") or confess("writing $cachefn $!"); - my $s = $json->encode($spots); - $fh->print($s); - $fh->close; + my $ta = [ gettimeofday ]; + $json->indent(1)->canonical(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 '$@'"); + return; + } + $json->indent(0)->canonical(0); + my $diff = _diffms($ta); + my $size = sprintf('%.3fKB', (length($s) / 1000)); + dbg("RBN:WRITE_CACHE size: $size time to write: $diff mS"); } sub check_cache @@ -700,10 +835,11 @@ sub check_cache eval {$spots = $json->decode($s)}; if ($spots && ref $spots) { if (exists $spots->{VERSION} && $spots->{VERSION} == $DATA_VERSION) { - # now clean out anything that is current + # 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]]; } @@ -712,8 +848,8 @@ sub check_cache return 1; } } + dbg("RBN::checkcache error decoding $@"); } - 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");