mqtt.esp

#define MAX_MQTT_BUFFER 2048

char mqttBuffer[MAX_MQTT_BUFFER];

struct MqttMessage {
	char command[20];
	char uid[20];
	char user[64];
	char door[20];
	char serializedMessage[MAX_MQTT_BUFFER];
	MqttMessage *nextMessage = NULL;
};

MqttMessage *mqttMessageQueue = NULL;

void connectToMqtt()
{
	if (!config.mqttEnabled || mqttClient.connected()) {
		return;
	}
#ifdef DEBUG
	Serial.println("[ INFO ] Connecting MQTT");
#endif
	mqttClient.connect();
}

void disconnectMqtt()
{
	if (!config.mqttEnabled || !mqttClient.connected()) {
		return;
	}
#ifdef DEBUG
	Serial.println("[ INFO ] Disconnecting MQTT");
#endif
	mqttClient.disconnect(true);
}

void onMqttDisconnect(AsyncMqttClientDisconnectReason reason)
{
	String reasonstr = "";
	switch (reason)
	{
	case (AsyncMqttClientDisconnectReason::TCP_DISCONNECTED):
		reasonstr = "TCP_DISCONNECTED";
		break;
	case (AsyncMqttClientDisconnectReason::MQTT_UNACCEPTABLE_PROTOCOL_VERSION):
		reasonstr = "MQTT_UNACCEPTABLE_PROTOCOL_VERSION";
		break;
	case (AsyncMqttClientDisconnectReason::MQTT_IDENTIFIER_REJECTED):
		reasonstr = "MQTT_IDENTIFIER_REJECTED";
		break;
	case (AsyncMqttClientDisconnectReason::MQTT_SERVER_UNAVAILABLE):
		reasonstr = "MQTT_SERVER_UNAVAILABLE";
		break;
	case (AsyncMqttClientDisconnectReason::MQTT_MALFORMED_CREDENTIALS):
		reasonstr = "MQTT_MALFORMED_CREDENTIALS";
		break;
	case (AsyncMqttClientDisconnectReason::MQTT_NOT_AUTHORIZED):
		reasonstr = "MQTT_NOT_AUTHORIZED";
		break;
	case (AsyncMqttClientDisconnectReason::ESP8266_NOT_ENOUGH_SPACE):
		reasonstr = "ESP8266_NOT_ENOUGH_SPACE";
		break;
	default:
		reasonstr = "Unknown";
		break;
	}
	writeEvent("WARN", "mqtt", "Disconnected from MQTT server", reasonstr);

	if (WiFi.isConnected())
	{
		mqttReconnectTimer.once(60, connectToMqtt);
	}
}

void mqttPublishEvent(JsonDocument *root, String topic)
{
	if (config.mqttEnabled && mqttClient.connected())
	{
		String stopic(config.mqttTopic);
		stopic = stopic + "/" + topic;
		String mqttBuffer;
		serializeJson(*root, mqttBuffer);
		mqttClient.publish(stopic.c_str(), 0, false, mqttBuffer.c_str());
#ifdef DEBUG
		Serial.print("[ INFO ] Mqtt Publish:");
		Serial.println(mqttBuffer);
#endif
	}
}

void mqttPublishEventHA(JsonDocument *root, String topic)
{
	if (config.mqttEnabled && mqttClient.connected())
	{
		String stopic = "homeassistant/";
		stopic = stopic + topic;
		String mqttBuffer;
		serializeJson(*root, mqttBuffer);
		mqttClient.publish(stopic.c_str(), 0, true, mqttBuffer.c_str());
#ifdef DEBUG
		Serial.print("[ INFO ] Mqtt Publish:");
		Serial.println(mqttBuffer);
#endif
	}
}

void mqttPublishEvent(JsonDocument *root)
{
	mqttPublishEvent(root, "send");
}

void mqttPublishAck(String command)
{
	DynamicJsonDocument root(512);
	root["type"] = command;
	root["ip"] = WiFi.localIP().toString();
	root["hostname"] = config.deviceHostname;
	mqttPublishEvent(&root);
}

void mqttPublishBoot(time_t boot_time)
{
	DynamicJsonDocument root(512);
	root["type"] = "boot";
	root["time"] = boot_time;
	root["uptime"] = "0";
	root["ip"] = WiFi.localIP().toString();
	root["hostname"] = config.deviceHostname;
	mqttPublishEvent(&root);
}

void mqttPublishDiscovery()
{
	String mtopic(config.mqttTopic);
	String topic;
	String deviceName = config.deviceHostname;
	String deviceIP = WiFi.localIP().toString();
	DynamicJsonDocument via(512);
	via["ids"] = WiFi.macAddress();

	DynamicJsonDocument dev(512);
	dev["ids"] = WiFi.macAddress();
	dev["name"] = config.deviceHostname;
	dev["mf"] = "esp-rfid";
	dev["sw"] = VERSION;
	String num;
	for (int n = 0; n < config.numRelays; n++)
	{
		DynamicJsonDocument root(512);
		num = String(n);
		topic = "lock/" + deviceName + "/lock" + num + "/config";
		root["name"] = deviceName + " Lock" + num;
		root["uniq_id"] = deviceName + "/lock" + num;
		root["stat_t"] = mtopic + "/io/lock" + num;
		root["cmd_t"] = mtopic + "/cmd";
		root["pl_unlk"] = "{cmd:'open', door:'" + num + "', doorip:'" + deviceIP + "'}";
		root["pl_lock"] = "{cmd:'close', door:'" + num + "', doorip:'" + deviceIP + "'}";
		root["avty_t"] = mtopic + "/avty";
		if (n == 0)
		{
			root["dev"] = dev;
		}
		else
		{
			root["dev"] = via;
		}
		mqttPublishEventHA(&root, topic);
	}

	if (config.doorstatpin != 255)
	{
		DynamicJsonDocument door(512);
		topic = "binary_sensor/" + deviceName + "/door/config";
		door["name"] = "Door";
		door["uniq_id"] = deviceName + "/door";
		door["stat_t"] = mtopic + "/io/door";
		door["avty_t"] = mtopic + "/avty";
		door["dev_cla"] = "door";
		door["dev"] = via;
		mqttPublishEventHA(&door, topic);

		DynamicJsonDocument tamper(512);
		topic = "binary_sensor/" + deviceName + "/tamper/config";
		tamper["name"] = "Door tamper";
		tamper["uniq_id"] = deviceName + "/tamper";
		tamper["stat_t"] = mtopic + "/io/tamper";
		tamper["avty_t"] = mtopic + "/avty";
		tamper["dev_cla"] = "safety";
		tamper["dev"] = via;
		mqttPublishEventHA(&tamper, topic);
	}

	if (config.doorbellpin != 255)
	{
		DynamicJsonDocument doorbell(512);
		topic = "binary_sensor/" + deviceName + "/doorbell/config";
		doorbell["name"] = "Doorbell";
		doorbell["uniq_id"] = deviceName + "/doorbell";
		doorbell["stat_t"] = mtopic + "/io/doorbell";
		doorbell["avty_t"] = mtopic + "/avty";
		doorbell["dev_cla"] = "sound";
		doorbell["icon"] = "mdi:bell";
		doorbell["dev"] = via;
		mqttPublishEventHA(&doorbell, topic);
	}

	DynamicJsonDocument tag(512);
	topic = "sensor/" + deviceName + "/tag/config";
	tag["name"] = "Tag";
	tag["uniq_id"] = deviceName + "/tag";
	tag["stat_t"] = mtopic + "/tag";
	tag["avty_t"] = mtopic + "/avty";
	tag["val_tpl"] = "{{ value_json.uid }}";
	tag["json_attr_t"] = mtopic + "/tag";
	tag["icon"] = "mdi:key";
	tag["dev"] = via;
	mqttPublishEventHA(&tag, topic);

	DynamicJsonDocument user(512);
	topic = "sensor/" + deviceName + "/user/config";
	user["name"] = "User";
	user["uniq_id"] = deviceName + "/name";
	user["stat_t"] = mtopic + "/tag";
	user["avty_t"] = mtopic + "/avty";
	user["val_tpl"] = "{{ value_json.username }}";
	user["json_attr_t"] = mtopic + "/tag";
	user["icon"] = "mdi:human";
	user["dev"] = via;
	mqttPublishEventHA(&user, topic);

	DynamicJsonDocument dellog(512);
	topic = "button/" + deviceName + "/dellog/config";
	dellog["name"] = deviceName + " Delete Log";
	dellog["uniq_id"] = deviceName + "/deletlog";
	dellog["cmd_t"] = mtopic + "/cmd";
	dellog["payload_press"] = "{cmd:'deletlog', doorip:'" + deviceIP + "'}";
	dellog["avty_t"] = mtopic + "/avty";
	dellog["dev"] = via;
	mqttPublishEventHA(&dellog, topic);
}

void mqttPublishAvty()
{
	String mtopic(config.mqttTopic);
	String avty_topic = mtopic + "/avty";
	String payloadString = "online";
	mqttClient.publish(avty_topic.c_str(), 0, true, payloadString.c_str());
#ifdef DEBUG
	Serial.println("[ INFO ] Mqtt Publish: online @ " + avty_topic);
#endif
}

void mqttPublishHeartbeat(time_t heartbeat, time_t uptime)
{
	DynamicJsonDocument root(512);
	root["type"] = "heartbeat";
	root["time"] = heartbeat;
	root["uptime"] = uptime;
	root["ip"] = WiFi.localIP().toString();
	root["hostname"] = config.deviceHostname;
	mqttPublishEvent(&root);
}

void mqttPublishAccess(time_t accesstime, String const &isknown, String const &type, String const &user, String const &uid)
{
	DynamicJsonDocument root(512);
	if (config.mqttEvents)
	{
		root["cmd"] = "log";
	}

	root["uid"] = uid;
	root["username"] = user;
	root["access"] = type;
	root["time"] = accesstime;
	root["doorName"] = config.doorName[0];

	if (!config.mqttHA)
	{
		root["type"] = "access";
		root["isKnown"] = isknown;
		root["hostname"] = config.deviceHostname;
		mqttPublishEvent(&root);
	}
	else
	{
		mqttPublishEvent(&root, "tag");
	}
}

void mqttPublishAccess(time_t accesstime, String const &isknown, int types[MAX_NUM_RELAYS], String const &user, String const &uid)
{
	DynamicJsonDocument root(512);
	JsonArray access = root.createNestedArray("access");
	JsonArray doorNames = root.createNestedArray("doorName");
	if (config.mqttEvents)
	{
		root["cmd"] = "log";
	}

	for (int i=0; i<config.numRelays; i++) {
		switch (types[i])
		{
		case 0:
			access.add("Disabled");
			break;
		case 1:
			access.add("Always");
			break;
		case 99:
			access.add("Admin");
			break;
		}
		doorNames.add(config.doorName[i]);
	}

	root["uid"] = uid;
	root["username"] = user;
	root["time"] = accesstime;

	if (!config.mqttHA)
	{
		root["type"] = "access";
		root["isKnown"] = isknown;
		root["hostname"] = config.deviceHostname;
		mqttPublishEvent(&root);
	}
	else
	{
		mqttPublishEvent(&root, "tag");
	}
}

void mqttPublishIo(String const &io, String const &state)
{
	if (config.mqttHA && config.mqttEnabled && mqttClient.connected())
	{
		String mtopic(config.mqttTopic);
		String topic = mtopic + "/io/" + io;

		mqttClient.publish(topic.c_str(), 0, false, state.c_str());

#ifdef DEBUG
		Serial.print("[ INFO ] Mqtt Publish: ");
		Serial.println(state + " @ " + topic);
#endif
	}
}

void onMqttPublish(uint16_t packetId)
{
	writeEvent("INFO", "mqtt", "MQTT publish acknowledged", String(packetId));
}

void getUserList()
{
	DynamicJsonDocument root(512);
	JsonArray users = root.createNestedArray("list");
	Dir dir = SPIFFS.openDir("/P/");
#ifdef DEBUG
	Serial.println("[ INFO ] getUserList");
#endif
	while (dir.next())
	{
		JsonObject item = users.createNestedObject();
		String uid = dir.fileName();
		uid.remove(0, 3);
		item["uid"] = uid;
		File f = SPIFFS.open(dir.fileName(), "r");
		size_t size = f.size();
		std::unique_ptr<char[]> buf(new char[size]);
		f.readBytes(buf.get(), size);
		DynamicJsonDocument json(512);
		auto error = deserializeJson(json, buf.get());
		if (!error)
		{
			mqttPublishEvent(&json);
		}
	}
}

void deleteAllUserFiles()
{
	Dir dir = SPIFFS.openDir("/P/");
	while (dir.next())
	{
		String uid = dir.fileName();
		uid.remove(0, 3);
		SPIFFS.remove(dir.fileName());
	}
}

void deleteUserID(const char *uid)
{
	// only do this if a user id has been provided
	if (uid)
	{
		Dir dir = SPIFFS.openDir("/P/");
		while (dir.next())
		{
			String user_id = dir.fileName();
			String myuid = uid;
			user_id.remove(0, 3);
			if (myuid == user_id)
			{
				SPIFFS.remove(dir.fileName());
			}
		}
	}
}

void onMqttMessage(char *topic, char *payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total)
{
	size_t n = 0;
	size_t i = index;
	while(n < len) {
		mqttBuffer[i] = payload[n];
		n++;
		i++;
	}
	if(index + len == total) {
		mqttBuffer[i] = '\0';
	} else {
		return;
	}
#ifdef DEBUG
	Serial.print("[ INFO ] JSON msg: ");
	Serial.println(mqttBuffer);
#endif

	DynamicJsonDocument mqttIncomingJson(2048);
	// cast to const char to avoid modification in place of mqttBuffer
	auto error = deserializeJson(mqttIncomingJson, (const char *)mqttBuffer);
	if (error)
	{
#ifdef DEBUG
		Serial.print("[ INFO ] Failed parsing MQTT message: ");
		Serial.println(mqttBuffer);
#endif
		return;
	}

	// Check if IP was sent with command because we only
	// accept commands where sent IP is equal to device IP
	if (!mqttIncomingJson.containsKey("doorip"))
	{
		return;
	}
	
	const char *ipadr = mqttIncomingJson["doorip"];
	String espIp = WiFi.localIP().toString();
	if (!((strcmp(ipadr, espIp.c_str()) == 0) && (ipadr != NULL)))
	{
#ifdef DEBUG
		Serial.print("[ INFO ] ESP  IP: ");
		Serial.println(espIp);
		Serial.print("[ INFO ] recv IP: ");
		Serial.println(ipadr);
#endif
		return;
	}

	if(ESP.getFreeHeap() < 2048)
	{
#ifdef DEBUG
		Serial.println("Dropping MQTT message, out of memory");
#endif
		writeEvent("ERRO", "mqtt", "Dropping MQTT message, out of memory", "");
		return;
	}

	MqttMessage* incomingMessage = new MqttMessage;
	if (mqttIncomingJson.containsKey("cmd")) {
		strlcpy(incomingMessage->command, mqttIncomingJson["cmd"], sizeof(incomingMessage->command));
	} else {
#ifdef DEBUG
		Serial.println("[ INFO ] No cmd key in MQTT message");
#endif
		return;
	}
	if (mqttIncomingJson.containsKey("uid")) {
		strlcpy(incomingMessage->uid, mqttIncomingJson["uid"], sizeof(incomingMessage->uid));
	}
	if (mqttIncomingJson.containsKey("user")) {
		strlcpy(incomingMessage->user, mqttIncomingJson["user"], sizeof(incomingMessage->user));
	}
	if (mqttIncomingJson.containsKey("door")) {
		strlcpy(incomingMessage->door, mqttIncomingJson["door"], sizeof(incomingMessage->door));
	}
	strlcpy(incomingMessage->serializedMessage, mqttBuffer, sizeof(mqttBuffer));

	MqttMessage* lastMessage = mqttMessageQueue;
	if(lastMessage == NULL)
	{
		mqttMessageQueue = incomingMessage;
	}
	else {
		while(lastMessage->nextMessage != NULL)
		{
			lastMessage = lastMessage->nextMessage;
		}
		lastMessage->nextMessage = incomingMessage;
	}
}

void processMqttMessage(MqttMessage *incomingMessage)
{
	char *command = incomingMessage->command;
	if (strcmp(command, "getuserlist") == 0)
	{
#ifdef DEBUG
		Serial.println("[ INFO ] Get User List");
#endif
		getUserList();
	}
	else if (strcmp(command, "opendoor") == 0)
	{
#ifdef DEBUG
		Serial.println("[ INFO ] Door open");
#endif
		writeLatest(" ", "MQTT", 1);
		mqttPublishAccess(epoch, "true", "Always", "MQTT", " ");
		for (int currentRelay = 0; currentRelay < config.numRelays; currentRelay++)
		{
			activateRelay[currentRelay] = true;
		}
		previousMillis = millis();
	}
	else if (strcmp(command, "open") == 0)
	{
#ifdef DEBUG
		Serial.println("[ INFO ] Door open");
#endif
		writeLatest(" ", "MQTT", 1);
		mqttPublishAccess(epoch, "true", "Always", "MQTT", " ");
		const char *door = incomingMessage->door;
		String stringDoor = String(door);
		int currentRelay = stringDoor.toInt();
		if (digitalRead(config.relayPin[currentRelay]) == !config.relayType[currentRelay])
		{
			activateRelay[currentRelay] = true;
		}
		previousMillis = millis();
	}
	else if (strcmp(command, "close") == 0)
	{
#ifdef DEBUG
		Serial.println("[ INFO ] Door close");
#endif
		writeLatest(" ", "MQTT", 1);
		mqttPublishAccess(epoch, "true", "Always", "MQTT", " ");
		const char *door = incomingMessage->door;
		String stringDoor = String(door);
		int currentRelay = stringDoor.toInt();
		if (digitalRead(config.relayPin[currentRelay]) == config.relayType[currentRelay])
		{
			activateRelay[currentRelay] = true;
		}
		previousMillis = millis();
	}
	else if (strcmp(command, "deletusers") == 0)
	{
#ifdef DEBUG
		Serial.println("[ INFO ] Delete all users");
#endif
		deleteAllUserFiles();
		mqttPublishAck(command);
	}
	else if (strcmp(command, "deletuid") == 0)
	{
#ifdef DEBUG
		Serial.println("[ INFO ] Delete a single user by uid");
#endif
		const char *uid = incomingMessage->uid;
		deleteUserID(uid);
		mqttPublishAck(command);
	}
	else if (strcmp(command, "adduser") == 0)
	{
#ifdef DEBUG
		Serial.print("[ INFO ] Add Users: ");
		const char *name = incomingMessage->user;
		Serial.println(name);
#endif
		const char *uid = incomingMessage->uid;
		String filename = "/P/";
		filename += uid;
		File f = SPIFFS.open(filename, "w+");
		// Check if we created the file
		if (f)
		{
			f.println(incomingMessage->serializedMessage);
		}
		f.close();
		mqttPublishAck(command);
	}
	else if (strcmp(command, "deletlog") == 0)
	{
		logMaintenance("delete", "/latestlog.json.1", nullptr);
		logMaintenance("delete", "/eventlog.json", nullptr);
		logMaintenance("rollover", "/latestlog.json", nullptr);
		mqttPublishAck(command);
	}
	else if (strcmp(command, "getconf") == 0)
	{
#ifdef DEBUG
		Serial.println("[ INFO ] Get configuration");
#endif
		File f = SPIFFS.open("/config.json", "r");
		char *buf;
		if (f)
		{
			int fileSize = f.size();
			buf = (char *)malloc(fileSize + 1);
			f.readBytes(buf, fileSize);
			f.close();
			buf[fileSize] = '\0';

			DynamicJsonDocument root(2048);
			root["type"] = "getconf";
			root["ip"] = WiFi.localIP().toString();
			root["hostname"] = config.deviceHostname;
			DynamicJsonDocument configFile(2048);
			deserializeJson(configFile, buf, fileSize + 1);
			root["configfile"] = configFile;
			mqttPublishEvent(&root);
			free(buf);
		}
	}
	else if (strcmp(command, "updateconf") == 0)
	{
#ifdef DEBUG
		Serial.println("[ INFO ] Update configuration");
#endif
		DynamicJsonDocument mqttIncomingJson(2048);
		auto error = deserializeJson(mqttIncomingJson, incomingMessage->serializedMessage);
		if (error)
		{
#ifdef DEBUG
			Serial.println("[ INFO ] Failed parsing MQTT message!!!");
#endif
			return;
		}
		File f = SPIFFS.open("/config.json", "w");
		if (f)
		{
			serializeJsonPretty(mqttIncomingJson["configfile"], f);
			f.close();
			mqttPublishAck("updateconf");
			shouldReboot = true;
		}
	}

	free(incomingMessage);
	return;
}

void onMqttSubscribe(uint16_t packetId, uint8_t qos)
{
#ifdef DEBUG
	Serial.println("[ INFO ] Subscribe acknowledged.");
	Serial.print("[ INFO ] packetId: ");
	Serial.println(packetId);
	Serial.print("[ INFO ] qos: ");
	Serial.println(qos);
#endif
}

void onMqttConnect(bool sessionPresent)
{
#ifdef DEBUG
	Serial.println("[ INFO ] MQTT Connected session");
#endif
	if (sessionPresent == true)
	{
#ifdef DEBUG
		Serial.println("[ INFO ] MQTT session Present: True");
#endif
		writeEvent("INFO", "mqtt", "Connected to MQTT Server", "Session Present");
	}
	mqttPublishBoot(epoch);

	String stopic(config.mqttTopic);
	stopic = stopic + "/cmd";
	mqttClient.subscribe(stopic.c_str(), 2);

	if (config.mqttHA)
	{
		mqttPublishDiscovery();
		mqttPublishAvty();
		if (config.doorstatpin != 255)
		{
			mqttPublishIo("door", digitalRead(config.doorstatpin) == HIGH ? "OFF" : "ON");
		}
		if (config.doorbellpin != 255)
		{
			mqttPublishIo("doorbell", digitalRead(config.doorbellpin) == HIGH ? "ON" : "OFF");
		}
		for (int i = 0; i < config.numRelays; i++)
		{
			mqttPublishIo("lock" + String(i), digitalRead(config.relayPin[i]) == config.relayType[i] ? "UNLOCKED" : "LOCKED");
		}
		mqttPublishIo("tamper", "OFF");
	}
}

void processMqttQueue()
{
	while (mqttMessageQueue != NULL)
	{
		MqttMessage *messageToProcess = mqttMessageQueue;
		mqttMessageQueue = messageToProcess->nextMessage;
		processMqttMessage(messageToProcess);
	}
}

void setupMqtt()
{
  if (!config.mqttEnabled)
	{
		return;
	}

#ifdef DEBUG
	Serial.println("[ INFO ] Trying to setup MQTT");
#endif
	if (config.mqttHA)
	{
		String stopic(config.mqttTopic);
		String topicString = stopic + "/avty";
		String payloadString = "offline";
		char *topicLWT = strdup(topicString.c_str());
		char *payloadLWT = strdup(payloadString.c_str());
		mqttClient.setWill(topicLWT, 2, true, payloadLWT);
	}
	mqttClient.setServer(config.mqttHost, config.mqttPort);
	mqttClient.setCredentials(config.mqttUser, config.mqttPass);
	mqttClient.onDisconnect(onMqttDisconnect);
	mqttClient.onPublish(onMqttPublish);
	mqttClient.onSubscribe(onMqttSubscribe);
	mqttClient.onConnect(onMqttConnect);
	mqttClient.onMessage(onMqttMessage);
}