popufare/busunit/paddlemgr/paddlemgr.c

/*
 * Copyright (c) 2019 Clementine Computing LLC.
 *
 * This file is part of PopuFare.
 *
 * PopuFare is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * PopuFare is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with PopuFare.  If not, see <https://www.gnu.org/licenses/>.
 *
 */

#include <sys/socket.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/un.h>

#include <netinet/in.h>
#include <arpa/inet.h>

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <poll.h>
#include <time.h>

#include <zlib.h>

#include "../common/common_defs.h"
#include "../commhub/commhub.h"
#include "../commhub/client_utils.h"
#include "../common/gpsmath.h"

// structure for loading stop definitions
//
typedef struct stop_struct {

  // scheduled arrival time
  //
  int hour;
  int min;

  // coordinates
  //
  double lat;
  double lon;

  int route;
  int trip;
  int stop;

  // human readable stop name
  //
  char name[STOP_NAME_LEN];

} stop;


int  commhub_fd = -1;

// Currently selected paddle
//
int   current_paddle_num = 0;

// Number of stops on said paddle
//
int   current_paddle_len = 0;

// Index of the active stop on this paddle
//
int   current_paddle_idx = 0;

// Data block to hold loaded paddle
//
stop   current_paddle[MAX_PADDLE_SIZE] = {{0}};


int load_paddle(int paddlenum) {
  char buffer[LINE_BUFFER_SIZE];
  char buffer2[LINE_BUFFER_SIZE];
  FILE *f;

  int i, eol;
  int n;

  sprintf(buffer, "%s%d.paddle", CONFIG_FILE_PATH, paddlenum);

  f = fopen(buffer, "rb");

  if (!f) {
    printf("Paddle not found: %s\n", buffer);
    return -1;
  }

  current_paddle_num = paddlenum;
  current_paddle_idx = 0;
  current_paddle_len = 0;
  n = 0;

  // For each line in the input file
  //
  while( fgets(buffer, LINE_BUFFER_SIZE, f) ) {

    if(current_paddle_idx >= MAX_PADDLE_SIZE) {
      fprintf(stderr, "Paddle %d has overflowed its maximum size of %d stops and has been truncated!\n", current_paddle_num, MAX_PADDLE_SIZE);
      break;
    }

    strip_crlf(buffer);  //get rid of any trailing CR/LF characters

    n++;

    i = eol = 0;

    i += get_field(buffer2, buffer + i, LINE_BUFFER_SIZE, &eol);

    // Skip any blank or comment lines
    //
    if(eol || buffer2[0] == '#') { continue; }

    // Clear this row
    //
    memset(current_paddle + current_paddle_len, 0, sizeof(stop));

    current_paddle[current_paddle_len].hour = strtol(buffer2, NULL, 10);

    i += get_field(buffer2, buffer + i, LINE_BUFFER_SIZE, &eol);
    if(eol) { printf("Line %d too short in paddle %d: \"%s\"\n", n, paddlenum, buffer); continue; }
    current_paddle[current_paddle_len].min = strtol(buffer2, NULL, 10);

    i += get_field(buffer2, buffer + i, LINE_BUFFER_SIZE, &eol);
    if(eol) { printf("Line %d too short in paddle %d: \"%s\"\n", n, paddlenum, buffer); continue; }
    current_paddle[current_paddle_len].lat = strtod(buffer2, NULL);

    i += get_field(buffer2, buffer + i, LINE_BUFFER_SIZE, &eol);
    if(eol) { printf("Line %d too short in paddle %d: \"%s\"\n", n, paddlenum, buffer); continue; }
    current_paddle[current_paddle_len].lon = strtod(buffer2, NULL);

    i += get_field(buffer2, buffer + i, LINE_BUFFER_SIZE, &eol);
    if(eol) { printf("Line %d too short in paddle %d: \"%s\"\n", n, paddlenum, buffer); continue; }
    current_paddle[current_paddle_len].route = strtol(buffer2, NULL, 10);

    i += get_field(buffer2, buffer + i, LINE_BUFFER_SIZE, &eol);
    if(eol) { printf("Line %d too short in paddle %d: \"%s\"\n", n, paddlenum, buffer); continue; }
    current_paddle[current_paddle_len].trip = strtol(buffer2, NULL, 10);

    i += get_field(buffer2, buffer + i, LINE_BUFFER_SIZE, &eol);
    if(eol) { printf("Line %d too short in paddle %d: \"%s\"\n", n, paddlenum, buffer); continue; }
    current_paddle[current_paddle_len].stop = strtol(buffer2, NULL, 10);

    i += get_field(buffer2, buffer + i, LINE_BUFFER_SIZE, &eol);
    strncpy(current_paddle[current_paddle_len].name, buffer2, STOP_NAME_LEN - 1);

    current_paddle_len++;
  }
  fclose(f);

  return paddlenum;
}

int clear_paddle() {

  current_paddle_num = 0;
  current_paddle_idx = 0;
  current_paddle_len = 0;

  return 0;
}

int where_the_hell_are_we() {
  int i;

  // by default, we will report that we are still at our last known stop
  //
  int found_stop = current_paddle_idx;

  struct tm temp;
  time_t now;
  time_t sched;

  now = time(NULL);

  if (current_paddle_num == 0) {
    return -1;
  }

  #ifdef ROLLOVER_FORWARD_ONLY
    // If we are in rollover-forward mode, we will not actively re-select any previous stops, nor the current one.
    // This allows two stops to have the same location but different route numbers for places where the route changes at a stop.
    //
    i = current_paddle_idx;
  #else
    i = 0;
  #endif

  for( ; i < current_paddle_len; i++) {

    // populate our time structure based on now
    // so the date will be correct
    //
    localtime_r(&now,&temp);

    // Set the expected arrival time
    //
    temp.tm_hour = current_paddle[i].hour;

    // Set the expected arrival time
    //
    temp.tm_min = current_paddle[i].min;

    // and convert it back to a scheduled arrival time in UTC unix timestamp format
    //
    sched = mktime( &temp );

    // First we do the time check, because that's cheap integer math
    // then GPS distance last because that's expensive trig
    //
    if( (abs(now - sched) <= ROLLOVER_TIME_WINDOW) &&
        (GPS_Dist(gps_stat.lat, gps_stat.lon, current_paddle[i].lat, current_paddle[i].lon) <= ROLLOVER_DISTANCE)) {

      // update our found_stop index to the matching stop
      //
      found_stop = i;

      #ifndef ROLLOVER_TO_FURTHEST_STOP

        // if ROLLOVER_TO_FURTHEST_STOP is NOT defined, we break as soon as
        // we've found ANY matching stop (even the one we started at)
        //
        break;

      #endif
    }
  }

  return found_stop;
}

int send_status_update() {
  struct message_record outgoing_msg;
  stop_status current = {0};

  if(commhub_fd < 0) {
    return -1;
  }

  current.paddle = current_paddle_num;

  if(current_paddle_num) {
    current.route = current_paddle[current_paddle_idx].route;
    current.trip = current_paddle[current_paddle_idx].trip;
    current.stop = current_paddle[current_paddle_idx].stop;

    current.lat = current_paddle[current_paddle_idx].lat;
    current.lon = current_paddle[current_paddle_idx].lon;

    strncpy(current.stopname, current_paddle[current_paddle_idx].name, STOP_NAME_LEN - 1);
  }

  prepare_message(&outgoing_msg, MAILBOX_STOP_STATUS, &current, sizeof(current));
  return send_message(commhub_fd, &outgoing_msg);
}

// This function sends an update to the driver and to the diagnostic log saying we're doing a stop rollover
//
int send_driver_update() {
  struct message_record outgoing_msg;

  if(commhub_fd < 0) { return -1; }

  if(current_paddle_num) {
    format_log_message(&outgoing_msg,
                       LOGLEVEL_DEBUG,
                       "%02d:%02d %s",
                       current_paddle[current_paddle_idx].hour,
                       current_paddle[current_paddle_idx].min,
                       current_paddle[current_paddle_idx].name);
    send_message(commhub_fd, &outgoing_msg);

    format_driver_message(&outgoing_msg,
                          LOGLEVEL_EVENT,
                          "%02d:%02d %s",
                          current_paddle[current_paddle_idx].hour,
                          current_paddle[current_paddle_idx].min,
                          current_paddle[current_paddle_idx].name);
    return send_message(commhub_fd, &outgoing_msg);
  }

  return -1;
}

int send_vault_drop() {
  struct message_record outgoing_msg;

  if(commhub_fd < 0) { return -1; }

  prepare_message(&outgoing_msg, MAILBOX_VAULT_DROP, "", 0);
  return send_message(commhub_fd, &outgoing_msg);
}

message_callback_return handle_status_req(struct message_record *msg, void *param) {
  send_status_update();
  return MESSAGE_HANDLED_CONT;
}

message_callback_return handle_gps_update(struct message_record *msg, void *param) {
  int tempidx;

  //The system callback will have already handled this message and put it into the correct data structure
  //

  //If either we have no current paddle, or no real GPS data, ignore this message
  //
  if( (!current_paddle_num) || (!gps_stat.gps_good) ) {
    return MESSAGE_HANDLED_CONT;
  }

  tempidx = where_the_hell_are_we();

  if(tempidx < 0) {
    return MESSAGE_HANDLED_CONT;
  }

  if(tempidx != current_paddle_idx) {
    current_paddle_idx = tempidx;
    send_status_update();
    send_driver_update();
    send_vault_drop();
  }

  return MESSAGE_HANDLED_CONT;
}

message_callback_return handle_next_req(struct message_record *msg, void *param) {

  if(current_paddle_idx < (current_paddle_len - 1)) {
    current_paddle_idx++;
    send_driver_update();
  }

  send_status_update();
  return MESSAGE_HANDLED_CONT;
}

message_callback_return handle_prev_req(struct message_record *msg, void *param) {

  if(current_paddle_idx > 0) {
    current_paddle_idx--;
    send_driver_update();
  }

  send_status_update();
  return MESSAGE_HANDLED_CONT;
}

message_callback_return handle_set_paddle_req(struct message_record *msg, void *param) {

  struct message_record outgoing_msg;
  set_paddle_req *req = (set_paddle_req *)msg->payload;

  clear_paddle();
  req->result = load_paddle(req->request);

  if(req->result > 0) {
    send_driver_update();
  }

  prepare_message(&outgoing_msg, MAILBOX_PADDLE_ACK, req, sizeof(set_paddle_req));
  send_message(commhub_fd, &outgoing_msg);

  send_status_update();
  return MESSAGE_HANDLED_CONT;
}


void maintain_ipc_hub_connect(char *progname) {
  struct message_record outgoing_msg;

  // if we have no connection to the communication hub
  //
  if(commhub_fd < 0) {

    // try and get one
    //
    commhub_fd = connect_to_message_server(progname);

    // if it worked
    //
    if(commhub_fd >= 0) {

      // Subscribe to the command mailboxes we act on
      //
      prepare_message(&outgoing_msg, MAILBOX_SUBSCRIBE, MAILBOX_SET_PADDLE, strlen(MAILBOX_SET_PADDLE));
      send_message(commhub_fd,&outgoing_msg);

      prepare_message(&outgoing_msg, MAILBOX_SUBSCRIBE, MAILBOX_NEXT_STOP, strlen(MAILBOX_NEXT_STOP));
      send_message(commhub_fd,&outgoing_msg);

      prepare_message(&outgoing_msg, MAILBOX_SUBSCRIBE, MAILBOX_PREV_STOP, strlen(MAILBOX_PREV_STOP));
      send_message(commhub_fd,&outgoing_msg);

      // Subscribe to the relevant status management mailboxes
      //
      subscribe_to_default_messages(commhub_fd);

      // Request updated status information...
      //
      prepare_message(&outgoing_msg, MAILBOX_STATUS_REQUEST, "", 0);    
      send_message(commhub_fd,&outgoing_msg);
    }
  }
}


int main(int argc, char **argv) {
  struct pollfd fds[2];
  int nfds = 0;
  int poll_return = 0;
  int read_return = 0;

  int i;

  struct message_record incoming_msg;

  long long int _usec_now, _usec_prv, _usec_del;
  _usec_now = get_usec_time();
  _usec_prv = _usec_now;
  _usec_del = 60000000;

  configure_signal_handlers(argv[0]);
  maintain_ipc_hub_connect(argv[0]);

  // Register our default keep-up-with-system status callbacks
  //
  register_system_status_callbacks();

  // Add our module-specific callbacks
  //
  register_dispatch_callback(MAILBOX_GPS_STATUS, CALLBACK_USER(1), handle_gps_update, NULL);
  register_dispatch_callback(MAILBOX_STATUS_REQUEST, CALLBACK_USER(2), handle_status_req, NULL);
  register_dispatch_callback(MAILBOX_SET_PADDLE, CALLBACK_USER(3), handle_set_paddle_req, NULL);
  register_dispatch_callback(MAILBOX_NEXT_STOP, CALLBACK_USER(4), handle_next_req, NULL);
  register_dispatch_callback(MAILBOX_PREV_STOP, CALLBACK_USER(5), handle_prev_req, NULL);

  while( exit_request_status == EXIT_REQUEST_NONE ) {

    RESET_WATCHDOG();

    //DEBUG
    _usec_now = get_usec_time();
    if ((_usec_now - _usec_prv) > _usec_del) {
      printf("[%lli] paddlemgr: heartbeat\n", get_usec_time());
      fflush(stdout);
      _usec_prv = _usec_now;
    }
    //DEBUG

    maintain_ipc_hub_connect(argv[0]);

    nfds=0;

    if(commhub_fd >= 0) {
      fds[nfds].fd = commhub_fd;
      fds[nfds].events = POLLIN;
      nfds++;
    }

    if(nfds > 0) {
      poll_return = poll(fds, nfds, POLL_TIMEOUT);
    }
    else {
      usleep(POLL_TIMEOUT * 1000);
      poll_return = 0;
    }

    if(poll_return <= 0) {
      continue;
    }

    //----  If we got a message

    for(i=0; i < nfds; i++) {

      if( fds[i].fd == commhub_fd ) {  

        //If we've lost connection, break this loop and poll all over again
        //
        if(fds[i].revents & (POLLERR | POLLHUP | POLLNVAL))  {
          close(commhub_fd);
          commhub_fd = -1;
          break;
        }

        if(fds[i].revents & POLLIN) {

          read_return = get_message(commhub_fd, &incoming_msg);

          if( read_return < 0 ) {
            close(commhub_fd);
            commhub_fd = -1;
            break;
          }

          // This passes the received message through the callback list
          //
          process_message(&incoming_msg);

        }

      }

    }

  }

  if(commhub_fd >= 0) {
    close(commhub_fd);
  }

  printf("Goodbye.\n");

  return 0;
}