StatsCrosslet.py

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

""" OPC-UA client for LOFAR stations crosslet stats

"""

# PyTango imports
import tango
from tango import DebugIt
from tango.server import run
from tango.server import Device
from tango.server import attribute, command
from tango.server import device_property
from tango import AttrQuality, DispLevel, DevState
from tango import AttrWriteType, PipeWriteType
# Additional import
# PROTECTED REGION ID(StatsCrosslet.additionnal_import) ENABLED START #
import opcua
import numpy
import traceback
import threading
from datetime import datetime
# PROTECTED REGION END #    //  StatsCrosslet.additionnal_import

__all__ = ["StatsCrosslet", "main"]


class StatsCrosslet(Device):
    """

    **Properties:**

    - Device Property
        OPC_Server_Name
            - Type:'DevString'
        OPC_Server_Port
            - Type:'DevULong'
        OPC_Time_out
            - Type:'DevDouble'
        Default_pause_time
            - Type:'DevDouble'
        Default_subband
            - Type:'DevULong'
        Default_integration_time
            - Type:'DevDouble'
    """
    # PROTECTED REGION ID(StatsCrosslet.class_variable) ENABLED START #
    client = 0
    ns = 0
    obj = 0
    record_cross = 0
    stop_data_read_loop = False
    data_acquisition_is_active = False
    data_read_loop = threading.Thread()

    def read_data(self):
        # This is the the thread that continuously reads the crosslet
        # statistics from the station and feeds it into the Tango
        # property system.
        self.debug_stream("Entering read_data loop.")
        while self.stop_data_read_loop is False:
            self.debug_stream("read_data is running...")
            threading.Event().wait(self._pause_time)
            if self.data_acquisition_is_active is True:
                try:
                    self.debug_stream("fetching data:  subband = %d, integration_time = %d", self._subband, self._integration_time)
                    t, visibilities_list, rcu_modes = self.obj.call_method(self.record_cross, self._subband, self._integration_time)
                    self.debug_stream("fetching data done: t = %s, len(visibilities_list) = %d, len(rcu_modes) = %d", t, len(visibilities_list), len(rcu_modes))
                    self._time_stamp = t
                    visibilities = numpy.array(visibilities_list)[0] + 1j * numpy.array(visibilities_list[1])
                    self._visibilities_real = visibilities.real
                    self._visibilities_imag = visibilities.imag
                    self._rcu_modes = rcu_modes
                    self.debug_stream("timestamp = %s, visibilities_real [0][1] = %f", self._time_stamp, self._visibilities_real[0][1])
                except KeyboardInterrupt:
                    self.debug_stream("KBD interrupt caught, leaving data read loop.")
                    self.stop_data_read_loop = True
                except Exception as e:
                    self.error_stream("Exception: %s.  Cannot call the method %s in the OPC-UA server %s.  Trace: %s", e, self.record_cross, self.OPC_Server_Name, traceback.format_exc())

    # PROTECTED REGION END #    //  StatsCrosslet.class_variable

    # -----------------
    # Device Properties
    # -----------------

    OPC_Server_Name = device_property(
        dtype='DevString',
        default_value="okeanos"
    )

    OPC_Server_Port = device_property(
        dtype='DevULong',
        default_value=55556
    )

    OPC_Time_out = device_property(
        dtype='DevDouble',
        default_value=1.0
    )

    Default_pause_time = device_property(
        dtype='DevDouble',
        default_value=1.0
    )

    Default_subband = device_property(
        dtype='DevULong',
        default_value=150
    )

    Default_integration_time = device_property(
        dtype='DevDouble',
        default_value=1.0
    )

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

    subband = attribute(
        dtype='DevUShort',
        access=AttrWriteType.READ_WRITE,
    )

    integration_time = attribute(
        dtype='DevDouble',
    )

    time_stamp = attribute(
        dtype='DevString',
    )

    pause_time = attribute(
        dtype='DevDouble',
        access=AttrWriteType.READ_WRITE,
    )

    rcu_modes = attribute(
        dtype=('DevString',),
        max_dim_x=96,
    )

    visibilities_imag = attribute(
        dtype=(('DevDouble',),),
        max_dim_x=96, max_dim_y=96,
    )

    visibilities_real = attribute(
        dtype=(('DevDouble',),),
        max_dim_x=96, max_dim_y=96,
    )

    # ---------------
    # General methods
    # ---------------

    def init_device(self):
        """Initialises the attributes and properties of the StatsCrosslet."""
        Device.init_device(self)
        # PROTECTED REGION ID(StatsCrosslet.init_device) ENABLED START #
        try:
            self.debug_stream("Connecting to OPC-UA server %s:%d...", self.OPC_Server_Name, self.OPC_Server_Port)
            self.client = opcua.Client("opc.tcp://{}:{}/".format(self.OPC_Server_Name, self.OPC_Server_Port), self.OPC_Time_out * 1000)
            self.client.connect()
            ns = self.client.get_namespace_index("http://lofar.eu")
            self.obj = self.client.get_root_node().get_child(["0:Objects", "{}:StationMetrics".format(ns),                                            "{}:RCU".format(ns)])
            self.record_cross = "{}:record_cross".format(ns)
            self.debug_stream("Connecting to OPC-UA server %s:%d done.", self.OPC_Server_Name, self.OPC_Server_Port)
            # Set default values in the read/write attributes.
            self._pause_time = self.Default_pause_time
            self._integration_time = self.Default_integration_time
            self._subband = self.Default_subband

            self.data_read_loop = threading.Thread(target = self.read_data)
            self.data_acquisition_is_active = False
            self.stop_data_read_loop = False
            self.data_read_loop.start()
        except Exception as e:
            self.error_stream("Cannot connect to the OPC-UA server %s.  Trace: %s" % (self.OPC_Server_Name, traceback.format_exc()))
            raise e
        # PROTECTED REGION END #    //  StatsCrosslet.init_device

    def always_executed_hook(self):
        """Method always executed before any TANGO command is executed."""
        # PROTECTED REGION ID(StatsCrosslet.always_executed_hook) ENABLED START #
        # PROTECTED REGION END #    //  StatsCrosslet.always_executed_hook

    def delete_device(self):
        """Hook to delete resources allocated in init_device.

        This method allows for any memory or other resources allocated in the
        init_device method to be released.  This method is called by the device
        destructor and by the device Init command.
        """
        # PROTECTED REGION ID(StatsCrosslet.delete_device) ENABLED START #
        self.debug_stream("Shutting down...")
        self.data_acquisition_is_active = False
        self.stop_data_read_loop = True
        self.debug_stream("Waiting for data acquisition thread to shut down...")
        self.data_read_loop.join()
        self.debug_stream("Waiting for data acquisition thread to shut down, done")
        if self.client is not None:
            self.client.close_session()
            self.client.close_secure_channel()
        self.debug_stream("Shutdown done.")
        # PROTECTED REGION END #    //  StatsCrosslet.delete_device
    # ------------------
    # Attributes methods
    # ------------------

    def read_subband(self):
        # PROTECTED REGION ID(StatsCrosslet.subband_read) ENABLED START #
        """Return the subband attribute."""
        return self._subband
        # PROTECTED REGION END #    //  StatsCrosslet.subband_read

    def write_subband(self, value):
        # PROTECTED REGION ID(StatsCrosslet.subband_write) ENABLED START #
        """Set the subband attribute."""
        self._subband = value
        # PROTECTED REGION END #    //  StatsCrosslet.subband_write

    def is_subband_allowed(self, attr):
        # PROTECTED REGION ID(StatsCrosslet.is_subband_allowed) ENABLED START #
        if attr==attr.READ_REQ:
            return self.get_state() not in [DevState.STANDBY]
        else:
            return self.get_state() not in [DevState.STANDBY]
        # PROTECTED REGION END #    //  StatsCrosslet.is_subband_allowed

    def read_integration_time(self):
        # PROTECTED REGION ID(StatsCrosslet.integration_time_read) ENABLED START #
        """Return the integration_time attribute."""
        return self._integration_time
        # PROTECTED REGION END #    //  StatsCrosslet.integration_time_read

    def is_integration_time_allowed(self, attr):
        # PROTECTED REGION ID(StatsCrosslet.is_integration_time_allowed) ENABLED START #
        return self.get_state() not in [DevState.ON,DevState.OFF,DevState.INIT]
        # PROTECTED REGION END #    //  StatsCrosslet.is_integration_time_allowed

    def read_time_stamp(self):
        # PROTECTED REGION ID(StatsCrosslet.time_stamp_read) ENABLED START #
        """Return the time_stamp attribute."""
        return self._time_stamp
        # PROTECTED REGION END #    //  StatsCrosslet.time_stamp_read

    def is_time_stamp_allowed(self, attr):
        # PROTECTED REGION ID(StatsCrosslet.is_time_stamp_allowed) ENABLED START #
        return self.get_state() not in [DevState.ON,DevState.OFF,DevState.INIT]
        # PROTECTED REGION END #    //  StatsCrosslet.is_time_stamp_allowed

    def read_pause_time(self):
        # PROTECTED REGION ID(StatsCrosslet.pause_time_read) ENABLED START #
        """Return the pause_time attribute."""
        return self._pause_time
        # PROTECTED REGION END #    //  StatsCrosslet.pause_time_read

    def write_pause_time(self, value):
        # PROTECTED REGION ID(StatsCrosslet.pause_time_write) ENABLED START #
        """Set the pause_time attribute."""
        self._pause_time = value
        # PROTECTED REGION END #    //  StatsCrosslet.pause_time_write

    def read_rcu_modes(self):
        # PROTECTED REGION ID(StatsCrosslet.rcu_modes_read) ENABLED START #
        """Return the rcu_modes attribute."""
        return self._rcu_modes
        # PROTECTED REGION END #    //  StatsCrosslet.rcu_modes_read

    def is_rcu_modes_allowed(self, attr):
        # PROTECTED REGION ID(StatsCrosslet.is_rcu_modes_allowed) ENABLED START #
        return self.get_state() not in [DevState.ON,DevState.OFF,DevState.INIT]
        # PROTECTED REGION END #    //  StatsCrosslet.is_rcu_modes_allowed

    def read_visibilities_imag(self):
        # PROTECTED REGION ID(StatsCrosslet.visibilities_imag_read) ENABLED START #
        """Return the visibilities_imag attribute."""
        return self._visibilities_imag
        # PROTECTED REGION END #    //  StatsCrosslet.visibilities_imag_read

    def is_visibilities_imag_allowed(self, attr):
        # PROTECTED REGION ID(StatsCrosslet.is_visibilities_imag_allowed) ENABLED START #
        return self.get_state() not in [DevState.ON,DevState.OFF,DevState.INIT]
        # PROTECTED REGION END #    //  StatsCrosslet.is_visibilities_imag_allowed

    def read_visibilities_real(self):
        # PROTECTED REGION ID(StatsCrosslet.visibilities_real_read) ENABLED START #
        """Return the visibilities_real attribute."""
        return self._visibilities_real
        # PROTECTED REGION END #    //  StatsCrosslet.visibilities_real_read

    def is_visibilities_real_allowed(self, attr):
        # PROTECTED REGION ID(StatsCrosslet.is_visibilities_real_allowed) ENABLED START #
        return self.get_state() not in [DevState.ON,DevState.OFF,DevState.INIT]
        # PROTECTED REGION END #    //  StatsCrosslet.is_visibilities_real_allowed

    # --------
    # Commands
    # --------

    @command(
    )
    @DebugIt()
    def start_acquisition(self):
        # PROTECTED REGION ID(StatsCrosslet.start_acquisition) ENABLED START #
        """
        Start the data acquisition of the station`s crosslet stats.

        :param argin: 'DevULong'

        :return:None
        """
        if is_start_acquisition_allowed() is True:
            self.data_acquisition_is_active = True
        # PROTECTED REGION END #    //  StatsCrosslet.start_acquisition

    def is_start_acquisition_allowed(self):
        # PROTECTED REGION ID(StatsCrosslet.is_start_acquisition_allowed) ENABLED START #
        return self.get_state() not in [DevState.ON,DevState.OFF,DevState.STANDBY,DevState.RUNNING]
        # PROTECTED REGION END #    //  StatsCrosslet.is_start_acquisition_allowed

    @command(
    )
    @DebugIt()
    def stop_acquisition(self):
        # PROTECTED REGION ID(StatsCrosslet.stop_acquisition) ENABLED START #
        """
        Stop the data acquisition.

        :return:None
        """
        if is_stop_acquisition_allowed() is True:
            self.data_acquisition_is_active = False
        # PROTECTED REGION END #    //  StatsCrosslet.stop_acquisition

    def is_stop_acquisition_allowed(self):
        # PROTECTED REGION ID(StatsCrosslet.is_stop_acquisition_allowed) ENABLED START #
        return self.get_state() not in [DevState.ON,DevState.INIT,DevState.RUNNING]
        # PROTECTED REGION END #    //  StatsCrosslet.is_stop_acquisition_allowed

# ----------
# Run server
# ----------


def main(args=None, **kwargs):
    """Main function of the StatsCrosslet module."""
    # PROTECTED REGION ID(StatsCrosslet.main) ENABLED START #
    return run((StatsCrosslet,), args=args, **kwargs)
    # PROTECTED REGION END #    //  StatsCrosslet.main


if __name__ == '__main__':
    main()