Skip to content
Snippets Groups Projects
Select Git revision
  • 3fea9d2a47ec233404596e8e02c5a4a8458ce500
  • master default protected
  • rap-1132-create-dynspec-step
  • svp_cobalt
  • gec-136-set-chain-info
  • integrate_predict_library
  • merl-105-dp3-python-move-not-copy
  • new-everybeam-interface
  • padre-filestream-input
  • ast-1238-use-xtensor-char
  • rap-1044-data-interpolation-step
  • fix-ddecal-docs
  • two-step-faraday-constraint
  • line-search
  • azelgeo-revised
  • SVP
  • test_everybeam_multifreq
  • clipper_baseline_selection
  • ast-1606-apply-calibration-solutions-to-facets-from-image-based-predict-rebased
  • gsl-antenna-solver
  • test-build-sanitizer
  • v6.4.1
  • v6.4
  • v6.3
  • v6.2.1
  • v6.2
  • v6.1
  • v6.0.1
  • v6.0
  • v5.3
  • v5.2
  • v5.1
  • v5.0
  • v4.2
  • v4.1
  • v4.0
  • LOFAR-Release-3_1_0
  • LOFAR-Release-3_1_1
  • LOFAR-Release-3_1_2
  • LOFAR-Release-3_1_3
  • LOFAR-Release-3_1_4
41 results

ParameterSet.cc

Blame
  • Code owners
    Assign users and groups as approvers for specific file changes. Learn more.
    Statistics.py 2.81 KiB
    # -*- coding: utf-8 -*-
    #
    # This file is part of the Statistics project
    #
    #
    #
    # Distributed under the terms of the APACHE license.
    # See LICENSE.txt for more info.
    
    """ Statistics Device Server for LOFAR2.0
    
    """
    
    # PyTango imports
    from tango.server import run
    from tango.server import device_property
    from tango import AttrWriteType
    # Additional import
    
    from clients.sst_client import sst_client
    
    from util.attribute_wrapper import attribute_wrapper
    from util.hardware_device import hardware_device
    
    from util.lofar_logging import device_logging_to_python, log_exceptions
    
    import numpy
    
    __all__ = ["SST", "main"]
    
    @device_logging_to_python({"device": "SST"})
    class SST(hardware_device):
    
        # -----------------
        # Device Properties
        # -----------------
    
        SST_Port = device_property(
            dtype='DevUShort',
            mandatory=True
        )
    
        # ----------
        # Attributes
        # ----------
        # --------
    
        # SST client annotation consists of a dict that contains the parameter name that needs to be read.
        # Example: comms_annotation={"parameter": "this_value_R"}
        packet_count_R = attribute_wrapper(comms_annotation={"parameter": "packet_count_R"}, datatype=numpy.int64)
        last_packet_timestamp_R = attribute_wrapper(comms_annotation={"parameter": "last_packet_timestamp_R"}, datatype=numpy.int64)
        queue_percentage_used_R = attribute_wrapper(comms_annotation={"parameter": "queue_percentage_used_R"}, datatype=numpy.double)
    
        # --------
        # overloaded functions
        def configure_for_off(self):
            """ user code here. is called when the state is set to OFF """
    
            # Stop keep-alive
            try:
                self.sst_client.stop()
            except Exception as e:
                self.warn_stream("Exception while stopping sst_client in configure_for_off function: {}. Exception ignored".format(e))
    
        @log_exceptions()
        def configure_for_initialise(self):
            """ user code here. is called when the sate is set to INIT """
            """Initialises the attributes and properties of the statistics device."""
    
            self.sst_client = sst_client("0.0.0.0", self.SST_Port, self.Fault, self)
    
            # map an access helper class
            for i in self.attr_list():
                try:
                    i.set_comm_client(self.sst_client)
                except Exception as e:
                    # use the pass function instead of setting read/write fails
                    i.set_pass_func()
                    self.warn_stream("error while setting the sst attribute {} read/write function. {}. using pass function instead".format(i, e))
                    pass
    
            self.sst_client.start()
    
        # --------
        # Commands
        # --------
    
    # ----------
    # Run server
    # ----------
    def main(args=None, **kwargs):
        """Main function of the Statistics Device module."""
        return run((SST,), args=args, **kwargs)
    
    
    if __name__ == '__main__':
        main()