diff --git a/CMakeLists.txt b/CMakeLists.txt index 128f58b1..f7fcde1f 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -197,6 +197,7 @@ file(COPY ${CMAKE_SOURCE_DIR}/default_plugins.json DESTINATION ${CMAKE_BINARY_DI add_subdirectory(plugins/modbus) add_subdirectory(plugins/mqtt) +add_subdirectory(plugins/thingsneural) add_subdirectory(plugins/ekuiper) add_subdirectory(plugins/file) add_subdirectory(plugins/monitor) diff --git a/default_plugins.json b/default_plugins.json index 9d03c259..a821aaeb 100644 --- a/default_plugins.json +++ b/default_plugins.json @@ -1,6 +1,7 @@ { "plugins": [ "libplugin-mqtt.so", + "libplugin-thingsneural.so", "libplugin-aws-iot.so", "libplugin-azure-iot.so", "libplugin-ekuiper.so", diff --git a/plugins/restful/adapter_handle.c b/plugins/restful/adapter_handle.c index 55edbc32..b93014d2 100644 --- a/plugins/restful/adapter_handle.c +++ b/plugins/restful/adapter_handle.c @@ -64,9 +64,9 @@ static bool tags_check_symbol(const char *tags) { for (size_t i = 0; i < strlen(tags); i++) { char c = tags[i]; - if (c == '[' || c == ']' || c == '{' || c == '}' || c == ';' || - c == ':' || c == '\'' || c == '"' || c == '<' || c == '>' || - c == '?' || c == '\\' || c == '|' || c == '`' || c == '\t' || + if (c == '[' || c == ']' || c == '{' || c == '}' || c == ';' || + c == ':' || c == '\'' || c == '"' || c == '<' || c == '>' || + c == '?' || c == '\\' || c == '|' || c == '`' || c == '\t' || c == '\n') { return false; } diff --git a/plugins/restful/normal_handle.c b/plugins/restful/normal_handle.c index ebcb68fd..9e67f5ed 100644 --- a/plugins/restful/normal_handle.c +++ b/plugins/restful/normal_handle.c @@ -479,6 +479,10 @@ static struct { .plugin = "MQTT", .schema = "mqtt", }, + { + .plugin = "ThingsNeural", + .schema = "thingsneural", + }, { .plugin = "NON A11", .schema = "nona11", diff --git a/plugins/thingsneural/CMakeLists.txt b/plugins/thingsneural/CMakeLists.txt new file mode 100644 index 00000000..cdf7c37c --- /dev/null +++ b/plugins/thingsneural/CMakeLists.txt @@ -0,0 +1,23 @@ +cmake_minimum_required(VERSION 3.12) + +project(plugin-thingsneural) + +set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins") + +file(COPY ${CMAKE_SOURCE_DIR}/plugins/thingsneural/thingsneural.json + DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/) + +add_library(${PROJECT_NAME} SHARED + thingsneural_config.c + thingsneural_handle.c + thingsneural_plugin.c + thingsneural_plugin_intf.c +) + +target_include_directories(${PROJECT_NAME} PRIVATE + ${CMAKE_SOURCE_DIR}/include/neuron + ${CMAKE_SOURCE_DIR}/plugins/thingsneural +) + +target_link_libraries(${PROJECT_NAME} neuron-base) +target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT}) diff --git a/plugins/thingsneural/thingsneural.json b/plugins/thingsneural/thingsneural.json new file mode 100644 index 00000000..3b2bb4d4 --- /dev/null +++ b/plugins/thingsneural/thingsneural.json @@ -0,0 +1,137 @@ +{ + "version": { + "name": "MQTT Version", + "name_zh": "MQTT 版本", + "type": "map", + "attribute": "optional", + "default": 4, + "valid": { + "map": [ + { + "key": "3.1.1", + "value": 4 + }, + { + "key": "5.0", + "value": 5 + } + ] + } + }, + "client-id": { + "name": "Client ID", + "name_zh": "客户端 ID", + "type": "string", + "attribute": "required", + "default": "neuron_${random_str}", + "valid": { + "length": 255 + } + }, + "qos": { + "name": "QoS Level", + "name_zh": "QoS 等级", + "type": "map", + "attribute": "optional", + "default": 0, + "valid": { + "map": [ + { + "key": "QoS 0", + "value": 0 + }, + { + "key": "QoS 1", + "value": 1 + }, + { + "key": "QoS 2", + "value": 2 + } + ] + } + }, + "host": { + "name": "Broker Host", + "name_zh": "服务器地址", + "attribute": "required", + "type": "string", + "default": "127.0.0.1", + "valid": { + "length": 255 + } + }, + "port": { + "name": "Broker Port", + "name_zh": "服务器端口", + "attribute": "required", + "type": "int", + "default": 1883, + "valid": { + "min": 1, + "max": 65535 + } + }, + "username": { + "name": "Username", + "name_zh": "用户名", + "attribute": "optional", + "type": "string", + "default": "", + "valid": { + "length": 255 + } + }, + "password": { + "name": "Password", + "name_zh": "密码", + "attribute": "optional", + "type": "string", + "default": "", + "valid": { + "length": 255 + } + }, + "ssl": { + "name": "SSL", + "name_zh": "SSL", + "attribute": "optional", + "type": "bool", + "default": false, + "valid": {} + }, + "product-key": { + "name": "Product Key", + "name_zh": "产品标识", + "description": "ThingsNeural productKey used in NLINK topics.", + "description_zh": "拼 NLINK 主题使用的 productKey。", + "attribute": "required", + "type": "string", + "valid": { + "length": 128 + } + }, + "device-key": { + "name": "Device Key", + "name_zh": "设备标识", + "description": "ThingsNeural deviceKey used in NLINK topics.", + "description_zh": "拼 NLINK 主题使用的 deviceKey。", + "attribute": "required", + "type": "string", + "valid": { + "length": 128 + } + }, + "points": { + "name": "Point Map", + "name_zh": "点位映射", + "description": "JSON array string. Each item maps a thing-model property to a south node, group and tag.", + "description_zh": "JSON 数组字符串。每一项把物模型属性映射到南向 node、group 和 tag。", + "attribute": "required", + "type": "string", + "default": "[]", + "valid": { + "length": 81960 + } + } +} diff --git a/plugins/thingsneural/thingsneural_config.c b/plugins/thingsneural/thingsneural_config.c new file mode 100644 index 00000000..124be75e --- /dev/null +++ b/plugins/thingsneural/thingsneural_config.c @@ -0,0 +1,279 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2026 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + **/ + +#include +#include + +#include "utils/asprintf.h" + +#include "thingsneural_plugin.h" + +static int copy_name(char *dst, size_t dst_len, const char *src) +{ + size_t len = 0; + + if (NULL == src || '\0' == src[0]) { + return -1; + } + len = strlen(src); + if (len >= dst_len) { + return -1; + } + memcpy(dst, src, len + 1); + return 0; +} + +static char *build_topic(const char *suffix, const char *product_key, + const char *device_key) +{ + char *topic = NULL; + if (neu_asprintf(&topic, "/iot/thing/%s/%s/property/%s", product_key, + device_key, suffix) <= 0) { + return NULL; + } + return topic; +} + +static void points_free(tn_point_t **points) +{ + tn_point_t *point = NULL; + tn_point_t *tmp = NULL; + + HASH_ITER(hh, *points, point, tmp) + { + HASH_DEL(*points, point); + free(point); + } +} + +void tn_points_free(tn_point_t **points) +{ + points_free(points); +} + +void tn_points_rename_node(tn_point_t *points, const char *node, + const char *new_name) +{ + tn_point_t *point = NULL; + tn_point_t *tmp = NULL; + + HASH_ITER(hh, points, point, tmp) + { + if (0 == strcmp(point->node, node)) { + snprintf(point->node, sizeof(point->node), "%s", new_name); + } + } +} + +void tn_points_rename_group(tn_point_t *points, const char *node, + const char *group, const char *new_name) +{ + tn_point_t *point = NULL; + tn_point_t *tmp = NULL; + + HASH_ITER(hh, points, point, tmp) + { + if (0 == strcmp(point->node, node) && + 0 == strcmp(point->group, group)) { + snprintf(point->group, sizeof(point->group), "%s", new_name); + } + } +} + +static int parse_points(neu_plugin_t *plugin, const char *text, + tn_point_t **points) +{ + json_error_t error; + json_t * root = NULL; + size_t index = 0; + json_t * item = NULL; + int rv = 0; + + *points = NULL; + if (NULL == text || '\0' == text[0]) { + return 0; + } + + root = json_loads(text, 0, &error); + if (NULL == root || !json_is_array(root)) { + plog_error(plugin, "points is not a JSON array"); + json_decref(root); + return -1; + } + + json_array_foreach(root, index, item) + { + tn_point_t *point = NULL; + tn_point_t *existing = NULL; + const char *property = NULL; + const char *node = NULL; + const char *group = NULL; + const char *tag = NULL; + + if (!json_is_object(item)) { + plog_error(plugin, "points[%zu] is not an object", index); + rv = -1; + break; + } + property = json_string_value(json_object_get(item, "property")); + node = json_string_value(json_object_get(item, "node")); + group = json_string_value(json_object_get(item, "group")); + tag = json_string_value(json_object_get(item, "tag")); + + point = calloc(1, sizeof(*point)); + if (NULL == point || + 0 != + copy_name(point->property, sizeof(point->property), property) || + 0 != copy_name(point->node, sizeof(point->node), node) || + 0 != copy_name(point->group, sizeof(point->group), group) || + 0 != copy_name(point->tag, sizeof(point->tag), tag)) { + plog_error(plugin, "points[%zu] missing property/node/group/tag", + index); + free(point); + rv = -1; + break; + } + + HASH_FIND_STR(*points, point->property, existing); + if (NULL != existing) { + plog_error(plugin, "duplicate property:%s", point->property); + free(point); + rv = -1; + break; + } + HASH_ADD_STR(*points, property, point); + } + + json_decref(root); + if (0 != rv) { + points_free(points); + } + return rv; +} + +static void take_optional_string(neu_json_elem_t *elem, char **out) +{ + if (NULL == elem->v.val_str || '\0' == elem->v.val_str[0]) { + free(elem->v.val_str); + elem->v.val_str = NULL; + return; + } + *out = elem->v.val_str; + elem->v.val_str = NULL; +} + +int tn_config_parse(neu_plugin_t *plugin, const char *setting, + tn_config_t *config, tn_point_t **points) +{ + int rv = 0; + neu_json_elem_t version = { .name = "version", + .t = NEU_JSON_INT, + .v.val_int = 4, + .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL }; + neu_json_elem_t qos = { .name = "qos", + .t = NEU_JSON_INT, + .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL }; + neu_json_elem_t client_id = { .name = "client-id", .t = NEU_JSON_STR }; + neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR }; + neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT }; + neu_json_elem_t username = { .name = "username", + .t = NEU_JSON_STR, + .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL }; + neu_json_elem_t password = { .name = "password", + .t = NEU_JSON_STR, + .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL }; + neu_json_elem_t ssl = { .name = "ssl", + .t = NEU_JSON_BOOL, + .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL }; + neu_json_elem_t product = { .name = "product-key", .t = NEU_JSON_STR }; + neu_json_elem_t device = { .name = "device-key", .t = NEU_JSON_STR }; + neu_json_elem_t points_el = { .name = "points", .t = NEU_JSON_STR }; + + memset(config, 0, sizeof(*config)); + *points = NULL; + + if (0 != + neu_parse_param(setting, NULL, 9, &version, &qos, &client_id, &host, + &port, &product, &device, &points_el, &ssl)) { + plog_error(plugin, "parse required setting fail"); + rv = -1; + goto end; + } + neu_parse_param(setting, NULL, 2, &username, &password); + + if (port.v.val_int <= 0 || port.v.val_int > 65535 || qos.v.val_int < 0 || + qos.v.val_int > 2 || + (version.v.val_int != NEU_MQTT_VERSION_V311 && + version.v.val_int != NEU_MQTT_VERSION_V5)) { + plog_error(plugin, "invalid port, qos or mqtt version"); + rv = -1; + goto end; + } + + config->version = version.v.val_int; + config->qos = qos.v.val_int; + config->port = (uint16_t) port.v.val_int; + config->ssl = ssl.v.val_bool; + config->client_id = client_id.v.val_str; + config->host = host.v.val_str; + config->product_key = product.v.val_str; + config->device_key = device.v.val_str; + client_id.v.val_str = NULL; + host.v.val_str = NULL; + product.v.val_str = NULL; + device.v.val_str = NULL; + take_optional_string(&username, &config->username); + take_optional_string(&password, &config->password); + + config->report_topic = + build_topic("report/request", config->product_key, config->device_key); + config->set_req_topic = + build_topic("set/request", config->product_key, config->device_key); + config->set_resp_topic = + build_topic("set/response", config->product_key, config->device_key); + if (NULL == config->report_topic || NULL == config->set_req_topic || + NULL == config->set_resp_topic) { + rv = -1; + goto end; + } + + if (0 != parse_points(plugin, points_el.v.val_str, points)) { + rv = -1; + goto end; + } + +end: + free(client_id.v.val_str); + free(host.v.val_str); + free(username.v.val_str); + free(password.v.val_str); + free(product.v.val_str); + free(device.v.val_str); + free(points_el.v.val_str); + if (0 != rv) { + tn_config_fini(config); + points_free(points); + } + return rv; +} + +void tn_config_fini(tn_config_t *config) +{ + free(config->client_id); + free(config->host); + free(config->username); + free(config->password); + free(config->product_key); + free(config->device_key); + free(config->report_topic); + free(config->set_req_topic); + free(config->set_resp_topic); + memset(config, 0, sizeof(*config)); +} diff --git a/plugins/thingsneural/thingsneural_config.h b/plugins/thingsneural/thingsneural_config.h new file mode 100644 index 00000000..1cc0305c --- /dev/null +++ b/plugins/thingsneural/thingsneural_config.h @@ -0,0 +1,47 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2026 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + **/ + +#ifndef NEURON_PLUGIN_THINGSNEURAL_CONFIG_H +#define NEURON_PLUGIN_THINGSNEURAL_CONFIG_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include "connection/mqtt_client.h" +#include "neuron.h" + +struct tn_point; + +typedef struct { + neu_mqtt_version_e version; + neu_mqtt_qos_e qos; + uint16_t port; + bool ssl; + char * client_id; + char * host; + char * username; + char * password; + char * product_key; + char * device_key; + char * report_topic; + char * set_req_topic; + char * set_resp_topic; +} tn_config_t; + +int tn_config_parse(neu_plugin_t *plugin, const char *setting, + tn_config_t *config, struct tn_point **points); +void tn_config_fini(tn_config_t *config); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/plugins/thingsneural/thingsneural_handle.c b/plugins/thingsneural/thingsneural_handle.c new file mode 100644 index 00000000..ae81fdf3 --- /dev/null +++ b/plugins/thingsneural/thingsneural_handle.c @@ -0,0 +1,436 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2026 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + **/ + +#include +#include +#include +#include + +#include "connection/mqtt_client.h" +#include "utils/asprintf.h" +#include "utils/time.h" + +#include "thingsneural_handle.h" +#include "thingsneural_plugin.h" + +#define TN_METHOD_REPORT "nlink.thing.property.report" +#define TN_METHOD_SET "nlink.thing.property.set" +#define TN_CODE_OK 200 +#define TN_CODE_BAD 400 +#define TN_CODE_FAIL 500 + +static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic, + uint8_t *payload, uint32_t len, void *data) +{ + neu_plugin_t *plugin = data; + (void) topic; + (void) len; + + if (0 == errcode) { + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL); + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, len, NULL); + } else { + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1, + NULL); + plog_error(plugin, "publish qos:%d fail:%d", qos, errcode); + } + free(payload); +} + +static int publish_json(neu_plugin_t *plugin, char *topic, char *payload) +{ + size_t len = strlen(payload); + int rv = neu_mqtt_client_publish(plugin->client, plugin->config.qos, topic, + (uint8_t *) payload, (uint32_t) len, + plugin, publish_cb); + if (0 != rv) { + plog_error(plugin, "publish %s fail", topic); + free(payload); + return -1; + } + return 0; +} + +static char *next_message_id(neu_plugin_t *plugin) +{ + char *id = NULL; + neu_asprintf(&id, "%" PRIu64 "-%" PRIu64, (uint64_t) neu_time_ms(), + ++plugin->msg_seq); + return id; +} + +static json_t *dvalue_to_json(neu_dvalue_t *value) +{ + switch (value->type) { + case NEU_TYPE_INT8: + return json_integer(value->value.i8); + case NEU_TYPE_UINT8: + case NEU_TYPE_BIT: + return json_integer(value->value.u8); + case NEU_TYPE_INT16: + return json_integer(value->value.i16); + case NEU_TYPE_UINT16: + case NEU_TYPE_WORD: + return json_integer(value->value.u16); + case NEU_TYPE_INT32: + return json_integer(value->value.i32); + case NEU_TYPE_UINT32: + case NEU_TYPE_DWORD: + return json_integer(value->value.u32); + case NEU_TYPE_INT64: + return json_integer(value->value.i64); + case NEU_TYPE_UINT64: + case NEU_TYPE_LWORD: + return json_integer((json_int_t) value->value.u64); + case NEU_TYPE_FLOAT: + if (isnan(value->value.f32)) { + return NULL; + } + return json_real(value->value.f32); + case NEU_TYPE_DOUBLE: + if (isnan(value->value.d64)) { + return NULL; + } + return json_real(value->value.d64); + case NEU_TYPE_BOOL: + return json_boolean(value->value.boolean); + case NEU_TYPE_STRING: + return json_string(value->value.str); + default: + return NULL; + } +} + +static const tn_point_t *find_uplink(tn_point_t *points, const char *node, + const char *group, const char *tag) +{ + tn_point_t *point = NULL; + tn_point_t *tmp = NULL; + + HASH_ITER(hh, points, point, tmp) + { + if (0 == strcmp(point->node, node) && + 0 == strcmp(point->group, group) && 0 == strcmp(point->tag, tag)) { + return point; + } + } + return NULL; +} + +static int json_to_dvalue(json_t *item, neu_dvalue_t *value) +{ + memset(value, 0, sizeof(*value)); + if (json_is_integer(item)) { + value->type = NEU_TYPE_INT64; + value->value.i64 = (int64_t) json_integer_value(item); + return 0; + } + if (json_is_real(item)) { + value->type = NEU_TYPE_DOUBLE; + value->value.d64 = json_real_value(item); + return 0; + } + if (json_is_boolean(item)) { + value->type = NEU_TYPE_BOOL; + value->value.boolean = json_boolean_value(item); + return 0; + } + if (json_is_string(item)) { + value->type = NEU_TYPE_STRING; + snprintf(value->value.str, sizeof(value->value.str), "%s", + json_string_value(item)); + return 0; + } + return -1; +} + +static int write_point(neu_plugin_t *plugin, const tn_point_t *point, + json_t *item) +{ + neu_reqresp_head_t header = { 0 }; + neu_req_write_tag_t cmd = { 0 }; + + if (0 != json_to_dvalue(item, &cmd.value)) { + return -1; + } + + header.type = NEU_REQ_WRITE_TAG; + cmd.driver = strdup(point->node); + cmd.group = strdup(point->group); + cmd.tag = strdup(point->tag); + if (NULL == cmd.driver || NULL == cmd.group || NULL == cmd.tag) { + free(cmd.driver); + free(cmd.group); + free(cmd.tag); + return -1; + } + + if (0 != neu_plugin_op(plugin, header, &cmd)) { + plog_error(plugin, "write %s/%s/%s fail", point->node, point->group, + point->tag); + free(cmd.driver); + free(cmd.group); + free(cmd.tag); + return -1; + } + return 0; +} + +static void publish_set_response(neu_plugin_t *plugin, const char *id, int code) +{ + json_t *root = NULL; + char * text = NULL; + + if (NULL == plugin->client || NULL == plugin->config.set_resp_topic) { + return; + } + + root = json_pack("{s:s, s:i, s:s, s:s}", "id", id ? id : "", "code", code, + "method", TN_METHOD_SET, "version", "1.0"); + if (NULL == root) { + return; + } + text = json_dumps(root, JSON_COMPACT); + json_decref(root); + if (NULL != text) { + publish_json(plugin, plugin->config.set_resp_topic, text); + } +} + +static void on_set_message(neu_mqtt_qos_e qos, const char *topic, + const uint8_t *payload, uint32_t len, void *data, + trace_w3c_t *trace) +{ + neu_plugin_t *plugin = data; + char * text = NULL; + + (void) qos; + (void) topic; + (void) trace; + + text = calloc(1, (size_t) len + 1); + if (NULL == text) { + return; + } + memcpy(text, payload, len); + tn_handle_set_payload(plugin, text); + free(text); +} + +int tn_subscribe_downlink(neu_plugin_t *plugin) +{ + if (NULL == plugin->client || NULL == plugin->config.set_req_topic) { + return -1; + } + return neu_mqtt_client_subscribe(plugin->client, plugin->config.qos, + plugin->config.set_req_topic, plugin, + on_set_message); +} + +int tn_handle_trans_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data) +{ + json_t *values = NULL; + json_t *params = NULL; + json_t *root = NULL; + char * text = NULL; + char * id = NULL; + char time_text[32] = { 0 }; + int n_value = 0; + + if (NULL == plugin->client || + !neu_mqtt_client_is_connected(plugin->client)) { + return NEU_ERR_PLUGIN_NOT_RUNNING; + } + + values = json_object(); + if (NULL == values) { + return NEU_ERR_EINTERNAL; + } + + pthread_mutex_lock(&plugin->lock); + utarray_foreach(data->tags, neu_resp_tag_value_meta_t *, tv) + { + const tn_point_t *point = NULL; + json_t * item = NULL; + + if (tv->value.type == NEU_TYPE_ERROR) { + continue; + } + point = find_uplink(plugin->points, data->driver, data->group, tv->tag); + if (NULL == point) { + continue; + } + item = dvalue_to_json(&tv->value); + if (NULL == item) { + continue; + } + json_object_set_new(values, point->property, item); + n_value += 1; + } + pthread_mutex_unlock(&plugin->lock); + + if (0 == n_value) { + json_decref(values); + return NEU_ERR_SUCCESS; + } + + snprintf(time_text, sizeof(time_text), "%" PRId64, neu_time_ms()); + id = next_message_id(plugin); + params = json_object(); + root = json_object(); + if (NULL == params || NULL == root) { + json_decref(values); + json_decref(params); + json_decref(root); + free(id); + return NEU_ERR_EINTERNAL; + } + json_object_set_new(params, "time", json_string(time_text)); + json_object_set_new(params, "values", values); + values = NULL; + json_object_set_new(root, "id", json_string(id ? id : "")); + json_object_set_new(root, "version", json_string("1.0")); + json_object_set_new(root, "ack", json_integer(1)); + json_object_set_new(root, "method", json_string(TN_METHOD_REPORT)); + json_object_set_new(root, "params", params); + params = NULL; + free(id); + + text = json_dumps(root, JSON_COMPACT); + json_decref(root); + if (NULL == text) { + return NEU_ERR_EINTERNAL; + } + if (0 != publish_json(plugin, plugin->config.report_topic, text)) { + return NEU_ERR_MQTT_PUBLISH_FAILURE; + } + return NEU_ERR_SUCCESS; +} + +int tn_handle_set_payload(neu_plugin_t *plugin, const char *payload) +{ + json_error_t error; + json_t * root = NULL; + json_t * params = NULL; + json_t * values = NULL; + const char * id = NULL; + const char * method = NULL; + const char * key = NULL; + json_t * item = NULL; + int code = TN_CODE_OK; + int n_ok = 0; + + root = json_loads(payload, 0, &error); + if (NULL == root || !json_is_object(root)) { + json_decref(root); + return NEU_ERR_BODY_IS_WRONG; + } + + id = json_string_value(json_object_get(root, "id")); + method = json_string_value(json_object_get(root, "method")); + params = json_object_get(root, "params"); + values = json_is_object(params) ? json_object_get(params, "values") : NULL; + if ((NULL != method && 0 != strcmp(method, TN_METHOD_SET)) || + !json_is_object(values)) { + publish_set_response(plugin, id, TN_CODE_BAD); + json_decref(root); + return NEU_ERR_BODY_IS_WRONG; + } + + json_object_foreach(values, key, item) + { + tn_point_t point = { 0 }; + tn_point_t *found = NULL; + int rv = 0; + + pthread_mutex_lock(&plugin->lock); + HASH_FIND_STR(plugin->points, key, found); + if (NULL != found) { + point = *found; + } + pthread_mutex_unlock(&plugin->lock); + + if (NULL == found) { + plog_notice(plugin, "property %s has no south mapping", key); + code = TN_CODE_BAD; + continue; + } + rv = write_point(plugin, &point, item); + if (0 != rv) { + code = TN_CODE_FAIL; + continue; + } + n_ok += 1; + } + + if (0 == n_ok && code == TN_CODE_OK) { + code = TN_CODE_BAD; + } + publish_set_response(plugin, id, code); + json_decref(root); + return code == TN_CODE_OK ? NEU_ERR_SUCCESS : NEU_ERR_BODY_IS_WRONG; +} + +int tn_handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub) +{ + plog_notice(plugin, "subscribe driver:%s group:%s", sub->driver, + sub->group); + free(sub->params); + free(sub->static_tags); + return NEU_ERR_SUCCESS; +} + +int tn_handle_update_subscribe(neu_plugin_t *plugin, neu_req_subscribe_t *sub) +{ + plog_notice(plugin, "update subscribe driver:%s group:%s", sub->driver, + sub->group); + free(sub->params); + free(sub->static_tags); + return NEU_ERR_SUCCESS; +} + +int tn_handle_unsubscribe_group(neu_plugin_t * plugin, + neu_req_unsubscribe_t *unsub) +{ + plog_notice(plugin, "unsubscribe driver:%s group:%s", unsub->driver, + unsub->group); + return NEU_ERR_SUCCESS; +} + +int tn_handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req) +{ + pthread_mutex_lock(&plugin->lock); + tn_points_rename_group(plugin->points, req->driver, req->group, + req->new_name); + pthread_mutex_unlock(&plugin->lock); + return NEU_ERR_SUCCESS; +} + +int tn_handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req) +{ + (void) plugin; + (void) req; + return NEU_ERR_SUCCESS; +} + +int tn_handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req) +{ + pthread_mutex_lock(&plugin->lock); + tn_points_rename_node(plugin->points, req->node, req->new_name); + pthread_mutex_unlock(&plugin->lock); + return NEU_ERR_SUCCESS; +} + +int tn_handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req) +{ + (void) plugin; + (void) req; + return NEU_ERR_SUCCESS; +} diff --git a/plugins/thingsneural/thingsneural_handle.h b/plugins/thingsneural/thingsneural_handle.h new file mode 100644 index 00000000..fc7229d1 --- /dev/null +++ b/plugins/thingsneural/thingsneural_handle.h @@ -0,0 +1,36 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2026 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + **/ + +#ifndef NEURON_PLUGIN_THINGSNEURAL_HANDLE_H +#define NEURON_PLUGIN_THINGSNEURAL_HANDLE_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include "neuron.h" + +int tn_subscribe_downlink(neu_plugin_t *plugin); +int tn_handle_trans_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data); +int tn_handle_set_payload(neu_plugin_t *plugin, const char *payload); +int tn_handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub); +int tn_handle_update_subscribe(neu_plugin_t *plugin, neu_req_subscribe_t *sub); +int tn_handle_unsubscribe_group(neu_plugin_t * plugin, + neu_req_unsubscribe_t *unsub); +int tn_handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req); +int tn_handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req); +int tn_handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req); +int tn_handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/plugins/thingsneural/thingsneural_plugin.c b/plugins/thingsneural/thingsneural_plugin.c new file mode 100644 index 00000000..9da6557f --- /dev/null +++ b/plugins/thingsneural/thingsneural_plugin.c @@ -0,0 +1,28 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2026 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + **/ + +#include "thingsneural_plugin_intf.h" + +#define DESCRIPTION \ + "Northbound ThingsNeural plugin using NLINK property messages." +#define DESCRIPTION_ZH "按 NLINK 物模型属性报文收发的北向应用插件" + +const neu_plugin_module_t neu_plugin_module = { + .version = NEURON_PLUGIN_VER_1_0, + .schema = "thingsneural", + .module_name = "ThingsNeural", + .module_descr = DESCRIPTION, + .module_descr_zh = DESCRIPTION_ZH, + .intf_funs = &tn_plugin_intf_funs, + .kind = NEU_PLUGIN_KIND_SYSTEM, + .type = NEU_NA_TYPE_APP, + .display = true, + .single = false, +}; diff --git a/plugins/thingsneural/thingsneural_plugin.h b/plugins/thingsneural/thingsneural_plugin.h new file mode 100644 index 00000000..182e2fec --- /dev/null +++ b/plugins/thingsneural/thingsneural_plugin.h @@ -0,0 +1,51 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2026 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + **/ + +#ifndef NEURON_PLUGIN_THINGSNEURAL_H +#define NEURON_PLUGIN_THINGSNEURAL_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#include "neuron.h" + +#include "thingsneural_config.h" + +typedef struct tn_point { + char property[NEU_TAG_NAME_LEN]; + char node[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; + char tag[NEU_TAG_NAME_LEN]; + UT_hash_handle hh; +} tn_point_t; + +struct neu_plugin { + neu_plugin_common_t common; + tn_config_t config; + neu_mqtt_client_t * client; + tn_point_t * points; + pthread_mutex_t lock; + uint64_t msg_seq; +}; + +void tn_points_free(tn_point_t **points); +void tn_points_rename_node(tn_point_t *points, const char *node, + const char *new_name); +void tn_points_rename_group(tn_point_t *points, const char *node, + const char *group, const char *new_name); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/plugins/thingsneural/thingsneural_plugin_intf.c b/plugins/thingsneural/thingsneural_plugin_intf.c new file mode 100644 index 00000000..a2d2622d --- /dev/null +++ b/plugins/thingsneural/thingsneural_plugin_intf.c @@ -0,0 +1,247 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2026 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + **/ + +#include +#include + +#include "thingsneural_config.h" +#include "thingsneural_handle.h" +#include "thingsneural_plugin.h" +#include "thingsneural_plugin_intf.h" + +extern const neu_plugin_module_t neu_plugin_module; + +static void connect_cb(void *data) +{ + neu_plugin_t *plugin = data; + plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED; + plog_notice(plugin, "plugin `%s` connected", neu_plugin_module.module_name); + if (0 != tn_subscribe_downlink(plugin)) { + plog_error(plugin, "subscribe property set topic fail"); + } +} + +static void disconnect_cb(void *data) +{ + neu_plugin_t *plugin = data; + plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED; + plog_notice(plugin, "plugin `%s` disconnected", + neu_plugin_module.module_name); +} + +static int open_client(neu_plugin_t *plugin, const tn_config_t *config) +{ + int rv = 0; + neu_mqtt_client_t *client = neu_mqtt_client_new(config->version); + + if (NULL == client) { + return -1; + } + + rv = neu_mqtt_client_set_zlog_category(client, plugin->common.log); + if (0 != rv) { + goto error; + } + rv = neu_mqtt_client_set_addr(client, config->host, config->port); + if (0 != rv) { + goto error; + } + rv = neu_mqtt_client_set_id(client, config->client_id); + if (0 != rv) { + goto error; + } + rv = neu_mqtt_client_set_connect_cb(client, connect_cb, plugin); + if (0 != rv) { + goto error; + } + rv = neu_mqtt_client_set_disconnect_cb(client, disconnect_cb, plugin); + if (0 != rv) { + goto error; + } + if (NULL != config->username) { + rv = neu_mqtt_client_set_user(client, config->username, + config->password); + if (0 != rv) { + goto error; + } + } + rv = neu_mqtt_client_set_tls(client, config->ssl, NULL, NULL, NULL, NULL); + if (0 != rv) { + goto error; + } + if (0 != neu_mqtt_client_open(client)) { + goto error; + } + + plugin->client = client; + return 0; + +error: + neu_mqtt_client_free(client); + return -1; +} + +static void close_client(neu_plugin_t *plugin) +{ + if (NULL == plugin->client) { + return; + } + if (NULL != plugin->config.set_req_topic) { + neu_mqtt_client_unsubscribe(plugin->client, + plugin->config.set_req_topic); + } + neu_mqtt_client_close(plugin->client); + neu_mqtt_client_free(plugin->client); + plugin->client = NULL; + plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED; +} + +neu_plugin_t *tn_plugin_open(void) +{ + neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t)); + neu_plugin_common_init(&plugin->common); + pthread_mutex_init(&plugin->lock, NULL); + return plugin; +} + +int tn_plugin_close(neu_plugin_t *plugin) +{ + pthread_mutex_destroy(&plugin->lock); + free(plugin); + return NEU_ERR_SUCCESS; +} + +int tn_plugin_init(neu_plugin_t *plugin, bool load) +{ + (void) load; + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, 5000); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 0); + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 0); + plog_notice(plugin, "initialize plugin `%s` success", + neu_plugin_module.module_name); + return NEU_ERR_SUCCESS; +} + +int tn_plugin_uninit(neu_plugin_t *plugin) +{ + close_client(plugin); + tn_config_fini(&plugin->config); + pthread_mutex_lock(&plugin->lock); + tn_points_free(&plugin->points); + pthread_mutex_unlock(&plugin->lock); + plog_notice(plugin, "uninitialize plugin `%s` success", + neu_plugin_module.module_name); + return NEU_ERR_SUCCESS; +} + +int tn_plugin_config(neu_plugin_t *plugin, const char *setting) +{ + tn_config_t config = { 0 }; + tn_point_t *points = NULL; + + if (0 != tn_config_parse(plugin, setting, &config, &points)) { + return NEU_ERR_NODE_SETTING_INVALID; + } + + close_client(plugin); + if (0 != open_client(plugin, &config)) { + tn_config_fini(&config); + tn_points_free(&points); + return NEU_ERR_MQTT_CONNECT_FAILURE; + } + + tn_config_fini(&plugin->config); + pthread_mutex_lock(&plugin->lock); + tn_points_free(&plugin->points); + plugin->points = points; + pthread_mutex_unlock(&plugin->lock); + plugin->config = config; + + if (0 != tn_subscribe_downlink(plugin)) { + plog_error(plugin, "subscribe property set topic fail"); + } + + plog_notice(plugin, "config plugin `%s` success, report:%s", + neu_plugin_module.module_name, plugin->config.report_topic); + return NEU_ERR_SUCCESS; +} + +int tn_plugin_start(neu_plugin_t *plugin) +{ + if (NULL == plugin->client) { + return NEU_ERR_MQTT_IS_NULL; + } + if (0 != neu_mqtt_client_open(plugin->client)) { + return NEU_ERR_MQTT_CONNECT_FAILURE; + } + if (0 != tn_subscribe_downlink(plugin)) { + plog_error(plugin, "subscribe property set topic fail"); + } + plog_notice(plugin, "start plugin `%s` success", + neu_plugin_module.module_name); + return NEU_ERR_SUCCESS; +} + +int tn_plugin_stop(neu_plugin_t *plugin) +{ + if (NULL != plugin->client && NULL != plugin->config.set_req_topic) { + neu_mqtt_client_unsubscribe(plugin->client, + plugin->config.set_req_topic); + neu_mqtt_client_close(plugin->client); + } + plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED; + plog_notice(plugin, "stop plugin `%s` success", + neu_plugin_module.module_name); + return NEU_ERR_SUCCESS; +} + +static int tn_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, + void *data) +{ + switch (head->type) { + case NEU_REQRESP_TRANS_DATA: + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL); + return tn_handle_trans_data(plugin, data); + case NEU_REQ_SUBSCRIBE_GROUP: + return tn_handle_subscribe_group(plugin, data); + case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: + return tn_handle_update_subscribe(plugin, data); + case NEU_REQ_UNSUBSCRIBE_GROUP: + return tn_handle_unsubscribe_group(plugin, data); + case NEU_REQ_UPDATE_GROUP: + return tn_handle_update_group(plugin, data); + case NEU_REQ_DEL_GROUP: + return tn_handle_del_group(plugin, data); + case NEU_REQ_UPDATE_NODE: + return tn_handle_update_driver(plugin, data); + case NEU_REQRESP_NODE_DELETED: { + neu_reqresp_node_deleted_t *req = data; + if (0 != strcmp(plugin->common.name, req->node)) { + return tn_handle_del_driver(plugin, data); + } + return NEU_ERR_SUCCESS; + } + default: + return NEU_ERR_PLUGIN_NOT_RUNNING; + } +} + +const neu_plugin_intf_funs_t tn_plugin_intf_funs = { + .open = tn_plugin_open, + .close = tn_plugin_close, + .init = tn_plugin_init, + .uninit = tn_plugin_uninit, + .start = tn_plugin_start, + .stop = tn_plugin_stop, + .setting = tn_plugin_config, + .request = tn_plugin_request, + .try_connect = NULL, +}; diff --git a/plugins/thingsneural/thingsneural_plugin_intf.h b/plugins/thingsneural/thingsneural_plugin_intf.h new file mode 100644 index 00000000..7a68c3bc --- /dev/null +++ b/plugins/thingsneural/thingsneural_plugin_intf.h @@ -0,0 +1,18 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2026 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + **/ + +#ifndef NEURON_PLUGIN_THINGSNEURAL_INTF_H +#define NEURON_PLUGIN_THINGSNEURAL_INTF_H + +#include "plugin.h" + +extern const neu_plugin_intf_funs_t tn_plugin_intf_funs; + +#endif diff --git a/tests/ft/app/test_thingsneural.py b/tests/ft/app/test_thingsneural.py new file mode 100644 index 00000000..0d2e3d0f --- /dev/null +++ b/tests/ft/app/test_thingsneural.py @@ -0,0 +1,204 @@ +import json +import time + +import pytest + +import neuron.api as api +import neuron.config as config +from neuron.common import description, process, random_port +from neuron.mqtt import Mock + +APP = "tn-device" +DRIVER_A = "modbus-a" +DRIVER_B = "modbus-b" +GROUP_A = "ga" +GROUP_B = "gb" +INTERVAL = 100 +PRODUCT_KEY = "pkdemo" +DEVICE_KEY = "dkdemo" +REPORT_TOPIC = f"/iot/thing/{PRODUCT_KEY}/{DEVICE_KEY}/property/report/request" +SET_REQ_TOPIC = f"/iot/thing/{PRODUCT_KEY}/{DEVICE_KEY}/property/set/request" +SET_RESP_TOPIC = f"/iot/thing/{PRODUCT_KEY}/{DEVICE_KEY}/property/set/response" +def tag(address): + return { + "name": "tag0", + "address": address, + "attribute": config.NEU_TAG_ATTRIBUTE_RW, + "type": config.NEU_TYPE_INT16, + } +POINTS = json.dumps( + [ + { + "property": "temperature", + "node": DRIVER_A, + "group": GROUP_A, + "tag": "tag0", + }, + { + "property": "humidity", + "node": DRIVER_B, + "group": GROUP_B, + "tag": "tag0", + }, + ] +) + + +@pytest.fixture(autouse=True, scope="class") +def neuron_process(): + process.remove_persistence() + neuron = process.NeuronProcess() + neuron.start() + deadline = time.time() + 10 + while time.time() < deadline: + try: + if api.ping().status_code == 200: + break + except Exception: + time.sleep(0.2) + else: + neuron.stop() + raise AssertionError("neuron did not listen on port 7000") + yield neuron + neuron.stop() + + +@pytest.fixture(autouse=True, scope="class") +def thingsneural_node(neuron_process): + api.add_node_check(APP, "ThingsNeural") + yield + api.del_node(APP) + + +@pytest.fixture(autouse=True, scope="class") +def modbus_nodes(thingsneural_node): + port = random_port() + simulator = process.start_simulator( + ["./modbus_simulator", "tcp", f"{port}", "ip_v4"] + ) + try: + for node, group, address in ( + (DRIVER_A, GROUP_A, "1!400001"), + (DRIVER_B, GROUP_B, "1!400002"), + ): + api.add_node_check(node, "Modbus TCP") + api.add_group_check(node, group, INTERVAL) + api.add_tags_check(node, group, tags=[tag(address)]) + api.subscribe_group_check(APP, node, group, params={}) + response = api.modbus_tcp_node_setting(node, port) + assert 200 == response.status_code + yield + finally: + api.del_node(DRIVER_A) + api.del_node(DRIVER_B) + process.stop_simulator(simulator) + + +@pytest.fixture(autouse=True, scope="class") +def mocker(): + mock = Mock() + yield mock + mock.done() + + +def setting(mocker): + return { + "client-id": mocker.rand_id(), + "qos": 0, + "host": "127.0.0.1", + "port": mocker.port, + "ssl": False, + "product-key": PRODUCT_KEY, + "device-key": DEVICE_KEY, + "points": POINTS, + } + + +def collect_reports(mocker): + messages = mocker.get(REPORT_TOPIC, num=6, timeout=8) + assert messages, "no NLINK property reports" + if isinstance(messages, dict): + messages = [messages] + return messages + + +class TestThingsNeural: + @description( + given="ThingsNeural plugin is loaded", + when="list plugins", + then="ThingsNeural is visible", + ) + def test_plugin_listed(self): + response = api.get_plugins() + assert 200 == response.status_code + names = [item["name"] for item in response.json().get("plugins", [])] + assert "ThingsNeural" in names + + @description( + given="two south groups mapped to thing-model properties", + when="tags are reported", + then="payload is NLINK and keys are property identifiers", + ) + def test_uplink_nlink(self, mocker): + api.node_setting_check(APP, setting(mocker)) + api.node_ctl(APP, config.NEU_CTL_START) + mocker.assert_client_connected() + + properties = set() + for message in collect_reports(mocker): + assert "node" not in message + assert "group" not in message + assert message["version"] == "1.0" + assert message["method"] == "nlink.thing.property.report" + assert message["ack"] == 1 + values = message["params"]["values"] + assert "tag0" not in values + properties.update(values.keys()) + + assert "temperature" in properties + assert "humidity" in properties + + @description( + given="a property set without node or group", + when="temperature and an unknown property are written", + then="only the mapped south tag changes", + ) + def test_downlink_reverse_map(self, mocker): + api.node_setting_check(APP, setting(mocker)) + api.node_ctl(APP, config.NEU_CTL_START) + mocker.assert_client_connected() + + response = mocker.reqresp( + SET_RESP_TOPIC, + SET_REQ_TOPIC, + { + "id": "set-1", + "version": "1.0", + "method": "nlink.thing.property.set", + "params": {"values": {"temperature": 1234, "humidity": 56}}, + }, + timeout=3, + ) + assert response is not None + assert response["id"] == "set-1" + assert response["code"] == 200 + assert response["method"] == "nlink.thing.property.set" + + time.sleep(0.5) + assert 1234 == api.read_tag(DRIVER_A, GROUP_A, "tag0") + assert 56 == api.read_tag(DRIVER_B, GROUP_B, "tag0") + + unknown = mocker.reqresp( + SET_RESP_TOPIC, + SET_REQ_TOPIC, + { + "id": "set-unknown", + "version": "1.0", + "method": "nlink.thing.property.set", + "params": {"values": {"pressure": 1}}, + }, + timeout=3, + ) + assert unknown is not None + assert unknown["code"] == 400 + assert 1234 == api.read_tag(DRIVER_A, GROUP_A, "tag0")