Skip to content

Instantly share code, notes, and snippets.

@jrast

jrast/client.py

Created February 16, 2022 13:17
Show Gist options
  • Save jrast/1befc0f62db27d7ede9985d6b1f7fb8d to your computer and use it in GitHub Desktop.
Save jrast/1befc0f62db27d7ede9985d6b1f7fb8d to your computer and use it in GitHub Desktop.
Download ZIP
Python OPC-UA Dummy Server + Client
Raw
client.py
import asyncio
from asyncua import Client, Node
import logging
logging.basicConfig(level=logging.WARN, format="%(asctime)s %(message)s")
_logger = logging.getLogger("dummy_client")
_logger.setLevel(level=logging.INFO)
class SubscriptionHandler:
def datachange_notification(self, node: Node, val, data):
_logger.info("Variable Value Changed: %s", val)
async def main():
client = Client("opc.tcp://localhost:4840")
async with client:
_logger.info("Connected!")
var = client.get_node("ns=2;i=4")
current_value = await var.read_value()
_logger.info("Current Variable Value: %s", current_value)
_logger.info("Creating Subscription")
handler = SubscriptionHandler()
subscription = await client.create_subscription(2000, handler)
sub_handle = await subscription.subscribe_data_change(var)
await asyncio.sleep(0.5)
_logger.info("Calling ChangeValue (1)")
methods = client.get_node("ns=2;i=2")
res = await methods.call_method("2:ChangeValue", current_value + 1)
_logger.info("Result from Change Value: %s", res)
_logger.info("Calling ChangeValue (2)")
methods = client.get_node("ns=2;i=2")
res = await methods.call_method("2:ChangeValue", current_value + 2)
_logger.info("Result from Change Value: %s", res)
await asyncio.sleep(1)
_logger.info("Delete subscription")
await subscription.unsubscribe(sub_handle)
await subscription.delete()
await asyncio.sleep(0.5)
_logger.info("Connection Closed")
if __name__ == "__main__":
asyncio.run(main())
Raw
server.py
import asyncio
from datetime import datetime
from asyncua import Server, uamethod, ua
async def main():
server = Server()
await server.init()
await server.set_build_info(
product_uri="https://github.com/jrast",
product_name="Dummy Server",
manufacturer_name="Jürg Rast",
software_version="beta",
build_number="---",
build_date=datetime.utcnow(),
)
server.set_endpoint("opc.tcp://0.0.0.0:4840")
ns_name = "https://github.com/jrast/dummy-server"
ns_idx = await server.register_namespace(ns_name)
# Create a new base object
dummy = await server.nodes.objects.add_object(ns_idx, "Dummy")
methods = await dummy.add_object(ns_idx, "Methods")
variables = await dummy.add_object(ns_idx, "Variables")
my_value = await variables.add_variable(ns_idx, "MyValue", 1.0)
@uamethod
async def change_value(parent, new_value):
await asyncio.sleep(0.5)
await my_value.write_value(new_value)
await methods.add_method(ns_idx, "ChangeValue", change_value, [ua.VariantType.Float], [])
async with server:
while True:
await asyncio.sleep(1)
if __name__ == "__main__":
import logging
logging.basicConfig(level=logging.INFO)
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
@jrast
Copy link
Author

jrast commented Feb 16, 2022

Sample output of "python client.py":

2022-02-16 14:15:41,698 Connected!
2022-02-16 14:15:41,700 Current Variable Value: 7.0
2022-02-16 14:15:41,701 Creating Subscription
2022-02-16 14:15:42,212 Calling ChangeValue (1)
2022-02-16 14:15:42,714 Result from Change Value: None
2022-02-16 14:15:42,714 Calling ChangeValue (2)
2022-02-16 14:15:43,229 Result from Change Value: None
2022-02-16 14:15:43,715 Variable Value Changed: 7.0
2022-02-16 14:15:43,716 Variable Value Changed: 8.0
2022-02-16 14:15:43,716 Variable Value Changed: 9.0
2022-02-16 14:15:44,246 Delete subscription
2022-02-16 14:15:44,767 Connection Closed

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment