agent_exec-2-0.cpp File Reference

#include "support/configCosmos.h"
#include "agent/agentclass.h"
#include "support/jsonlib.h"
#include "support/convertlib.h"
#include "support/datalib.h"
#include "agent/command_queue.h"
#include <iostream>
#include <iomanip>
#include <list>
#include <fstream>
#include <sstream>
#include <mutex>

Include dependency graph for agent_exec-2-0.cpp:

Classes
struct	boot_flags

Functions
void	move_and_compress_exec ()
int32_t	get_last_offset ()
int32_t	get_flags ()
int32_t	request_get_queue_size (string &request, string &response, Agent *agent)
int32_t	request_get_event (string &request, string &response, Agent *agent)
int32_t	request_get_command (string &request, string &response, Agent *agent)
int32_t	request_del_command (string &request, string &response, Agent *agent)
int32_t	request_del_command_id (string &request, string &response, Agent *agent)
int32_t	request_add_command (string &request, string &response, Agent *agent)
int32_t	request_remote_command (string &request, string &response, Agent *agent)
int32_t	request_reopen_exec (string &request, string &response, Agent *agent)
int32_t	request_set_logstride_exec (string &request, string &response, Agent *agent)
int32_t	request_get_logstride_exec (string &request, string &response, Agent *agent)
int32_t	request_reopen_soh (string &request, string &response, Agent *agent)
int32_t	request_set_logperiod (string &request, string &response, Agent *agent)
int32_t	request_get_logperiod (string &request, string &response, Agent *agent)
int32_t	request_set_logstride_soh (string &request, string &response, Agent *agent)
int32_t	request_get_logstride_soh (string &request, string &response, Agent *agent)
int32_t	request_set_logstring (string &request, string &response, Agent *agent)
int32_t	request_get_logstring (string &request, string &response, Agent *agent)
int32_t	request_save_command (string &request, string &response, Agent *)
void	collect_data_loop () noexcept
void	move_and_compress_soh ()
void	move_and_compress_beacon ()
int32_t	get_power_mode ()
int	main (int argc, char *argv[])

Variables
static Agent *	agent
static CommandQueue	cmd_queue
static double	logdate_exec =0.
static double	newlogstride_exec = 300. / 86400.
static double	logstride_exec = 0.
static std::mutex	exec_mutex
static string	jjstring
static string	myjstring
static thread	cdthread
static string	logstring
static vector< jsonentry * >	logtable
static double	logdate_soh =0.
static double	newlogperiod = 30. / 86400.
static double	logperiod = 0
static double	newlogstride_soh = 600. / 86400.
static double	logstride_soh = 0.
static std::mutex	soh_mutex
static std::mutex	beacon_mutex
static vector< eventstruc >	eventdict
static vector< eventstruc >	events
static beatstruc	iscbeat
static struct boot_flags	boot_flags
static double	correcttime
static double	epsilon
static double	delta
static string	incoming_dir
static string	outgoing_dir
static string	immediate_dir
static string	temp_dir

Function Documentation

void move_and_compress_exec

(

)

                               {
    exec_mutex.lock();
    cmd_queue.join_event_threads();
    log_move(agent->getNode(), "exec");
    exec_mutex.unlock();
}

int32_t get_last_offset

(

)

{
    int32_t offset = 0;
    FILE *fp = fopen(("/cosmos/nodes/" + agent->nodeName + "/last_offset").c_str(), "r");
    if (fp != nullptr)
    {
        fscanf(fp, "%d", &offset);
        fclose(fp);
    }
    return offset;
}

int32_t get_flags

(

)

{
    int8_t launched = 0;
    int8_t bootcheck = 0;
    int8_t deployed = 0;
    FILE *fp = fopen(("/cosmos/nodes/" + agent->nodeName + "/boot_flags").c_str(), "r");
    if (fp != nullptr)
    {
        fscanf(fp, "%hhd %hhd %hhd", &launched, &deployed, &bootcheck);
        if (launched)
        {
            boot_flags.launched = true;
            agent->cinfo->node.flags |= NODE_FLAG_LAUNCHED;
        }
        else
        {
            boot_flags.launched = false;
            agent->cinfo->node.flags &= ~NODE_FLAG_LAUNCHED;
        }
        if (bootcheck)
        {
            boot_flags.bootcheck = true;
            agent->cinfo->node.flags |= NODE_FLAG_BOOTCHECK;
        }
        else
        {
            boot_flags.bootcheck = false;
            agent->cinfo->node.flags &= ~NODE_FLAG_BOOTCHECK;
        }
        if (deployed)
        {
            boot_flags.deployed = true;
            agent->cinfo->node.flags |= NODE_FLAG_DEPLOYED;
        }
        else
        {
            boot_flags.deployed = false;
            agent->cinfo->node.flags &= ~NODE_FLAG_DEPLOYED;
        }
        fclose(fp);
        return 1;
    }
    else
    {
        return 0;
    }
}

int32_t request_get_queue_size	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    response = std::to_string(cmd_queue.get_command_size());
    return 0;
}

int32_t request_get_event	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    std::ostringstream ss;

    if(cmd_queue.get_event_size()==0)   {
        ss << "[Empty]";
    }
    else {
        int j;
        int32_t iretn = sscanf(request.c_str(),"%*s %d",&j);

        // if valid index then return event
        if (iretn == 1) {
            if(j >= 0 && j < static_cast<int>(cmd_queue.get_event_size()))
                ss << cmd_queue.get_event(j);
            else
                ss << "<" << j << "> is not a valid event queue index (current range between 0 and "
                   << cmd_queue.get_event_size()-1 << ")";
        }

        // if no index given, return the entire queue
        else if (iretn ==  -1) {
            for(int i = 0; i < static_cast<int>(cmd_queue.get_event_size()); ++i) {
                Event cmd = cmd_queue.get_event(i);
                ss << "[" << i << "]" << "[" << mjd2iso8601(cmd.getUtc()) << "]" << cmd << endl;
            }
        }
        // if the user supplied something that couldn't be turned into an integer
        else if (iretn == 0) { ss << "Usage:\tget_event [ index ]\t"; }
    }

    response = ss.str();
    return 0;
}

int32_t request_get_command	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    std::ostringstream ss;

    if(cmd_queue.get_command_size()==0) {
        ss << "[Empty]";
    }
    else {
        int j;
        int32_t iretn = sscanf(request.c_str(),"%*s %d",&j);

        // if valid index then return command
        if (iretn == 1) {
            if(j >= 0 && j < static_cast<int>(cmd_queue.get_command_size()))
                ss << cmd_queue.get_command(j);
            else
                ss << "<" << j << "> is not a valid command queue index (current range between 0 and "
                   << cmd_queue.get_command_size()-1 << ")";
        }

        // if no index given, return the entire queue
        else if (iretn ==  -1) {
            for(int i = 0; i < static_cast<int>(cmd_queue.get_command_size()); ++i) {
                Event cmd = cmd_queue.get_command(i);
                ss << "[" << i << "]" << "[" << mjd2iso8601(cmd.getUtc()) << "]" << cmd << endl;
            }
        }
        // if the user supplied something that couldn't be turned into an integer
        else if (iretn == 0) { ss << "Usage:\tget_command [ index ]\t"; }
    }

    response = ss.str();
    return 0;
}

int32_t request_del_command	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    Event cmd;
    string line(request);

    // Check for valid request
    if (line.find(" ") == string::npos)
    {
        return GENERAL_ERROR_UNDEFINED;
    }

    line.erase(0, line.find(" ")+1);
    cmd.set_command(line);

    //delete command
    int n = cmd_queue.del_command(cmd);

    if(!cmd.is_command()) {
        response =  "Not a valid command: " + line;
    }
    else {
        response = std::to_string(n) + " commands deleted from the queue";
    }

    return 0;
}

int32_t request_del_command_id	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    Event cmd;
    std::ostringstream ss;

    if(cmd_queue.get_command_size()==0) {
        ss << "the command queue is empty";
    }
    else {
        int j;
        int32_t iretn = sscanf(request.c_str(),"%*s %d",&j);

        // if valid index then return command
        if (iretn == 1) {
            if(j >= 0 && j < static_cast<int>(cmd_queue.get_command_size())) {
                response = std::to_string(cmd_queue.del_command(j)) + " commands deleted from the queue";
            } else {
                ss << "<" << j << "> is not a valid command queue index (current range between 0 and " << cmd_queue.get_command_size()-1 << ")";
            }
        }
        // if the user supplied something that couldn't be turned into an integer
        else if (iretn == 0)    {
            ss << "Usage:\tdel_command_id [ index ]\t";
        }
    }

    response = ss.str();
    return 0;
}

int32_t request_add_command	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    Event cmd;
    string line(request);

    // Check for valid request
    if (line.find(" ") == string::npos)
    {
        return GENERAL_ERROR_UNDEFINED;
    }

    line.erase(0, line.find(" ")+1);
    cmd.set_command(line);

    // add command
    if(cmd.is_command()) {
        cmd_queue.add_command(cmd);
        response =  "Command added to queue: " + line;
    }
    else {
        response =  "Not a valid command: " + line;
    }

    // sort the queue
    cmd_queue.sort();
    return 0;
}

int32_t request_remote_command	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
//    char request_re[AGENTMAXBUFFER + 5];
    int32_t iretn = 0;
//    FILE *pd;
    int i;

    // Locate where our command starts.
    bool flag = false;
    for (i=0; i<AGENTMAXBUFFER-1; i++) {
        if (flag) {
            if (request[i] != ' ')
                break;
        }
        else if (request[i] == ' ') {
            flag = true;
        }
    }

    if (i == AGENTMAXBUFFER-1) {
        response = "Unable to find an appropriate command.";
    }
    else {
        // Run the process and create a pipe.
        iretn = data_execute(request.substr(request.find(" ") + 1), response);
    }

    return iretn;
}

int32_t request_reopen_exec	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    // Do not wait for log stride, move data from temp to incoming immediately.
    logdate_exec = agent->cinfo->node.loc.utc;
    move_and_compress_exec();
    return 0;
}

int32_t request_set_logstride_exec	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    sscanf(request.c_str(),"%*s %lf",&newlogstride_exec);
    return 0;
}

int32_t request_get_logstride_exec	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    response =  std::to_string(logstride_exec);
    return 0;
}

int32_t request_reopen_soh	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    // Do not wait for logstride, push soh log files out of temp and into incoming.
    logdate_soh = agent->cinfo->node.loc.utc;
    move_and_compress_soh();
    return 0;
}

int32_t request_set_logperiod	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    sscanf(request.c_str(),"%*s %lf",&newlogperiod);
    newlogperiod /= 86400.;
    return 0;
}

int32_t request_get_logperiod	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    response =  std::to_string(86400. * logperiod);
    return 0;
}

int32_t request_set_logstride_soh	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    sscanf(request.c_str(),"%*s %lf",&newlogstride_soh);
    newlogstride_soh /= 86400.;
    return 0;
}

int32_t request_get_logstride_soh	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    response =  std::to_string(86400. * logstride_soh);
    return 0;
}

int32_t request_set_logstring	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    request.erase(0, request.find(" ")+1);
    logtable.clear();
    json_table_of_list(logtable, request.c_str(), agent->cinfo);
    return 0;
}

int32_t request_get_logstring	(	string &	request,
		string &	response,
		Agent *	agent
	)

{
    response = logstring;
    return 0;
}

int32_t request_save_command	(	string &	request,
		string &	response,
		Agent *
	)

{
    vector<string> args = string_split(request, " ");
    if (args.size() == 2)
    {
        cmd_queue.save_commands(temp_dir, args[1]);
    }
    else
    {
        cmd_queue.save_commands(temp_dir);
    }
    return 0;
}

void collect_data_loop ( )

noexcept

{
    int32_t iretn;
    while (agent->running())
    {
        // Collect new data
        Agent::messstruc mess;
        iretn = agent->readring(mess, Agent::AgentMessage::ALL, 5., Agent::Where::TAIL, "", agent->cinfo->node.name);
        if (iretn >= 0)
        {
            if (mess.meta.type < Agent::AgentMessage::BINARY)
            {
                json_parse(mess.adata, agent->cinfo);
                agent->cinfo->node.utc = currentmjd(0.);

                for (devicestruc device: agent->cinfo->device)
                {
                    if (device.utc > agent->cinfo->node.utc)
                    {
                        agent->cinfo->node.utc = device.utc;
                    }
                }
            }
        }
    }
    return;
}

void move_and_compress_soh

(

)

                              {
    soh_mutex.lock();
    log_move(agent->getNode(), "soh");
    soh_mutex.unlock();
}

void move_and_compress_beacon

(

)

                                 {
    string beacon_string;
    beacon_mutex.lock();
    log_write(agent->cinfo->node.name, DATA_LOG_TYPE_BEACON, logdate_soh, json_of_beacon(beacon_string, agent->cinfo));
    log_move(agent->getNode(), "beacon");
    beacon_mutex.unlock();
}

int32_t get_power_mode

(

)

{
    int32_t powermode = -1;
    uint16_t tindex;
    string tdata;
    int32_t iretn = data_execute("power_mode_get", tdata);
    if(iretn<0) {
        return powermode;
    } else  {
        if (sscanf(tdata.c_str(), "%u\n", &tindex) == 1)
        {
            if (tindex > 0)
            {
                powermode = tindex;
            }
        }
    //    FILE *fp = popen("/cosmos/scripts/power_mode_get", "r");
//    if (fp == nullptr)
//    {
//        return -errno;
//    }
//    char tdata[100];
//    uint16_t tindex;
//    if ((fgets(tdata, 100, fp)) == tdata)
//    {
//        if (sscanf(tdata, "%u\n", &tindex) == 1)
//        {
//            if (tindex > 0)
//            {
//                powermode = tindex;
//            }
//        }
//    }
//    pclose(fp);
        return powermode;
    }
}

int main	(	int	argc,
		char *	argv[]
	)

{
    vector<eventstruc> events, eventdict;
    string jjstring, myjstring;
    double llogmjd, dlogmjd, clogmjd;
    int32_t iretn;

    // Set node name to first argument
    if (argc == 2)
    {
        agent = new Agent(argv[1], "exec", 0.);
    }
    else
    {
        agent = new Agent("", "exec", 0.);
    }

    if ((iretn = agent->wait()) < 0)
    {
        fprintf(agent->get_debug_fd(), "%16.10f %s Failed to start Agent %s on Node %s Dated %s : %s\n",currentmjd(), mjd2iso8601(currentmjd()).c_str(), agent->getAgent().c_str(), agent->getNode().c_str(), utc2iso8601(data_ctime(argv[0])).c_str(), cosmos_error_string(iretn).c_str());
        exit(iretn);
    }
    else
    {
        fprintf(agent->get_debug_fd(), "%16.10f %s Started Agent %s on Node %s Dated %s\n",currentmjd(), mjd2iso8601(currentmjd()).c_str(), agent->getAgent().c_str(), agent->getNode().c_str(), utc2iso8601(data_ctime(argv[0])).c_str());
    }

    // Fix time
    iretn = agent->get_agent_time(correcttime, epsilon, delta, "gps");
    if (iretn >= 0)
    {
        printf("Initialized time from time GPS: Delta %f %f Epsilon %f\n", 86400. * (correcttime - currentmjd()), 86400.*delta, 86400.*epsilon);
        set_local_clock(correcttime);
    }
    else
    {
        FILE *fp = fopen(("/cosmos/nodes/" + agent->nodeName + "/last_date").c_str(), "r");
        if (fp != nullptr)
        {
            calstruc date;
            fscanf(fp, "%02d%02d%02d%02d%04d%*c%02d\n", &date.month, &date.dom, &date.hour, &date.minute, &date.year, &date.second);
            fclose(fp);
            int32_t offset = get_last_offset();
            date.second += offset;
            double delta = cal2mjd(date) -  currentmjd();
            if (delta > 3.5e-4)
            {
                delta = set_local_clock(cal2mjd(date));
                printf("Initialized time from file: Delta %f\n", delta);
            }
        }
    }

    // Check for launched
    get_flags();

    agent->cinfo->node.utc = 0.;
    agent->cinfo->agent[0].aprd = 1.;
    cout<<"  started."<<endl;

    // Establish Executive functions

    // Set the immediate, incoming, outgoing, and temp directories
    immediate_dir = data_base_path(agent->getNode(), "immediate", "exec") + "/";
    if (immediate_dir.empty())
    {
        cout<<"unable to create directory: <"<<(agent->getNode()+"/immediate")+"/exec"<<"> ... exiting."<<endl;
        exit(1);
    }

    incoming_dir = data_base_path(agent->getNode(), "incoming", "exec") + "/";
    if (incoming_dir.empty())
    {
        cout<<"unable to create directory: <"<<(agent->getNode()+"/incoming")+"/exec"<<"> ... exiting."<<endl;
        exit(1);
    }

    outgoing_dir = data_base_path(agent->getNode(), "outgoing", "exec") + "/";
    if (outgoing_dir.empty())
    {
        cout<<"unable to create directory: <"<<(agent->getNode()+"/outgoing")+"/exec"<<"> ... exiting."<<endl;
        exit(1);
    }

    temp_dir = data_base_path(agent->getNode(), "temp", "exec") + "/";
    if (temp_dir.empty())
    {
        cout<<"unable to create directory: <"<<(agent->getNode()+"/temp")+"/exec"<<"> ... exiting."<<endl;
        exit(1);
    }

    // Event related request functions
    if ((iretn=agent->add_request("getqueuesize", request_get_queue_size, "", "returns the current size of the command queue")))
        exit (iretn);
    if ((iretn=agent->add_request("delcommand", request_del_command, "entry-string", "deletes the specified command event from the queue according to its JSON string")))
        exit (iretn);
    if ((iretn=agent->add_request("delcommandid", request_del_command_id, "entry #", "deletes the specified command event from the queue according to its position")))
        exit (iretn);
    if ((iretn=agent->add_request("getcommand", request_get_command, "[ entry # ]", "returns the requested command queue entry (or all if none specified)")))
        exit (iretn);
    if ((iretn=agent->add_request("getqueue", request_get_command, "[ entry # ]", "returns the requested command queue entry (or all if none specified)")))
        exit (iretn);
    if ((iretn=agent->add_request("addcommand", request_add_command, "{\"event_name\":\"\"}{\"event_utc\":0}{\"event_utcexec\":0}{\"event_flag\":0}{\"event_type\":0}{\"event_data\":\"\"}{\"event_condition\":\"\"}", "adds the specified command event to the queue")))
        exit (iretn);
    if ((iretn=agent->add_request("runremotecommand", request_remote_command, "command-to-run", "run a command local to the agent and return its output")))
        exit (iretn);
    if ((iretn=agent->add_request("getevent", request_get_event, "[ entry # ]", "returns the requested event queue entry (or all if none specified)")))
        exit (iretn);
    if ((iretn=agent->add_request("reopenexec", request_reopen_exec, "", "flushes exec log files out of temp directory")))
        exit (iretn);
    if ((iretn=agent->add_request("setlogstrideexec", request_set_logstride_exec, "logstride", "modify how frequently we flush exec log files out of temp directory")))
        exit (iretn);
    if ((iretn=agent->add_request("getlogstrideexec", request_get_logstride_exec, "", "return how frequently we flush exec log files out of temp directory")))
        exit (iretn);
    if ((iretn=agent->add_request("savecommand", request_save_command, "{file}", "Save current command list to optional file name (default: .queue)")))
        exit (iretn);
    if ((iretn=agent->add_request("savequeue", request_save_command, "{file}", "Save current command list to optional file name (default: .queue)")))
        exit (iretn);

    // SOH related request functions
    if ((iretn=agent->add_request("reopensoh", request_reopen_soh, "", "flushes soh log files out of temp directory")))
        exit (iretn);
    if ((iretn=agent->add_request("setlogperiod" ,request_set_logperiod, "logperiod", "set how often we log our SOH data")))
        exit (iretn);
    if ((iretn=agent->add_request("getlogperiod" ,request_get_logperiod, "", "return how often we log our SOH data")))
        exit (iretn);
    if ((iretn=agent->add_request("setlogstring", request_set_logstring, "logstring", "set what parts of our SOH that we log.")))
        exit (iretn);
    if ((iretn=agent->add_request("getlogstring", request_get_logstring, "", "return what portions of our SOH that we log")))
        exit (iretn);
    if ((iretn=agent->add_request("setlogstride_soh", request_set_logstride_soh, "logstride", "modify how frequently we flush our soh log files out of temp directory")))
        exit (iretn);
    if ((iretn=agent->add_request("getlogstride_soh", request_get_logstride_soh, "", "return how frequently we flush our soh log files out of temp directory")))
        exit (iretn);

    load_dictionary(eventdict, agent->cinfo, "events.dict");

    // Reload existing queue
    cmd_queue.restore_commands(temp_dir);

    // Start thread to collect SOH data
    bool log_data_flag = true;
    vector <beatstruc> servers = agent->find_agents(1.);
    for (beatstruc &i : servers)
    {
        if (strcmp(i.node, agent->nodeName.c_str()) && !strcmp(i.proc, "exec"))
        {
            log_data_flag = false;
            break;
        }
    }

    if (log_data_flag)
    {
        printf("Primary: SOH logger\n");
    }
    else
    {
        printf("Secondary: Not logging SOH\n");
    }

    cdthread = thread([=] { collect_data_loop(); });

    // Create default logstring
    logstring = json_list_of_soh(agent->cinfo);
    printf("===\nlogstring: %s\n===\n", logstring.c_str()); fflush(stdout);
    json_table_of_list(logtable, logstring.c_str(), agent->cinfo);

    agent->set_sohstring(logstring);

    // Start performing the body of the agent
    agent->post(Agent::AgentMessage::REQUEST, "postsoh");
    COSMOS_SLEEP(10.);
    llogmjd = currentmjd();
    clogmjd =  currentmjd();
    logdate_exec = 0.;
    logstride_exec = 0.;
    agent->start_active_loop();
    agent->debug_level = 0;
    ElapsedTime postet;
    ElapsedTime savet;
    while(agent->running())
    {
        int32_t next_power_mode = get_power_mode();
        agent->cinfo->node.powmode = next_power_mode;

        // Fix time
//        iretn = agent->get_agent_time(correcttime, epsilon, delta, "gps");
//        if (iretn >= 0 && fabs(delta*86400.) > 30.)
//        {
//            printf("Initialized time from time GPS: Delta %f %f Epsilon %f\n", 86400. * (correcttime - currentmjd()), 86400.*delta, 86400.*epsilon);
//            set_local_clock(correcttime);
//            llogmjd = currentmjd();
//            clogmjd =  currentmjd();
//            logdate_exec = 0.;
//            logstride_exec = 0.;
//        }
//        else
//        {
//            iretn = agent->get_agent_time(correcttime, epsilon, delta, "duplex");
//            if (iretn >= 0 && fabs(delta*86400.) > 30.)
//            {
//                printf("Initialized time from time DUPLEX: Delta %f %f Epsilon %f\n", 86400. * (correcttime - currentmjd()), 86400.*delta, 86400.*epsilon);
//                set_local_clock(correcttime);
//                llogmjd = currentmjd();
//                clogmjd =  currentmjd();
//                logdate_exec = 0.;
//                logstride_exec = 0.;
//            }
//        }

        dlogmjd = (clogmjd-llogmjd)*86400.;
        agent->cinfo->node.utc = clogmjd = currentmjd();
        agent->cinfo->node.downtime = get_last_offset();

        // Check if exec logstride has changed
        if (fabs(newlogstride_exec - logstride_exec) > std::numeric_limits<double>::epsilon())
        {
            logstride_exec = newlogstride_exec;
            logdate_exec = currentmjd(0.);
            move_and_compress_exec();
        }

        // Check if exec logstride has expired
        if (logstride_exec != 0. && floor(clogmjd/logstride_exec)*logstride_exec > logdate_exec)
        {
            logdate_exec = floor(clogmjd/logstride_exec)*logstride_exec;
            move_and_compress_exec();
        }

        // Check if the SOH logperiod has changed
        if (newlogperiod != logperiod )
        {
            logperiod = newlogperiod;
            logdate_soh = agent->cinfo->node.utc;

            if (log_data_flag)
            {
                move_and_compress_soh();
            }
        }

        if (fabs(newlogstride_soh - logstride_soh) > std::numeric_limits<double>::epsilon())
        {
            logstride_soh = newlogstride_soh;
            logdate_soh = currentmjd(0.);


            if (log_data_flag)
            {
                move_and_compress_soh();
                move_and_compress_beacon();
            }
        }

        if (logstride_soh != 0. && floor(clogmjd/logstride_soh)*logstride_soh > logdate_soh)
        {
            logdate_soh = floor(clogmjd/logstride_soh)*logstride_soh;
            if (log_data_flag)
            {
                move_and_compress_soh();
                move_and_compress_beacon();
            }
        }

        // Check if SOH logperiod has expired
        if (dlogmjd-logperiod > -logperiod/20.)
        {
            llogmjd = clogmjd;
            if (log_data_flag && agent->cinfo->node.utc != 0. && logstring.size())
            {
                log_write(agent->cinfo->node.name, DATA_LOG_TYPE_SOH, logdate_soh, json_of_table(jjstring, logtable, agent->cinfo));
//                log_write(agent->cinfo->node.name, DATA_LOG_TYPE_SOH, logdate_soh, json_of_table(jjstring, logtable, agent->cinfo), static_cast <string>("immediate"));
            }
        }

        // Perform SOH specific functions
        if (log_data_flag && logstride_soh > 0. && agent->cinfo->node.utc != 0. && postet.split() >= 43200. * logperiod)
        {
            postet.reset();
            agent->post(Agent::AgentMessage::REQUEST, "postsoh");

            loc_update(&agent->cinfo->node.loc);
            update_target(agent->cinfo);
            //            agent->post(Agent::AgentMessage::SOH, json_of_table(myjstring, logtable, agent->cinfo));
            calc_events(eventdict, agent->cinfo, events);
            for (uint32_t k=0; k<events.size(); ++k)
            {
                memcpy(&agent->cinfo->event[0],&events[k],sizeof(eventstruc));
                strcpy(agent->cinfo->event[0].condition,agent->cinfo->emap[events[k].handle.hash][events[k].handle.index].text);
                log_write(agent->cinfo->node.name,DATA_LOG_TYPE_EVENT,logdate_soh, json_of_event(jjstring, agent->cinfo));
            }
        }

        // Perform Executive specific functions
        cmd_queue.load_commands(immediate_dir);
        cmd_queue.load_commands(incoming_dir);
        cmd_queue.join_event_threads();
        cmd_queue.run_commands(agent, agent->getNode(), logdate_exec);

        if (savet.split() > 60.)
        {
            FILE *fp = fopen(("/cosmos/nodes/" + agent->nodeName + "/last_date").c_str(), "w");
            if (fp)
            {
                savet.reset();
                calstruc date = mjd2cal(currentmjd());
                fprintf(fp, "%02d%02d%02d%02d%04d.59\n", date.month, date.dom, date.hour, date.minute, date.year);
                fclose(fp);
            }
        }
        agent->finish_active_loop();
    }

    agent->shutdown();
    if (log_data_flag)
    {
        cdthread.join();
    }
}

Variable Documentation

Agent* agent

static

CommandQueue cmd_queue

static

double logdate_exec =0.

static

double newlogstride_exec = 300. / 86400.

static

double logstride_exec = 0.

static

std::mutex exec_mutex

static

string jjstring

static

string myjstring

static

thread cdthread

static

string logstring

static

vector<jsonentry*> logtable

static

double logdate_soh =0.

static

double newlogperiod = 30. / 86400.

static

double logperiod = 0

static

double newlogstride_soh = 600. / 86400.

static

double logstride_soh = 0.

static

std::mutex soh_mutex

static

std::mutex beacon_mutex

static

vector<eventstruc> eventdict

static

vector<eventstruc> events

static

beatstruc iscbeat

static

struct boot_flags boot_flags

static

double correcttime

static

double epsilon

static

double delta

static

string incoming_dir

static

string outgoing_dir

static

string immediate_dir

static

string temp_dir

static