# pylint: disable=abstract-method
# pylint: disable=missing-module-docstring
# pylint: disable=missing-class-docstring
# pylint: disable=missing-function-docstring
# pylint: disable=W0223, R0903, C0115, C0116, W0212, W0221
import logging
from typing import List
import tango
from ska_tango_base import SKABaseDevice
from ska_tango_base.base.component_manager import BaseComponentManager
from tango.server import command
# Details:
# https://tango-controls.readthedocs.io/projects/rfc/en/
# latest/14/Logging.html#logging-levels
_TANGO_LOG_LEVEL_TO_PYTHON = {
"OFF": logging.CRITICAL, # there is no "off"
"FATAL": logging.CRITICAL,
"ERROR": logging.ERROR,
"WARN": logging.WARNING,
"WARNING": logging.WARNING,
"INFO": logging.INFO,
"DEBUG": logging.DEBUG,
}
[docs]class LogComponentManager(BaseComponentManager):
def __init__(self, logger: logging.Logger) -> None:
"""
Update logging config so that certain parts can be overridden
:return: An instance of LogComponentManager
:rtype: LogComponentManager
"""
super().__init__(logger, None, None)
class TangoDeviceTagsFilter(logging.Filter):
"""Reset the log record components if a TLS log"""
@classmethod
def filter(cls, record):
# Log a TLS log
if hasattr(record, "device_name"):
record.tags = f"tango-device:{record.device_name}"
record.filename = "unknown_file"
record.threadName = "unknown_thread"
record.funcName = record.src_funcName
record.created = record.timestamp
record.lineno = 0
return True
self.logger.addFilter(TangoDeviceTagsFilter())
[docs] def log(
self,
timestamp: str,
tango_log_level: str,
tango_device: str,
message: str,
) -> None:
"""Override log components and log to stdout.
:param timestamp: The millisecond since epoch (01.01.1970)
:type timestamp: str
:param tango_log_level: The log level
:type tango_log_level: str
:param tango_device: The tango device
:type tango_device: str
:param message: The message to log
:type message: str
"""
try:
function_name = ""
if " - " in message:
function_name, message = message.split(" - ", 1)
log_level = _TANGO_LOG_LEVEL_TO_PYTHON[tango_log_level]
log_timestamp = float(timestamp) / 1000
self.logger.log(
log_level,
message,
extra={
"device_name": tango_device,
"src_funcName": function_name,
"timestamp": log_timestamp,
},
)
except Exception as e:
self.logger.exception(e)
raise
[docs]class DishLogger(SKABaseDevice):
[docs] def create_component_manager(self):
"""Create the component manager LogComponentManager
:return: Instance of LogComponentManager
:rtype: LogComponentManager
"""
return LogComponentManager(self.logger)
[docs] @command(dtype_in=[str], doc_out="Consume a log message from TLS")
def Log(self, log_message: List[str]):
"""Write the log to stdout as received from TLS
Sample log:
['1650964795495', 'ERROR', 'mid-dish/dish-manager/SKA001',
'TangoUtils::DeviceAttributeToCorbaAny() - A Message',
'', '@7f48dcc80700 [7]']
Details of the list items here:
https://tango-controls.readthedocs.io/projects/rfc/
en/latest/14/Logging.html#log-consumer
:param log_message: Parts of the TLS log message
:type log_message: List[str]
"""
timestamp, tango_log_level, tango_device, message, _, _ = log_message
self.component_manager.log(timestamp, tango_log_level, tango_device, message)
[docs] @command(dtype_in=str, doc_in="name of the device to add new logging target")
def SetDishLoggerTarget(self, device_name: str) -> None:
"""Add DishLogger as a logging target destination on device"""
logging_device = tango.DeviceProxy(device_name)
logging_device.add_logging_target(f"device::{self.get_name()}")
[docs] @command(dtype_in=str, doc_in="name of the device to remove logging target")
def RemoveDishLoggerTarget(self, device_name: str) -> None:
"""Remove DishLogger as a logging target destination on device"""
logging_device = tango.DeviceProxy(device_name)
logging_device.remove_logging_target(f"device::{self.get_name()}")
def run():
DishLogger.run_server()
if __name__ == "__main__":
run()