```
import asyncio
import random
import math
from datetime import datetime
from opcua import Server
# Constants
BOOL_COUNT = 100
RAMP_COUNT = 100
SINUSOIDAL_COUNT = 100
ASYMPTOTIC_COUNT = 100
TOTAL_ANALOG = 400
VALUE_RANGE = (0, 2000.0)
# Setup server
server = Server()
server.set_endpoint("opc.tcp://localhost:4840/freeopcua/server/")
server.set_server_name("Custom OPC UA Server with 500 Points")
# Create address space
idx = server.register_namespace("http://example.org")
boolean_group = server.nodes.objects.add_object(idx, "Booleans")
analog_group = server.nodes.objects.add_object(idx, "Analog")
# Create nodes
bool_nodes = [boolean_group.add_variable(idx, f"Boolean{i}", False) for i in range(BOOL_COUNT)]
for node in bool_nodes:
node.set_writable()
analog_nodes = [analog_group.add_variable(idx, f"Analog{i}", 0.0) for i in range(TOTAL_ANALOG)]
for node in analog_nodes:
node.set_writable()
# Asymptotic and linked values
asymptotic_nodes = analog_nodes[ASYMPTOTIC_COUNT : ASYMPTOTIC_COUNT * 2]
linked_nodes = analog_nodes[ASYMPTOTIC_COUNT * 2 : ASYMPTOTIC_COUNT * 3]
# Update functions
async def update_booleans():
async def toggle_boolean(node):
while True:
delay = random.randint(3, 300)
await asyncio.sleep(delay)
node.set_value(not node.get_value())
# Create a task for each boolean node
await asyncio.gather(*(toggle_boolean(node) for node in bool_nodes))
async def update_ramps():
async def ramp_value(node):
while True:
ramp_time = random.randint(1, 60) * 60 # Ramp time in seconds
step = random.uniform(1, 10)
value = node.get_value()
target = random.uniform(*VALUE_RANGE)
steps = int(ramp_time / step)
for _ in range(steps):
value += step if value < target else -step
node.set_value(min(max(value, VALUE_RANGE[0]), VALUE_RANGE[1]))
await asyncio.sleep(step)
if abs(value - target) < 0.01:
break
# Create a task for each ramp node
await asyncio.gather(*(ramp_value(node) for node in analog_nodes[:RAMP_COUNT]))
async def update_sinusoidal():
async def sinusoidal_value(node):
while True:
period = random.randint(60, 3600) # Seconds
amplitude = (VALUE_RANGE[1] - VALUE_RANGE[0]) / 2
offset = sum(VALUE_RANGE) / 2
for t in range(period):
value = offset + amplitude * math.sin(2 * math.pi * t / period)
node.set_value(value)
await asyncio.sleep(1)
# Create a task for each sinusoidal node
await asyncio.gather(*(sinusoidal_value(node) for node in analog_nodes[RAMP_COUNT:RAMP_COUNT + SINUSOIDAL_COUNT]))
async def update_asymptotic():
async def asymptotic_value(node, linked_node):
while True:
value = node.get_value()
target = random.uniform(*VALUE_RANGE)
time_to_target = random.randint(600, 3600)
steps = time_to_target
step = (target - value) / steps
for _ in range(steps):
value += step
node.set_value(value)
linked_node.set_value(target)
await asyncio.sleep(1)
if abs(value - target) < 0.01:
break
# Create a task for each asymptotic node and its linked node
await asyncio.gather(
*(
asymptotic_value(node, linked_node)
for node, linked_node in zip(asymptotic_nodes, linked_nodes)
)
)
# Start server and tasks
async def main():
stop_event = asyncio.Event()
def stop_server():
print("Stopping server...")
stop_event.set()
try:
server.start()
print(f"Server started at {server.endpoint}")
tasks = [
asyncio.create_task(update_booleans()),
asyncio.create_task(update_ramps()),
asyncio.create_task(update_sinusoidal()),
asyncio.create_task(update_asymptotic()),
]
await stop_event.wait() # Wait indefinitely until the event is set
except KeyboardInterrupt:
stop_server()
finally:
server.stop()
print("Server stopped")
if __name__ == "__main__":
asyncio.run(main())
```
This was made in 20 minutes in discussion with ChatGPT.
At first, it didn't update all the values, just a few. Here's what the query to ChatGPT was initially:
"I need an OPC UA server with 500 points. It should be written in python. of the 500 points, there will be 5 groups. 100 booleans, each turning on and off at a random interval of between 3 and 300 seconds. 400 analog values, these will be within the range of 0 to 2000.00. 100 of these values will ramp up and down with a random ramping period of between 1 and 60 minutes. 100 values will follow varying sinusoidal curves within the previously indicated range. 100 of the values will start at a random value, then grow or shrink asymptotically over between 10 and 60 minutes to approach a second random value. Once they reach 99% of the second value, they approach a new random value in the same manner as the first. The last hundred values will be the 'ending' points of the previous set of 100 values, so they will appear to make a step change that the previously described value is approaching."