perl/RBN.pm

   1 #
   2 # The RBN connection system
   3 #
   4 # Copyright (c) 2020 Dirk Koopman G1TLH
   5 #
   6
   7 use warnings;
   8 use strict;
   9
  10 package RBN;
  11
  12 use 5.10.1;
  13
  14 use lib qw {.};
  15
  16 use DXDebug;
  17 use DXUtil;
  18 use DXLog;
  19 use DXUser;
  20 use DXChannel;
  21 use Math::Round qw(nearest nearest_floor);
  22 use Date::Parse;
  23 use Time::HiRes qw(gettimeofday);
  24 use Spot;
  25 use DXJSON;
  26 use IO::File;
  27
  28 use constant {
  29                           ROrigin => 0,
  30                           RQrg => 1,
  31                           RCall => 2,
  32                           RMode => 3,
  33                           RStrength => 4,
  34                           RTime => 5,
  35                           RUtz => 6,
  36                           Respot => 7,
  37                           RQra => 8,
  38                           RSpotData => 9,
  39                          };
  40
  41 use constant {
  42                           SQrg => 0,
  43                           SCall => 1,
  44                           STime => 2,
  45                           SComment => 3,
  46                           SOrigin => 4,
  47                           SZone => 11,
  48                          };
  49 use constant {
  50                           OQual => 0,
  51                           OAvediff => 1,
  52                           OSpare => 2,
  53                           ODiff => 3,
  54                          };
  55 use constant {
  56                           CTime => 0,
  57                           CQual => 1,
  58                           CData => 2,
  59                          };
  60
  61 use constant {
  62                           DScore => 0,
  63                           DGood => 1,
  64                           DBad => 2,
  65                           DLastin => 3,
  66                           DEviants => 4,
  67                          };
  68
  69
  70 our $DATA_VERSION = 1;
  71
  72 our @ISA = qw(DXChannel);
  73
  74 our $startup_delay = 5*60;              # don't send anything out until this timer has expired
  75                                 # this is to allow the feed to "warm up" with duplicates
  76                                 # so that the "big rush" doesn't happen.
  77
  78 our $respottime = 3*60;         # the time between respots of a callsign - if a call is
  79                                 # still being spotted (on the same freq) and it has been
  80                                 # spotted before, it's spotted again after this time
  81                                 # until the next respottime has passed.
  82
  83
  84 our $beacontime = 5*60;                 # same as minspottime, but for beacons (and shorter)
  85
  86 our $dwelltime = 10;                    # the amount of time to wait for duplicates before issuing
  87                                 # a spot to the user (no doubt waiting with bated breath).
  88
  89 our $limbotime = 5*60;                  # if there are fewer than $minqual candidates and $dwelltime
  90                                 # has expired then allow this spot to live a bit longer. It may
  91                                 # simply be that it is not in standard spot coverage. (ask G4PIQ
  92                                 # about this).
  93
  94 our $cachetime = 60*60;                 # The length of time spot data is cached
  95
  96 our $filterdef = $Spot::filterdef; # we use the same filter as the Spot system. Can't think why :-).
  97
  98 my $spots;                                              # the GLOBAL spot cache
  99 my $qrg;                                                # the GlOBAL (ephemeral) qrg cache (generated on re-read of cache)
 100
 101
 102 my %runtime;                                    # how long each channel has been running
 103
 104 our $cachefn = localdata('rbn_cache');
 105 our $cache_valid = 4*60;                # The cache file is considered valid if it is not more than this old
 106
 107 our $maxqrgdiff = 10;                   # the maximum
 108 our $minqual = 2;                               # the minimum quality we will accept for output
 109 our $maxqual = 9;                               # if there is enough quality, then short circuit any remaining dwelltime.
 110
 111 my $json;
 112 my $noinrush = 0;                               # override the inrushpreventor if set
 113 our $maxdeviants = 5;                   # the number of deviant QRGs to record for skimmer records
 114
 115 our %seeme;                                     # the list of users that want to see themselves
 116
 117
 118 sub init
 119 {
 120         $json = DXJSON->new;
 121         $json->canonical(0);
 122         if (check_cache()) {
 123                 $noinrush = 1;
 124         } else {
 125                 $spots = {VERSION=>$DATA_VERSION};
 126         }
 127         if (defined $DB::VERSION) {
 128                 $noinrush = 1;
 129                 $json->indent(1);
 130         }
 131
 132 }
 133
 134 sub new
 135 {
 136         my $self = DXChannel::alloc(@_);
 137
 138         # routing, this must go out here to prevent race condx
 139         my $pkg = shift;
 140         my $call = shift;
 141
 142         $self->{last} = 0;
 143         $self->{noraw} = 0;
 144         $self->{nospot} = 0;
 145         $self->{nouser} = {};
 146         $self->{norbn} = 0;
 147         $self->{noraw10} = 0;
 148         $self->{nospot10} = 0;
 149         $self->{nouser10} = {};
 150         $self->{norbn10} = 0;
 151         $self->{nospothour} = 0;
 152         $self->{nouserhour} = {};
 153         $self->{norbnhour} = 0;
 154         $self->{norawhour} = 0;
 155         $self->{sort} = 'N';
 156         $self->{lasttime} = $main::systime;
 157         $self->{respottime} = $respottime;
 158         $self->{beacontime} = $beacontime;
 159         $self->{showstats} = 0;
 160         $self->{pingint} = 0;
 161         $self->{nopings} = 0;
 162         $self->{queue} = {};
 163
 164         return $self;
 165 }
 166
 167 sub start
 168 {
 169         my ($self, $line, $sort) = @_;
 170         my $user = $self->{user};
 171         my $call = $self->{call};
 172         my $name = $user->{name};
 173
 174         # log it
 175         unless ($self->{hostname}) {
 176                 $self->{hostname} = $self->{conn}->peerhost || 'unknown';
 177         }
 178
 179         $self->{name} = $name ? $name : $call;
 180         $self->state('prompt');         # a bit of room for further expansion, passwords etc
 181         $self->{lang} = $user->lang || $main::lang || 'en';
 182         if ($line =~ /host=/) {
 183                 my ($h) = $line =~ /host=(\d+\.\d+\.\d+\.\d+)/;
 184                 $line =~ s/\s*host=\d+\.\d+\.\d+\.\d+// if $h;
 185                 unless ($h) {
 186                         ($h) = $line =~ /host=([\da..fA..F:]+)/;
 187                         $line =~ s/\s*host=[\da..fA..F:]+// if $h;
 188                 }
 189                 if ($h) {
 190                         $h =~ s/^::ffff://;
 191                         $self->{hostname} = $h;
 192                 }
 193         }
 194         $self->{width} = 80 unless $self->{width} && $self->{width} > 80;
 195         $self->{consort} = $line;       # save the connection type
 196
 197         LogDbg('err', "$call connected from $self->{hostname}");
 198
 199         # set some necessary flags on the user if they are connecting
 200         $self->{registered} = 1;
 201         # sort out privilege reduction
 202         $self->{priv} = 0;
 203
 204         # get the filters
 205 #       $self->{inrbnfilter} = Filter::read_in('rbn', $call, 1)
 206 #               || Filter::read_in('rbn', 'node_default', 1);
 207
 208         Filter::load_dxchan($self, 'rbn', 1);
 209
 210         # clean up qra locators
 211         my $qra = $user->qra;
 212         $qra = undef if ($qra && !DXBearing::is_qra($qra));
 213         unless ($qra) {
 214                 my $lat = $user->lat;
 215                 my $long = $user->long;
 216                 $user->qra(DXBearing::lltoqra($lat, $long)) if (defined $lat && defined $long);
 217         }
 218
 219         # if we have been running and stopped for a while
 220         # if the cache is warm enough don't operate the inrush preventor
 221         $self->{inrushpreventor} = exists $runtime{$call} && $runtime{$call} > $startup_delay || $noinrush ?  0 : $main::systime + $startup_delay;
 222         dbg("RBN: noinrush: $noinrush, setting inrushpreventor on $self->{call} to $self->{inrushpreventor}");
 223 }
 224
 225 my @queue;                                              # the queue of spots ready to send
 226
 227 sub normal
 228 {
 229         my $self = shift;
 230         my $line = shift;
 231         my @ans;
 232         my $dbgrbn = isdbg('rbn');
 233
 234         # remove leading and trailing spaces
 235         chomp $line;
 236         $line =~ s/^\s*//;
 237         $line =~ s/\s*$//;
 238
 239         # add base RBN
 240
 241         my $now = $main::systime;
 242
 243         # parse line
 244         dbg "RBN:RAW,$line" if isdbg('rbnraw');
 245         return unless $line=~/^DX\s+de/;
 246
 247         my (undef, undef, $origin, $qrg, $call, $mode, $s, $m, $spd, $u, $sort, $t, $tx) = split /[:\s]+/, $line;
 248
 249         # fix up FT8 spots from 7001
 250         $t = $u, $u = '' if !$t && is_ztime($u);
 251         $t = $sort, $sort = '' if !$t && is_ztime($sort);
 252         my $qra = $spd, $spd = '' if is_qra($spd);
 253         $u = $qra if $qra;
 254
 255         # is this anything like a callsign?
 256         unless (is_callsign($call)) {
 257                 dbg("RBN: ERROR $call from $origin on $qrg is invalid, dumped");
 258                 return;
 259         }
 260
 261         # is it 'baddx'
 262         if ($DXProt::baddx->in($call)) {
 263                 dbg("RBN: Bad DX spot '$call', ignored");
 264                 dbg($line) if isdbg('nologchan');
 265                 return;
 266         }
 267
 268
 269         # remove all extraneous crap from the origin - just leave the base callsign
 270         my $norigin = basecall($origin);
 271         unless ($norigin) {
 272                 dbg("RBN: ERROR '$origin' is an invalid callsign, dumped");
 273                 return;
 274         }
 275         $origin = $norigin;
 276
 277         # is this callsign in badspotter list?
 278         if ($DXProt::badspotter->in($origin) || $DXProt::badnode->in($origin)) {
 279                 dbg("RBN: ERROR $origin is a bad spotter/node, dumped");
 280                 return;
 281         }
 282
 283         # is the qrg valid
 284         unless ($qrg =~ /^\d+\.\d{1,3}$/) {
 285                 dbg("RBN: ERROR qrg $qrg from $origin invalid, dumped");
 286                 return;
 287         }
 288
 289         $sort ||= '';
 290         $tx ||= '';
 291         $qra ||= '';
 292     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 && isdbg('rbn');
 293
 294         ++$self->{noraw};
 295         ++$self->{noraw10};
 296         ++$self->{norawhour};
 297
 298         my $b;
 299
 300         if ($t || $tx) {
 301
 302                 # fix up times for things like 'NXDXF B' etc
 303                 if ($tx && is_ztime($t)) {
 304                         if (is_ztime($tx)) {
 305                                 $b = $t;
 306                                 $t = $tx;
 307                         } else {
 308                                 dbg "RBN:ERR,$line";
 309                                 return (0);
 310                         }
 311                 }
 312                 if ($sort && $sort eq 'NCDXF') {
 313                         $mode = 'DXF';
 314                         $t = $tx;
 315                 }
 316                 if ($sort && $sort eq 'BEACON') {
 317                         $mode = 'BCN';
 318                 }
 319                 if ($mode =~ /^PSK/) {
 320                         $mode = 'PSK';
 321                 }
 322                 if ($mode eq 'RTTY') {
 323                         $mode = 'RTT';
 324                 }
 325
 326                 # The main de-duping key is [call, $frequency], but we probe a bit around that frequency to find a
 327                 # range of concurrent frequencies that might be in play.
 328
 329                 # The key to this is deducing the true callsign by "majority voting" (the greater the number of spotters
 330         # the more effective this is) together with some lexical analsys probably in conjuction with DXSpider
 331                 # data sources (for singleton spots) to then generate a "centre" from and to zone (whatever that will mean if it isn't the usual one)
 332                 # and some heuristical "Kwalitee" rating given distance from the zone centres of spotter, recipient user
 333         # and spotted. A map can be generated once per user and spotter as they are essentially mostly static.
 334                 # The spotted will only get a coarse position unless other info is available. Programs that parse
 335                 # DX bulletins and the online data online databases could be be used and then cached.
 336
 337                 # Obviously users have to opt in to receiving RBN spots and other users will simply be passed over and
 338                 # ignored.
 339
 340                 # Clearly this will only work in the 'mojo' branch of DXSpider where it is possible to pass off external
 341                 # data requests to ephemeral or semi resident forked processes that do any grunt work and the main
 342                 # process to just the standard "message passing" which has been shown to be able to sustain over 5000
 343                 # per second (limited by the test program's output and network speed, rather than DXSpider's handling).
 344
 345                 my $search = 5;
 346                 my $nqrg = nearest(1, $qrg * 10);  # normalised to nearest Khz
 347                 my $sp = "$call|$nqrg";           # hopefully the skimmers will be calibrated at least this well!
 348
 349                 # deal with the unix time
 350                 my ($hh,$mm) = $t =~ /(\d\d)(\d\d)Z$/;
 351                 my $utz = $hh*3600 + $mm*60 + $main::systime_daystart; # possible issue with late spot from previous day
 352                 $utz -= 86400 if $utz > $now+3600;                                         # too far ahead, drag it back one day
 353
 354                 #
 355                 # But before we do anything, if this call is in the seeme hash then just send the spot to them
 356                 #
 357                 if (exists $seeme{$call} && (my $ref = $seeme{$call})) {
 358                         foreach my $rcall ( @$ref) {
 359                                 my $uchan = DXChannel::get($rcall);
 360                                 if ($uchan) {
 361                                         if ($uchan->is_user) {
 362                                                 if (isdbg('seeme')) {
 363                                                         dbg("seeme: $line");
 364                                                         dbg( qq{seemme: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});
 365                                                 }
 366                                                 my @s =  Spot::prepare($qrg, $call, $utz, sprintf("%-3s %2ddB **SEEME**", $mode, $s), $origin.'-#');
 367                                                 my $buf = $uchan->format_dx_spot(@s);
 368                                                 dbg("seeme: result '$buf'") if isdbg('seeme');
 369                                                 $uchan->local_send('S', $buf);
 370                                         } else {
 371                                                 LogDbg('err',"RBN Someone is playing silly persons $rcall is not a user and cannot do 'seeme', ignored and reset");
 372                                                 del_seeme($rcall);
 373                                         }
 374                                 }
 375                         }
 376                 }
 377
 378                 # find it?
 379                 my $cand = $spots->{$sp};
 380                 unless ($cand) {
 381                         my ($i, $new);
 382                         for ($i = $nqrg; !$cand && $i <= $nqrg+$search; $i += 1) {
 383                                 $new = "$call|$i";
 384                                 $cand = $spots->{$new}, last if exists $spots->{$new};
 385                         }
 386                         if ($cand) {
 387                                 my $diff = $i - $nqrg;
 388                                 dbg(qq{RBN: QRG Diff using $new (+$diff) for $sp for qrg $qrg}) if (isdbg('rbnqrg') || ($dbgrbn && isdbg('rbn')));
 389                                 $sp = $new;
 390                         }
 391                 }
 392                 unless ($cand) {
 393                         my ($i, $new);
 394                         for ($i = $nqrg; !$cand && $i >= $nqrg-$search; $i -= 1) {
 395                                 $new = "$call|$i";
 396                                 $cand = $spots->{$new}, last if exists $spots->{$new};
 397                         }
 398                         if ($cand) {
 399                                 my $diff = $nqrg - $i;
 400                                 dbg(qq{RBN: QRG Diff using $new (-$diff) for $sp for qrg $qrg}) if (isdbg('rbnqrg') || ($dbgrbn && isdbg('rbn')));
 401                                 $sp = $new;
 402                         }
 403                 }
 404
 405                 # if we have one and there is only one slot and that slot's time isn't expired for respot then return
 406                 my $respot = 0;
 407                 if ($cand && ref $cand) {
 408                         if (@$cand <= CData) {
 409                                 if ($self->{respottime} > 0 && $now - $cand->[CTime] < $self->{respottime}) {
 410                                         dbg("RBN: key: '$sp' call: $call qrg: $qrg DUPE \@ ". atime(int $cand->[CTime])) if $dbgrbn && isdbg('rbn');
 411                                         return;
 412                                 }
 413
 414                                 dbg("RBN: key: '$sp' RESPOTTING call: $call qrg: $qrg last seen \@ ". atime(int $cand->[CTime])) if $dbgrbn && isdbg('rbn');
 415                                 $cand->[CTime] = $now;
 416                                 ++$respot;
 417                         }
 418
 419                         # otherwise we have a spot being built up at the moment
 420                 } elsif ($cand) {
 421                         dbg("RBN: key '$sp' = '$cand' not ref");
 422                         return;
 423                 } else {
 424                         # new spot / frequency
 425                         $spots->{$sp} = $cand = [$now, 0];
 426                         dbg("RBN: key: '$sp' call: $call qrg: $qrg NEW" . ($respot ? ' RESPOT' : '')) if $dbgrbn && isdbg('rbn');
 427                 }
 428
 429                 # add me to the display queue unless we are waiting for initial in rush to finish
 430                 return unless $noinrush || $self->{inrushpreventor} < $main::systime;
 431
 432                 # build up a new record and store it in the buildup
 433                 # create record and add into the buildup
 434                 my $r = [$origin, nearest(.1, $qrg), $call, $mode, $s, $t, $utz, $respot, $u];
 435                 my @s =  Spot::prepare($r->[RQrg], $r->[RCall], $r->[RUtz], '', $r->[ROrigin]);
 436                 if ($s[5] == 666) {
 437                         dbg("RBN: ERROR invalid prefix/callsign $call from $origin-# on $qrg, dumped");
 438                         return;
 439                 }
 440
 441                 if ($self->{inrbnfilter}) {
 442                         my ($want, undef) = $self->{inrbnfilter}->it($s);
 443                         return unless $want;
 444                 }
 445                 $r->[RSpotData] = \@s;
 446
 447                 ++$self->{queue}->{$sp};# unless @$cand>= CData; # queue the KEY (not the record)
 448
 449                 dbg("RBN: key: '$sp' ADD RECORD call: $call qrg: $qrg origin: $origin respot: $respot") if $dbgrbn && isdbg('rbn');
 450
 451                 push @$cand, $r;
 452
 453         } else {
 454                 dbg "RBN:DATA,$line" if $dbgrbn && isdbg('rbn');
 455         }
 456 }
 457
 458 # we should get the spot record minus the time, so just an array of record (arrays)
 459 sub send_dx_spot
 460 {
 461         my $self = shift;
 462         my $quality = shift;
 463         my $cand = shift;
 464
 465         ++$self->{norbn};
 466         ++$self->{norbn10};
 467         ++$self->{norbnhour};
 468
 469         # $r = [$origin, $qrg, $call, $mode, $s, $utz, $respot];
 470
 471         my $mode = $cand->[CData]->[RMode]; # as all the modes will be the same;
 472
 473         my @dxchan = DXChannel::get_all();
 474
 475         foreach my $dxchan (@dxchan) {
 476                 next unless $dxchan->is_user;
 477                 my $user = $dxchan->{user};
 478                 next unless $user &&  $user->wantrbn;
 479
 480                 # does this user want this sort of spot at all?
 481                 my $want = 0;
 482                 ++$want if $user->wantbeacon && $mode =~ /^BCN|DXF/;
 483                 ++$want if $user->wantcw && $mode =~ /^CW/;
 484                 ++$want if $user->wantrtty && $mode =~ /^RTT/;
 485                 ++$want if $user->wantpsk && $mode =~ /^PSK|FSK|MSK/;
 486                 ++$want if $user->wantft && $mode =~ /^FT/;
 487
 488                 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",
 489                                         $user->wantrbn,
 490                                         $user->wantft,
 491                                         $user->wantbeacon,
 492                                         $user->wantcw,
 493                                         $user->wantpsk,
 494                                         $user->wantrtty,
 495                                    )) if isdbg('rbnll');
 496
 497                 # send one spot to one user out of the ones that we have
 498                 $self->dx_spot($dxchan, $quality, $cand) if $want;
 499         }
 500 }
 501
 502 sub dx_spot
 503 {
 504         my $self = shift;
 505         my $dxchan = shift;
 506         my $quality = shift;
 507         my $cand = shift;
 508         my $call = $dxchan->{call};
 509         my $strength = 100;             # because it could if we talk about FTx
 510         my $saver;
 511         my %zone;
 512         my $respot;
 513         my $qra;
 514
 515         ++$self->{nousers}->{$call};
 516         ++$self->{nousers10}->{$call};
 517         ++$self->{nousershour}->{$call};
 518
 519         my $filtered;
 520         my $rf = $dxchan->{rbnfilter} || $dxchan->{spotsfilter};
 521         my $comment;
 522
 523         foreach my $r (@$cand) {
 524                 # $r = [$origin, $qrg, $call, $mode, $s, $t, $utz, $respot, $qra];
 525                 # Spot::prepare($qrg, $call, $utz, $comment, $origin);
 526                 next unless $r && ref $r;
 527
 528                 $qra = $r->[RQra] if !$qra && $r->[RQra] && is_qra($r->[RQra]);
 529
 530                 $comment = sprintf "%-3s %2ddB $quality", $r->[RMode], $r->[RStrength];
 531                 my $s = $r->[RSpotData];                # the prepared spot
 532                 $s->[SComment] = $comment;              # apply new generated comment
 533
 534                 ++$zone{$s->[SZone]};           # save the spotter's zone
 535
 536                 # save the lowest strength one
 537                 if ($r->[RStrength] < $strength) {
 538                         $strength = $r->[RStrength];
 539                         $saver = $s;
 540                         dbg("RBN: STRENGTH spot: $s->[SCall] qrg: $s->[SQrg] origin: $s->[SOrigin] dB: $r->[RStrength] < $strength") if isdbg 'rbnll';
 541                 }
 542
 543                 if ($rf) {
 544                         my ($want, undef) = $rf->it($s);
 545                         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';
 546                         next unless $want;
 547                         $filtered = $s;
 548                 }
 549         }
 550
 551         if ($rf) {
 552                 $saver = $filtered;             # if nothing passed the filter's lips then $saver == $filtered == undef !
 553         }
 554
 555         if ($saver) {
 556                 my $buf;
 557                 # create a zone list of spotters
 558                 delete $zone{$saver->[SZone]};  # remove this spotter's zone (leaving all the other zones)
 559                 my $z = join ',', sort {$a <=> $b} keys %zone;
 560
 561                 # alter spot data accordingly
 562                 $saver->[SComment] .= " Z:$z" if $z;
 563
 564                 send_final($dxchan, $saver);
 565
 566                 ++$self->{nospot};
 567                 ++$self->{nospot10};
 568                 ++$self->{nospothour};
 569
 570                 if ($qra) {
 571                         my $user = DXUser::get_current($saver->[SCall]) || DXUser->new($saver->[SCall]);
 572                         unless ($user->qra && is_qra($user->qra)) {
 573                                 $user->qra($qra);
 574                                 dbg("RBN: update qra on $saver->[SCall] to $qra");
 575                         }
 576                         # update lastseen if nothing else
 577                         $user->put;
 578                 }
 579         }
 580 }
 581
 582 sub send_final
 583 {
 584         my $dxchan = shift;
 585         my $saver = shift;
 586         my $call = $dxchan->{call};
 587         my $buf;
 588
 589         dbg("RBN: SENDING to $call spot: $saver->[SCall] qrg: $saver->[SQrg] origin: $saver->[SOrigin] $saver->[SComment]") if isdbg 'rbnll';
 590         if ($dxchan->{ve7cc}) {
 591                 my $call = $saver->[SOrigin];
 592                 $saver->[SOrigin] .= '-#';
 593                 $buf = VE7CC::dx_spot($dxchan, @$saver);
 594                 $saver->[SOrigin] = $call;
 595         } else {
 596                 my $call = $saver->[SOrigin];
 597                 $saver->[SOrigin] = substr($call, 0, 6);
 598                 $saver->[SOrigin] .= '-#';
 599                 $buf = $dxchan->format_dx_spot(@$saver);
 600                 $saver->[SOrigin] = $call;
 601         }
 602         $dxchan->local_send('R', $buf);
 603 }
 604
 605 # per second
 606 sub process
 607 {
 608         my $rbnskim = isdbg('rbnskim');
 609
 610         foreach my $dxchan (DXChannel::get_all()) {
 611                 next unless $dxchan->is_rbn;
 612
 613                 # At this point we run the queue to see if anything can be sent onwards to the punter
 614                 my $now = $main::systime;
 615                 my $ta = [gettimeofday];
 616                 my $items = 0;
 617
 618                 # now run the waiting queue which just contains KEYS ($call|$qrg)
 619                 foreach my $sp (keys %{$dxchan->{queue}}) {
 620                         my $cand = $spots->{$sp};
 621                         ++$items;
 622
 623                         unless ($cand && $cand->[CTime]) {
 624                                 dbg "RBN Cand $sp " . ($cand ? 'def' : 'undef') . " [CTime] " . ($cand->[CTime] ? 'def' : 'undef') . " dwell $dwelltime";
 625                                 delete $spots->{$sp};
 626                                 delete $dxchan->{queue}->{$sp};    # remove
 627                                 next;
 628                         }
 629
 630                         my $ctime = $cand->[CTime];
 631                         my $quality = @$cand - CData;
 632                         my $dwellsecs =  $now - $ctime;
 633                         if ($quality >= $maxqual || $dwellsecs >= $dwelltime || $dwellsecs >= $limbotime) {
 634                                 # we have a candidate, create qualitee value(s);
 635                                 unless (@$cand > CData) {
 636                                         dbg "RBN: QUEUE key '$sp' MISSING RECORDS, IGNORED" . dd($cand) if isdbg 'rbnqueue';
 637                                         delete $spots->{$sp}; # don't remember it either - this means that a spot HAS to come in with sufficient spotters to be processed.
 638                                         delete $dxchan->{queue}->{$sp};
 639                                         next;
 640                                 }
 641                                 dbg "RBN: QUEUE PROCESSING key: '$sp' $now >= $cand->[CTime]" if isdbg 'rbnqueue';
 642                                 my $spotters = $quality;
 643
 644                                 # dump it and remove it from the queue if it is of unadequate quality, but only if it is no longer in Limbo and can be reasonably passed on to its demise
 645                                 my $r = $cand->[CData];
 646                                 if ($dwellsecs > $limbotime && $quality < $minqual) {
 647                                         if ( $rbnskim && isdbg('rbnskim')) {
 648                                                 $r = $cand->[CData];
 649                                                 if ($r) {
 650                                                         my $lastin = difft($ctime, $now, 2);
 651                                                         my $s = "RBN:SKIM time in Limbo exceeded DUMPED (lastin: $lastin Q:$quality < Q:$minqual) key: '$sp' = $r->[RCall] on $r->[RQrg] by $r->[ROrigin] \@ $r->[RTime] route: $dxchan->{call}";
 652                                                         dbg($s);
 653                                                 }
 654                                         }
 655                                         delete $spots->{$sp}; # don't remember it either - this means that a spot HAS to come in with sufficient spotters to be processed.
 656                                         delete $dxchan->{queue}->{$sp};
 657                                         next;
 658                                 }
 659
 660                                 # we have a possible removal from Limbo, check for more than one skimmer and reset the quality if required
 661                                 # DOES THIS TEST CAUSE RACES?
 662                                 if (!$r->[Respot] && $quality >= $minqual && $dwellsecs > $dwelltime+1) {
 663
 664                                         # because we don't need to check for repeats by the same skimmer in the normal case, we do here
 665                                         my %seen;
 666                                         my @origin;
 667                                         foreach my $wr (@$cand) {
 668                                                 next unless ref $wr;
 669                                                 push @origin, $wr->[ROrigin];
 670                                                 if (exists $seen{$wr->[ROrigin]}) {
 671                                                         next;
 672                                                 }
 673                                                 $seen{$wr->[ROrigin]} = $wr;
 674                                         }
 675                                         # reset the quality to ignore dupes
 676                                         my $oq = $quality;
 677                                         $quality = keys %seen;
 678                                         if ($quality >= $minqual) {
 679                                                 if ( $rbnskim && isdbg('rbnskim')) {
 680                                                         my $lastin = difft($ctime, $now, 2);
 681                                                         my $sk = join ' ', keys %seen;
 682                                                         my $or = join ' ', @origin;
 683                                                         my $s = "RBN:SKIM promoted from Limbo - key: '$sp' (lastin: $lastin Q now: $quality was $oq skimmers now: $sk";
 684                                                         $s .= " was $or" if $or ne $sk;
 685                                                         $s .= ')';
 686                                                         dbg($s);
 687                                                 }
 688                                         } elsif ($oq != $quality) {
 689                                                 if ( $rbnskim && isdbg('rbnskim')) {
 690                                                         my $lastin = difft($ctime, $now, 2);
 691                                                         my $sk = join ' ', keys %seen;
 692                                                         my $or = join ' ', @origin;
 693                                                         my $s = "RBN:SKIM quality reset key: '$sp' (lastin: $lastin Q now: $quality was $oq skimmers now: $sk was: $or)";
 694                                                         dbg($s);
 695                                                 }
 696                                                 # remove the excess
 697                                                 my @ncand = (@$cand[CTime, CQual], values %seen);
 698                                                 $spots->{$sp} = \@ncand;
 699                                         }
 700                                 }
 701
 702                                 # we now kick this spot into Limbo
 703                                 if ($quality < $minqual) {
 704                                         next;
 705                                 }
 706
 707                                 $quality = 9 if $quality > 9;
 708                                 $cand->[CQual] = $quality if $quality > $cand->[CQual];
 709
 710                                 # this scores each candidate according to its skimmer's QRG score (i.e. how often it agrees with its peers)
 711                                 # what happens is hash of all QRGs in candidates are incremented by that skimmer's reputation for "accuracy"
 712                                 # or, more exactly, past agreement with the consensus. This score can be from -5 -> +5.
 713                                 my %qrg = ();
 714                                 my $skimmer;
 715                                 my $sk;
 716                                 my $band;
 717                                 my %seen = ();
 718                                 foreach $r (@$cand) {
 719                                         next unless ref $r;
 720                                         if (exists $seen{$r->[ROrigin]}) {
 721                                                 $r = 0;
 722                                                 next;
 723                                         }
 724                                         $seen{$r->[ROrigin]} = 1;
 725                                         $band ||= int $r->[RQrg] / 1000;
 726                                         $sk = "SKIM|$r->[ROrigin]|$band"; # thus only once per set of candidates
 727                                         $skimmer = $spots->{$sk};
 728                                         unless ($skimmer) {
 729                                                 $skimmer = $spots->{$sk} = [1, 0, 0, $now, []]; # this first time, this new skimmer gets the benefit of the doubt on frequency.
 730                                                 dbg("RBN:SKIM new slot $sk " . $json->encode($skimmer)) if  $rbnskim && isdbg('rbnskim');
 731                                         }
 732                                         $qrg{$r->[RQrg]} += ($skimmer->[DScore] || 1);
 733                                 }
 734
 735                                 # determine the most likely qrg and then set it - NOTE (-)ve votes, generated by the skimmer scoring system above, are ignored
 736                                 my @deviant;
 737                                 my $c = 0;
 738                                 my $mv = 0;
 739                                 my $qrg = 0;
 740                                 while (my ($k, $votes) = each %qrg) {
 741                                         if ($votes >= $mv) {
 742                                                 $qrg = $k;
 743                                                 $mv = $votes;
 744                                         }
 745                                         ++$c;
 746                                 }
 747
 748                                 # Ignore possible spots with 0 QRG score - as determined by the skimmer scoring system above -  as they are likely to be wrong
 749                                 unless ($qrg > 0) {
 750                                         if ( $rbnskim && isdbg('rbnskim')) {
 751                                                 my $keys;
 752                                                 while (my ($k, $v) = (each %qrg)) {
 753                                                         $keys .= "$k=>$v, ";
 754                                                 }
 755                                                 $keys =~ /,\s*$/;
 756                                                 my $i = 0;
 757                                                 foreach $r (@$cand) {
 758                                                         next unless $r && ref $r;
 759                                                         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";
 760                                                         ++$i;
 761                                                 }
 762                                         }
 763                                         delete $spots->{$sp}; # get rid
 764                                         delete $dxchan->{queue}->{$sp};
 765                                         next;
 766                                 }
 767
 768                                 # detemine and spit out the deviants. Then adjust the scores according to whether it is a deviant or good
 769                                 # NOTE: deviant nodes can become good (or less bad), and good nodes bad (or less good) on each spot that
 770                                 # they generate. This is based solely on each skimmer's agreement (or not) with the "consensus" score generated
 771                                 # above ($qrg). The resultant score + good + bad is stored per band and will be used the next time a spot
 772                                 # appears on this band from each skimmer.
 773                                 foreach $r (@$cand) {
 774                                         next unless $r && ref $r;
 775                                         my $diff = $c > 1 ? nearest(.1, $r->[RQrg] - $qrg) : 0;
 776                                         $sk = "SKIM|$r->[ROrigin]|$band";
 777                                         $skimmer = $spots->{$sk};
 778                                         if ($diff) {
 779                                                 ++$skimmer->[DBad] if $skimmer->[DBad] < $maxdeviants;
 780                                                 --$skimmer->[DGood] if $skimmer->[DGood] > 0;
 781                                                 push @deviant, sprintf("$r->[ROrigin]:%+.1f", $diff);
 782                                                 push @{$skimmer->[DEviants]}, $diff;
 783                                                 shift @{$skimmer->[DEviants]} while @{$skimmer->[DEviants]} > $maxdeviants;
 784                                         } else {
 785                                                 ++$skimmer->[DGood] if $skimmer->[DGood] < $maxdeviants;
 786                                                 --$skimmer->[DBad] if $skimmer->[DBad] > 0;
 787                                                 shift @{$skimmer->[DEviants]};
 788                                         }
 789                                         $skimmer->[DScore] = $skimmer->[DGood] - $skimmer->[DBad];
 790                                         if ($rbnskim && isdbg('rbnskim')) {
 791                                                 my $lastin = difft($skimmer->[DLastin], $now, 2);
 792                                                 my $difflist = join(', ', @{$skimmer->[DEviants]});
 793                                                 $difflist = " band qrg diffs: $difflist" if $difflist;
 794                                                 dbg("RBN:SKIM key $sp slot $sk $r->[RQrg] - $qrg = $diff Skimmer score: $skimmer->[DGood] - $skimmer->[DBad] = $skimmer->[DScore] lastseen:$lastin ago$difflist");
 795                                         }
 796                                         $skimmer->[DLastin] = $now;
 797                                         $r->[RSpotData]->[SQrg] = $qrg if $qrg && $c > 1; # set all the QRGs to the agreed value
 798                                 }
 799
 800                                 $qrg = (sprintf "%.1f",  $qrg)+0;
 801                                 $r = $cand->[CData];
 802                                 $r->[RQrg] = $qrg;
 803                                 my $squality = "Q:$cand->[CQual]";
 804                                 $squality .= '*' if $c > 1;
 805                                 $squality .= '+' if $r->[Respot];
 806
 807                                 if (isdbg('progress')) {
 808                                         my $rt = difft($ctime, $now, 2);
 809                                         my $s = "RBN: SPOT key: '$sp' = $r->[RCall] on $r->[RQrg] by $r->[ROrigin] \@ $r->[RTime] $squality route: $dxchan->{call} dwell:$rt";
 810                                         my $td = @deviant;
 811                                         $s .= " QRGScore: $mv Deviants: $td/$spotters";
 812                                         $s .= ' (' . join(', ', sort @deviant) . ')' if $td;
 813                                         dbg($s);
 814                                 }
 815
 816                                 # finally send it out to any waiting public
 817                                 send_dx_spot($dxchan, $squality, $cand);
 818
 819                                 # clear out the data and make this now just "spotted", but no further action required until respot time
 820                                 dbg "RBN: QUEUE key '$sp' cleared" if isdbg 'rbn';
 821
 822                                 delete $dxchan->{queue}->{$sp};
 823
 824                                 # calculate new sp (which will be 70% likely the same as the old one)
 825                                 # 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.
 826                                 # and we want to store the key that corresponds to majority opinion.
 827                                 my $nqrg = nearest(1, $qrg * 10);  # normalised to nearest Khz
 828                                 my $nsp = "$r->[RCall]|$nqrg";
 829                                 if ($sp ne $nsp) {
 830                                         dbg("RBN:SKIM CHANGE KEY sp '$sp' -> '$nsp' for storage") if  $rbnskim && isdbg('rbnskim');
 831                                         delete $spots->{$sp};
 832                                         $spots->{$nsp} = [$now, $cand->[CQual]];
 833                                 } else {
 834                                         $spots->{$sp} = [$now, $cand->[CQual]];
 835                                 }
 836                         }
 837                         else {
 838                                 dbg sprintf("RBN: QUEUE key: '$sp' SEND time not yet reached %.1f secs left", $cand->[CTime] + $dwelltime - $now) if isdbg 'rbnqueue';
 839                         }
 840                 }
 841                 if (isdbg('rbntimer')) {
 842                         my $diff = _diffus($ta);
 843                         dbg "RBN: TIMER process queue for call: $dxchan->{call} $items spots $diff uS";
 844                 }
 845         }
 846 }
 847
 848 sub per_minute
 849 {
 850         foreach my $dxchan (DXChannel::get_all()) {
 851                 next unless $dxchan->is_rbn;
 852                 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');
 853                 if ($dxchan->{noraw} == 0 && $dxchan->{lasttime} > 60) {
 854                         LogDbg('err', "RBN: no input from $dxchan->{call}, disconnecting");
 855                         $dxchan->disconnect;
 856                 }
 857                 $dxchan->{noraw} = $dxchan->{norbn} = $dxchan->{nospot} = 0; $dxchan->{nousers} = {};
 858                 $runtime{$dxchan->{call}} += 60;
 859         }
 860
 861         # save the spot cache
 862         write_cache() unless $main::systime + $startup_delay < $main::systime;;
 863 }
 864
 865 sub per_10_minute
 866 {
 867         my $count = 0;
 868         my $removed = 0;
 869         while (my ($k,$cand) = each %{$spots}) {
 870                 next if $k eq 'VERSION';
 871                 next if $k =~ /^O\|/;
 872                 next if $k =~ /^SKIM\|/;
 873
 874                 if ($main::systime - $cand->[CTime] > $cachetime) {
 875                         delete $spots->{$k};
 876                         ++$removed;
 877                 }
 878                 else {
 879                         ++$count;
 880                 }
 881         }
 882         dbg "RBN:STATS spot cache remain: $count removed: $removed"; # if isdbg('rbn');
 883         foreach my $dxchan (DXChannel::get_all()) {
 884                 next unless $dxchan->is_rbn;
 885                 my $nq = keys %{$dxchan->{queue}};
 886                 my $pc = $dxchan->{noraw10} ? sprintf("%.1f%%",$dxchan->{norbn10}*100/$dxchan->{noraw10}) : '0.0%';
 887                 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}};
 888                 $dxchan->{noraw10} = $dxchan->{norbn10} = $dxchan->{nospot10} = 0; $dxchan->{nousers10} = {};
 889         }
 890 }
 891
 892 sub per_hour
 893 {
 894         foreach my $dxchan (DXChannel::get_all()) {
 895                 next unless $dxchan->is_rbn;
 896                 my $nq = keys %{$dxchan->{queue}};
 897                 my $pc = $dxchan->{norawhour} ? sprintf("%.1f%%",$dxchan->{norbnhour}*100/$dxchan->{norawhour}) : '0.0%';
 898                 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}};
 899                 $dxchan->{norawhour} = $dxchan->{norbnhour} = $dxchan->{nospothour} = 0; $dxchan->{nousershour} = {};
 900         }
 901 }
 902
 903 sub finish
 904 {
 905         write_cache();
 906 }
 907
 908 sub write_cache
 909 {
 910         return unless $json;
 911
 912         my $ta = [ gettimeofday ];
 913
 914         $json->indent(1)->canonical(1) if isdbg 'rbncache';
 915         my $s = eval {$json->encode($spots)};
 916         if ($s) {
 917                 my $fh = IO::File->new(">$cachefn") or confess("writing $cachefn $!");
 918                 $fh->print($s);
 919                 $fh->close;
 920         } else {
 921                 dbg("RBN:Write_cache error '$@'");
 922                 return;
 923         }
 924         $json->indent(0)->canonical(0);
 925         my $diff = _diffms($ta);
 926         my $size = sprintf('%.3fKB', (length($s) / 1000));
 927         dbg("RBN:WRITE_CACHE size: $size time to write: $diff mS");
 928 }
 929
 930 sub check_cache
 931 {
 932         if (-e $cachefn) {
 933                 my $mt = (stat($cachefn))[9];
 934                 my $t = $main::systime - $mt || 1;
 935                 my $p = difft($mt, 2);
 936                 if ($t < $cache_valid) {
 937                         dbg("RBN:check_cache '$cachefn' spot cache exists, created $p ago and not too old");
 938                         my $fh = IO::File->new($cachefn);
 939                         my $s;
 940                         if ($fh) {
 941                                 local $/ = undef;
 942                                 $s = <$fh>;
 943                                 dbg("RBN:check_cache cache read size " . length $s);
 944                                 $fh->close;
 945                         } else {
 946                                 dbg("RBN:check_cache file read error $!");
 947                                 return undef;
 948                         }
 949                         if ($s) {
 950                                 eval {$spots = $json->decode($s)};
 951                                 if ($spots && ref $spots) {
 952                                         if (exists $spots->{VERSION} && $spots->{VERSION} == $DATA_VERSION) {
 953                                                 # now clean out anything that has spot build ups in progress
 954                                                 while (my ($k, $cand) = each %$spots) {
 955                                                         next if $k eq 'VERSION';
 956                                                         next if $k =~ /^O\|/;
 957                                                         next if $k =~ /^SKIM\|/;
 958                                                         if (@$cand > CData) {
 959                                                                 $spots->{$k} = [$cand->[CTime], $cand->[CQual]];
 960                                                         }
 961                                                 }
 962                                                 dbg("RBN:check_cache spot cache restored");
 963                                                 return 1;
 964                                         }
 965                                 }
 966                                 dbg("RBN::checkcache error decoding $@");
 967                         }
 968                 } else {
 969                         my $d = difft($main::systime-$cache_valid);
 970                         dbg("RBN::checkcache '$cachefn' created $p ago is too old (> $d), ignored");
 971                 }
 972         } else {
 973                 dbg("RBN:check_cache '$cachefn' spot cache not present");
 974         }
 975
 976         return undef;
 977 }
 978
 979 sub add_seeme
 980 {
 981         my $call = shift;
 982         my $base = basecall($call);
 983         my $ref = $seeme{$base} || [];
 984         push @$ref, $call unless grep $_ eq $call, @$ref;
 985         $seeme{$base} = $ref;
 986 }
 987
 988 sub del_seeme
 989 {
 990         my $call = shift;
 991         my $base = basecall($call);
 992         my $ref = $seeme{$base};
 993         return unless $ref && @$ref;
 994
 995         @$ref =  grep {$_ ne $call} @$ref;
 996         if (@$ref) {
 997                 $seeme{$base} = $ref;
 998         } else {
 999                 delete $seeme{basecall($call)};
1000         }
1001 }
1002 1;