it2s-itss-facilities/src/config.c

#include "config.h"
#include "facilities.h"

#include <syslog.h>
#include <string.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <it2s-config.h>
#include <dirent.h>
#include <camv2/ProtectedCommunicationZone.h>
#include <zmq.h>

#include <itss-management/ManagementRequest.h>
#include <itss-management/ManagementReply.h>

#define syslog_info(msg, ...) syslog(LOG_INFO, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__)
#define syslog_emerg(msg, ...) syslog(LOG_EMERG, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__)
#define syslog_err(msg, ...) syslog(LOG_ERR, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__)

#ifndef NDEBUG
#define syslog_debug(msg, ...) syslog(LOG_DEBUG, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__)
#else
#define syslog_debug(msg, ...)
#endif

static int fetch_target_address(char** addresses, uint16_t addresses_len) {
    int index = -1;

    for (int i = 0; i < addresses_len; ++i) {
        char* addr = addresses[i];

        // If TCP, ignore receiver addresses, e.g. tcp://*:[port]
        if (!memcmp(addr, "tcp", 3)) {
            bool found_astk = false;
            for (int j = 0; j < strlen(addr); ++j) {
                if (addr[j] == '*') {
                    found_astk = true;
                    break;
                }
            }

            if (found_astk) continue;
        }

        index = i;
        break;
    }

    return index;
}

int facilities_config(void* facilities_s) {
    int rv = 0;

    facilities_t *facilities = (facilities_t*) facilities_s;
    
    it2s_config_t* config = calloc(1, sizeof(it2s_config_t));
    rv = it2s_config_read("/etc/it2s/itss.toml", config);
    if (rv) goto cleanup;

    facilities->zmq.responders = calloc(config->facilities.zmq.addresses_len, sizeof(zmq_pollitem_t));
    facilities->zmq.n_responders = 0;

    for (int i = 0; i < config->facilities.zmq.addresses_len; ++i) {
        char* addr = config->facilities.zmq.addresses[i];

        // IPC
        if (!memcmp(addr, "ipc", 3)) {
          
            // Create dir 
            int lp = 0;
            for (int j = 0; j < strlen(addr); ++j) {
                if (addr[j] == '/') lp = j;
            }

            char dir[256];
            memcpy(dir, addr+6, lp-6);
            dir[lp-6] = 0;

            struct stat st = {0};
            if (stat(dir, &st) == -1) {
                mkdir(dir, 0777);
            }

            // Bind
            void* socket = zmq_socket(facilities->zmq.ctx, ZMQ_REP);
            zmq_bind(socket, addr);

            facilities->zmq.responders[facilities->zmq.n_responders].socket = socket;
            facilities->zmq.responders[facilities->zmq.n_responders].events = ZMQ_POLLIN;
            ++facilities->zmq.n_responders;

        } else if (!memcmp(addr, "tcp", 3)) {
            
            bool found_astk = false; 
            for (int j = 0; j < strlen(addr); ++j) {
                if (addr[j] == '*') {
                    found_astk = true; 
                    break;
                }
            }

            if (found_astk) {
                // Bind
                void* socket = zmq_socket(facilities->zmq.ctx, ZMQ_REP);
                zmq_bind(socket, addr);

                facilities->zmq.responders[facilities->zmq.n_responders].socket = socket;
                facilities->zmq.responders[facilities->zmq.n_responders].events = ZMQ_POLLIN;
                ++facilities->zmq.n_responders;
            }
        }
    }
    if (!facilities->zmq.n_responders) {
        syslog_info("[facilities] [config] a valid address to listen to was not found, exiting now");
        rv = 1;
        goto cleanup;
    }

    // Fetch [transport] address
    int index = fetch_target_address(config->transport.zmq.addresses, config->transport.zmq.addresses_len);
    if (index != -1) {
        facilities->zmq.transport_address = malloc(strlen(config->transport.zmq.addresses[index])+1);
        strcpy(facilities->zmq.transport_address, config->transport.zmq.addresses[index]);
    } else {
        syslog_err("[facilities] [config] a valid address for [transport] was not found");
        rv = 1;
        goto cleanup;
    }

    // Fetch [applications] address
    index = fetch_target_address(config->applications.zmq.addresses, config->applications.zmq.addresses_len);
    if (index != -1) {
        facilities->zmq.applications_address = malloc(strlen(config->applications.zmq.addresses[index])+1);
        strcpy(facilities->zmq.applications_address, config->applications.zmq.addresses[index]);
    } else {
        syslog_err("[facilities] [config] a valid address for [applications] was not found");
        rv = 1;
        goto cleanup;
    }

    // Fetch [security] address
    index = fetch_target_address(config->security.zmq.addresses, config->security.zmq.addresses_len);
    if (index != -1) {
        facilities->zmq.security_address = malloc(strlen(config->security.zmq.addresses[index])+1);
        strcpy(facilities->zmq.security_address, config->security.zmq.addresses[index]);
    } else {
        syslog_err("[facilities] [config] a valid address for [security] was not found");
        rv = 1;
        goto cleanup;
    }

    // Fetch [management] address
    index = fetch_target_address(config->management.zmq.addresses, config->management.zmq.addresses_len);
    if (index != -1) {
        facilities->zmq.management_address = malloc(strlen(config->management.zmq.addresses[index])+1);
        strcpy(facilities->zmq.management_address, config->management.zmq.addresses[index]);
    } else {
        syslog_err("[facilities] [config] a valid address for [management] was not found");
        rv = 1;
        goto cleanup;
    }

    // Values
    // General
    if (!strcmp("obu", config->general.itss_type)) {
        facilities->station_type = 5;
    } else if (!strcmp("rsu", config->general.itss_type)) {
        facilities->station_type = 15;
    } else if (!strcmp("unknown", config->general.itss_type)) {
        facilities->station_type = 0;
    } else if (!strcmp("pedestrian", config->general.itss_type)) {
        facilities->station_type = 1;
    } else if (!strcmp("cyclist", config->general.itss_type)) {
        facilities->station_type = 2;
    } else if (!strcmp("moped", config->general.itss_type)) {
        facilities->station_type = 3;
    } else if (!strcmp("motorcycle", config->general.itss_type)) {
        facilities->station_type = 4;
    } else if (!strcmp("passengerCar", config->general.itss_type)) {
        facilities->station_type = 5;
    } else if (!strcmp("bus", config->general.itss_type)) {
        facilities->station_type = 6;
    } else if (!strcmp("lightTruck", config->general.itss_type)) {
        facilities->station_type = 7;
    } else if (!strcmp("heavyTruck", config->general.itss_type)) {
        facilities->station_type = 8;
    } else if (!strcmp("trailer", config->general.itss_type)) {
        facilities->station_type = 9;
    } else if (!strcmp("specialVehicles", config->general.itss_type)) {
        facilities->station_type = 10;
    } else if (!strcmp("tram", config->general.itss_type)) {
        facilities->station_type = 11;
    } else if (!strcmp("roadSideUnit", config->general.itss_type)) {
        facilities->station_type = 15;
    } else {
        syslog_err("[facilities] [config] unrecognized ITSS type, running as OBU");
        facilities->station_type = 5;
    }

    facilities->use_security = config->security.use_security;

    pthread_mutex_init(&facilities->id.lock, NULL);
    pthread_mutex_init(&facilities->id.change.lock, NULL);

    facilities->id.change.random = config->security.identity.random;
    if (facilities->id.change.random) {
        srand(time(NULL));
        facilities->id.value = rand();
    } else {
        facilities->id.value = config->security.identity.station_id;
    }
    // Inform management
    ManagementRequest_t* mreq_set = calloc(1, sizeof(ManagementRequest_t));
    mreq_set->present = ManagementRequest_PR_attributes;
    mreq_set->choice.attributes.present = ManagementRequestAttributes_PR_set;
    mreq_set->choice.attributes.choice.set.stationID = malloc(sizeof(long));
    *mreq_set->choice.attributes.choice.set.stationID = facilities->id.value;
    uint8_t b_oer[128];
    asn_enc_rval_t enc = asn_encode_to_buffer(NULL, ATS_CANONICAL_OER, &asn_DEF_ManagementRequest, mreq_set, b_oer, 128);
    if (enc.encoded != -1) {
        void* management_socket = zmq_socket(facilities->zmq.ctx, ZMQ_REQ);
        zmq_connect(management_socket, facilities->zmq.management_address);
        zmq_send(management_socket, b_oer, enc.encoded, 0);
        uint8_t code;
        zmq_recv(management_socket, &code, 1, 0);
        zmq_close(management_socket);
    }
    ASN_STRUCT_FREE(asn_DEF_ManagementRequest, mreq_set);

    // DENM
    facilities->den->n_max_events = config->facilities.denm.nmax_active_events;

    // CAM
    facilities->lightship->active = config->facilities.cam.activate; 
    facilities->lightship->vehicle_gen_min = config->facilities.cam.obu_period_min;
    facilities->lightship->vehicle_gen_max = config->facilities.cam.obu_period_max;
    facilities->lightship->rsu_gen_min = config->facilities.cam.rsu_period_min;
    facilities->lightship->rsu_vehicle_permanence = config->facilities.cam.rsu_vehicle_permanence;

    // IVIM
    facilities->infrastructure->n_max_services = config->facilities.ivim.nmax_active_services;
    facilities->infrastructure->replay_interval = config->facilities.ivim.replay_interval;
    facilities->infrastructure->default_service_duration = config->facilities.ivim.default_service_duration * 60000;

    // CPM
    facilities->dissemination->active = config->facilities.cpm.activate;
    facilities->dissemination->T_GenCpmMin = config->facilities.cpm.rsu_obu_period_min;
    facilities->dissemination->T_GenCpmMax = config->facilities.cpm.rsu_obu_period_max;
    facilities->dissemination->radar_rotation = config->applications.its_center.rotation;

    // Replay
    facilities->replay = config->networking.replay.activate;

    // PZ
    if (facilities->station_type == 15) {

        int i = 0;

        DIR *d = opendir(config->facilities.protected_zones.path);
        struct dirent *dir;
        char file[256];
        char pz_xml[2048];
        if (d) {
            while ((dir = readdir(d)) != NULL && i < 16) {
                if (dir->d_name[0] == '.') continue;
                sprintf(file, "%s/%s", config->facilities.protected_zones.path, dir->d_name);
                FILE *fp = fopen(file, "r");
                if (!fp) continue;
                fseek(fp, 0, SEEK_END);
                uint16_t size = ftell(fp);
                fseek(fp, 0, SEEK_SET);
                if (!size) {
                    fclose(fp);
                    continue;
                }
                if (!fread(pz_xml, 1, size, fp)) {
                    fclose(fp);
                    continue;
                }
                fclose(fp);
               
                ProtectedCommunicationZone_t *zone = calloc(1, sizeof(ProtectedCommunicationZone_t));

                asn_dec_rval_t dec = xer_decode(NULL, &asn_DEF_ProtectedCommunicationZone, (void**) &zone, pz_xml, size);
                if (!dec.code) {
                    facilities->lightship->pz[i] = zone;
                    ++facilities->lightship->pz_len;
                    ++i;
                    syslog_debug("[facilities] [config] loaded protection zone @ (%ld, %ld)", zone->protectedZoneLatitude, zone->protectedZoneLongitude);
                } else {
                    ASN_STRUCT_FREE(asn_DEF_ProtectedCommunicationZone, zone);
                }

            }
            closedir(d);
        }
    }

    pthread_mutex_init(&facilities->epv.space.lock, NULL);
    pthread_mutex_init(&facilities->epv.time.lock, NULL);
    facilities->epv.time.resolution = TIME_MILLISECONDS;

    ManagementRequest_t* mreq = calloc(1, sizeof(ManagementRequest_t));
    mreq->present = ManagementRequest_PR_attributes;
    mreq->choice.attributes.present = ManagementRequestAttributes_PR_get;
    mreq->choice.attributes.choice.get.coordinates = 1;
    mreq->choice.attributes.choice.get.altitude = 1;
    mreq->choice.attributes.choice.get.heading = 1;
    mreq->choice.attributes.choice.get.speed = 1;
    mreq->choice.attributes.choice.get.gpsType = 1;
    mreq->choice.attributes.choice.get.clockType = 1;
    mreq->choice.attributes.choice.get.clock = 1;
    mreq->choice.attributes.choice.get.clockOffset = 1;
    void* management_socket = zmq_socket(facilities->zmq.ctx, ZMQ_REQ);
    zmq_connect(management_socket, facilities->zmq.management_address);
    uint8_t buffer[256];
    enc = oer_encode_to_buffer(&asn_DEF_ManagementRequest, NULL, mreq, buffer, 256);

    zmq_send(management_socket, buffer, enc.encoded, 0);
    zmq_recv(management_socket, buffer, 256, 0);

    ManagementReply_t* mrep = calloc(1, sizeof(ManagementReply_t));
    oer_decode(NULL, &asn_DEF_ManagementReply, (void**) &mrep, buffer, 256);
    long lat, lon, alt, alt_conf;
    if (mrep->returnCode == ManagementReplyReturnCode_accepted && mrep->attributes &&
            mrep->attributes->coordinates && 
            mrep->attributes->altitude && 
            mrep->attributes->heading && 
            mrep->attributes->speed &&
            mrep->attributes->clockType &&
            mrep->attributes->clock &&
            mrep->attributes->clockOffset &&
            mrep->attributes->gpsType) {
        facilities->epv.space.latitude = mrep->attributes->coordinates->latitude;
        facilities->epv.space.longitude = mrep->attributes->coordinates->longitude;
        facilities->epv.space.altitude = mrep->attributes->altitude->altitudeValue;
        facilities->epv.space.altitude_conf = mrep->attributes->altitude->altitudeConfidence;
        facilities->epv.space.heading = mrep->attributes->heading->headingValue;
        facilities->epv.space.heading_conf = mrep->attributes->heading->headingConfidence;
        facilities->epv.space.speed = mrep->attributes->speed->speedValue;
        facilities->epv.space.speed_conf = mrep->attributes->speed->speedConfidence;

        facilities->epv.space.type = *mrep->attributes->gpsType;
        facilities->epv.time.type = *mrep->attributes->clockType;

        asn_INTEGER2ulong(mrep->attributes->clock, &facilities->epv.time.clock);
        asn_INTEGER2ulong(mrep->attributes->clockOffset, &facilities->epv.time.offset);
    } else {
        syslog_err("[applications] rejected MR attribute request");
        rv = 1;
        goto cleanup;
    }
    ASN_STRUCT_FREE(asn_DEF_ManagementRequest, mreq);
    ASN_STRUCT_FREE(asn_DEF_ManagementReply, mrep);
    zmq_close(management_socket);

    facilities->vehicle.length = config->facilities.vehicle.length;
    facilities->vehicle.width = config->facilities.vehicle.width;
    facilities->vehicle.role = config->facilities.vehicle.role;

cleanup:
    it2s_config_free(config);

    return rv;
}