Accessing webportal as client not server.

opcua-to-http-gw/opcua-to-http-gw.py

@@ -5,6 +5,7 @@ Basic OPC UA <-> HTTP gateway server.
 
 import sys
 import asyncio
+from asyncua import Client, ua
 import asyncua
 from dataclasses import dataclass, field
 import json
@@ -13,6 +14,7 @@ import urllib
 import argparse
 import logging
 import __main__
+import random
 
 # command line handling
 parser = argparse.ArgumentParser(description='Run OPCUA Server.')
@@ -35,6 +37,9 @@ erp5_url = args.erp5_url
 erp5_username = args.erp5_username
 erp5_password = args.erp5_password
 ERP5_REQUEST_API = "ERP5Site_handleOPCUARequest"
+read_backend = False
+bool_server = False
+bool_client = True
 
 # ERP5 backend storage for OPCUA Document
 @dataclass(frozen=True)
@@ -61,9 +66,9 @@ class ERP5Handler(asyncua.common.subscription.SubHandler):
     self.session.auth = (erp5_username, erp5_password)
 
     if http_method == "POST":
-        self.session.post(f"{self.uri}?data={params}")
+      self.session.post(f"{self.uri}?data={params}")
     elif http_method == "GET":
-        return self.session.get(f"{self.uri}")
+      return self.session.get(f"{self.uri}")
 
   def datachange_notification(self, node, val, data):
     self.call(node=node, val=val, data=data)
@@ -83,46 +88,115 @@ class InternalSession(asyncua.server.internal_session.InternalSession):
 # Start OPCUA Server
 async def main():
   _logger = logging.getLogger(__name__)
-
-  # setup our server
-  server = asyncua.Server()
-  await server.init()
-
-  if bool(int(ipv6_enabled)):
-    server.set_endpoint(f"opc.tcp://[{ipv6}]:{port}/freeopcua/server/")
-  #else:
-  #  server.set_endpoint(f"opc.tcp://{ipv4}:{port}/freeopcua/server/")
-
-  if xml is not None:
-    await server.import_xml(xml)
-
-  # read previous state as saved in ERP5 backend
-  erp5_json = erp5_handler.call(http_method="GET").json()
-  _logger.error(erp5_json)
-  for k,v in erp5_json.items():
-    # set
-    node = server.get_node(k)
-    _logger.debug("Init from ERP5. Set %s = %s at %s" %(k, v, node))
-    await node.write_value(v)
-
-  subscription = await server.create_subscription(1000, erp5_handler)
-  await subscription.subscribe_events()
-  nodes = await asyncua.common.ua_utils.get_nodes_of_namespace(server)
-  await subscription.subscribe_data_change(nodes)
-
-  def create_session(name,
-                    user=asyncua.server.users.User(role=asyncua.server.users.UserRole.Anonymous),
-                    external=False):
-    self = server.iserver
-    return InternalSession(self, self.aspace, self.subscription_service, name, user=user, external=external)
-
-  server.iserver.create_session = create_session
-
-  # start OPCUA server
-  _logger.info("Starting server!")
-  async with server:
-    while True:
-      await asyncio.sleep(1)
+  if bool_client:
+    #VariantType=<VariantType.Boolean
+    #VariantType=<VariantType.Int64
+    # Example of creating and sending data to the HTTP backend
+    def send_custom_data(node, val, data_type):
+        custom_data = {
+            'node': node,
+            'val': val,
+            'data': data_type
+        }
+        # Send the data to the ERP5 backend
+        erp5_handler.call(http_method="POST", **custom_data)
+        _logger.info("Sent custom data to ERP5 backend")
+    # FHI Server
+    ipv6 = "2001:67c:1254:108:404e::a"
+    url = f"opc.tcp://[{ipv6}]:{port}/PwDC"
+    _logger.info("Connecting to %s ...", url)
+    async with Client(url=url, timeout=2) as client:
+      while True:
+        await asyncio.sleep(1)
+        # Get the root node
+        root = client.get_root_node()
+
+        # Get the objects node
+        objects = await root.get_child("0:Objects")
+
+        # Get the Powerfuse object
+        powerfuse_obj = await objects.get_child("1:Powerfuse")
+
+        # Get the variables
+        machine_status_rot = await powerfuse_obj.get_child("1:Powerfuse_Maschinenstatus_Status_Rot")
+        machine_status_gelb = await powerfuse_obj.get_child("1:Powerfuse_Maschinenstatus_Status_Gelb")
+
+        # Read values
+        value_rot = await machine_status_rot.read_value()
+        value_gelb = await machine_status_gelb.read_value()
+        _logger.info(f"Initial values: Rot={value_rot}, Gelb={value_gelb}")
+
+        # Write values
+        await machine_status_rot.write_value(True)
+        await machine_status_gelb.write_value(False)
+
+        # Start sending custom data in the background
+        send_custom_data(node = "Powerfuse_Maschinenstatus_Status_Rot", 
+                              val = value_rot, 
+                              data_type = "VariantType=<VariantType.String")
+
+  if bool_server:
+    # setup our server
+    server = asyncua.Server()
+    await server.init()
+
+    if bool(int(ipv6_enabled)):
+      # todo change name of /freeopcua/server
+      _logger.debug(f"Setting endpoint to: opc.tcp://[{ipv6}]:{port}/freeopcua/server/")
+      server.set_endpoint(f"opc.tcp://[{ipv6}]:{port}/freeopcua/server/")
+    #else:
+    #  server.set_endpoint(f"opc.tcp://{ipv4}:{port}/freeopcua/server/")
+
+    if xml is not None:
+      await server.import_xml(xml)
+
+    # read previous state as saved in ERP5 backend
+    #todo: BadNodeIdUnknown is due to the fact that when the server is started the node is not yet created - first create node than read it from backend
+    if read_backend:
+      erp5_json = erp5_handler.call(http_method="GET").json()
+      _logger.error(erp5_json)
+      for k,v in erp5_json.items():
+        # set
+        node = server.get_node(k)
+        _logger.debug("Init from ERP5. Set %s = %s at %s" %(k, v, node))
+        await node.write_value(v)
+
+    subscription = await server.create_subscription(1000, erp5_handler)
+    await subscription.subscribe_events()
+    nodes = await asyncua.common.ua_utils.get_nodes_of_namespace(server)
+    await subscription.subscribe_data_change(nodes)
+
+    def create_session(name,
+                      user=asyncua.server.users.User(role=asyncua.server.users.UserRole.Anonymous),
+                      external=False):
+      self = server.iserver
+      return InternalSession(self, self.aspace, self.subscription_service, name, user=user, external=external)
+
+    server.iserver.create_session = create_session
+
+    # Example of creating and sending data to the HTTP backend
+    async def send_custom_data():
+      while True:
+        # Create your custom data
+        custom_data = {
+            'node': 'ns=2;s=MyCustomNode',
+            'val': random.randint(0, 19),
+            'data': 'VariantType=<VariantType.Int64'
+        }
+        # Send the data to the ERP5 backend
+        erp5_handler.call(http_method="POST", **custom_data)
+        _logger.info("Sent custom data to ERP5 backend")
+        await asyncio.sleep(10)  # Wait for 10 seconds before sending next data
+
+    # Start sending custom data in the background
+    asyncio.create_task(send_custom_data())
+
+    # start OPCUA server
+    _logger.info("Starting server!")
+    async with server:
+      # End creating nodes for 
+      while True:
+        await asyncio.sleep(1)
 
 logging.basicConfig(level=logging.DEBUG)
 asyncio.run(main(), debug=True)