test_device.py

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

""" test Device Server
"""

# PyTango imports
from tango.server import run
from tango.server import device_property
# Additional import

<<<<<<< HEAD
from clients.SNMP_client import SNMP_client
from src.attribute_wrapper import *
from src.hardware_device import *
=======
from clients.test_client import test_client
from util.attribute_wrapper import *
from util.hardware_device import *
>>>>>>> master

__all__ = ["example_device", "main"]


class example_device(hardware_device):


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

    SNMP_community = b"public"
    SNMP_host = "127.0.0.1"
    SNMP_timeout = 5.0

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

    # simple scalar
    attr1 = attribute_wrapper(comms_annotation={"oids": "1.3.6.1.2.1.1.6.0"}, datatype=numpy.bool_, access=AttrWriteType.READ_WRITE)
    # simple scalar with host
    attr2 = attribute_wrapper(comms_annotation={"oids": "1.3.6.1.2.1.1.5.0"}, datatype=numpy.bool_, access=AttrWriteType.READ_WRITE)

    #spectrum with all elements
    attr3 = attribute_wrapper(comms_annotation={"oids": ["1.3.6.1.2.1.1.5.1", "1.3.6.1.2.1.1.5.2", "1.3.6.1.2.1.1.5.3"]}, dims=(3,), datatype=numpy.bool_)
    #inferred spectrum
    attr4 = attribute_wrapper(comms_annotation={"oids": ["1.3.6.1.2.1.1.5.0"]}, dims=(3,), datatype=numpy.bool_)


    def always_executed_hook(self):
        """Method always executed before any TANGO command is executed."""
        pass

    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 (a Tango built-in).
		"""
        self.debug_stream("Shutting down...")

        self.Off()
        self.debug_stream("Shut down.  Good bye.")

    # --------
    # overloaded functions
    # --------
    def initialise(self):
        """ user code here. is called when the state is set to STANDBY """

        # set up the SNMP ua client
        self.snmp_manager = SNMP_client(self.SNMP_community, self.SNMP_host, self.SNMP_timeout, self.Fault, self)

<<<<<<< HEAD
        # map the attributes to the OPC ua comm client
        for i in self.attr_list():
            i.set_comm_client(self.snmp_manager)

        self.snmp_manager.start()

# --------
# Commands
# --------
=======
        # set up the test client
        self.test_client = test_client(self.Fault, self)

        # map an access helper class
        for i in self.attr_list():
            i.set_comm_client(self.test_client)

        self.test_client.start()
>>>>>>> master


# ----------
# Run server
# ----------
def main(args=None, **kwargs):
    """Main function of the PCC module."""
    return run((example_device,), args=args, **kwargs)


if __name__ == '__main__':
    main()