path: root/mfi-mqtt/mfi-mqtt.c

#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <signal.h>
#include <getopt.h>
#include <pthread.h>
#include <libgen.h>

#include <json-c/json.h>
#include <mosquitto.h>

#include "mosquitto.h"
#include "reporting.h"

#define XSTR(s) STR(s)
#define STR(s) #s

static char *APP_VERSION = XSTR(VERSION);
static volatile sig_atomic_t closing_time = 0;

typedef struct control_message {
    int output;
    int state;
} control_message;

control_message * parse_control_message(const struct mosquitto_message *message) {
    struct json_tokener *tok;
    enum json_tokener_error jerr;
    control_message *out_message = NULL;
    json_object *msg, *state_key, *output_key;

    tok = json_tokener_new();
    msg = json_tokener_parse_ex(tok, message->payload, message->payloadlen);
    jerr = json_tokener_get_error(tok);
    if (jerr != json_tokener_success) {
        fprintf(stderr, "Invalid message format: %s\n", json_tokener_error_desc(jerr));
        goto cleanup;
    }

    output_key = json_object_object_get(msg, "output");
    if (!output_key) {
        fprintf(stderr, "Invalid message format: no output key\n");
        goto cleanup;
    }

    state_key = json_object_object_get(msg, "state");
    if (!state_key) {
        fprintf(stderr, "Invalid message format: no state key\n");
        goto cleanup;
    }

    out_message = calloc(sizeof(control_message), 1);
    out_message->state = json_object_get_int(state_key);
    out_message->output = json_object_get_int(output_key);
    // Clamp the value to on or off
    out_message->state = out_message->state > 0 ? 1 : 0;

cleanup:
    if (msg) json_object_put(msg);
    json_tokener_free(tok);

    return out_message;
}

void set_relay(control_message *msg) {
    FILE *f;
    char *basename;
    char filename[255];

    basename = "/proc/power/relay";

    memset(filename, 0, sizeof(filename));
    snprintf(filename, strlen(basename) + 2, "%s%i", basename, msg->output);

    f = fopen(filename, "w");
    if (!f) {
        fprintf(stderr, "Failed to open relay file %s\n", filename);
        return;
    }

    fprintf(f, "%i\n", msg->state);
    fclose(f);
}

void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message)
{
    control_message *msg;

    msg = parse_control_message(message);
    if (!msg) {
        return;
    }

    if (msg->output > get_output_count() || msg->output <= 0) {
        fprintf(stderr, "Invalid output number: %i\n", msg->output);
        goto cleanup;
    }

    fprintf(stderr, "Set output %i to %i\n", msg->output, msg->state);
    set_relay(msg);

cleanup:
    free(msg);
}

void my_connect_callback(struct mosquitto *mosq, void *obj, int result, int flags)
{
    char *topic;
    char *prefix;
    char hostname[255];

    memset(hostname, 0, sizeof(hostname));
    gethostname(hostname, 255);

    prefix = "/mfi/devices";
    topic = calloc(sizeof(char), strlen(prefix) + strlen(hostname) + 2);
    sprintf(topic, "%s/%s", prefix, hostname);

    fprintf(stderr, "Subscribed to topic %s\n", topic);

	if (!result) {
        mosquitto_subscribe(mosq, NULL, topic, 0);
	} else {
        fprintf(stderr, "%s\n", mosquitto_connack_string(result));
		mosquitto_disconnect(mosq);
	}
}

static void signal_handler(int signo) {
    closing_time = 1;
}

void set_signal_handler() {
    struct sigaction action;

    action.sa_handler = signal_handler;
    action.sa_flags = SA_SIGINFO | SA_RESTART;

    if (sigaction(SIGINT, &action, NULL) == -1) {
        perror("Error connecting signal");
        exit(-1);
    }
}

void mask_sig() {
    sigset_t mask;
    sigemptyset(&mask);
    sigaddset(&mask, SIGINT);
    pthread_sigmask(SIG_BLOCK, &mask, NULL);
}

void * control_thread(struct mosquitto *mosq) {
    mask_sig();
	mosquitto_loop_forever(mosq, 1000*86400, 1);
    return NULL;
}

void * pinger_thread(struct mosquitto *mosq) {
    power_statistics **stats;
    int report_count;
    char *output2;

    mask_sig();

    do {
        if (closing_time) {
            break;
        }

        report_count = get_stats_all_outputs(&stats);
        output2 = format_report_all_outputs(stats, report_count);

        mosquitto_publish(mosq, NULL, "/mfi/reports", strlen(output2), output2, 0, false);

        free(output2);
        free(stats);

        sleep(2);
    } while(true);

    return NULL;
}

// Stop all the blinking!
void * light_management_thread(void *arg) {
    FILE *light_fp;
    FILE *freq_fp;

    do {
        if (closing_time) {
            break;
        }

        light_fp = fopen("/proc/led/status", "w");
        if (light_fp) {
            fprintf(light_fp, "1\n");
            fclose(light_fp);
        }

        freq_fp = fopen("/proc/led/freq", "w");
        if (freq_fp) {
            fprintf(freq_fp, "0\n");
            fclose(freq_fp);
        }

        sleep(1);
    } while(true);

    return NULL;
}

// Doing this the "C way" is a real pain in the ass, just shell out and forget
// about it
void cleanup_crap_processes() {
    // Otherwise init will continue to respawn them
    system("sed -i "
        "-e '/ubnt-websockets/s/^/#/' "
        "-e '/telnetd/s/^/#/' "
        "-e '/mca[-d]/s/^/#/' "
        "-e '/lighttpd/s/^/#/' "
        "/etc/inittab");

    system("kill -HUP 1");

    // Most of these kill cleanly but a few are stubborn so don't ask, tell.
    system("pkill -9 ubnt-websockets");
    system("pkill -9 lighttpd");
    system("pkill upnpd");
    system("pkill telnetd");
    system("pkill mca-monitor");
    system("pkill mcad");
    system("pkill avahi-daemon");
}

int real_main(char *broker, int broker_port) {
    struct mosquitto *mosq;
    pthread_t pinger_thread_h, control_thread_h, light_management_thread_h;

    cleanup_crap_processes();

    set_signal_handler();

    mosq = connect_to_broker(broker, broker_port);
    if (!mosq) {
        return 1;
    }
    fprintf(stderr, "Connected to broker\n");

	mosquitto_connect_with_flags_callback_set(mosq, my_connect_callback);
	mosquitto_message_callback_set(mosq, my_message_callback);

    pthread_create(&pinger_thread_h, NULL, (void * (*)(void *))pinger_thread, mosq);
    pthread_create(&control_thread_h, NULL, (void * (*)(void *))control_thread, mosq);
    pthread_create(&light_management_thread_h, NULL, (void * (*)(void *))light_management_thread, NULL);

    do {
        sleep(2);
        if (closing_time) {
            fprintf(stderr, "Shutting down\n");

            mosquitto_disconnect(mosq);
            pthread_join(pinger_thread_h, NULL);
            pthread_join(control_thread_h, NULL);
            break;
        }
    } while(true);

    shutdown_broker(mosq);

	return 0;
}

void print_usage(char *name) {
    printf("usage: %s [--help|-h] [--version|-V] [--broker|-b] [--port|-p]\n", name);
    printf("Control mFi Power Strip via MQTT messages\n\n");

    printf("    -b, --broker    Broker hostname or IP address [REQUIRED]\n");
    printf("    -p, --port      Broker port, default: 1883\n");
    printf("    -h, --help      Show this message\n");
    printf("    -V, --version   Show application version info\n");

    free(name);
}

int main(int argc, char *argv[])
{
    int c;
    int rc;
    char *broker;
    int port = 1883;
    int option_index = 0;

    while (true) {
        static struct option long_options[] = {
            {"help",    no_argument,       0, 'h'},
            {"version", no_argument,       0, 'V'},
            {"broker",  required_argument, 0, 'b'},
            {"port",    required_argument, 0, 'p'},
            {0, 0, 0, 0}
        };

        if ((c = getopt_long(argc, argv, "hVb:p:", long_options, &option_index)) == -1) {
            break;
        }

        switch (c) {
            case 'h':
                print_usage(strdup(basename(argv[0])));
                return 0;

            case 'V':
                fprintf(stdout, "%s version %s\n", basename(argv[0]), APP_VERSION);
                return 0;

            case 'b':
                broker = strdup(optarg);
                break;

            case 'p':
                port = atoi(optarg);
                if (port > 65536 || port <= 0) {
                    fprintf(stderr, "error: invalid port specified\n");
                    return 1;
                }
                break;
        }
    }

    if (!broker) {
        fprintf(stderr, "error: no broker configured\n");
        return 1;
    }

    rc = real_main(broker, port);
    free(broker);

    return rc;
}