Registering multiple notifications on the AdsTestServer

[TurboThommy]

Hello,

I tried to use the AdsTestServer with AdvancedHandler to mock the behaviour of a PLC.
Everything worked great up until the point where I tried to create multiple device notifications:

Code

  from ctypes import sizeof
  import struct
  import pyads
  from pyads.testserver import AdsTestServer, AdvancedHandler, PLCVariable
  from pyads import constants
  
  from time import sleep
  
  
  def main():
      handler = AdvancedHandler()
      test_server = AdsTestServer(handler=handler)
  
  
      test1 = PLCVariable(
          'Test1',
          value=0,
          ads_type=constants.ADST_UINT32,
          symbol_type='DWORD',
          index_group=61445,
          index_offset=10000
      )
      handler.add_variable(test1)
  
  
      test2 = PLCVariable(
          'Test2',
          value=0,
          ads_type=constants.ADST_UINT32,
          symbol_type='DWORD',
          index_group=61445,
          index_offset=10001
      )
      handler.add_variable(test2)
  
  
      sleep(1)
  
  
      test_server.start()
      plc = pyads.Connection('127.0.0.1.1.1')
  
  
      def callback1(notification, data):
          try:
              _, _, value = plc.parse_notification(
                  notification=notification,
                  plc_datatype=pyads.PLCTYPE_DWORD
              )
              print(f'Callback 1: {data}: {value}')
          except:
              print('Error in callback1.')
  
  
      def callback2(notification, data):
          try:
              _, _, value = plc.parse_notification(
                  notification=notification,
                  plc_datatype=pyads.PLCTYPE_DWORD
              )
              print(f'Callback 2: {data}: {value}')
          except:
              print('Error in callback2.')
  
  
      with plc:
          symbol1 = handler.get_variable_by_name('Test1')
          symbol2 = handler.get_variable_by_name('Test2')
  
          attr = pyads.NotificationAttrib(sizeof(pyads.PLCTYPE_DWORD))
          plc.add_device_notification(
              data='Test1',
              attr=attr,
              callback=callback1
          )
  
          attr = pyads.NotificationAttrib(sizeof(pyads.PLCTYPE_DWORD))
          plc.add_device_notification(
              data='Test2',
              attr=attr,
              callback=callback2
          )
  
          symbol1.write(struct.pack('<I', 1))
          symbol2.write(struct.pack('<I', 10))
  
      sleep(1)
      test_server.stop()
      sleep(1)
  
  
  if __name__ == "__main__":
      main()

I expected following output:

Expected Output

2025-07-10T17:36:32+0200 Info: Connected to 127.0.0.1
Callback 1: Test1: 1
Callback 2: Test2: 10
2025-07-10T17:36:32+0200 Info: connection closed by remote

But instead I got:

Actual Output

2025-07-10T17:36:32+0200 Info: Connected to 127.0.0.1
Callback 2: Test2: 1
Callback 2: Test2: 10
2025-07-10T17:36:32+0200 Info: connection closed by remote

To me it seems like only the callback registered last is used. Unfortunately I don't have real hardware at hand right now to test if this is only the case for the test server.

The application is run on Ubuntu using the WSL in case that makes any difference.

[chrisbeardy]

I think this may be a limitation of the test server at the moment, I need to double check. In any case I'll leave this open as a potential feature.

EDIT: Looks like I am wrong and this should be handled.

[chrisbeardy]

@TurboThommy thanks for the report, looks like your expected and actual output are the same, can you edit please.

[TurboThommy]

@chrisbeardy They are not the same. As seen in the actual output the first printed line differs. Instead of printing"Callback 1" and the correct variable name, "Callback 2" is printed as well as the name of the variable (Test2) the second callback was registered for.

Since yesterday I tried using the integer tuple to specify the index_group and index_offet of both variables but results were the same. Furthermore I tried reading the variables directly after registering the callbacks which worked just fine.
Today I also got the chance to test this on real hardware and with a real PLC I was not able to reproduce the issue so far.

[chrisbeardy]

you are correct, my mistake. There must be an issue in the Test Server, will need to investigate.

[ASil-Scanway]

Hey.
I found how to make workaround or fix for this issue.
The root cause is PLCVariable register_notification method using notification_count as instance field instead of class field.

It seems original intent was to use notification_count in static way

class PLCVariable:
    """Storage item for named data.

    Also include variable type so it can be retrieved later.
    This basically mirrors SAdsSymbolEntry or AdsSymbol, however we want to
    avoid using those directly since they are test subjects.
    """

    handle_count = 10000  # Keep track of the latest awarded handle
    notification_count = 10  # Keep track the latest notification handle

and in constructor, it's used this way

        PLCVariable.handle_count += 1

but register_notification uses it as instance field

    def register_notification(self) -> int:
        """Register a new notification."""

        handle = PLCVariable.notification_count
        self.notifications.append(handle)
        PLCVariable.notification_count += 1
        return handle

This leads to the situation when the same value is reused for multiple notifications if user tries to register notification for different variable (other instance of PLCVariable)

Diff to fix the issue is below. Note that I didn't test it through and through.

diff --git a/src/pyads/testserver/advanced_handler.py b/src/pyads/testserver/advanced_handler.py
index 7ba5187..92a1b6c 100644
--- a/src/pyads/testserver/advanced_handler.py
+++ b/src/pyads/testserver/advanced_handler.py
@@ -164,9 +164,9 @@ class PLCVariable:
     def register_notification(self) -> int:
         """Register a new notification."""
 
-        handle = self.notification_count
+        handle = PLCVariable.notification_count
         self.notifications.append(handle)
-        self.notification_count += 1
+        PLCVariable.notification_count += 1
         return handle
 
     def unregister_notification(self, handle: int = None):