Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
O
opcua-asyncio
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
1
Merge Requests
1
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
Nikola Balog
opcua-asyncio
Commits
5e173792
Commit
5e173792
authored
Feb 20, 2022
by
oroulet
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
update some xml examples
parent
878f892d
Changes
2
Hide whitespace changes
Inline
Side-by-side
Showing
2 changed files
with
66 additions
and
43 deletions
+66
-43
examples/simple-client-server-xml/client_minimal.py
examples/simple-client-server-xml/client_minimal.py
+18
-10
examples/simple-client-server-xml/server.py
examples/simple-client-server-xml/server.py
+48
-33
No files found.
examples/simple-client-server-xml/client_minimal.py
View file @
5e173792
import
asyncio
import
logging
from
asyncua
import
Client
...
...
@@ -5,29 +8,34 @@ class HelloClient:
def
__init__
(
self
,
endpoint
):
self
.
client
=
Client
(
endpoint
)
def
__
enter__
(
self
):
self
.
client
.
connect
()
async
def
__a
enter__
(
self
):
await
self
.
client
.
connect
()
return
self
.
client
def
__
exit__
(
self
,
exc_type
,
exc_val
,
exc_tb
):
self
.
client
.
disconnect
()
async
def
__a
exit__
(
self
,
exc_type
,
exc_val
,
exc_tb
):
await
self
.
client
.
disconnect
()
if
__name__
==
'__main__'
:
with
HelloClient
(
"opc.tcp://localhost:40
840/freeopcua/server/"
)
as
client
:
async
def
main
()
:
async
with
HelloClient
(
"opc.tcp://localhost:4
840/freeopcua/server/"
)
as
client
:
root
=
client
.
nodes
.
root
print
(
"Root node is: "
,
root
)
objects
=
client
.
nodes
.
objects
print
(
"Objects node is: "
,
objects
)
hellower
=
objects
.
get_child
(
"0:Hellower"
)
hellower
=
await
objects
.
get_child
(
"0:Hellower"
)
print
(
"Hellower is: "
,
hellower
)
resulting_text
=
hellower
.
call_method
(
"0:SayHello"
,
False
)
resulting_text
=
await
hellower
.
call_method
(
"0:SayHello"
,
False
)
print
(
resulting_text
)
resulting_text
=
hellower
.
call_method
(
"1:SayHello2"
,
True
)
resulting_text
=
await
hellower
.
call_method
(
"1:SayHello2"
,
True
)
print
(
resulting_text
)
resulting_array
=
hellower
.
call_method
(
"1:SayHelloArray"
,
False
)
resulting_array
=
await
hellower
.
call_method
(
"1:SayHelloArray"
,
False
)
print
(
resulting_array
)
if
__name__
==
"__main__"
:
logging
.
basicConfig
(
level
=
logging
.
INFO
)
asyncio
.
run
(
main
())
examples/simple-client-server-xml/server.py
View file @
5e173792
import
os.path
try
:
from
IPython
import
embed
except
ImportError
:
import
code
def
embed
():
vars
=
globals
()
vars
.
update
(
locals
())
shell
=
code
.
InteractiveConsole
(
vars
)
shell
.
interact
()
import
asyncio
import
logging
from
asyncua
import
ua
,
uamethod
,
Server
...
...
@@ -47,39 +39,62 @@ def say_hello_array(parent, happy):
class
HelloServer
:
def
__init__
(
self
,
endpoint
,
name
,
model_filepath
):
self
.
server
=
Server
()
self
.
model_filepath
=
model_filepath
self
.
server
.
set_server_name
(
name
)
self
.
server
.
set_endpoint
(
endpoint
)
# This need to be imported at the start or else it will overwrite the data
self
.
server
.
import_xml
(
model_filepath
)
async
def
init
(
self
):
await
self
.
server
.
init
(
)
self
.
server
.
set_endpoint
(
endpoint
)
self
.
server
.
set_server_name
(
name
)
# This need to be imported at the start or else it will overwrite the data
await
self
.
server
.
import_xml
(
self
.
model_filepath
)
objects
=
self
.
server
.
nodes
.
objects
freeopcua_namespace
=
self
.
server
.
get_namespace_index
(
"urn:freeopcua:python:server"
)
hellower
=
objects
.
get_child
(
"0:Hellower"
)
hellower_say_hello
=
hellower
.
get_child
(
"0:SayHello"
)
freeopcua_namespace
=
await
self
.
server
.
get_namespace_index
(
"urn:freeopcua:python:server"
)
hellower
=
await
objects
.
get_child
(
"0:Hellower"
)
hellower_say_hello
=
await
hellower
.
get_child
(
"0:SayHello"
)
self
.
server
.
link_method
(
hellower_say_hello
,
say_hello_xml
)
hellower
.
add_method
(
freeopcua_namespace
,
"SayHello2"
,
say_hello
,
[
ua
.
VariantType
.
Boolean
],
[
ua
.
VariantType
.
String
])
hellower
.
add_method
(
freeopcua_namespace
,
"SayHelloArray"
,
say_hello_array
,
[
ua
.
VariantType
.
Boolean
],
[
ua
.
VariantType
.
String
])
def
__enter__
(
self
):
self
.
server
.
start
()
await
hellower
.
add_method
(
freeopcua_namespace
,
"SayHello2"
,
say_hello
,
[
ua
.
VariantType
.
Boolean
],
[
ua
.
VariantType
.
String
],
)
await
hellower
.
add_method
(
freeopcua_namespace
,
"SayHelloArray"
,
say_hello_array
,
[
ua
.
VariantType
.
Boolean
],
[
ua
.
VariantType
.
String
],
)
async
def
__aenter__
(
self
):
await
self
.
init
()
await
self
.
server
.
start
()
return
self
.
server
def
__
exit__
(
self
,
exc_type
,
exc_val
,
exc_tb
):
self
.
server
.
stop
()
async
def
__a
exit__
(
self
,
exc_type
,
exc_val
,
exc_tb
):
await
self
.
server
.
stop
()
if
__name__
==
'__main__'
:
async
def
main
()
:
script_dir
=
os
.
path
.
dirname
(
__file__
)
with
HelloServer
(
"opc.tcp://0.0.0.0:40840/freeopcua/server/"
,
"FreeOpcUa Example Server"
,
os
.
path
.
join
(
script_dir
,
"test_saying.xml"
))
as
server
:
embed
()
async
with
HelloServer
(
"opc.tcp://0.0.0.0:4840/freeopcua/server/"
,
"FreeOpcUa Example Server"
,
os
.
path
.
join
(
script_dir
,
"test_saying.xml"
),
)
as
server
:
while
True
:
await
asyncio
.
sleep
(
1
)
if
__name__
==
"__main__"
:
logging
.
basicConfig
(
level
=
logging
.
INFO
)
asyncio
.
run
(
main
())
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment