#
# This file is part of the SKA Tango Base project
#
# Distributed under the terms of the BSD 3-clause new license.
# See LICENSE.txt for more info.
"""
This module implements a generic base model and device for SKA.
It exposes the generic attributes, properties and commands of an SKA device.
"""
import logging
import os
import threading
import typing
from typing import Any
import ska_ser_logging
from ska_control_model import LoggingLevel
from ska_control_model import __name__ as control_model_name
from ska_control_model import __version__ as control_model_version
from tango import DebugIt
from tango.server import Device, attribute, command, device_property
from . import release
from .base.logging import (
_LMC_TO_PYTHON_LOGGING_LEVEL,
_LMC_TO_TANGO_LOGGING_LEVEL,
LoggingUtils,
)
from .faults import LoggingLevelError
__all__ = ["SKADevice"]
module_logger = logging.getLogger(__name__)
_STATE_MACHINE_LOGS_ENV_VAR = "STB_STATE_MACHINE_VERBOSE"
[docs]
class SKADevice(Device):
"""A generic base device for SKA."""
_logging_config_lock = threading.Lock()
_logging_configured = False
def _init_logging(self) -> None:
"""Initialize the logging mechanism, using default properties."""
class EnsureTagsFilter(logging.Filter):
"""
Ensure all records have a "tags" field.
The tag will be the empty string if a tag is not provided.
"""
def filter(self, record: logging.LogRecord) -> bool: # noqa: A003
if not hasattr(record, "tags"):
record.tags = ""
return True
# There may be multiple devices in a single device server - these will all be
# starting at the same time, so use a lock to prevent race conditions, and
# a flag to ensure the SKA standard logging configuration is only applied once.
with SKADevice._logging_config_lock:
if not SKADevice._logging_configured:
ska_ser_logging.configure_logging(tags_filter=EnsureTagsFilter)
SKADevice._logging_configured = True
transitions_verbose = os.environ.get(_STATE_MACHINE_LOGS_ENV_VAR)
if transitions_verbose is None or transitions_verbose == "OFF":
logging.getLogger("transitions").setLevel(logging.WARNING)
device_name = self.get_name()
self.logger = logging.getLogger(device_name)
# device may be reinitialised, so remove existing handlers and filters
for handler in list(self.logger.handlers):
self.logger.removeHandler(handler)
for filt in list(self.logger.filters):
self.logger.removeFilter(filt)
# add a filter with this device's name
device_name_tag = f"tango-device:{device_name}"
class TangoDeviceTagsFilter(logging.Filter):
"""Filter that adds tango device name to the emitted record."""
def filter(self, record: logging.LogRecord) -> bool: # noqa: A003
record.tags = device_name_tag
return True
self.logger.addFilter(TangoDeviceTagsFilter())
# before setting targets, give Python logger a reference to the log4tango logger
# to support the TangoLoggingServiceHandler target option
self.logger.tango_logger = self.get_logger() # type: ignore[attr-defined]
# initialise using defaults in device properties
self._logging_level: LoggingLevel | None = None
self.set_logging_level(self.LoggingLevelDefault)
self.set_logging_targets(self.LoggingTargetsDefault)
self.logger.debug("Logger initialised")
# monkey patch Tango Logging Service streams so they go to the Python
# logger instead
def _debug_patch(
*args: Any,
source: str | None = None,
**kwargs: Any,
) -> None:
self.logger.debug(*args, **kwargs)
self.debug_stream = _debug_patch
def _info_patch(
*args: Any,
source: str | None = None,
**kwargs: Any,
) -> None:
self.logger.info(*args, **kwargs)
self.info_stream = _info_patch
def _warn_patch(
*args: Any,
source: str | None = None,
**kwargs: Any,
) -> None:
self.logger.warning(*args, **kwargs)
self.warn_stream = _warn_patch
def _error_patch(
*args: Any,
source: str | None = None,
**kwargs: Any,
) -> None:
self.logger.error(*args, **kwargs)
self.error_stream = _error_patch
def _fatal_patch(
*args: Any,
source: str | None = None,
**kwargs: Any,
) -> None:
self.logger.critical(*args, **kwargs)
self.fatal_stream = _fatal_patch
# -----------------
# Device Properties
# -----------------
LoggingLevelDefault = device_property(
dtype="uint16",
doc="Default logging level at device startup.",
default_value=LoggingLevel.INFO,
)
"""
Device property.
Default logging level at device startup. See
:py:class:`~ska_control_model.LoggingLevel`
"""
LoggingTargetsDefault = device_property(
dtype="DevVarStringArray",
doc="Default logging targets at device startup.",
default_value=["tango::logger"],
)
"""
Device property.
Default logging targets at device startup. See the project readme
for details.
"""
# -----------
# Init device
# -----------
[docs]
def init_device(self) -> None:
"""
Initialise the tango device after startup.
Subclasses that have no need to override the default
implementation of state management may leave ``init_device()``
alone. Override the ``do()`` method on the nested class
``InitCommand`` instead.
"""
super().init_device()
self._init_logging()
self._build_state = f"{release.name}, {release.version}, {release.description}"
self._version_id = release.version
if hasattr(self, "add_version_info"):
self.add_version_info(release.name, release.version)
self.add_version_info(control_model_name, control_model_version)
if hasattr(self, "__version__"):
self.add_version_info(self.__class__.__name__, self.__version__)
# ---------------
# General methods
# ---------------
[docs]
def set_logging_level(self, value: LoggingLevel) -> None:
"""
Set the logging level for the device.
Both the Python logger and the Tango logger are updated.
:param value: Logging level for logger
:raises LoggingLevelError: for invalid value
"""
try:
lmc_logging_level = LoggingLevel(value)
except ValueError as value_error:
raise LoggingLevelError(
f"Invalid level - {value} - must be one of "
f"{list(LoggingLevel.__members__.values())} "
) from value_error
self._logging_level = lmc_logging_level
self.logger.setLevel(_LMC_TO_PYTHON_LOGGING_LEVEL[lmc_logging_level])
self.logger.tango_logger.set_level( # type: ignore[attr-defined]
_LMC_TO_TANGO_LOGGING_LEVEL[lmc_logging_level]
)
self.logger.info(
"Logging level set to %s on Python and Tango loggers", lmc_logging_level
)
[docs]
def set_logging_targets(self, targets: list[str]) -> None:
"""
Set the additional logging targets for the device.
Note that this excludes the handlers provided by the ska_ser_logging
library defaults.
:param targets: Logging targets for logger
"""
device_name = self.get_name()
valid_targets = LoggingUtils.sanitise_logging_targets(targets, device_name)
LoggingUtils.update_logging_handlers(valid_targets, self.logger)
# ----------
# Attributes
# ----------
@attribute(dtype="str")
def buildState(self) -> str:
"""
Read the Build State of the device.
:return: the build state of the device
"""
return self._build_state
_build_state: str
"""Value returned by the :py:func:`buildState` attribute.
Should be set by subclasses during device initialisation.
Typically this will include the name of the python package, the version number
of that package and a brief description of that package.
:meta public:
"""
@attribute(dtype="str")
def versionId(self) -> str:
"""
Read the Version Id of the device.
:return: the version id of the device
"""
return self._version_id
_version_id: str
"""Value returned by the :py:func:`versionId` attribute.
Should be set by subclasses during device initialisation.
Must be of the form ``"<major>.<minor>.<patch>"``.
:meta public:
"""
@attribute(dtype=LoggingLevel)
def loggingLevel(self) -> LoggingLevel:
"""
Read the logging level of the device.
Initialises to LoggingLevelDefault on startup.
See :py:class:`~ska_control_model.LoggingLevel`
:return: Logging level of the device.
"""
return self._logging_level
@loggingLevel.write # type: ignore[no-redef]
def loggingLevel(self, value: LoggingLevel) -> None:
"""
Set the logging level for the device.
Both the Python logger and the Tango logger are updated.
:param value: Logging level for logger
"""
self.set_logging_level(value)
@attribute(dtype=("str",), max_dim_x=4)
def loggingTargets(self) -> list[str]:
"""
Read the additional logging targets of the device.
Note that this excludes the handlers provided by the ska_ser_logging
library defaults - initialises to LoggingTargetsDefault on startup.
:return: Logging level of the device.
"""
return [str(handler.name) for handler in self.logger.handlers]
@loggingTargets.write # type: ignore[no-redef]
def loggingTargets(self, value: list[str]) -> None:
"""
Set the additional logging targets for the device.
Note that this excludes the handlers provided by the ska_ser_logging
library defaults.
:param value: Logging targets for logger
"""
self.set_logging_targets(value)
# --------
# Commands
# --------
@command(doc_in="Get the version information of the device.", dtype_out=("str",))
@DebugIt()
def GetVersionInfo(self) -> list[str]:
"""
Return the version information of the device.
To modify behaviour for this command, modify the do() method of
the command class.
:return: The device class name and its package information.
"""
return [f"{self.__class__.__name__}, {self._build_state}"]
# ----------
# Run server
# ----------
def main(*args: str, **kwargs: str) -> int:
"""
Entry point for module.
:param args: positional arguments
:param kwargs: named arguments
:return: exit code
"""
return typing.cast(int, SKADevice.run_server(args=args or None, **kwargs))
if __name__ == "__main__":
main()