66 lines
1.8 KiB
Python
66 lines
1.8 KiB
Python
from typing import Dict
|
|
|
|
from .abstract_plugin import AbstractCommand, AbstractPlugin, AbstractCommandCompiler
|
|
import logging
|
|
|
|
|
|
class LogCommand(AbstractCommand):
|
|
LEVELS = { # _nameToLevel would be nicer, but we are limiting options here
|
|
"DEBUG": logging.DEBUG,
|
|
"INFO": logging.INFO,
|
|
"WARNING": logging.WARNING
|
|
# Programs are not allowed to print error level
|
|
}
|
|
|
|
def __init__(self, program_logger: logging.Logger, level: str, message: str):
|
|
|
|
self._program_logger = program_logger
|
|
|
|
try:
|
|
self._level = self.LEVELS[level]
|
|
except KeyError:
|
|
raise ValueError(f"Unknown or not allowed log level: {level}")
|
|
|
|
self._message = message
|
|
|
|
def execute(self):
|
|
self._program_logger.log(self._level, self._message)
|
|
|
|
def abort(self):
|
|
# There is really nothing to abort here
|
|
pass
|
|
|
|
def describe(self) -> dict:
|
|
return {
|
|
"command": "log",
|
|
"params": {
|
|
"level": logging.getLevelName(self._level),
|
|
"message": self._message
|
|
}
|
|
}
|
|
|
|
|
|
class LogCompiler(AbstractCommandCompiler):
|
|
|
|
def __init__(self, program_logger: logging.Logger):
|
|
self._program_logger = program_logger
|
|
|
|
def compile(self, level: str, message: str) -> AbstractCommand:
|
|
return LogCommand(self._program_logger, level, message)
|
|
|
|
|
|
class LogPlugin(AbstractPlugin):
|
|
plugin_name = "log"
|
|
|
|
def __init__(self):
|
|
self._logger = logging.getLogger("plugin").getChild("log")
|
|
self._program_logger = self._logger.getChild("program")
|
|
|
|
def load_compilers(self) -> Dict[str, AbstractCommandCompiler]:
|
|
return {
|
|
"log": LogCompiler(self._program_logger)
|
|
}
|
|
|
|
def close(self):
|
|
pass
|