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