Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from kervi.plugin.kervi_plugin import KerviPlugin
class WebSocketIPCPlugin(KerviPlugin):
def __init__(self, config, manager):
KerviPlugin.__init__(self, "WSIPC", config, manager)
from kervi.plugin.ipc.websocket.socket_spine import SocketSpine
self._socket_spine = SocketSpine(self.global_config)
self.spine.register_command_handler("startWebSocket", self._start_socket)
self.spine.trigger_event(
"websocketReady",
None,
{
}
)
def _start_socket(self):
self._socket_spine.start_socket()
from kervi.plugin.kervi_plugin import KerviPlugin
class WebServerPlugin(KerviPlugin):
def __init__(self, name, config, manager):
KerviPlugin.__init__(self, name, config, manager)
def server_ready(self, use_ssl, ip, port):
self.spine.trigger_event(
"WebAppReady",
None,
{
"ip": ip,
"port": port,
"ssl": use_ssl
}
)
def start(self):
pass
for key, value in inspect.getmembers(obj)
if not key.startswith("__")
and not inspect.isabstract(value)
and not inspect.isbuiltin(value)
and not inspect.isfunction(value)
and not inspect.isgenerator(value)
and not inspect.isgeneratorfunction(value)
and not inspect.ismethod(value)
and not inspect.ismethoddescriptor(value)
and not inspect.isroutine(value)
)
return self.default(data)
return obj
from kervi.plugin.kervi_plugin import KerviPlugin
class StoragePlugin(KerviPlugin):
def __init__(self, name, config, manager):
KerviPlugin.__init__(self, name, config, manager)
@property
def storage_type(self):
return "persisted"
def store_value(self, value_id, value, persist=False):
raise NotImplementedError
def get_value_data(self, date_from, date_to, limit):
raise NotImplementedError
def store_setting(self, group, name, value):
raise NotImplementedError
from kervi.plugin.kervi_plugin import KerviPlugin
class MessagePlugin(KerviPlugin):
def __init__(self, name, configuration, manager):
KerviPlugin.__init__(self, name, configuration, manager)
@property
def message_type(self):
raise NotImplementedError
def send_message(self, addresses, subject, **kwargs):
raise NotImplementedError
#stream_id, stream_event, observer_id, handler=None, name="stream", observer_type="stream_observer"
class _FileChunkObserver(StreamObserver):
def __init__(self, plugin, file_id, file):
StreamObserver.__init__(self, file_id, None,file_id + ".observer")
self._plugin = plugin
self._file_id = file_id
self._file = file
def on_event(self, stream_event, data=None):
#print("x", stream_event)
if stream_event == "FILE_CHUNK":
self._file.write(data)
if stream_event == "FILE_CLOSE":
self._plugin.close_file(self._file_id)
class FilesPlugin(KerviPlugin):
def __init__(self, name, configuration, manager):
KerviPlugin.__init__(self, name, configuration, manager)
self._root = self.plugin_config.root
self.spine.register_query_handler("files_get_dir", self.get_dir)
self.spine.register_query_handler("files_get_file", self.get_file)
self.spine.register_command_handler("files_save_file", self.save_file)
self.spine.register_query_handler("files_get_thumbnail", self.get_thumbnail)
self.spine.register_command_handler("files_open_file", self.open_file)
#self.spine.register_command_handler("files_write_file", self.write_file)
#self.spine.register_command_handler("files_close_file", self.close_file)
self._open_files = {}
def get_default_config(self):
return {
"root": "media",