diff options
| -rw-r--r-- | NeoRuntime/Runtime/luxcena_neo/strip.py | 2 | ||||
| -rw-r--r-- | NeoRuntime/Runtime/neo_runtime.py | 141 | ||||
| -rw-r--r-- | src/NeoRuntimeManager/IPC.js | 178 | ||||
| -rw-r--r-- | src/NeoRuntimeManager/RuntimeProcess.js | 78 | ||||
| -rw-r--r-- | src/NeoRuntimeManager/index.js | 34 | ||||
| -rw-r--r-- | src/UserData/index.js | 3 | ||||
| -rw-r--r-- | src_frontend/Components/MainControls/ControlComponents.svelte | 2 |
7 files changed, 321 insertions, 117 deletions
diff --git a/NeoRuntime/Runtime/luxcena_neo/strip.py b/NeoRuntime/Runtime/luxcena_neo/strip.py index a65b3f0..bfe2bbc 100644 --- a/NeoRuntime/Runtime/luxcena_neo/strip.py +++ b/NeoRuntime/Runtime/luxcena_neo/strip.py @@ -59,7 +59,7 @@ class Strip: self.__brightness = 255 self.__actual_brightness = self.__brightness - self.__globvars_path = path.join(path.split(path.dirname(path.abspath(__file__)))[0], "globvars.json") + self.__globvars_path = path.join(path.split(path.dirname(path.abspath(__file__)))[0], "state.json") if path.exists(self.__globvars_path): try: with open(self.__globvars_path, "r") as f: diff --git a/NeoRuntime/Runtime/neo_runtime.py b/NeoRuntime/Runtime/neo_runtime.py index e5941e2..4ecbc97 100644 --- a/NeoRuntime/Runtime/neo_runtime.py +++ b/NeoRuntime/Runtime/neo_runtime.py @@ -11,7 +11,8 @@ import time import threading import select import traceback -from os import path +import socket +from os import path, remove from luxcena_neo.strip import Strip @@ -41,7 +42,6 @@ def init_package(package_path, entry_module, strip): # Make the strip instance available in our modules setattr(module, "strip", strip) - module_entry_instance.declare_variables() return module_entry_instance def exec_module(module_executor_loop_func): @@ -53,10 +53,12 @@ def exec_module(module_executor_loop_func): class NeoRuntime: - def __init__(self, package_path, entry_module, strip_config_file): + def __init__(self, package_path, entry_module, strip_config_file, socket_file): self.__strip = init_strip(strip_config_file) self.__module_entry_instance = init_package(package_path, entry_module, self.__strip) self.__module_th = None + self.__socket_file = socket_file + self.__send_strip_buffer = False def start(self): @@ -66,49 +68,113 @@ class NeoRuntime: # This will run in this thread. print("> Starting to listen on stdin") + self.__s = None try: - self.__command_listener_loop() + self.__bind_socket() + self.__socket_listener() except KeyboardInterrupt: print("Exiting...") except Exception as e: traceback.print_exc() - + finally: + self.__close_socket() + + def __bind_socket(self): + if path.exists(self.__socket_file): + remove(self.__socket_file) + + self.__s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.__s.bind(self.__socket_file) + self.__s.listen(1) - def __command_listener_loop(self): + def __socket_listener(self): + self.__s_clients = [] last_send = time.perf_counter() + while True: if not self.__module_th.is_alive(): break - while sys.stdin in select.select([sys.stdin], [], [], 0)[0]: - line = sys.stdin.readline() - if line: - line = line.replace("\n", "") - if (line[0:10] == ":::setvar:"): - name, value = (line.split(" ", 1)[1]).replace("\"", "").split(":", 1) - if name in self.__module_entry_instance.vars: - self.__module_entry_instance.vars[name] = value - elif (line[0:11] == ":::setglob:"): - name, value = (line.split(" ", 1)[1]).replace("\"", "").split(":", 1) - if name == "brightness": - self.__strip.brightness = int(value) - elif name == "power_on": - self.__strip.power_on = value == "true" - else: - print(f"Unknown globvar \"{name}\"") - else: - if (time.perf_counter() - last_send) > 0.5: - _vars = "{" - for name, var in self.__module_entry_instance.vars: - _vars += f" \"{name}\" : {{ \"value\": \"{var.value}\", \"var_type\": \"{var.var_type}\" }}, " - if len(_vars) > 2: - _vars = _vars[0:-2] - _vars += "}" - - globvars = "{ \"power_on\": " + str(self.__strip.power_on).lower() + ", " - globvars += " \"brightness\":" + str(self.__strip.brightness) + " }" - print(f"{{ \":::data:\": {{ \"variables\": {_vars}, \"globvars\": {globvars} }} }}") - last_send = time.perf_counter() + r, w, e = select.select([self.__s, *self.__s_clients], self.__s_clients, [], 0) + + if (time.perf_counter() - last_send) > 0.5: + states = { + "variables": self.__module_entry_instance.var.to_dict(), + "globvars": { + "power_on": self.__strip.power_on, + "brightness": self.__strip.brightness + } + } + buf = bytes([1]) + bytes(json.dumps(states), "ascii") + + for ws in w: + try: + ws.send(buf) + except BrokenPipeError: + self.__s_clients.remove(ws) + ws.close() + + last_send = time.perf_counter() + + for rs in r: + if rs is self.__s: + c, a = self.__s.accept() + self.__s_clients.append(c) + else: + data = rs.recv(128) + if not data: + self.__s_clients.remove(rs) + rs.close() + else: + self.__execute_command(data) + def __close_socket(self): + if (self.__s is None): return + r, w, e = select.select([self.__s, *self.__s_clients], self.__s_clients, [], 0) + for ws in w: + try: + ws.shutdown(socket.SHUT_RDWR) + except BrokenPipeError: + ws.close() + self.__s_clients.remove(ws) + ws.close() + self.__s.close() + + + + def __execute_command(self, command): + """ + command should be of type bytes + first byte indicates command type (currently setglob or setvar) + + for command type 1 + byte 1 indicates which globvar + byte 2 indicates value + for command type 2 + first 32 bytes are the var name + + """ + # print(command.hex(" ")) + if command[0] == 0: + if command[1] == 0: + self.__strip.power_on = (command[2] == 1) + print(f"Strip power: {self.__strip.power_on}") + elif command[1] == 1: + self.__strip.brightness = command[2] + print(f"Strip brightness: {self.__strip.brightness}") + else: + print(f"Unknown globvar {command[1]}.") + elif command[0] == 1: + name = command[3:3+command[1]].decode("ascii") + value = command[3+command[1]:3+command[1]+command[2]].decode("ascii") + if name in self.__module_entry_instance.var: + self.__module_entry_instance.var[name] = value + else: + print(f"Unknown variable {name}") + elif command[0] == 2: + self.__send_strip_buffer = (command[1] == 1) + else: + print("UNKNOWN COMMAND") + def __module_loop(self): self.__module_entry_instance.on_start() @@ -153,11 +219,14 @@ if __name__ == "__main__": parser.add_argument('--strip-config', help='Path to the strip config file.') parser.add_argument('--mode-path', help='Path of the folder the mode is in.') parser.add_argument('--mode-entry', help='Path of the module that is the entry-point of the module.') + parser.add_argument('--socket-file', help='The socket file the runtime will use to allow communication [default: /tmp/neo_runtime.sock].', default='/tmp/neo_runtime.sock') + parser.add_argument('--socket-enable', help='Wether to enable socket communication [default: true].', default=True) args = parser.parse_args() args.strip_config = args.strip_config.replace("\"", "") args.mode_path = args.mode_path.replace("\"", "") args.mode_entry = args.mode_entry.replace("\"", "") + args.socket_file = args.socket_file.replace("\"", "") if not path.exists(args.strip_config): print(f"Strip config not found ({args.strip_config}).") sys.exit(1) @@ -172,6 +241,6 @@ if __name__ == "__main__": print(f"Module : {args.mode_path}/{args.mode_entry}") print(f"> Starting \"{args.mode_path}\" in NeoRuntime.") - runtime = NeoRuntime(args.mode_path, args.mode_entry, args.strip_config) + runtime = NeoRuntime(args.mode_path, args.mode_entry, args.strip_config, args.socket_file) runtime.start() print ("> NeoRuntime exited...") diff --git a/src/NeoRuntimeManager/IPC.js b/src/NeoRuntimeManager/IPC.js new file mode 100644 index 0000000..817c049 --- /dev/null +++ b/src/NeoRuntimeManager/IPC.js @@ -0,0 +1,178 @@ +/** + * This module is used to communicate with a python NeoRuntime instance. + * + * @author jakobst1n. + * @since 3.10.2021 + */ + +const net = require("net"); +let logger = require(__basedir + "/src/logger"); + +/** @type {int} How long wait between each reconnection attempt */ +const RECONNECT_INTERVAL = 1000; +/** @type {Object} ENUM-ish for command that can be sent to neoruntime */ +const COMMAND = Object.freeze({SET_GLOB : 0, + SET_VAR : 1, + SET_SEND_STRIP_BUF: 2}); +/** @type {Object} ENUM-ish for globvars */ +const GLOBVAR = Object.freeze({POWER_ON : 0, + BRIGHTNESS: 1}); +/** @type {Object} ENUM-ish for what type of data neoruntime sends */ +const DATATYPE = Object.freeze({STATES : 1, + STRIP_BUF: 2}); + +/** + * class that will keep a active connection to a socket if possible, and + * automatically reconnect. It will emit events when data is received, + * and it will send commands to the process. */ +class IPC { + + constructor(_socketFile, _eventEmitter) { + this.socketFile = _socketFile; + this.eventEmitter = _eventEmitter; + + this.client; + this.connected = false; + this.reconnectInterval = false; + + this.globvars = {}; + this.variables = {}; + + this.reconnect(); + } + + /** + * If we are not already attempting to reconnect, this will start a + * interval that tries to reconnect. */ + reconnect() { + if (this.reconnectInterval === false) { + this.reconnectInterval = setInterval(this.tryOpenSocketConnection.bind(this), RECONNECT_INTERVAL); + } + } + + /** + * This will attempt to connect to the socket, and then setup all listeners + * if it succedes. */ + tryOpenSocketConnection() { + // logger.info("Attempting to start IPC"); + + this.client = net.createConnection(this.socketFile) + .on('connect', () => { + clearInterval(this.reconnectInterval); + this.reconnectInterval = false; + // logger.info("IPC Connected."); + }) + .on("ready", () => { + this.connected = true; + }) + .on('data', (data) => { + switch (data[0]) { + case DATATYPE.STATES: + let json_data; + try { + json_data = JSON.parse(data.toString("ascii", 1)); + } catch (e) { + logger.warning("Could not parse json data from neoruntime"); + return; + } + + if (json_data.hasOwnProperty("globvars")) { + forEachDiff(json_data["globvars"], this.globvars, (key, newVal) => { + this.eventEmitter.emit("change", key, newVal); + }); + this.globvars = json_data["globvars"]; + } + if (json_data.hasOwnProperty("variables")) { + forEachDiff(json_data["variables"], this.variables, (key, newVal) => { + this.eventEmitter.emit("change", `variable/${key}`, newVal); + }); + this.variables = json_data["variables"]; + } + break; + + default: + logger.info(data); + } + + }) + .on("timeout", () => { + logger.info("IPC Timeout"); + }) + .on("close", (hadError) => { + // logger.info("IPC Close, hadError: ", hadError); + this.connected = false; + this.reconnect(); + }) + .on("end", () => { + // logger.info("IPC End"); + this.connected = false; + }) + .on('error', (data) => { + // logger.info('IPC Server not active.'); + this.connected = false; + this.reconnect(); + }) + ; + } + + /** + * Will send a command to the socket if we have a active connection, + * if not it will just drop the command. there is no queue implemented + * for such events. */ + sendCommand(commandType, name, value) { + if (this.connected) { + let buf = Buffer.allocUnsafe(128); // It's fine, we know what we are doing + // let buf = Buffer.alloc(128); + + switch (commandType) { + case (COMMAND.SET_GLOB): + buf[1] = name; + buf[2] = value; + break; + case (COMMAND.SET_VAR): + if (name.length > 32) { return {success: false, reason: "name too long", detail: "max size of name is 32 bytes"}; } + if (name.length > 93) { return {success: false, reason: "value too long", detail: "max size of value is 93 bytes"}; } + buf[1] = name.length; + buf[2] = value.length; + buf.write(name, 3, name.length, "ascii"); + buf.write(value, 3+name.length, value.length, "ascii"); + break; + case (COMMAND.SET_SEND_STRIP_BUF): + buf[1] = (name) ? 1 : 0; + default: + logger.info(`IPC UNKNOWN COMMANDTYPE ${commandType}`) + return; + } + + buf[0] = commandType; + this.client.write(buf); + return {success: true} + } + return {success: false, reason: "socket not connected", detail: "This usually means the python script is not running"}; + } + +} + +const isObject = v => v && typeof v === 'object'; + +/** + * Will call callback on all the differences between the dicts + */ +function forEachDiff(dict1, dict2, callback) { + for (const key of new Set([...Object.keys(dict1), ...Object.keys(dict2)])) { + if (isObject(dict1[key]) && isObject(dict2[key])) { + if (dict1[key].value !== dict2[key].value) { + callback(key, dict1[key]); + } + } else if (dict1[key] !== dict2[key]) { + if (isObject(dict2[key]) && (dict1[key] == null)) { + dict2[key].value = null; + callback(key, dict2[key]) + } else { + callback(key, dict1[key]); + } + } + } +} + +module.exports = {IPC, COMMAND, GLOBVAR};
\ No newline at end of file diff --git a/src/NeoRuntimeManager/RuntimeProcess.js b/src/NeoRuntimeManager/RuntimeProcess.js index 60f6a28..60c1de9 100644 --- a/src/NeoRuntimeManager/RuntimeProcess.js +++ b/src/NeoRuntimeManager/RuntimeProcess.js @@ -3,10 +3,11 @@ let spawn = require("child_process"); class RuntimeProcess { - constructor(_modePath, _onVarChange, _eventEmitter) { + constructor(_modePath, _eventEmitter) { this.modePath = _modePath; this.logfile = `${this.modePath}/mode.log`; - + this.errfile = `${this.modePath}/mode.error`; + this.stdout = ""; this.stderr = ""; @@ -16,9 +17,6 @@ class RuntimeProcess { this.isRunning = false; this.exitCode = null; - this.variables = {}; - this.globvars = {}; - this.onVarChange = _onVarChange; this.eventEmitter = _eventEmitter; } @@ -29,7 +27,7 @@ class RuntimeProcess { } this.isRunning = true; this.proc = spawn.spawn( - "python3", + "python3", [ "-u", // This makes us able to get real-time output `${__basedir}/NeoRuntime/Runtime/neo_runtime.py`, @@ -44,28 +42,13 @@ class RuntimeProcess { }); fs.ensureFileSync(this.logfile); + fs.ensureFileSync(this.errfile); this.eventEmitter.emit("proc:start"); this.proc.stdout.on('data', (_stdout) => { let stdout_str = _stdout.toString(); - - let regex = /{ ":::data:": { (.*) } }/gi; - let data = stdout_str.match(regex); - stdout_str = stdout_str.replace(regex, () => ""); - - if ((data != null) && (data.length > 0)) { - try { - this.processVarData(data) - } catch {} - } - - if (stdout_str.replace("\n", "").replace(" ", "") == "") { - // In this case, we want to ignore the data. - } else { - // stdout_str = stdout_str.replace(/\n$/, "") - fs.appendFile(this.logfile, "\n====stdout====\n" + stdout_str); - this.eventEmitter.emit("proc:stdout", stdout_str); - } + fs.appendFile(this.logfile, `[${timestamp()}]: ` + stdout_str); + this.eventEmitter.emit("proc:stdout", stdout_str); }); this.proc.stdout.on('end', () => { @@ -73,9 +56,8 @@ class RuntimeProcess { }); this.proc.stderr.on('data', (_stderr) => { - // let stderr_str = _stderr.toString().replace(/\n$/, "") - let stderr_str = _stderr.toString() - fs.appendFile(this.logfile, "\n====stderr====\n" + stderr_str); + let stderr_str = _stderr.toString(); + fs.appendFile(this.errfile, `[${timestamp()}]: ` + stderr_str); this.eventEmitter.emit("proc:stderr", stderr_str); }); @@ -85,7 +67,7 @@ class RuntimeProcess { this.proc.on('close', (code) => { if (code) { - fs.appendFile(this.logfile, "\n====close====\nScript exited with code " + code.toString()); + fs.appendFile(this.logfile, `[${timestamp()}]: ` + "Script exited with code " + code.toString()); } this.eventEmitter.emit("proc:exit", 0); this.isRunning = false; @@ -106,45 +88,15 @@ class RuntimeProcess { console.log(err); } } - - processVarData(data) { - data = JSON.parse(data)[":::data:"]; - if (data.hasOwnProperty("globvars")) { - forEachDiff(data["globvars"], this.globvars, (key, newVal) => { - this.onVarChange("globvars", key, newVal); - }); - this.globvars = data["globvars"]; - } - if (data.hasOwnProperty("variables")) { - forEachDiff(data["variables"], this.variables, (key, newVal) => { - this.onVarChange("variables", key, newVal); - }); - this.variables = data["variables"]; - } - } - } -const isObject = v => v && typeof v === 'object'; - /** - * Will call callback on all the differences between the dicts + * Creates and returns a timestamp that can be used in logfiles. + * + * @return {string} timestamp */ -function forEachDiff(dict1, dict2, callback) { - for (const key of new Set([...Object.keys(dict1), ...Object.keys(dict2)])) { - if (isObject(dict1[key]) && isObject(dict2[key])) { - if (dict1[key].value !== dict2[key].value) { - callback(key, dict1[key]); - } - } else if (dict1[key] !== dict2[key]) { - if (isObject(dict2[key]) && (dict1[key] == null)) { - dict2[key].value = null; - callback(key, dict2[key]) - } else { - callback(key, dict1[key]); - } - } - } +function timestamp() { + return (new Date()).toISOString(); } module.exports = RuntimeProcess; diff --git a/src/NeoRuntimeManager/index.js b/src/NeoRuntimeManager/index.js index 62acb8a..6238323 100644 --- a/src/NeoRuntimeManager/index.js +++ b/src/NeoRuntimeManager/index.js @@ -8,6 +8,7 @@ const fs = require("fs"); const fsPromises = fs.promises; const RuntimeProcess = require("./RuntimeProcess"); +const IPC = require("./IPC"); let logger = require(__basedir + "/src/logger"); const EventEmitter = require('events'); @@ -20,6 +21,8 @@ let modeId = null; let modeExitCode = 0; /** @type {RuntimeProcess} This is the current RuntimeProcess instance */ let runtimeProcess = null; +/** @type {IPC} The IPC instance, used to communicate with the script */ +let ipc = null; /** @type {EventEmitter} This is used to emit events when things change */ const eventEmitter = new EventEmitter(); /** @type {boolean} If this is true, we will not do things the usual way */ @@ -83,13 +86,6 @@ function setMode(_modeId) { return {success: false, reason: "unknown modeId"}; } logger.info(`Changing mode to "${_modeId}".`); - - let globvarsTmp = {}; - let variablesTmp = {}; - if (runtimeProcess != null) { - globvarsTmp = runtimeProcess.globvars; - variablesTmp = runtimeProcess.variables; - } stopMode(); @@ -97,10 +93,9 @@ function setMode(_modeId) { neoModules.userData.config.activeMode = modeId; eventEmitter.emit("change", "mode", modeId); - runtimeProcess = new RuntimeProcess(getModePath(_modeId), onVariableChange, eventEmitter); - runtimeProcess.globvars = globvarsTmp; - runtimeProcess.variables = variablesTmp; + runtimeProcess = new RuntimeProcess(getModePath(_modeId), eventEmitter); startMode(); + return {success: true} }; @@ -194,7 +189,7 @@ function onVariableChange(location, name, newValue) { */ function getGlobvars() { if (!modeRunning()) { return {}; } - return runtimeProcess.globvars; + return ipc.globvars; } /** @@ -207,8 +202,15 @@ function getGlobvars() { */ function setGlobvar(name, value) { if (!modeRunning()) { return; } - runtimeProcess.proc.stdin.write(`:::setglob: ${name}:${value}\n`); - return {success: true} + + switch(name) { + case "power_on": + return ipc.sendCommand(IPC.COMMAND.SET_GLOB, IPC.GLOBVAR.POWER_ON, (value) ? 1 : 0); + case "brightness": + return ipc.sendCommand(IPC.COMMAND.SET_GLOB, IPC.GLOBVAR.BRIGHTNESS, value); + default: + return {success: false, reason: "unknown globvar", detail: name}; + } } /** @@ -218,7 +220,7 @@ function setGlobvar(name, value) { */ function getVariables() { if (!modeRunning()) { return {}; } - return runtimeProcess.variables; + return ipc.variables; } /** @@ -231,8 +233,7 @@ function getVariables() { */ function setVariable(name, value) { if (!modeRunning()) { return; } - runtimeProcess.proc.stdin.write(`:::setvar: ${name}:${value}\n`); - return {success: true} + return ipc.sendCommand(IPC.COMMAND.SET_VAR, name, value); } /** @@ -281,6 +282,7 @@ function stopDebugger() { module.exports = (_neoModules) => { neoModules = _neoModules; + ipc = new IPC.IPC(neoModules.userData.config.neoRuntimeIPC.socketFile, eventEmitter); return { event: eventEmitter, modes: listModes, diff --git a/src/UserData/index.js b/src/UserData/index.js index e5318c9..704c5d5 100644 --- a/src/UserData/index.js +++ b/src/UserData/index.js @@ -40,6 +40,9 @@ function ensureMainConfig() { if (config.DiscoveryServer.address == null) { config.DiscoveryServer.address = "https://erj46s.deta.dev"; } if (config.DiscoveryServer.broadcastSelf == null) { config.DiscoveryServer.broadcastSelf = false; } + if (config.neoRuntimeIPC == null) { config.neoRuntimeIPC = {}; } + if (config.neoRuntimeIPC.socketFile == null) { config.neoRuntimeIPC.socketFile = "/tmp/neo_runtime.sock"; } + fse.writeFileSync(__datadir + "/config/config.ini", ini.encode(config)) } diff --git a/src_frontend/Components/MainControls/ControlComponents.svelte b/src_frontend/Components/MainControls/ControlComponents.svelte index 5f6d165..65bd1c4 100644 --- a/src_frontend/Components/MainControls/ControlComponents.svelte +++ b/src_frontend/Components/MainControls/ControlComponents.svelte @@ -45,7 +45,7 @@ } name = name.replace("variable/", ""); - switch (value.var_type) { + switch (value.type) { case "COLOR": if (value.value == null) { delete colorVariables[name]; |