FilesExpand file tree

master

/

client-events.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

57 lines (42 loc) · 1.62 KB

master

/

client-events.py

Copy path

Top

File metadata and controls

• Code
• Blame

57 lines (42 loc) · 1.62 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

import sys

sys.path.insert(0, "..")

try:

from IPython import embed

except ImportError:

import code

def embed():

vars = globals()

vars.update(locals())

shell = code.InteractiveConsole(vars)

shell.interact()

from opcua import Client

class SubHandler(object):

"""

Subscription Handler. To receive events from server for a subscription

data_change and event methods are called directly from receiving thread.

Do not do expensive, slow or network operation there. Create another

thread if you need to do such a thing

"""

def event_notification(self, event):

print("New event recived: ", event)

if __name__ == "__main__":

client = Client("opc.tcp://localhost:4840/freeopcua/server/")

# client = Client("opc.tcp://admin@localhost:4840/freeopcua/server/") #connect using a user

try:

client.connect()

# Client has a few methods to get proxy to UA nodes that should always be in address space such as Root or Objects

root = client.get_root_node()

print("Objects node is: ", root)

# Now getting a variable node using its browse path

obj = root.get_child(["0:Objects", "2:MyObject"])

print("MyObject is: ", obj)

myevent = root.get_child(["0:Types", "0:EventTypes", "0:BaseEventType", "2:MyFirstEvent"])

print("MyFirstEventType is: ", myevent)

msclt = SubHandler()

sub = client.create_subscription(100, msclt)

handle = sub.subscribe_events(obj, myevent)

embed()

sub.unsubscribe(handle)

sub.delete()

finally:

client.disconnect()