Skip to content
Snippets Groups Projects
Select Git revision
  • 17aa5689ee3f7fa10ba476abc04055bc4ca6500a
  • master default protected
  • L2SS-1914-fix_job_dispatch
  • TMSS-3170
  • TMSS-3167
  • TMSS-3161
  • TMSS-3158-Front-End-Only-Allow-Changing-Again
  • TMSS-3133
  • TMSS-3319-Fix-Templates
  • test-fix-deploy
  • TMSS-3134
  • TMSS-2872
  • defer-state
  • add-custom-monitoring-points
  • TMSS-3101-Front-End-Only
  • TMSS-984-choices
  • SDC-1400-Front-End-Only
  • TMSS-3079-PII
  • TMSS-2936
  • check-for-max-244-subbands
  • TMSS-2927---Front-End-Only-PXII
  • Before-Remove-TMSS
  • LOFAR-Release-4_4_318 protected
  • LOFAR-Release-4_4_317 protected
  • LOFAR-Release-4_4_316 protected
  • LOFAR-Release-4_4_315 protected
  • LOFAR-Release-4_4_314 protected
  • LOFAR-Release-4_4_313 protected
  • LOFAR-Release-4_4_312 protected
  • LOFAR-Release-4_4_311 protected
  • LOFAR-Release-4_4_310 protected
  • LOFAR-Release-4_4_309 protected
  • LOFAR-Release-4_4_308 protected
  • LOFAR-Release-4_4_307 protected
  • LOFAR-Release-4_4_306 protected
  • LOFAR-Release-4_4_304 protected
  • LOFAR-Release-4_4_303 protected
  • LOFAR-Release-4_4_302 protected
  • LOFAR-Release-4_4_301 protected
  • LOFAR-Release-4_4_300 protected
  • LOFAR-Release-4_4_299 protected
41 results

radb_common_testing.py

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    CentralLogger.py 3.00 KiB
    #!/usr/bin/env python
    # -*- coding: utf-8 -*-
    
    """CentralLogger device server receiving logs from SendingDS server"""
    import sys
    import time
    import logging
    from PyTango import server, DeviceProxy, Database, DbDevInfo, DevState, DebugIt, AttrQuality, AttrWriteType, DispLevel 
    from PyTango.server import Device, DeviceMeta, attribute, command, run, device_property
    
    
    """CentralLogger device server class"""
    class CentralLogger(Device):
        __metaclass__ = DeviceMeta
    
        def init_device(self):
            Device.init_device(self)
            self.set_state(DevState.STANDBY)
            self.info_stream("Init CentralLogger Device.")
    
        @command(dtype_in='DevVarLongStringArray', dtype_out=None)
        def SetCentralLoggingLevel(self, element_devices_and_levels):
            CentralLoggingLevel = element_devices_and_levels[0][:]
            CentralLoggingDevice = element_devices_and_levels[1][:]
            i = 0
            while i < len(CentralLoggingLevel[:]):
                self.info_stream("%s,%s", CentralLoggingLevel[i], CentralLoggingDevice[i])
                dev1 = DeviceProxy(CentralLoggingDevice[i])
                dev1.centralLoggingLevel = CentralLoggingLevel[i]
                property_names = ["logging_level",
                                  "logging_target",
                                 ]
                dev_properties = dev1.get_property(property_names)
                dev_properties["logging_level"] = ["DEBUG"]
                dev_properties["logging_target"].append("device::central/cdev/cdev1")
                dev1.put_property(dev_properties)
                dev1.add_logging_target("device::central/cdev/cdev1")
                i+=1
    
        """Logs are received from all element devices at all levels.
           Filtering happens within the log function depending upon
           CentralLoggingLevel of each element device.
           NOTE : This fiteration logic is still under consideration."""
        @command(dtype_in='DevVarStringArray', dtype_out=None)
        def log(self, details):
            cmessage = details[3]
            clevel = details[1]
            tango_log_level = {"FATAL": 1, "ERROR": 2, "WARN": 3, "INFO": 4, "DEBUG": 5}
            level_number = tango_log_level[clevel]
            clogsource = details[2]
            if clogsource == details[2]:
                device = DeviceProxy(clogsource)
                deviceLogLevel = device.centralLoggingLevel
    
                if clevel == "FATAL" and level_number <= deviceLogLevel:
                    self.fatal_stream("%s : %s", clogsource, cmessage)
                elif clevel == "ERROR" and level_number <= deviceLogLevel:
                    self.error_stream("%s : %s", clogsource, cmessage)
                elif clevel == "WARN" and level_number <= deviceLogLevel:
                    self.warn_stream("%s : %s", clogsource, cmessage)
                elif clevel == "INFO" and level_number <= deviceLogLevel:
                    self.info_stream("%s : %s", clogsource, cmessage)
                elif clevel == "DEBUG" and level_number <= deviceLogLevel:
                    self.debug_stream("%s : %s", clogsource, cmessage)
                else:
                    pass
            else:
                pass
    
    run((CentralLogger,))