Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"port": nethelper.get_free_port([9434, 9435, 9436])
},
"log" : {
"levels":["fatal", "error", "warning", "information", "debug"],
"level":"verbose",
"console_level":"verbose",
"file": "kervi.log",
"max_file_size": 1000000,
"resetLog": False
},
"modules":[],
"network" : {
"ip": nethelper.get_ip_address(),
"http_port": nethelper.get_free_port([80, 8080, 8000]),
"ws_port": nethelper.get_free_port([9000, 9001, 9002]),
"ipc_root_port": nethelper.get_free_port([9500]),
"ipc_root_address": nethelper.get_ip_address()
},
"plugin_manager":{
"plugin_types":{
"default": {
"managed": False,
"own_process": True
},
"messaging": {
"own_process": False,
"managed": True
},
"ui": {
"own_process": True,
"managed": False
},
try:
from kervi.version import VERSION
except:
VERSION="0.0.0"
self._logger.verbose("kervi version: %s", VERSION)
import kervi.hal as hal
hal_driver = hal._load(self.config.platform.driver)
if hal_driver:
self._logger.verbose("platform driver: %s", hal_driver)
self._logger.verbose("board: %s", hal.get_board_name())
from kervi.plugin.message_bus.bus_manager import BusManager
self.bus_manager = BusManager(self._log_queue)
self.config.network.ipc_root_port = nethelper.get_free_port([self.config.network.ipc_root_port])
self.bus_manager.load("kervi-main", self.config.network.ipc_root_port, None, self.config.network.ipc_root_address)
from kervi import spine
self.spine = spine.Spine()
self.spine.register_query_handler("GetApplicationInfo", self._get_application_info)
self.spine.register_query_handler("getProcessInfo", self.get_process_info)
self.spine.register_event_handler("modulePing", self._module_ping)
self.spine.register_event_handler("processReady", self._process_ready, scope="app-" + self.config.application.id)
self.spine.register_event_handler("WebAppReady", self._webapp_ready, scope="app-" + self.config.application.id)
self.spine.register_event_handler("websocketReady", self._websocket_ready, scope="app-" + self.config.application.id)
self._module_processes = []
self._process_info = []
self._process_info_lock = threading.Lock()
self._kervi_modules = []
self._kervi_modules_lock = threading.Lock()
self.spine.send_command("startThreads", local_only=True)
time.sleep(.5)
module_port = self.config.network.ipc_root_port
self._process_info_lock.acquire()
self._process_info = [{"id":"IPC", "ready":False}]
self._process_info_lock.release()
module_port += 1
self._module_processes += [
process._start_process(
"app-" + self.config.application.id,
"IPC",
self.config,
nethelper.get_free_port([module_port]),
app_helpers._KerviSocketIPC
)
]
for module in self.config.modules:
self._process_info_lock.acquire()
self._process_info += [{"id":module, "ready":False}]
self._process_info_lock.release()
module_port += 1
self._module_processes += [
process._start_process(
"app-" + self.config.application.id,
module,
self.config,
nethelper.get_free_port([module_port]),