Creating a Visual Feedback Dashboard that Executes a Ping Script Continuously

Here you go, for what it's worth.
I run this on a gateway timer event set to 30,000 ms. This timer value depends on the number of pings to be executed so may need to be extended as each ping is done sequentially.

The list of devices to be pinged is obtained from a database query.
The results are written to a memory tag. A couple of other tags keep track of the script status.

# Gateway Timer Script - Ping test
# Author:	Transistor
# Date:		2020-11-28
# Desc:		A series of ping tests is carried out on the devices included in resultsTag.
#			The ping tests are carried out from the gateway - not the client.
#			The ping tests are carried out by a gateway timer.
#			The ping test timer MUST BE RUN AS A DEDICATED THREAD.
import datetime
import os			# Required for ping operation
import re			# Required to strip leading zeros from IP address octets.

# Run a query on the routers table every 30s or so. Be awware that enough time should be allowed to run the maximum number of pings.
# Generate a dataset of all the form
#[
#['10.166.28.200', 'Area 1', 'Router', 'Machine 1'],
#['10.166.28.201', 'Area 1', 'PLC',    'Machine 1'],
#... ]
# from the router table. Initialise the dataset with the last seen and status columns.
pingList = system.db.runQuery("SELECT wan_ip, location, device, system, 0 AS 'last ping', 0 AS status FROM routers WHERE ping > 0", 'dbLAN')

# Read in the resultsTag. We'll need this to get the previous ping time
prevScanResults = system.dataset.toPyDataSet(system.tag.read('[default]BAPL_ping/Results').value)

# Prepare the dataset for output.
headers = ['wan_ip', 'location', 'device', 'system', 'last ping', 'status']
data = []

# Record the script start time.
startTime = datetime.datetime.now()
system.tag.write('[default]BAPL_Ping/LastPingStartTime', startTime)
system.tag.write('[default]BAPL_Ping/PingBusy', True)

for row in pingList:
	ip = row['wan_ip']
	shortIp = re.sub('(^|\.)0+(?=[^.])', r'\1', ip)			# Strip the leading zeros from each octet or Windows ping won't find it!	
	# For Windows: -n 1 = ping once. -w 200 = timeout value.
	# For Linux:   -c 1 = ping once. -w 1 = 1 s timeout.
	r = os.system("ping -n 1 -w 150 " + shortIp) == 0
	if r:										# Good result.
		lastPing = 'OK at ' + str(datetime.datetime.now())[10:19]
		status = 1
	else:										# No response.
		lastPing = 'Unknown'					# Default.
		status = -1								# Default.
		for row_p in prevScanResults:
			if row['wan_ip'] == row_p['wan_ip']:# This device was on the previous scan list.
				lastPing = row_p['last ping']
				status = 0
				break

	data.append([row['wan_ip'], row['location'], row['device'], row['system'], lastPing, status])

# Prepare the dataset for output.
dataOut = system.dataset.toDataSet(headers, data)
# Update the resultsTag for availability for all clients.
system.tag.write('[default]BAPL_Ping/Results', dataOut)

system.tag.write('[default]BAPL_Ping/PingBusy', False)
# Record the completion timestamp.
system.tag.write('[default]BAPL_Ping/LastPingEndTime', datetime.datetime.now())
timeDelta = datetime.datetime.now() - startTime
system.tag.write('[default]BAPL_Ping/TotalPingTime', float(timeDelta.seconds))		# Convert to seconds.
1 Like