[Python] Python opcua library -> get_value() function is not reading value, traceback show...

I am trying to read 4 nodes values. I wrote a test for a check kepserver connection and get_value from nodes and it works. Here is the test codes :

def kepware_test():
client = opcua.Client("opc.tcp://xxx.xxx.xxx.xxx:49320")
try:
client.connect()

# Read the value of a node
node = client.get_node(PLC_NAMESPACE_PREFIX + "TestChannel.S7_1500.Product.Tag_Presence")
value = node.get_value()
print("Node value:", value)

# Subcribe to a node
subs = client.create_subscription(100, MySubHandler())
subs.subscribe_data_change(node)

while True:
# Wait for data change notifications
# dv = ua.DataValue(ua.Variant(True, ua.VariantType.Boolean))
# node.set_value(dv) # Simulate a data change
value = node.get_value()
print("Node value:", value)
time.sleep(5)

except Exception as e:
print(e)

if __name__ == "__main__":
kepware_test()


Returns node value correctly and subscription is working also. But then I wrote some code to get tag path from sql and opcserver link. And I implemented like that

class KepwareHelper():
def __init__(self, message_handler=None, qImage_handler=None):
self.connected = False
self.client = None
self.subs = None
self.message_handler = message_handler
self.qImage_handler = qImage_handler

print("KepwareHelper initialized")
self.app_data = AppData()
self.take_image_handler = None

def connect_kepware(self):
print("Connecting to kepware...")
self.client = opcua.Client(self.app_data.station_info["OpcAddress"])
self.client.connect()
print("Connected kepware")
self.connected = True

def disconnect_kepware(self):
if self.client:
self.client.disconnect()
self.connected = False

def initialize_nodes(self):
station_info = self.app_data.get_station_info()
if not station_info:
print("Station information not found.")
return

print("Node lar olusuyor...")
self.dummy_tag = self.client.get_node(PLC_NAMESPACE_PREFIX + station_info["DummyTag"])
self.model_tag = self.client.get_node(PLC_NAMESPACE_PREFIX + station_info["ModelTag"])
self.test_status_tag = self.client.get_node(PLC_NAMESPACE_PREFIX + station_info["TestStatusTag"])
self.test_data_tag = self.client.get_node(PLC_NAMESPACE_PREFIX + station_info["TestDataTag"])
self.test_result_tag = self.client.get_node(PLC_NAMESPACE_PREFIX + station_info["TestResultTag"])
self.tag_presence_tag = self.client.get_node(PLC_NAMESPACE_PREFIX + station_info["TagPresenceTag"])
self.pallet_presence_tag = self.client.get_node(PLC_NAMESPACE_PREFIX + station_info["PalletPresenceTag"])
print("Node lar olustu...")

def subscribe(self):
print("Subscribing to nodes...")
self.subs = self.client.create_subscription(1000, MySubHandler(self.handle_datachange))
# Subscribe to the nodes for data changes
self.subs.subscribe_data_change(self.tag_presence_tag)
self.subs.subscribe_data_change(self.pallet_presence_tag)
self.subs.subscribe_data_change(self.dummy_tag)
self.subs.subscribe_data_change(self.model_tag)
print("Subscribed to nodes...")

def unsubscribe(self):
if self.subs:
print("Unsubscribing from nodes...")
self.subs.delete()
self.subs = None

def handle_datachange(self):
print("Data change detected")
is_ok = asyncio.run(self.read_product_arrived_check_nodes())
print("Product arrived check nodes:", is_ok)
if is_ok:
if self.take_image_handler is None:
self.take_image_handler = TakeImageHandler(self.handle_message, self.handle_product_arrived_bit, self.handle_qimage)
print("TakeImageHandler initialized")
self.take_image_handler.run()

def handle_message(self, message):
if self.message_handler:
self.message_handler(message)

def handle_product_arrived_bit(self, value):
self.write_to_image_taken_finished(value)

def handle_qimage(self, q_image):
self.qImage_handler(q_image)

async def read_product_arrived_check_nodes(self):
# DummyTag dolu olmali, ModelTagDolu olmali
# TagPresenceTag dolu olmali, PalletPresenceTag dolu olmali
try:
if self.connected:
tag_presence_value = await self.tag_presence_tag.get_value()
print("Tag Presence Value:", tag_presence_value)
pallet_presence_value = await self.pallet_presence_tag.get_value()
print("Pallet Presence Value:", pallet_presence_value)
dummy_value = await self.dummy_tag.get_value()
print("Dummy Value:", dummy_value)
model_value = await self.model_tag.get_value()
print("Model Value:", model_value)

return dummy_value and model_value and tag_presence_value and pallet_presence_value
else:
print("Not connected to Kepware server, Could not read from node.")
return False
except Exception as e:
import traceback
traceback.print_exc()
print(f"Error reading from node: {e}")
return False


def write_to_image_taken_finished(self, value : bool):
try:
if self.connected:
dv = ua.DataValue(ua.Variant(value, ua.VariantType.Boolean))
# self.image_taken_finished.set_value(dv)
else:
print("Not connected to Kepware server, Could not write to node.")
except Exception as e:
print(f"Error writing to node: {e}")

class MySubHandler(object):
"""
Subscription Handler. To receive events from server for a subscription
This class is just a sample class. Whatever class having these methods can be used
"""
def __init__(self, handle_datachange=None):
self.handle_datachange = handle_datachange

def datachange_notification(self, node, val, data):
print("Data change notification received")
self.handle_datachange()

def data_change(self, handle, node, val, attr):
"""
Deprecated, use datachange_notification
"""
pass

def event_notification(self, event):
"""
called for every event notification from server
"""
pass


def status_change_notification(self, status):
"""
called for every status change notification from server
"""
pass

• AppData class has singleton pattern to keep values for whole project. I am fetching data and setting into AppData class to reach data quickly.
• Station info data is keeps opcserver link and tag paths. And stores data correctly and return data correctly
• I can connect the opcserver succesfully
• I can create nodes successfully
• I can subscribe nodes successfully

But I can't read node value with using get_value() or get_values() methods.

import traceback
traceback.print_exc()


Exception catch something but catch don't return error messaje but tracing throws me this error :

Traceback (most recent call last):
File "D:\Projects\button-rotation-detection\project\src\kepware_helper.py", line 90, in read_product_arrived_check_nodes
tag_presence_value = await self.tag_presence_tag.get_value()
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^
File "C:\Users\XXX\AppData\Roaming\Python\Python313\site-packages\opcua\common\node.py", line 155, in get_value
result = self.get_data_value()
File "C:\Users\XXX\AppData\Roaming\Python\Python313\site-packages\opcua\common\node.py", line 164, in get_data_value
return self.get_attribute(ua.AttributeIds.Value)
~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\XXX\AppData\Roaming\Python\Python313\site-packages\opcua\common\node.py", line 275, in get_attribute
result = self.server.read(params)
File "C:\Users\XXX\AppData\Roaming\Python\Python313\site-packages\opcua\client\ua_client.py", line 347, in read
data = self._uasocket.send_request(request)
File "C:\Users\XXX\AppData\Roaming\Python\Python313\site-packages\opcua\client\ua_client.py", line 83, in send_request
data = future.result(self.timeout)
File "C:\Python313\Lib\concurrent\futures\_base.py", line 458, in result
raise TimeoutError()
TimeoutError


Why is this happen?

Continue reading...