Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if self._as_service:
signal.signal(signal.SIGINT, handler_stop_signals)
signal.signal(signal.SIGTERM, handler_stop_signals)
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(
for plugin_module in plugin_modules:
self._process_info.append(
{"id":plugin_module, "ready": False, "pid": None}
)
self._process_info_lock.release()
module_port = pluginManager.load_plugins(module_port+1)
for module in self.config.modules:
self._process_info_lock.acquire()
self._process_info += [{"id":module, "ready":False, "pid":None}]
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]),
app_helpers._KerviModuleLoader,
log_queue = self._log_queue
)
]
#print("wait for ready")
try:
while not self.char_list and not self._is_ready():
#print(self.char_list)
time.sleep(1)
except KeyboardInterrupt:
pass
def _start_plugin_process(self, module_port):
process._start_process(
"plugin-" + self._manager._config.application.id + "." + self._plugin_module,
"plugin_" + self._plugin_module,
self._manager._config,
nethelper.get_free_port([module_port]),
_KerviPluginProcess,
plugin_module=self._plugin_module,
plugin_config=self._config,
log_queue = self._manager._log_queue,
load_silent = self._manager._load_silent
)
def _deep_update(d, u):
"""Update a nested dictionary or similar mapping.
Modify ``source`` in place.
"""
for k, v in u.items():
if isinstance(v, collections.Mapping):
r = _deep_update(d.get(k, {}), v)
d[k] = r
else:
d[k] = u[k]
return d
class _KerviModuleLoader(process._KerviProcess):
""" Private class that starts a seperate process that loads a module in the Kervi application """
def init_process(self, **kwargs):
print("load:", self.name)
#import kervi.core_sensors.cpu_sensors
try:
import kervi.hal as hal
hal._load()
__import__(self.name, fromlist=[''])
except ImportError:
self.spine.log.exception("load module:{0}", self.name)
#import kervi.utility.storage
self.spine.send_command("startThreads", local_only=True)
self.spine.trigger_event(
"moduleLoaded",
self.name,