pyopcxmlda
A Python3 implementation of OPC XML DA.
Installation
pip3 install pyopcxmlda
Supported functions
- Browse
- browse server
- Get Properties
- get item properties for a set of items
- Get Status
- get status of server, vendor specific info, etc
- Read
- read a set of item values
- Subscribe
- subscribe a set of item values
- Subscription Cancel
- cancel a subscription
- Subscription Polled Refresh
- poll subscribed items
- Write
- write a set of item values
How to use pyopcxmlda
1. Connect and Send Commands
from datetime import datetime
from pyopcxmlda import Client
from pyopcxmlda.tag import Tag
from pyopcxmlda.constants import DataType
"""
Typical data types:
* boolean
* byte
* short
* unsignedShort
* int
* unsignedInt
* float
* double
* long
* unsignedLong
* decimal
* string
* dateTime
"""
__ITEMS = [
Tag(itemName="//path/to/node/item1", value=-10, type=DataType.INT),
Tag(itemName="//path/to/node/item2", value=10, type=DataType.UINT),
Tag(itemName="//path/to/node/item3", value=-10.0, type=DataType.FLOAT),
Tag(itemName="//path/to/node/item4", value=-1000, type=DataType.LONG)
]
__HOST = '192.168.1.15' # REQUIRED
__PORT = 9500 # OPTIONAL: default is 9500
__NAMESPACE = 'ns0' # OPTIONAL: default is 'ns0'
with Client(host=__HOST, port=__PORT, namespace=__NAMESPACE) as plc:
"""
Browse OPC nodes
example: browse specified path
Args:
namespace:str="",
itemPath:str="",
clientRequestHandle:str="",
itemName:str="",
continuationPoint:str="",
maxElementsReturned:int=0,
browseFilter:str="all",
elementNameFilter:str="",
vendorFilter:str="",
returnAllProperties:bool=False,
returnPropertyValues:bool=False,
returnErrorText:bool=False
Returns:
list[elements]: list of Elements
example:
[
Elements(
xsiType='ns1:BrowseElement',
hasChildren=False,
isItem=True,
name='item',
itemName='//path/to/valid/item',
itemPath='',
fault=None
)
]
Notes:
Elements has a sub Fault class that is filled on error
example:
[
Elements(
xsiType='',
hasChildren=False,
isItem=False,
name='',
itemName='',
itemPath='',
fault=Fault(
faultcode='ns1:E_UNKNOWNITEMNAME',
faultstring='E_UNKNOWNITEMNAME',
detail='Browse invalid/unknown item path/name=//path/to/invalid/item'
)
)
to access the fields of each entry in result
for element in result:
print((f'xsiType:{element.xsiType}, '
f'hasChildren:{element.HasChildren}, '
f'isItem:{element.IsItem}, '
f'name:{element.Name}, '
f'itemName:{element.ItemName}, '
f'itemPath:{element.ItemPath}'))
"""
# to browse root
response = plc.browse()
# to specified path
response = plc.browse(itemName="//path/")
"""
Get Properties of an item(s)
Args:
itemList:list=[],
namespace:str="",
localeID:str="",
clientRequestHandle:str="",
itemPath:str="",
returnAllProperties:bool=True,
returnPropertyValues:bool=True,
returnErrorText:bool=True
Returns:
list[Properties]: list of named tuple
example:
[
Properties(
itemPath='',
itemName='//path/to/node',
dataType='unsignedInt',
value=3620759,
timestamp='2022-08-31T18:23:04Z',
accessRights='readable',
scanRate='10',
quality='good',
euType='noEnum',
error=''
)
]
"""
response = plc.getProperties(itemList=__ITEMS)
"""
Get the server status
Args:
clientRequestHandle(str): request/response identifier
localeID(str): language settings
Returns:
Status: tuple that contains server information
example:
Status(
statusInfo="Running",
vendorInfo='Test',
supportedLocaleIDs="en",
supportedInterfaceVersions="XML_DA_Version_1_0",
productVersion="MyMachine"
)
"""
response = plc.getStatus()
"""
Read mixed item values
example: read randomly mixed data types
Args:
itemList(list[Tag])[Required]: list of data class Tag
example: itemList=__ITEMS
namespace(str): override namespace
example: namespace="ns1"
Returns:
list[Tag]: list of Tag
example:
[
Tag(
itemName='//path/to/node/item2',
value=3556464,
itemPath='',
type='unsignedInt',
error=''
)
]
Notes:
look at __ITEMS to understand named tuple setup
to access the fields of each entry in result
for tag in read_result:
print((f'device:{tag.device}, '
f'value:{tag.value}, '
f'data_type:{tag.type}, '
f'status:{tag.error}'))
"""
read_result = plc.read(itemList=__ITEMS)
"""
Subscribe a set of item values
Args:
itemList:list=[],
namespace:str="",
returnValuesOnReply:bool=False,
subscriptionPingRate:int=0,
returnErrorText:bool=True,
returnDiagnosticInfo:bool=False,
returnItemTime:bool=False,
returnItemPath:bool=False,
returnItemName:bool=False,
requestDeadline:str="",
clientRequestHandle:str="",
localeID:str="",
itemPath:str="",
deadband:float=0.0,
requestedSamplingRate:int=0,
enableBuffering:bool=False,
Returns:
Subscription: tuple that contains subscription information
example:
Subscription(
ServerSubHandle="1657752351",
error='',
items=[
{
'ClientItemHandle': '//path/to/node',
'type': 'unsignedInt',
'value': 6229
}
]
)
"""
response = plc.subscribe(itemList=__ITEMS)
"""
Cancel specified subscription
Args:
serverSubHandle:str,
clientRequestHandle:str="",
namespace:str=""
Returns:
Fault: tuple that contains fault information
example: fault exists on invalid handle
Fault(
faultcode="ns1:E_NOSUBSCRIPTION",
faultstring="E_NOSUBSCRIPTION",
detail="Subscription not found"
)
"""
response = plc.subscriptionCancel(serverSubHandle="3621660")
"""
Poll subscribed items
Args:
subscriptions:list[Subscription],
holdTime:str="",
waitTime:int=0,
returnAllItems:bool=False,
returnErrorText:bool=True,
returnDiagnosticInfo:bool=False,
returnItemTime:bool=False,
returnItemPath:bool=False,
returnItemName:bool=False,
requestDeadline:str="",
clientRequestHandle:str="",
localeID:str="",
namespace:str=""
Returns:
Fault: tuple that contains fault information
example: fault exists on invalid handle
[
Subscription(
serverSubHandle='1657752363',
items=[
{
'itemName': '//path/to/node1',
'type': 'unsignedInt',
'value': 3624471
},
{
'itemName': '//path/to/node2',
'type': 'unsignedInt',
'value': 6241
}
]
),
error='')
]
"""
response = plc.subscriptionPolledRefresh(subscriptions=__SUBSCRIPTIONS)
"""
Write mixed item values
example: write randomly mixed data types
Args:
itemList(list[Tag])[Required]: list of data class Tag
example: itemList=__ITEMS
namespace(str): override namespace
example: namespace="ns1"
Returns:
list[Tag]: list of incorrectly defined Tag or access
example:
[
Tag(
itemName='//path/to/item',
value='',
itemPath='',
type='',
error='E_READONLY'
)
]
Notes:
look at __ITEMS to understand named tuple setup
"""
response = plc.write(itemList=__ITEMS)