SST.py

# -*- coding: utf-8 -*-
#
# This file is part of the SST project
#
#
#
# Distributed under the terms of the APACHE license.
# See LICENSE.txt for more info.

""" SST Device Server for LOFAR2.0

"""

# PyTango imports
from tango.server import run
from tango.server import device_property, attribute
from tango import AttrWriteType
# Additional import

from clients.sst_client import sst_client, SST_collector

from util.attribute_wrapper import attribute_wrapper
from util.hardware_device import hardware_device
from util.lofar_git import get_version

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,
        default_value=5001
    )

    # ----------
    # Attributes
    # ----------

    version_R = attribute(dtype = str, access = AttrWriteType.READ, fget = lambda self: get_version())

    # number of UDP packets that were received
    nof_packets_received_R  = attribute_wrapper(comms_annotation={"type": "udp", "parameter": "nof_packets_received"}, datatype=numpy.uint64)
    # number of UDP packets that were dropped because we couldn't keep up with processing
    nof_packets_dropped_R   = attribute_wrapper(comms_annotation={"type": "udp", "parameter": "nof_packets_dropped"}, datatype=numpy.uint64)
    # last packet we processed
    last_packet_R           = attribute_wrapper(comms_annotation={"type": "udp", "parameter": "last_packet"}, dims=(9000,), datatype=numpy.uint8)
    # when last packet was received
    last_packet_timestamp_R = attribute_wrapper(comms_annotation={"type": "udp", "parameter": "last_packet_timestamp"}, datatype=numpy.uint64)

    # number of UDP packets that were processed
    nof_packets_processed_R = attribute_wrapper(comms_annotation={"type": "sst", "parameter": "nof_packets"}, datatype=numpy.uint64)
    # queue fill percentage, as reported by the consumer
    queue_fill_percentage_R = attribute_wrapper(comms_annotation={"type": "queue", "parameter": "fill_percentage"}, datatype=numpy.uint64)

    # number of invalid (non-SST) packets received
    nof_invalid_packets_R   = attribute_wrapper(comms_annotation={"type": "sst", "parameter": "nof_invalid_packets"}, datatype=numpy.uint64)
    # last packet that could not be parsed
    last_invalid_packet_R   = attribute_wrapper(comms_annotation={"type": "sst", "parameter": "last_invalid_packet"}, dims=(9000,), datatype=numpy.uint8)
    # number of packets with valid payloads
    nof_valid_payloads_R    = attribute_wrapper(comms_annotation={"type": "sst", "parameter": "nof_valid_payloads"}, dims=(SST_collector.MAX_INPUTS,), datatype=numpy.uint64)
    # number of packets with invalid payloads
    nof_payload_errors_R    = attribute_wrapper(comms_annotation={"type": "sst", "parameter": "nof_payload_errors"}, dims=(SST_collector.MAX_INPUTS,), datatype=numpy.uint64)
    # latest SSTs
    sst_R                   = attribute_wrapper(comms_annotation={"type": "sst", "parameter": "sst_values"}, dims=(SST_collector.MAX_SUBBANDS, SST_collector.MAX_INPUTS), datatype=numpy.uint64)
    # reported timestamp for each row in the latest SSTs
    sst_timestamp_R         = attribute_wrapper(comms_annotation={"type": "sst", "parameter": "sst_timestamps"}, dims=(SST_collector.MAX_INPUTS,), datatype=numpy.uint64)
    # integration interval for each row in the latest SSTs
    integration_interval_R  = attribute_wrapper(comms_annotation={"type": "sst", "parameter": "integration_intervals"}, dims=(SST_collector.MAX_INPUTS,), datatype=numpy.float32)

    # --------
    # 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 SST Device module."""

    from util.lofar_logging import configure_logger
    import logging
    configure_logger(logging.getLogger())

    return run((SST,), args=args, **kwargs)


if __name__ == '__main__':
    main()