Merge branch 'master' into L2SS-234-inspect-archive-from-python

CDB/LOFAR_ConfigDb.json

@@ -715,8 +715,17 @@
                 "SST": {
                     "LTS/SST/1": {
                         "properties": {
-                            "SST_Port": [
+                            "SST_Client_Port": [
                                 "5001"
+                            ],
+                            "OPC_Server_Name": [
+                                "dop36.astron.nl"
+                            ],
+                            "OPC_Server_Port": [
+                                "4840"
+                            ],
+                            "OPC_Time_Out": [
+                                "5.0"
                             ]
                         }
                     }

CDB/sdp-sim-config.json

@@ -18,6 +18,28 @@
                     }
                 }
             }
+        },
+        "SST": {
+            "LTS": {
+                "SST": {
+                    "LTS/SST/1": {
+                        "properties": {
+                            "SST_Client_Port": [
+                                "5001"
+                            ],
+                            "OPC_Server_Name": [
+                                "sdptr-sim"
+                            ],
+                            "OPC_Server_Port": [
+                                "4840"
+                            ],
+                            "OPC_Time_Out": [
+                                "5.0"
+                            ]
+                        }
+                    }
+                }
+            }
         }
     }
 }

CDB/thijs_ConfigDb.json

@@ -94,8 +94,17 @@
                 "SST": {
                     "LTS/SST/1": {
                         "properties": {
-                            "SST_Port": [
+                            "SST_Client_Port": [
                                 "5001"
+                            ],
+                            "OPC_Server_Name": [
+                                "dop36.astron.nl"
+                            ],
+                            "OPC_Server_Port": [
+                                "4840"
+                            ],
+                            "OPC_Time_Out": [
+                                "5.0"
                             ]
                         }
                     }

CDB/thomas_ConfigDb.json

@@ -25,6 +25,19 @@
                     }
                 }
             }
+        },
+        "SST": {
+            "LTS": {
+                "SST": {
+                    "LTS/SST/1": {
+                        "properties": {
+                            "OPC_Server_Name": [
+                                "okeanos"
+                            ]
+                        }
+                    }
+                }
+            }
         }
     }
 }

CDB/thomas_arm64_ConfigDb.json

@@ -25,6 +25,19 @@
                     }
                 }
             }
+        },
+        "SST": {
+            "LTS": {
+                "SST": {
+                    "LTS/SST/1": {
+                        "properties": {
+                            "OPC_Server_Name": [
+                                "okeanos"
+                            ]
+                        }
+                    }
+                }
+            }
         }
     }
 }

devices/clients/attribute_wrapper.py

@@ -14,11 +14,12 @@ class attribute_wrapper(attribute):
     Wraps all the attributes in a wrapper class to manage most of the redundant code behind the scenes
     """
 
-    def __init__(self, comms_annotation=None, datatype=None, dims=(1,), access=AttrWriteType.READ, init_value=None, **kwargs):
+    def __init__(self, comms_id=None, comms_annotation=None, datatype=None, dims=(1,), access=AttrWriteType.READ, init_value=None, **kwargs):
         """
         wraps around the tango Attribute class. Provides an easier interface for 1d or 2d arrays. Also provides a way to abstract
         managing the communications interface.
 
+        comms_id: user-supplied identifier that is attached to this object, to identify which communication class will need to be attached
         comms_annotation: data passed along to the attribute. can be given any form of data. handling is up to client implementation
         datatype: any numpy datatype
         dims: dimensions of the attribute as a tuple, or (1,) for a scalar.
@@ -31,6 +32,7 @@ class attribute_wrapper(attribute):
         if "numpy" not in str(datatype) and datatype != str:
             raise TypeError("Attribute needs to be a Tango-supported numpy or str type, but has type \"%s\"" % (datatype,))
 
+        self.comms_id = comms_id # store data that can be used to identify the comms interface to use. not used by the wrapper itself
         self.comms_annotation = comms_annotation  # store data that can be used by the comms interface. not used by the wrapper itself
         self.numpy_type = datatype  # tango changes our attribute to their representation (E.g numpy.int64 becomes "DevLong64")
 

devices/clients/comms_client.py

@@ -65,6 +65,9 @@ class CommClient(Thread):
                 # signal that we're disconnected
                 self.fault_func()
 
+                # don't enter a spam-connect loop if faults immediately occur
+                time.sleep(self.try_interval)
+
     def ping(self):
         return
 

devices/clients/opcua_client.py

@@ -225,7 +225,37 @@ class ProtocolAttribute:
             # make sure it is a python array
             value = value.tolist() if type(value) == numpy.ndarray else value
 
+        try:
             self.node.set_data_value(opcua.ua.uatypes.Variant(value=value, varianttype=self.ua_type))
+            raise TypeError
+        except (TypeError, opcua.ua.uaerrors.BadTypeMismatch) as e:
+            # A type conversion went wrong or there is a type mismatch.
+            #
+            # This is either the conversion us -> opcua in our client, or client -> server.
+            # Report all types involved to allow assessment of the location of the error.
+            if type(value) == list:
+                our_type = "list({dtype}) x ({dimensions})".format(
+                    dtype=(type(value[0]).__name__ if value else ""),
+                    dimensions=len(value))
+            else:
+                our_type = "{dtype}".format(
+                    dtype=type(value))
+
+            is_scalar = (self.dim_x + self.dim_y) == 1
+
+            if is_scalar:
+                expected_server_type = "{dtype} (scalar)".format(
+                    dtype=self.ua_type)
+            else:
+                expected_server_type = "{dtype} x ({dim_x}, {dim_y})".format(
+                    dtype=self.ua_type,
+                    dim_x=self.dim_x,
+                    dim_y=self.dim_y)
 
+            actual_server_type = "{dtype} {dimensions}".format(
+                dtype=self.node.get_data_type_as_variant_type(),
+                dimensions=(self.node.get_array_dimensions() or "???"))
 
+            attribute_name = self.node.get_display_name().to_string()
 
+            raise TypeError(f"Cannot write value to OPC-UA attribute '{attribute_name}': tried to convert data type {our_type} to expected server type {expected_server_type}, server reports type {actual_server_type}") from e

devices/common/lofar_logging.py

@@ -9,6 +9,9 @@ hostname = socket.gethostname()
 def configure_logger(logger: logging.Logger, log_extra=None):
     logger.setLevel(logging.DEBUG)
 
+    # remove spam from the OPC-UA client connection
+    logging.getLogger("opcua").setLevel(logging.WARN)
+
     try:
         from logstash_async.handler import AsynchronousLogstashHandler, LogstashFormatter
 
@@ -27,7 +30,7 @@ def configure_logger(logger: logging.Logger, log_extra=None):
         # easily grep'ed, be parsed with a couple of shell commands and
         # easily fed into an Kibana/Elastic search system.
         handler = logging.StreamHandler()
-        formatter = logging.Formatter(fmt = '%(asctime)s.%(msecs)d %(levelname)s - HOST="{}" PID="%(process)d" TNAME="%(threadName)s" TID="%(thread)d" FILE="%(pathname)s" LINE="%(lineno)d" FUNC="%(funcName)s" MSG="%(message)s"'.format(hostname), datefmt = '%Y-%m-%dT%H:%M:%S')
+        formatter = logging.Formatter(fmt = '%(asctime)s.%(msecs)d %(levelname)s - HOST="{}" PID="%(process)d" TNAME="%(threadName)s" TID="%(thread)d" LOGGER="%(name)s" FILE="%(pathname)s" LINE="%(lineno)d" FUNC="%(funcName)s" MSG="%(message)s"'.format(hostname), datefmt = '%Y-%m-%dT%H:%M:%S')
         handler.setFormatter(formatter)
         logger.addHandler(handler)
     except Exception:

devices/devices/hardware_device.py

@@ -90,7 +90,7 @@ class hardware_device(Device):
         self.set_state(DevState.STANDBY)
 
     @command()
-    @only_in_states([DevState.STANDBY])
+    @only_in_states([DevState.STANDBY, DevState.ON])
     @DebugIt()
     @fault_on_error()
     @log_exceptions()
@@ -100,6 +100,11 @@ class hardware_device(Device):
 
         :return:None
         """
+        if self.get_state() == DevState.ON:
+            # Already on. Don't complain.
+            logger.warning("Requested to go to ON state, but am already in ON state.")
+            return
+
         self.configure_for_on()
         self.set_state(DevState.ON)
 
@@ -125,7 +130,7 @@ class hardware_device(Device):
         self.set_state(DevState.OFF)
 
     @command()
-    @only_in_states([DevState.ON, DevState.INIT, DevState.STANDBY])
+    @only_in_states([DevState.ON, DevState.INIT, DevState.STANDBY, DevState.FAULT])
     @DebugIt()
     @log_exceptions()
     def Fault(self):
@@ -138,6 +143,11 @@ class hardware_device(Device):
 
         :return:None
         """
+        if self.get_state() == DevState.FAULT:
+            # Already faulting. Don't complain.
+            logger.warning("Requested to go to FAULT state, but am already in FAULT state.")
+            return
+
         self.configure_for_fault()
         self.set_state(DevState.FAULT)
 

devices/devices/sdp/sdp.py

@@ -94,16 +94,6 @@ class SDP(hardware_device):
     FPGA_sdp_info_observation_id_RW = attribute_wrapper(comms_annotation=["2:FPGA_sdp_info_observation_id_RW"], datatype=numpy.uint32, dims=(16,), access=AttrWriteType.READ_WRITE)
     FPGA_sdp_info_station_id_R = attribute_wrapper(comms_annotation=["2:FPGA_sdp_info_station_id_R"], datatype=numpy.uint32, dims=(16,))
     FPGA_sdp_info_station_id_RW = attribute_wrapper(comms_annotation=["2:FPGA_sdp_info_station_id_RW"], datatype=numpy.uint32, dims=(16,), access=AttrWriteType.READ_WRITE)
-    FPGA_sst_offload_enable_R = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_enable_R"], datatype=numpy.bool_, dims=(16,))
-    FPGA_sst_offload_enable_RW = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_enable_RW"], datatype=numpy.bool_, dims=(16,), access=AttrWriteType.READ_WRITE)
-    FPGA_sst_offload_hdr_eth_destination_mac_R = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_hdr_eth_destination_mac_R"], datatype=numpy.str_, dims=(16,))
-    FPGA_sst_offload_hdr_eth_destination_mac_RW = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_hdr_eth_destination_mac_RW"], datatype=numpy.str_, dims=(16,), access=AttrWriteType.READ_WRITE)
-    FPGA_sst_offload_hdr_ip_destination_address_R = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_hdr_ip_destination_address_R"], datatype=numpy.str_, dims=(16,))
-    FPGA_sst_offload_hdr_ip_destination_address_RW = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_hdr_ip_destination_address_RW"], datatype=numpy.str_, dims=(16,), access=AttrWriteType.READ_WRITE)
-    FPGA_sst_offload_hdr_udp_destination_port_R = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_hdr_udp_destination_port_R"], datatype=numpy.uint16, dims=(16,))
-    FPGA_sst_offload_hdr_udp_destination_port_RW = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_hdr_udp_destination_port_RW"], datatype=numpy.uint16, dims=(16,), access=AttrWriteType.READ_WRITE)
-    FPGA_sst_offload_selector_R = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_selector_R"], datatype=numpy.bool_, dims=(16,))
-    FPGA_sst_offload_selector_RW = attribute_wrapper(comms_annotation=["2:FPGA_sst_offload_selector_RW"], datatype=numpy.bool_, dims=(16,), access=AttrWriteType.READ_WRITE)
     FPGA_status_R = attribute_wrapper(comms_annotation=["2:FPGA_status_R"], datatype=numpy.bool_, dims=(16,))
     FPGA_temp_R = attribute_wrapper(comms_annotation=["2:FPGA_temp_R"], datatype=numpy.float_, dims=(16,))
     FPGA_version_R = attribute_wrapper(comms_annotation=["2:FPGA_version_R"], datatype=numpy.str_, dims=(16,))

devices/devices/sdp/sst.py

@@ -25,6 +25,7 @@ from tango import AttrWriteType
 # Additional import
 
 from clients.sst_client import sst_client, SST_collector
+from clients.opcua_connection import OPCUAConnection
 from clients.attribute_wrapper import attribute_wrapper
 
 from devices.hardware_device import hardware_device
@@ -43,7 +44,32 @@ class SST(hardware_device):
     # Device Properties
     # -----------------
 
-    SST_Port = device_property(
+    OPC_Server_Name = device_property(
+        dtype='DevString',
+        mandatory=True
+    )
+
+    OPC_Server_Port = device_property(
+        dtype='DevULong',
+        mandatory=True
+    )
+
+    OPC_Time_Out = device_property(
+        dtype='DevDouble',
+        mandatory=True
+    )
+
+    SST_Client_Name = device_property(
+        dtype='DevString',
+        mandatory=True
+    )
+
+    SST_Client_MAC = device_property(
+        dtype='DevString',
+        mandatory=True
+    )
+
+    SST_Client_Port = device_property(
         dtype='DevUShort',
         mandatory=True
     )
@@ -54,34 +80,46 @@ class SST(hardware_device):
 
     version_R = attribute(dtype = str, access = AttrWriteType.READ, fget = lambda self: get_version())
 
+    # FPGA control points for SSTs
+    FPGA_sst_offload_enable_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_enable_RW"], datatype=numpy.bool_, dims=(16,), access=AttrWriteType.READ_WRITE)
+    FPGA_sst_offload_enable_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_enable_R"], datatype=numpy.bool_, dims=(16,))
+    FPGA_sst_offload_hdr_eth_destination_mac_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_hdr_eth_destination_mac_RW"], datatype=numpy.str_, dims=(16,), access=AttrWriteType.READ_WRITE)
+    FPGA_sst_offload_hdr_eth_destination_mac_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_hdr_eth_destination_mac_R"], datatype=numpy.str_, dims=(16,))
+    FPGA_sst_offload_hdr_ip_destination_address_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_hdr_ip_destination_address_RW"], datatype=numpy.str_, dims=(16,), access=AttrWriteType.READ_WRITE)
+    FPGA_sst_offload_hdr_ip_destination_address_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_hdr_ip_destination_address_R"], datatype=numpy.str_, dims=(16,))
+    FPGA_sst_offload_hdr_udp_destination_port_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_hdr_udp_destination_port_RW"], datatype=numpy.uint16, dims=(16,), access=AttrWriteType.READ_WRITE)
+    FPGA_sst_offload_hdr_udp_destination_port_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_hdr_udp_destination_port_R"], datatype=numpy.uint16, dims=(16,))
+    FPGA_sst_offload_selector_RW = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_selector_RW"], datatype=numpy.bool_, dims=(16,), access=AttrWriteType.READ_WRITE)
+    FPGA_sst_offload_selector_R = attribute_wrapper(comms_id=OPCUAConnection, comms_annotation=["2:FPGA_sst_offload_selector_R"], datatype=numpy.bool_, dims=(16,))
+
     # number of UDP packets that were received
-    nof_packets_received_R  = attribute_wrapper(comms_annotation={"type": "udp", "parameter": "nof_packets_received"}, datatype=numpy.uint64)
+    nof_packets_received_R  = attribute_wrapper(comms_id=sst_client, 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)
+    nof_packets_dropped_R   = attribute_wrapper(comms_id=sst_client, 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)
+    last_packet_R           = attribute_wrapper(comms_id=sst_client, 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)
+    last_packet_timestamp_R = attribute_wrapper(comms_id=sst_client, 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)
+    nof_packets_processed_R = attribute_wrapper(comms_id=sst_client, 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)
+    queue_fill_percentage_R = attribute_wrapper(comms_id=sst_client, 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)
+    nof_invalid_packets_R   = attribute_wrapper(comms_id=sst_client, 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)
+    last_invalid_packet_R   = attribute_wrapper(comms_id=sst_client, 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)
+    nof_valid_payloads_R    = attribute_wrapper(comms_id=sst_client, 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)
+    nof_payload_errors_R    = attribute_wrapper(comms_id=sst_client, 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)
+    sst_R                   = attribute_wrapper(comms_id=sst_client, 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)
+    sst_timestamp_R         = attribute_wrapper(comms_id=sst_client, 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)
+    integration_interval_R  = attribute_wrapper(comms_id=sst_client, comms_annotation={"type": "sst", "parameter": "integration_intervals"}, dims=(SST_collector.MAX_INPUTS,), datatype=numpy.float32)
 
     # --------
     # overloaded functions
@@ -94,17 +132,27 @@ class SST(hardware_device):
         except Exception as e:
             self.warn_stream("Exception while stopping sst_client in configure_for_off function: {}. Exception ignored".format(e))
 
+        try:
+            self.opcua_connection.stop()
+        except Exception as e:
+            self.warn_stream("Exception while stopping OPC UA connection 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)
+        self.sst_client = sst_client("0.0.0.0", self.SST_Client_Port, self.Fault, self)
+
+        self.OPCua_client = OPCUAConnection("opc.tcp://{}:{}/".format(self.OPC_Server_Name, self.OPC_Server_Port), "http://lofar.eu", self.OPC_Time_Out, self.Fault, self)
 
         # map an access helper class
         for i in self.attr_list():
             try:
+                if i.comms_id == sst_client:
                     i.set_comm_client(self.sst_client)
+                if i.comms_id == OPCUAConnection:
+                    i.set_comm_client(self.OPCUA_client)
             except Exception as e:
                 # use the pass function instead of setting read/write fails
                 i.set_pass_func()
@@ -113,6 +161,8 @@ class SST(hardware_device):
 
         self.sst_client.start()
 
+        self.OPCua_client.start()
+
     # --------
     # Commands
     # --------

devices/tox.ini

@@ -24,9 +24,13 @@ commands =
 ;    doc8 doc/source/ README.rst
     flake8
 
-[testenv:bandit]
+[testenv:bandit];
+; B104: hardcoded_bind_all_interfaces
+;           - We disable this warning as Docker serves as our firewall.
+;             It thus matters what interfaces Docker will bind our
+;             containers to, not what our containers listen on.
 commands =
-    bandit -r devices/ clients/ common/ examples/ util/ -n5 -ll
+    bandit -r devices/ clients/ common/ examples/ util/ -n5 -ll -s B104
 
 [flake8]
 filename = *.py,.stestr.conf,.txt

docker-compose/jupyter/Dockerfile

@@ -8,6 +8,8 @@ ARG CONTAINER_EXECUTION_UID=1000
 RUN sudo pip3 install jupyter
 RUN sudo pip3 install ipykernel
 RUN sudo pip3 install jupyter_bokeh
+# Install matplotlib, jupyterplot
+RUN sudo pip3 install matplotlib jupyterplot
 
 # Configure jupyter_bokeh
 RUN sudo mkdir -p /usr/share/jupyter /usr/etc