diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000000000000000000000000000000000000..8b65da80f55da549ea10337b53b72ff59b99ec3c
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,16 @@
+*log
+__pycache__
+*pyc
+*.bak
+.*.s??
+*.o
+*.a
+*.in
+.deps*
+Makefile
+configure
+config.*
+config/
+aclocal.m4
+autom4te.cache/
+sdptr
\ No newline at end of file
diff --git a/test/py/Client.py b/test/py/Client.py
index bf56e3e30e44e8791834f720e1da05b5b024aafa..aa4e99d2fc65d7038bb8c1e53619614828397708 100644
--- a/test/py/Client.py
+++ b/test/py/Client.py
@@ -19,11 +19,23 @@
# ##########################################################################
# Author:
-# . [name]
+# . Leon Hiemstra, Pieter Donker
# Purpose:
-# . " "
+# . script to test (control/monitor) opc-ua server
# Description:
-# . " "
+# Client.py is used to connect (TCP/IP) to an opc-ua server and test (read/write)
+# opc-ua points (registers on a fpga).
+# For LTS the server is running on DOP36 port 4840, this client can not run
+# on DOP36 itself (too old system).
+#
+# In this version the following is working:
+# . --info, read info from server, list all available points on the server.
+# . --all, read all *_R point and print needed time.
+# . --wg, turn off WG or setup for XST (to generate crosslets plot).
+# . --setup, setup SST, BST and XST stream.
+# . --stream, turn off or on SST, BST or XST stream.
+#
+# . run ./Client.py -h for help
# ##########################################################################
import sys
@@ -199,88 +211,134 @@ def read_subband_weights(obj):
vals = var.get_value()
-@timing
-def write_subband_weights(obj):
- weights = [8192, 0, 0, 0] * N_SUB * N_NODES * (S_PN//4)
- #weights = [8192] * N_SUB * N_NODES * S_PN
- print(len(weights))
+def setup_wg_xst_mode(obj):
+ '''
+ setup wg for xst stream
+ '''
+ # Write WG configuration with phases from 0 - 360 and 1 to 1/12 amplitudes
+ wg_ampl = [0.01 * (1 - (i / S_PN)) for i in range(S_PN)] * N_NODES
+ var = obj.get_child("2:FPGA_wg_amplitude_RW")
+ var.set_value(ua.Variant(value=list(wg_ampl), varianttype=ua.VariantType.Double))
+
+ wg_phase = [i * (360 / S_PN) for i in range(S_PN)] * N_NODES
+ var = obj.get_child("2:FPGA_wg_phase_RW")
+ var.set_value(ua.Variant(value=list(wg_phase), varianttype=ua.VariantType.Double))
+
+ wg_freq = [(102 / 1024) * 200e6 for i in range(S_PN)] * N_NODES
+ var = obj.get_child("2:FPGA_wg_frequency_RW")
+ var.set_value(ua.Variant(value=list(wg_freq), varianttype=ua.VariantType.Double))
+
+ print("wg on")
+ enable = [True for i in range(S_PN)] * N_NODES # enable wg
+ var = obj.get_child("2:FPGA_wg_enable_RW")
+ var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
+
+
+def turn_wg_off(obj):
+ print("wg off")
+ enable = [False for i in range(S_PN)] * N_NODES # enable wg
+ var = obj.get_child("2:FPGA_wg_enable_RW")
+ var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
+
+
+def setup_sst_stream(obj):
+ '''
+ setup bst stream, use git/upe_gear/peripherals/pi_satistics_stream.py for recording and plotting.
+ - pi_statistic_stream.py --unb2 2 --pn2 0 --cmd 4 -s=BST
+ '''
+ port = [5001] * N_NODES # use port 5002
+ var = obj.get_child("2:FPGA_sst_offload_hdr_udp_destination_port_RW")
+ var.set_value(ua.Variant(value=list(port), varianttype=ua.VariantType.UInt16))
+
+ dest_mac = ["00:1B:21:71:76:B9"] * N_NODES # use mac of dop36
+ var = obj.get_child("2:FPGA_sst_offload_hdr_eth_destination_mac_RW")
+ var.set_value(ua.Variant(value=list(dest_mac), varianttype=ua.VariantType.String))
+
+ dest_addr = ["10.99.0.254"] * N_NODES # use addr of dop36
+ var = obj.get_child("2:FPGA_sst_offload_hdr_ip_destination_address_RW")
+ var.set_value(ua.Variant(value=list(dest_addr), varianttype=ua.VariantType.String))
+
+ weights = [8192] * N_SUB * N_NODES * S_PN # set weights for subbands to default
var = obj.get_child("2:FPGA_subband_weights_RW")
- #dv = ua.DataValue(ua.Variant(value=weights, varianttype=ua.VariantType.UInt32))
- #dv.ServerTimestamp = None
- #dv.SourceTimestamp = None
var.set_value(ua.Variant(value=list(weights), varianttype=ua.VariantType.UInt32))
-
-def write_xst_subband_select(obj):
- subsel = [7, 102, 103, 104, 105, 106, 107, 108] * N_NODES # fill all fields
- var = obj.get_child("2:FPGA_xst_subband_select_RW")
- var.set_value(ua.Variant(value=list(subsel), varianttype=ua.VariantType.UInt32))
+ enable = [True] * N_NODES # enable weighted subbands
+ var = obj.get_child("2:FPGA_sst_offload_weighted_subbands_RW")
+ var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
-def write_xst_wg_setup(obj, on=False):
+def setup_bst_stream(obj):
'''
- setup wg for xst stream
+ setup bst stream, use git/upe_gear/peripherals/pi_satistics_stream.py for recording and plotting.
+ - pi_statistic_stream.py --unb2 2 --pn2 0 --cmd 4 -s=BST
'''
- if on is True:
- # Write WG configuration with phases from 0 - 360 and 1 to 1/12 amplitudes
- wg_ampl = [0.01 * (1 - (i / S_PN)) for i in range(S_PN)] * N_NODES
- var = obj.get_child("2:FPGA_wg_amplitude_RW")
- var.set_value(ua.Variant(value=list(wg_ampl), varianttype=ua.VariantType.Double))
- print(len(wg_ampl))
- wg_phase = [i * (360 / S_PN) for i in range(S_PN)] * N_NODES
- var = obj.get_child("2:FPGA_wg_phase_RW")
- var.set_value(ua.Variant(value=list(wg_phase), varianttype=ua.VariantType.Double))
-
- wg_freq = [(102 / 1024) * 200e6 for i in range(S_PN)] * N_NODES
- var = obj.get_child("2:FPGA_wg_frequency_RW")
- var.set_value(ua.Variant(value=list(wg_freq), varianttype=ua.VariantType.Double))
-
- enable = [True for i in range(S_PN)] * N_NODES # enable wg
- var = obj.get_child("2:FPGA_wg_enable_RW")
- var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
- else:
- enable = [False for i in range(S_PN)] * N_NODES # enable wg
- var = obj.get_child("2:FPGA_wg_enable_RW")
- var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
+ port = [5002] * N_NODES # use port 5002
+ var = obj.get_child("2:FPGA_bst_offload_hdr_udp_destination_port_RW")
+ var.set_value(ua.Variant(value=list(port), varianttype=ua.VariantType.UInt16))
+ dest_mac = ["00:1B:21:71:76:B9"] * N_NODES # use mac of dop36
+ var = obj.get_child("2:FPGA_bst_offload_hdr_eth_destination_mac_RW")
+ var.set_value(ua.Variant(value=list(dest_mac), varianttype=ua.VariantType.String))
-def write_xst_stream_setup(obj, on=False):
+ dest_addr = ["10.99.0.254"] * N_NODES # use addr of dop36
+ var = obj.get_child("2:FPGA_bst_offload_hdr_ip_destination_address_RW")
+ var.set_value(ua.Variant(value=list(dest_addr), varianttype=ua.VariantType.String))
+
+
+def setup_xst_stream(obj):
'''
setup xst stream, use git/upe_gear/peripherals/pi_satistics_stream.py for recording and plotting.
- pi_statistic_stream.py --unb2 2 --pn2 0 --cmd 4 -s=XST
'''
- if on is True:
- port = [5003] * N_NODES # use port 5001
- var = obj.get_child("2:FPGA_xst_offload_hdr_udp_destination_port_RW")
- var.set_value(ua.Variant(value=list(port), varianttype=ua.VariantType.UInt16))
+ port = [5003] * N_NODES # use port 5001
+ var = obj.get_child("2:FPGA_xst_offload_hdr_udp_destination_port_RW")
+ var.set_value(ua.Variant(value=list(port), varianttype=ua.VariantType.UInt16))
+
+ dest_mac = ["00:1B:21:71:76:B9"] * N_NODES # use mac of dop36
+ var = obj.get_child("2:FPGA_xst_offload_hdr_eth_destination_mac_RW")
+ var.set_value(ua.Variant(value=list(dest_mac), varianttype=ua.VariantType.String))
- dest_mac = ["00:1B:21:71:76:B9"] * N_NODES # use mac of dop36
- var = obj.get_child("2:FPGA_xst_offload_hdr_eth_destination_mac_RW")
- var.set_value(ua.Variant(value=list(dest_mac), varianttype=ua.VariantType.String))
+ dest_addr = ["10.99.0.254"] * N_NODES # use addr of dop36
+ var = obj.get_child("2:FPGA_xst_offload_hdr_ip_destination_address_RW")
+ var.set_value(ua.Variant(value=list(dest_addr), varianttype=ua.VariantType.String))
- dest_addr = ["10.99.0.254"] * N_NODES # use addr of dop36
- var = obj.get_child("2:FPGA_xst_offload_hdr_ip_destination_address_RW")
- var.set_value(ua.Variant(value=list(dest_addr), varianttype=ua.VariantType.String))
+ subsel = [0, 102, 0, 0, 0, 0, 0, 0] * N_NODES # use subband 102 only
+ var = obj.get_child("2:FPGA_xst_subband_select_RW")
+ var.set_value(ua.Variant(value=list(subsel), varianttype=ua.VariantType.UInt32))
- subsel = [0, 102, 0, 0, 0, 0, 0, 0] * N_NODES # use subband 102 only
- var = obj.get_child("2:FPGA_xst_subband_select_RW")
- var.set_value(ua.Variant(value=list(subsel), varianttype=ua.VariantType.UInt32))
+ interval = [1.0] * N_NODES # use fixed interval 1.0 second
+ var = obj.get_child("2:FPGA_xst_integration_interval_RW")
+ var.set_value(ua.Variant(value=list(interval), varianttype=ua.VariantType.Double))
- interval = [1.0] * N_NODES # use fixed interval 1.0 second
- var = obj.get_child("2:FPGA_xst_integration_interval_RW")
- var.set_value(ua.Variant(value=list(interval), varianttype=ua.VariantType.Double))
+ processing = [True] * N_NODES # enable processing
+ var = obj.get_child("2:FPGA_xst_processing_enable_RW")
+ var.set_value(ua.Variant(value=list(processing), varianttype=ua.VariantType.Boolean))
- processing = [True] * N_NODES # enable processing
- var = obj.get_child("2:FPGA_xst_processing_enable_RW")
- var.set_value(ua.Variant(value=list(processing), varianttype=ua.VariantType.Boolean))
- enable = [True] * N_NODES # enable offload
+def set_enable_stream(obj, mode):
+ if mode == 'OFF':
+ enable = [False] * N_NODES # disable offload
+ var = obj.get_child("2:FPGA_sst_offload_enable_RW")
+ var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
+ var = obj.get_child("2:FPGA_bst_offload_enable_RW")
+ var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
var = obj.get_child("2:FPGA_xst_offload_enable_RW")
var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
- else:
- enable = [False] * N_NODES # enable offload
+ return
+
+ enable = [True] * N_NODES # enable offload
+ if mode == 'SST':
+ var = obj.get_child("2:FPGA_sst_offload_enable_RW")
+ var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
+ elif mode == 'BST':
+ var = obj.get_child("2:FPGA_bst_offload_enable_RW")
+ var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
+ elif mode == 'XST':
var = obj.get_child("2:FPGA_xst_offload_enable_RW")
var.set_value(ua.Variant(value=list(enable), varianttype=ua.VariantType.Boolean))
+ else:
+ print('Wrong mode "{}"'.format(mode))
+ return
if __name__ == "__main__":
# Parse command line arguments
@@ -288,8 +346,9 @@ if __name__ == "__main__":
parser.add_argument('-n', '--nodes', dest='nodes', type=int, nargs='+', help="nodes to use")
parser.add_argument('-i', '--info', dest='info', action='store_true', help="print point infor from server")
parser.add_argument('-a', '--all', dest='all_r', action='store_true', help="recv all*_R points and show time")
- parser.add_argument('--wgxst', dest='wg_xst_mode', type=bool, help="turn on/off wg for xst")
- parser.add_argument('--offloadxst', dest='offload_xst', type=bool, help="turn on/off xst offload")
+ parser.add_argument('--wg', dest='wg_mode', type=str, choices=['OFF', 'XST'], help="turn wg off/on for xst")
+ parser.add_argument('--setup', dest='offload_setup', type=str, choices=['SST', 'BST', 'XST'], help="setup offload for selected mode")
+ parser.add_argument('--stream', dest='stream', type=str, choices=['OFF', 'SST', 'BST', 'XST'], help="turn off/on selected offload stream")
parser.add_argument('--verbosity', default='INFO', help="stdout log level can be [ERROR | WARNING | INFO | DEBUG]")
args = parser.parse_args()
@@ -299,9 +358,13 @@ if __name__ == "__main__":
# logger = logging.getLogger("KeepAlive")
# logger.setLevel(logging.DEBUG)
+ host = 'dop36'
+ port = 4840
+ client = Client("opc.tcp://{}:{}/".format(host, port))
+
# client = Client("opc.tcp://LAPTOP-N0VQ3UDT:4840/")
# client = Client("opc.tcp://192.168.137.102:4840/")
- client = Client("opc.tcp://dop36:4840/")
+ # client = Client("opc.tcp://dop36:4840/")
# client = Client("opc.tcp://169.254.91.66:4840/")
# I hope this is secure, because I have zero clue about security
# client.set_security_string("Basic256Sha256,SignAndEncrypt,certificate-example.der,private-key-example.pem")
@@ -309,15 +372,15 @@ if __name__ == "__main__":
cnt = 1
weights = [-1 for i in range(8000)] # 187392)
# 0x7fff
-
+
mask = [True for i in range(16)]
- scraps = [i+3 for i in range(2048)]
+ scraps = [i + 3 for i in range(2048)]
clientRunning = True
while clientRunning:
try:
client.connect()
- print("I'm in")
+ print("Connected to {}:{}".format(host, port))
################
# this section contains some code about navigating around the address space
@@ -346,11 +409,22 @@ if __name__ == "__main__":
for i in range(len(children)):
print("\t\tchild", i, ": ", children[i].get_browse_name())
- if args.wg_xst_mode:
- write_xst_wg_setup(Object, on=args.wg_xst_mode)
-
- if args.offload_xst:
- write_xst_stream_setup(Object, on=args.offload_xst)
+ if args.wg_mode is not None:
+ if args.wg_mode == 'OFF':
+ turn_wg_off(obj)
+ elif args.wg_mode == 'XST':
+ setup_wg_xst_mode(Object)
+
+ if args.offload_setup is not None:
+ if args.offload_setup == 'SST':
+ setup_sst_stream(Object)
+ elif args.offload_setup == 'BST':
+ setup_bst_stream(Object)
+ elif args.offload_setup == 'XST':
+ setup_xst_stream(Object)
+
+ if args.stream is not None:
+ set_enable_stream(Object, mode=args.stream)
if args.all_r is True:
check_get_all_R_time(Object)