GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-29 15:09:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat:add modbus-tcp simulator

Browse Source
This commit is contained in:
hxy7yx
2025-11-24 02:09:43 -08:00
parent 811ba04408
commit c9bf291fa2
15 changed files with 1733 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+3 -1
View File
@@ -169,7 +169,9 @@ set(NEURON_SOURCES
plugins/restful/system_handle.c
plugins/restful/datalayers_handle.c
plugins/restful/rest.c
plugins/restful/user.c)
plugins/restful/user.c
plugins/restful/simulator_handle.c
simulator/modbus_tcp_simulator.c)
set(CMAKE_BUILD_RPATH ./)
add_executable(neuron)
include/neuron/errcodes.h
+1
View File
@@ -53,6 +53,7 @@ typedef enum {
NEU_ERR_USER_NOT_EXISTS = 1020,
NEU_ERR_INVALID_USER_LEN = 1021,
NEU_ERR_USER_NO_PERMISSION = 1022,
NEU_ERR_PORT_IN_USE = 1023,
NEU_ERR_NODE_EXIST = 2002,
NEU_ERR_NODE_NOT_EXIST = 2003,
include/neuron/modbus_tcp_simulator.h
+58
View File
@@ -0,0 +1,58 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2025 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.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#ifndef NEU_MODBUS_TCP_SIMULATOR_H
#define NEU_MODBUS_TCP_SIMULATOR_H
#include <stdbool.h>
#include <stdint.h>
int neu_modbus_simulator_init(const char *config_dir);
void neu_modbus_simulator_fini(void);
int neu_modbus_simulator_start(const char *ip, uint16_t port);
void neu_modbus_simulator_stop(void);
void neu_modbus_simulator_apply_persist(void);
typedef enum {
NEU_SIM_TAG_NONE = 0,
NEU_SIM_TAG_SINE = 1, // float, [-100,100], period 60s
NEU_SIM_TAG_SAW = 2, // int16, [0,100], period 100s
NEU_SIM_TAG_SQUARE = 3, // int16, [-10,10], period 10s
NEU_SIM_TAG_RANDOM = 4 // int16, [0,100]
} neu_sim_tag_type_e;
int neu_modbus_simulator_config_tags(const char **names, const char **addr_strs,
const uint16_t * addresses,
const neu_sim_tag_type_e *types,
int tag_count);
char *neu_modbus_simulator_export_drivers_json(void);
char *neu_modbus_simulator_list_tags_json(void);
typedef struct {
bool running;
char ip[64];
uint16_t port;
int tag_count;
int error;
} neu_modbus_simulator_status_t;
void neu_modbus_simulator_get_status(neu_modbus_simulator_status_t *out);
#endif
neuron.json
+3 -2
View File
@@ -4,5 +4,6 @@
"disable_auth": 0,
"syslog_host": "",
"syslog_port": 541,
"sub_filter_error": 0
}
"sub_filter_error": 0,
"modbus_simulator": { "port":502 }
}
persistence/0114_2.14.0_modbus_simulator.sql
+6
View File
@@ -0,0 +1,6 @@
CREATE TABLE IF NOT EXISTS modbus_tcp_simulator (
id INTEGER PRIMARY KEY CHECK (id = 1),
enabled INTEGER NOT NULL DEFAULT 0,
tags_json TEXT NOT NULL DEFAULT '[]',
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
plugins/restful/handle.c
+55
View File
@@ -44,6 +44,7 @@
#include "version_handle.h"
#include "handle.h"
#include "simulator_handle.h"
struct neu_rest_handle_ctx {
void *plugin;
@@ -190,6 +191,24 @@ static struct neu_http_handler cors_handler[] = {
{
.url = "/api/v2/auth/basic/security_policies",
},
{
.url = "/api/v2/simulator/status",
},
{
.url = "/api/v2/simulator/start",
},
{
.url = "/api/v2/simulator/stop",
},
{
.url = "/api/v2/simulator/config",
},
{
.url = "/api/v2/simulator/tags",
},
{
.url = "/api/v2/simulator/export",
},
};
static struct neu_http_handler rest_handlers[] = {
@@ -686,6 +705,42 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/auth/basic/security_policies",
.value.handler = handle_server_security_policy,
},
{
.method = NEU_HTTP_METHOD_GET,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/simulator/status",
.value.handler = handle_simulator_status,
},
{
.method = NEU_HTTP_METHOD_GET,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/simulator/tags",
.value.handler = handle_simulator_list_tags,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/simulator/start",
.value.handler = handle_simulator_start,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/simulator/stop",
.value.handler = handle_simulator_stop,
},
{
.method = NEU_HTTP_METHOD_POST,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/simulator/config",
.value.handler = handle_simulator_set_config,
},
{
.method = NEU_HTTP_METHOD_GET,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/simulator/export",
.value.handler = handle_simulator_export,
},
};
void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size)
plugins/restful/simulator_handle.c
+217
View File
@@ -0,0 +1,217 @@
#include <ctype.h>
#include <jansson.h>
#include <stdlib.h>
#include <string.h>
#include "handle.h"
#include "modbus_tcp_simulator.h"
#include "utils/http.h"
#include "utils/log.h"
#include "json/neu_json_fn.h"
static int extract_hold_start_from_addr(const char *addr_str,
uint16_t * out_idx0)
{
if (!addr_str || !out_idx0)
return -1;
const char *p = strstr(addr_str, "!4");
if (!p)
return -1;
p += 2;
if (!isdigit((unsigned char) *p))
return -1;
long v = strtol(p, NULL, 10);
if (v <= 0 || v > 65535)
return -1;
*out_idx0 = (uint16_t)(v - 1);
return 0;
}
void handle_simulator_status(nng_aio *aio)
{
NEU_VALIDATE_JWT(aio);
neu_modbus_simulator_status_t st = { 0 };
neu_modbus_simulator_get_status(&st);
char buf[256] = { 0 };
snprintf(buf, sizeof(buf),
"{\"running\":%d,\"port\":%u,\"tag_count\":%d,\"error\":%d}",
st.running ? 1 : 0, st.port, st.tag_count, st.error);
neu_http_ok(aio, buf);
}
void handle_simulator_start(nng_aio *aio)
{
NEU_VALIDATE_JWT(aio);
int rc = neu_modbus_simulator_start(NULL, 0);
neu_modbus_simulator_status_t st = { 0 };
neu_modbus_simulator_get_status(&st);
if (rc != 0 || (!st.running && st.error != 0)) {
char buf[64] = { 0 };
snprintf(buf, sizeof(buf), "{\"error\":%d}", st.error);
neu_http_ok(aio, buf);
return;
}
neu_http_ok(aio, "{\"error\":0}");
}
void handle_simulator_list_tags(nng_aio *aio)
{
NEU_VALIDATE_JWT(aio);
char *out = neu_modbus_simulator_list_tags_json();
if (!out) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, NEU_ERR_EINTERNAL, result_error);
});
return;
}
neu_http_ok(aio, out);
free(out);
}
void handle_simulator_stop(nng_aio *aio)
{
NEU_VALIDATE_JWT(aio);
neu_modbus_simulator_stop();
neu_http_ok(aio, "{\"error\":0}");
}
void handle_simulator_set_config(nng_aio *aio)
{
NEU_VALIDATE_JWT(aio);
char * body = NULL;
size_t sz = 0;
if (neu_http_get_body(aio, (void **) &body, &sz) != 0 || !body) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, NEU_ERR_PARAM_IS_WRONG, result_error);
});
return;
}
json_error_t jerr;
json_t * root = json_loads(body, 0, &jerr);
free(body);
if (!root) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, NEU_ERR_PARAM_IS_WRONG, result_error);
});
return;
}
json_t *tags = json_object_get(root, "tags");
if (!tags || !json_is_array(tags)) {
json_decref(root);
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, NEU_ERR_PARAM_IS_WRONG, result_error);
});
return;
}
int count = 0;
for (size_t i = 0; i < json_array_size(tags); ++i) {
json_t *obj = json_array_get(tags, i);
if (!obj || !json_is_object(obj))
continue;
json_t *jaddr = json_object_get(obj, "address");
json_t *jtype = json_object_get(obj, "type");
if (!jaddr || !json_is_string(jaddr) || !jtype)
continue;
const char *addr_str = json_string_value(jaddr);
uint16_t idx0 = 0;
if (extract_hold_start_from_addr(addr_str, &idx0) != 0)
continue;
count++;
}
char ** names = NULL;
char ** addresses = NULL;
uint16_t * indices = NULL;
neu_sim_tag_type_e *types = NULL;
if (count > 0) {
names = calloc((size_t) count, sizeof(char *));
addresses = calloc((size_t) count, sizeof(char *));
indices = calloc((size_t) count, sizeof(uint16_t));
types = calloc((size_t) count, sizeof(neu_sim_tag_type_e));
if (!names || !addresses || !indices || !types) {
free(names);
free(addresses);
free(indices);
free(types);
json_decref(root);
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, NEU_ERR_EINTERNAL, result_error);
});
return;
}
}
int m = 0;
for (size_t i = 0; i < json_array_size(tags); ++i) {
json_t *obj = json_array_get(tags, i);
if (!obj || !json_is_object(obj))
continue;
json_t *jname = json_object_get(obj, "name");
json_t *jaddr = json_object_get(obj, "address");
json_t *jtype = json_object_get(obj, "type");
if (!jaddr || !json_is_string(jaddr) || !jtype)
continue;
const char *addr_str = json_string_value(jaddr);
uint16_t idx0 = 0;
if (extract_hold_start_from_addr(addr_str, &idx0) != 0)
continue;
neu_sim_tag_type_e tp = NEU_SIM_TAG_NONE;
if (json_is_string(jtype)) {
const char *ts = json_string_value(jtype);
if (ts) {
if (strcmp(ts, "sine") == 0)
tp = NEU_SIM_TAG_SINE;
else if (strcmp(ts, "saw") == 0)
tp = NEU_SIM_TAG_SAW;
else if (strcmp(ts, "square") == 0)
tp = NEU_SIM_TAG_SQUARE;
else if (strcmp(ts, "random") == 0)
tp = NEU_SIM_TAG_RANDOM;
}
} else if (json_is_integer(jtype)) {
tp = (neu_sim_tag_type_e) json_integer_value(jtype);
}
if (tp == NEU_SIM_TAG_NONE || idx0 >= 1000)
continue;
indices[m] = idx0;
types[m] = tp;
names[m] = jname && json_is_string(jname)
? strdup(json_string_value(jname))
: strdup("");
addresses[m] = strdup(addr_str);
m++;
}
json_decref(root);
neu_modbus_simulator_config_tags(
(const char **) names, (const char **) addresses, indices, types, m);
for (int i = 0; i < m; ++i) {
free(names[i]);
free(addresses[i]);
}
free(names);
free(addresses);
free(indices);
free(types);
neu_http_ok(aio, "{\"error\":0}");
}
void handle_simulator_export(nng_aio *aio)
{
NEU_VALIDATE_JWT(aio);
char *drivers = neu_modbus_simulator_export_drivers_json();
if (!drivers) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, NEU_ERR_EINTERNAL, result_error);
});
return;
}
neu_http_response_file(
aio, drivers, strlen(drivers),
"attachment; filename=\"simulator_modbus_south.json\"");
free(drivers);
}
plugins/restful/simulator_handle.h
+13
View File
@@ -0,0 +1,13 @@
#ifndef NEU_PLUGIN_REST_SIMULATOR_HANDLE_H
#define NEU_PLUGIN_REST_SIMULATOR_HANDLE_H
#include <nng/nng.h>
void handle_simulator_status(nng_aio *aio);
void handle_simulator_start(nng_aio *aio);
void handle_simulator_stop(nng_aio *aio);
void handle_simulator_set_config(nng_aio *aio);
void handle_simulator_export(nng_aio *aio);
void handle_simulator_list_tags(nng_aio *aio);
#endif
simulator/modbus_tcp_simulator.c
+1191
View File
File diff suppressed because it is too large Load Diff
src/main.c
+8
View File
@@ -28,6 +28,7 @@
#include <unistd.h>
#include "core/manager.h"
#include "modbus_tcp_simulator.h"
#include "utils/log.h"
#include "utils/time.h"
@@ -143,6 +144,8 @@ static int neuron_run(const neu_cli_args_t *args)
rv = neu_persister_create(args->config_dir);
assert(rv == 0);
neu_modbus_simulator_apply_persist();
zlog_notice(neuron, "neuron start, daemon: %d, version: %s (%s %s)",
args->daemonized, NEURON_VERSION,
NEURON_GIT_REV NEURON_GIT_DIFF, NEURON_BUILD_DATE);
@@ -190,6 +193,10 @@ int main(int argc, char *argv[])
neuron = zlog_get_category("neuron");
zlog_level_switch(neuron, default_log_level);
if (neu_modbus_simulator_init(args.config_dir) != 0) {
nlog_warn("modbus simulator init failed");
}
if (neuron_already_running()) {
if (args.stop) {
rv = neuron_stop();
@@ -242,6 +249,7 @@ int main(int argc, char *argv[])
main_end:
neu_cli_args_fini(&args);
neu_modbus_simulator_fini();
zlog_fini();
return rv;
}
src/utils/http.c
+1
View File
@@ -339,6 +339,7 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_USER_NOT_EXISTS:
case NEU_ERR_USER_NO_PERMISSION:
case NEU_ERR_INVALID_USER_LEN:
case NEU_ERR_PORT_IN_USE:
status = NNG_HTTP_STATUS_BAD_REQUEST;
break;
case NEU_ERR_FILE_NOT_EXIST:
tests/ft/neuron/api.py
+34
View File
@@ -544,3 +544,37 @@ def server_security_policy(app_name, policy_name, jwt=config.default_jwt):
return requests.post(url=config.BASE_URL + "/api/v2/auth/basic/security_policies",
headers={"Authorization": jwt},
json={"node": app_name, "policyName": policy_name})
def simulator_status(jwt=config.default_jwt):
return requests.get(url=config.BASE_URL + "/api/v2/simulator/status",
headers={"Authorization": jwt})
@gen_check
def simulator_start(jwt=config.default_jwt):
return requests.post(url=config.BASE_URL + "/api/v2/simulator/start",
headers={"Authorization": jwt})
@gen_check
def simulator_stop(jwt=config.default_jwt):
return requests.post(url=config.BASE_URL + "/api/v2/simulator/stop",
headers={"Authorization": jwt})
@gen_check
def simulator_set_config(tags, jwt=config.default_jwt):
return requests.post(url=config.BASE_URL + "/api/v2/simulator/config",
headers={"Authorization": jwt},
json={"tags": tags})
def simulator_export(jwt=config.default_jwt):
return requests.get(url=config.BASE_URL + "/api/v2/simulator/export",
headers={"Authorization": jwt})
def simulator_list_tags(jwt=config.default_jwt):
return requests.get(url=config.BASE_URL + "/api/v2/simulator/tags",
headers={"Authorization": jwt})
tests/ft/neuron/process.py
+12 -4
View File
@@ -38,7 +38,9 @@ def start_neuron_disable_auth(dir='build/'):
process = subprocess.Popen(
command, stderr=subprocess.PIPE, cwd=dir)
time.sleep(1)
assert process.poll() is None
if process.poll() is not None:
stderr = process.stderr.read().decode()
assert False, "Neuron failed to start. Stderr: " + stderr
return process
def start_neuron_debug(dir='build/'):
@@ -46,7 +48,9 @@ def start_neuron_debug(dir='build/'):
process = subprocess.Popen(
command, stderr=subprocess.PIPE, cwd=dir)
time.sleep(1)
assert process.poll() is None
if process.poll() is not None:
stderr = process.stderr.read().decode()
assert False, "Neuron failed to start. Stderr: " + stderr
return process
def start_neuron_sub_filter_err(dir='build/'):
@@ -54,14 +58,18 @@ def start_neuron_sub_filter_err(dir='build/'):
process = subprocess.Popen(
command, stderr=subprocess.PIPE, cwd=dir)
time.sleep(1)
assert process.poll() is None
if process.poll() is not None:
stderr = process.stderr.read().decode()
assert False, "Neuron failed to start. Stderr: " + stderr
return process
def start_neuron(dir='build/'):
process = subprocess.Popen(
['./neuron'], stderr=subprocess.PIPE, cwd=dir)
time.sleep(1)
assert process.poll() is None
if process.poll() is not None:
stderr = process.stderr.read().decode()
assert False, "Neuron failed to start. Stderr: " + stderr
return process
tests/ft/simulator/__init__.py
+1
View File
@@ -0,0 +1 @@
tests/ft/simulator/test_simulator.py
+130
View File
@@ -0,0 +1,130 @@
import json
from pathlib import Path
import socket
import time
import neuron.api as api
import neuron.error as error
import neuron.config as config
from neuron.common import *
def _update_build_neuron_json(port):
roots = [Path("build"), Path(__file__).parents[3] / "build"]
for path in (r / s / "neuron.json" for r in roots for s in ["config", ""]):
if path.exists():
with path.open("r+", encoding="utf-8") as f:
data = json.load(f)
sim = data.get("modbus_simulator") or {}
sim["port"] = int(port)
if not sim.get("ip") or sim["ip"] == "0.0.0.0":
sim["ip"] = "127.0.0.1"
data["modbus_simulator"] = sim
f.seek(0)
json.dump(data, f, ensure_ascii=False, indent=1)
f.truncate()
return
assert False, "neuron.json not found"
class TestSimulator:
def _check_status(self, running=True, port=None, tag_count=None):
st = api.simulator_status().json()
assert set(st.keys()) >= {"running", "port", "tag_count", "error"}
assert st["error"] == 0
assert st["running"] == (1 if running else 0)
if port is not None:
assert st["port"] == port
if tag_count is not None:
assert st["tag_count"] == tag_count
return st
@description(given="load neuron.json",
when="start and stop simulator",
then="status reflects running and correct port")
def test_start_stop_status(self, class_setup_and_teardown):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
sim_port = s.getsockname()[1]
_update_build_neuron_json(sim_port)
class_setup_and_teardown.restart()
api.simulator_start()
self._check_status(running=True, port=sim_port, tag_count=0)
api.simulator_stop()
self._check_status(running=False, port=sim_port)
@description(given="set three tags using full name/address strings",
when="list tags",
then="returned tags")
def test_set_config_and_list_tags(self, class_setup_and_teardown):
api.simulator_start()
tags = [
{"type": "sine", "name": "t1", "address": "1!400001"},
{"type": "square", "name": "t3", "address": "1!400005"},
{"type": "saw", "name": "t2", "address": "1!400010"},
]
api.simulator_set_config(tags)
lt = api.simulator_list_tags().json()
assert "tags" in lt
assert isinstance(lt["tags"], list) and len(lt["tags"]) == 3
names = [t["name"] for t in lt["tags"]]
addrs = [t["address"] for t in lt["tags"]]
types = [t["type"] for t in lt["tags"]]
assert addrs == ["1!400001", "1!400005", "1!400010"]
assert names == ["t1", "t3", "t2"]
assert types == ["sine", "square", "saw"]
@description(given="simulator configured and running",
when="export configuration",
then="exports as attachment with expected structure and correct port")
def test_export_attachment(self):
st = api.simulator_status().json()
port = st.get("port", 0)
resp = api.simulator_export()
assert 200 == resp.status_code
dispo = resp.headers.get("Content-Disposition", "")
assert "attachment;" in dispo
assert "simulator_modbus_south.json" in dispo
data = json.loads(resp.text)
assert "nodes" in data and isinstance(data["nodes"], list)
assert len(data["nodes"]) >= 1
node0 = data["nodes"][0]
assert "plugin" in node0 and node0["plugin"] == "Modbus TCP"
assert "params" in node0 and isinstance(node0["params"], dict)
assert node0["params"].get("port") == port
assert "groups" in node0 and isinstance(node0["groups"], list) and len(node0["groups"]) >= 1
assert node0["groups"][0].get("group") == "group1"
@description(given="simulator started and tags configured",
when="restart neuron with new port in config",
then="enabled state and tags restored from DB, port updated from neuron.json")
def test_persistence_after_restart(self, class_setup_and_teardown):
api.simulator_start()
tags = [{"type": "sine", "name": "p1", "address": "1!400001"},
{"type": "sine", "name": "p2", "address": "1!400011"}]
api.simulator_set_config(tags)
st_before = self._check_status(running=True, tag_count=2)
port_before = st_before["port"]
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
new_port = s.getsockname()[1]
if new_port == port_before:
new_port += 1
_update_build_neuron_json(new_port)
class_setup_and_teardown.restart()
self._check_status(running=True, port=new_port, tag_count=2)
lt = api.simulator_list_tags().json()
assert len(lt["tags"]) == 2
assert lt["tags"][0]["name"] == "p1"
assert lt["tags"][1]["name"] == "p2"