Getting started¶
Installation¶
Requires: Python>=3.7.6
$ pip install -U PyPlcnextRsc
Warning
This project is under development stage, please always get the latest version at
Create connection¶
In order to use RSC service , you need to get an already connected Device instance. see
Device
To do this , there are two code-styles :
from PyPlcnextRsc import Device, GUISupplierExample
# 1st. Use with-block , which will automatically connect and dispose the connection
with Device('192.168.1.10', secureInfoSupplier=GUISupplierExample) as device:
... # do rsc operation
# ---------------------------------------------------------------------------------
# 2nd. Use traditional way , which you should connect and dispose the connection explicitly
device = Device('192.168.1.10', secureInfoSupplier=GUISupplierExample)
device.connect()
... # do rsc operation
device.dispose()
If PLCnect’s User Authentication is disabled , just ignore secureInfoSupplier , like this:
with Device('192.168.1.10') as device:
... # do rsc operation
Note
If running the package locally , use IP 127.0.0.1
This package use TLS socket in default , if you want the fastest speed in development stage, you can use it without TLS :
from PyPlcnextRsc import Device, ExtraConfigure, GUISupplierExample
conf = ExtraConfigure()
conf.useTls = False
with Device('192.168.1.10', secureInfoSupplier=GUISupplierExample, port=41111, config=conf) as device:
...
Tip
Users should also create a new TCPGateWay to make it valid :
Edit file /etc/plcnext/device/System/RscGateway/RscGateway.settings , add a new label in <Gateways>..</Gateways>.
For example:
<TcpGatewaySettings gatewayId=”2” tcpPort=”41111” sessionTimeout=”300000” encrypted=”false” />
After you getting the connected instance of Device , then just use it to get raw services:
from PyPlcnextRsc.Arp.Plc.Gds.Services import IDataAccessService, ISubscriptionService, IForceService
from PyPlcnextRsc.Arp.Device.Interface.Services import IDeviceInfoService
service1 = IDataAccessService(device)
service2 = ISubscriptionService(device)
service3 = IForceService(device)
service4 = IDeviceInfoService(device)
Secure Info Supplier¶
The secureInfoSupplier parameter in Device is a callback function that
developer should supply for their application, is used to get the username and password for login.
But if the user-authentication is disabled, this function will not be called, so you can leave it with None (default).
There are two demos of supplier already in the package, check the source code of them:
As you can see the source code of these example is quite simple, when the server (device) need to authentication, the function will be invoked, developer should implement their own event to get the secure info from end-user.
This function should be defined with one or zero parameter in its’ signature. if one parameter is defined, then the information about the target device will be collected and passed to you as dict
“General.SerialNumber” - The “SerialNumber” parameter indicates the serial number of the device.
“General.ArticleName” - The “ArticleName” parameter indicates the device name.
“General.Firmware.Version” - The “FirmwareVersion” parameter indicates the firmware version of the device.
“Notification” - A message for display to users before accessing the device.
“Address” - The ip address and port that has be configured before.
Developer could do further checking by using this, or just display to end-users. No parameter in the signature is also ok , in this case those information will not be gathered for this function.
The return of this function must be a tuple with username and password in str type
Note
Only for develop(debug) period :
Write a lambda like this is ok in develop stage, but not for release !
secureInfoSupplier = lambda:("admin","my_pass")
Warning
Always note that it is not a good habit to write the username and password directly in code, that is very dangerous.
Function demos¶
Note
The underlying type of each RscVariant returned by RSC service calls should never be assumed, but should always be checked before performing any other operation on that object.
PLC Basic¶
Read Status and Infos¶
The device interface services provide a range of functions for accessing properties of the operating system and the controller hardware. You can call the information with the following interfaces.
Link to relative service :
Tip
Get all identifiers for these services to query, visit : # TODO
from PyPlcnextRsc.Arp.Device.Interface.Services import IDeviceInfoService,IDeviceStatusService
...
device_info_service = IDeviceInfoService(device)
device_status_service = IDeviceStatusService(device)
info_items = [
"General.ArticleName",
"General.ArticleNumber",
"General.SerialNumber",
"General.Firmware.Version",
"General.Hardware.Version",
"Interfaces.Ethernet.1.0.Mac"
]
status_items = [
"Status.Cpu.0.Load.Percent",
"Status.Memory.Usage.Percent",
"Status.ProgramMemoryIEC.Usage.Percent",
"Status.DataMemoryIEC.Usage.Percent",
"Status.Board.Temperature.Centigrade",
"Status.Board.Temperature.Centigrade",
"Status.Board.Humidity"
]
for identifier, result in zip(info_items, device_info_service.GetItems(info_items)):
print(identifier.rjust(40) + " : " + str(result.GetValue()))
for identifier, result in zip(status_items, device_status_service.GetItems(status_items)):
print(identifier.rjust(40) + " : " + str(result.GetValue()))
>>>
General.ArticleName : AXC F 2152
General.ArticleNumber : 2404267
General.SerialNumber : 1357546529
General.Firmware.Version : 2021.0.5 LTS (21.0.5.35585)
General.Hardware.Version : 02
Interfaces.Ethernet.1.0.Mac : 00:A0:45:A2:C0:5A
Status.Cpu.0.Load.Percent : 5
Status.Memory.Usage.Percent : 38
Status.ProgramMemoryIEC.Usage.Percent : 1
Status.DataMemoryIEC.Usage.Percent : 1
Status.Board.Temperature.Centigrade : 52
Status.Board.Temperature.Centigrade : 52
Status.Board.Humidity : 14
Querying notifications¶
Notifications that have been saved by means of the Notification Logger can be queried via RSC interfaces, and one that is already provided with the PLCnext Runtime is this INotificationLoggerService. You can use filter criteria for the query in order to specify the query. You can query all archives or the notifications of one specific archive.
For Detail Instructions, please refer to
A basic demo example of reading notifications:
from PyPlcnextRsc import Device, GUISupplierExample
from PyPlcnextRsc.Arp.Services.NotificationLogger.Services import INotificationLoggerService, NotificationFilter, SortOrder
if __name__ == "__main__":
with Device('192.168.1.10', secureInfoSupplier=GUISupplierExample) as device:
notification_logger_service = INotificationLoggerService(device)
# leave all filter field ignore
_filter = NotificationFilter(
StoredIdLowerLimit=0,
StoredIdUpperLimit=0,
NotificationNameRegExp="",
SenderNameRegExp="",
TimestampBefore="",
TimestampAfter="",
SeverityLowerLimit="",
SeverityUpperLimit=""
)
sn = notification_logger_service.QueryStoredNotifications(
archives=["default"],
Filter=_filter,
limit=10,
sortOrder=SortOrder.TimestampDesc,
language="en"
)
for n in sn:
print(f"ID:{n.Id}\t\t Severity={n.Severity}\t\t Sender={n.SenderName}\t\t Payload={n.Payload}")
>>>
ID:466 Severity=Info Sender=System Manager Payload=('SystemManager state changed: Running, error=false, warning=false',)
ID:465 Severity=Info Sender=PLC Manager Payload=('Plc state changed: Stop (warm) ==> Running',)
ID:464 Severity=Info Sender=Device Interface Payload=('Link state changed: interface 1, port 1, status: Up',)
ID:463 Severity=Info Sender=System Manager Payload=('SystemManager state changed: Stop, error=false, warning=false',)
ID:462 Severity=Info Sender=PLC Manager Payload=('Plc state changed: Ready ==> Stop (warm)',)
ID:461 Severity=Internal Sender=PROFINET Controller Payload=('Led state changed: Arp.Io.PnC (Controller), BF=Off, SF=Off',)
ID:460 Severity=Info Sender=PLC Manager Payload=('Plc state changed: ==> Ready',)
ID:459 Severity=Internal Sender=PROFINET Controller Payload=('Led state changed: Arp.Io.PnC (Controller), BF=On, SF=Off',)
ID:458 Severity=Internal Sender=PROFINET Device Payload=('Led state changed: Arp.Io.PnD (Device), BF=On, SF=Off',)
ID:457 Severity=Info Sender=System Manager Payload=('SystemManager state changed: Ready, error=false, warning=false',)
Control device¶
#TODO doc
Setting device¶
#TODO doc
Variables¶
Primitive¶
#TODO doc
Complex¶
Here is a helper tool for user to operate Complex data type, it is
DataTypeStore
The DataTypeStore is mainly used to create the instance of Complex variable, make value in python can be equivalent
to IEC61131.
Have a see to the basic typical example:
from PyPlcnextRsc import Device, RscVariant, GUISupplierExample
from PyPlcnextRsc.Arp.Plc.Gds.Services import IDataAccessService, WriteItem
from PyPlcnextRsc.tools import DataTypeStore
if __name__ == "__main__":
TypeStore = DataTypeStore.fromString(
"""
TYPE
DemoStruct : STRUCT
Field1 : INT;
Field2 : BOOL;
END_STRUCT
DemoArray : ARRAY[0..10] OF INT;
END_TYPE
""")
with Device('192.168.1.10', secureInfoSupplier=GUISupplierExample) as device:
# create DemoStruct
demo1 = TypeStore.NewSchemaInstance("DemoStruct")
demo1.Field1 = 123
demo1.Field2 = True
# create DemoArray
demo2 = TypeStore.NewSchemaInstance("DemoArray")
demo2[:] = [i * 2 for i in range(11)]
# get raw data access service
data_access_service = IDataAccessService(device)
# Write demo1 to PLCnext
data_access_service.Write((WriteItem("Arp.Plc.Eclr/demo1", RscVariant.of(demo1)),))
# Read demo1
read_item = data_access_service.Read(("Arp.Plc.Eclr/demo1",))[0]
rcv_demo1 = TypeStore.ReceiveAsSchemaInstance("DemoStruct", read_item.Value)
print(rcv_demo1)
# ---------------
data_access_service.Write((WriteItem("Arp.Plc.Eclr/demo2", RscVariant.of(demo2)),))
read_item = data_access_service.Read(("Arp.Plc.Eclr/demo2",))[0]
rcv_demo2 = TypeStore.ReceiveAsSchemaInstance("DemoArray", read_item.Value)
print(rcv_demo2)
Let’s now start with DataTypeStore.
There are two typical way to create it. The example showed above is directly using str value, another way to create us using file:
MyTypeStore = DataTypeStore.fromFile('MyType.txt')
The code for creating the DataTypeStore is the same to
PLCnext Engineer, so for most case just copy the DataType code from PLCnext Engineer is ok.
Note
Known limits:
Enum and User-defined String is not implement yet! please use its’ origin type instead.
If a type is referenced another type, the type which is referenced must be declared before, this is a little different from PLCnext Engineer
User default value is not supported , but will come soon.
DataTypeStore also support nested variable ,
so you can use any DateType in theory, such as:
TYPE
UnusualArray : ARRAY[-10..10] OF INT;
MyArray :ARRAY[0..10] OF INT;
MyArray2 : array [0..2] OF MyArray;
MyArray3 : ARRAY[0..1] OF MyArray2;
MyArray3plus: ARRAY[0..1] OF array [0..2] OF ARRAY[0..10] OF INT;
MyStruct0 : STRUCT
Field1 : INT;
Field2 : BOOL;
END_STRUCT
ArrayOfStruct : ARRAY[0..2] OF MyStruct0;
StructWithArray : STRUCT
Field1 : ARRAY[0..10] OF INT;
Field2 : BOOL;
END_STRUCT;
OtherStruct : STRUCT
Field1 : INT;
Field2 : BOOL;
F3:MyStruct0;
F4:MyArray;
END_STRUCT
// Support Comment
/*
Foobar : STRUCT
Field1 : ARRAY[0..10] OF INT;
Field2 : BOOL;
END_STRUCT;
*/
# Also you can use pound sign to comment , this is very convenient for Python IDE to Auto comment lines.
END_TYPE
Let’s study with this struct:
The following code is the typical way for writing it to PLCnext
TypeStore = DataTypeStore.fromString(
"""
TYPE
St_Basic : STRUCT
Field1 : INT;
Field2 : BOOL;
Field3 : DWORD;
END_STRUCT
St_Basic2 : STRUCT
Field1 : St_Basic;
Field2 : St_Basic;
END_STRUCT
Arr3_Str : ARRAY[0..2] OF STRING;
Arr3_StBasic: ARRAY[0..2] OF St_Basic;
Arr3_Time: ARRAY[0..3] OF TIME;
MyComplexStruct : STRUCT
xBool : BOOL;
xINT : INT;
xULINT : ULINT;
xDWORD : DWORD;
xStr : STRING;
xTIME : TIME;
stBasic: St_Basic;
arrINT : ARRAY[0..2] OF INT; //Anonymous array
arrStr : Arr3_Str;
stStBasic2 : St_Basic2;
END_STRUCT
END_TYPE
"""
)
with Device('192.168.1.10', secureInfoSupplier=GUISupplierExample) as device:
MyComplexStruct = TypeStore.NewSchemaInstance("MyComplexStruct")
MyComplexStruct.xBool = True
MyComplexStruct.xINT = 2152
MyComplexStruct.xULINT = 18446744073709551615
MyComplexStruct.xDWORD = 0xAABBCCDD # this is just int actually
MyComplexStruct.xStr = 'Hello World!!!'
MyComplexStruct.xTIME = 1 * 86400000 + 2 * 3600000 + 3 * 60000 + 4 * 1000 + 567 # T#1d2h3m4.567s
MyComplexStruct.stBasic.Field1 = 123
MyComplexStruct.stBasic.Field2 = True
MyComplexStruct.stBasic.Field3 = 0x00112233
MyComplexStruct.arrINT[0] = 100
MyComplexStruct.arrINT[1] = 200
MyComplexStruct.arrINT[2] = 300
MyComplexStruct.arrStr[0] = "Hello"
MyComplexStruct.arrStr[2] = "World"
MyComplexStruct.stStBasic2.Field1.Field1 = 1000
MyComplexStruct.stStBasic2.Field1.Field2 = False
MyComplexStruct.stStBasic2.Field1.Field3 = 0xABCD0123
MyComplexStruct.stStBasic2.Field2.Field1 = 1000
MyComplexStruct.stStBasic2.Field2.Field2 = True
MyComplexStruct.stStBasic2.Field2.Field3 = 0x0123ABCD
# Note : write struct or array must use **IDataAccessService.Write**
# the **IDataAccessService.WriteSingle** not support for complex data
err = IDataAccessService(device).Write((WriteItem("Arp.Plc.Eclr/myComplexStruct", RscVariant.of(MyComplexStruct)),))[0]
if err == DataAccessError.NONE:
print("Success")
For senior programmer, the following code is more friendly and high efficiency to you :
MyComplexStruct = TypeStore.NewSchemaInstance("MyComplexStruct")
Make_St_Basic = lambda: TypeStore.NewSchemaInstance("St_Basic")
Make_St_Basic2 = lambda: TypeStore.NewSchemaInstance("St_Basic2")
MyComplexStruct.stBasic = Make_St_Basic()(1, True, Field3=0x00112233) # __call__ for a struct will fill all the fields
# Direct set array data by slice
# Note: MyComplexStruct.arrINT = [100, 200, 300] is a syntactic sugar in
# 'PyPlcnextRsc.common.objects.rsc_struct.RscStructBuilder' since V0.1.4 , see the source code to learn more
#
# Here the array auto created is a Subclass of python list
# so you can use it just like list, this is the reason why you can use slice
MyComplexStruct.arrINT = [100, 200, 300] # This is equal to MyComplexStruct.arrINT[:] = [100, 200, 300]
MyComplexStruct.arrStr = ["Hello", "", "World"]
# since V0.1.4 this way is supported
MyComplexStruct.stStBasic2 = { # this example use dict, and also can use tuple or list
"Field1": (1, # use tuple or list *MUST* have the same order and size
True,
0x121212
),
"Field2": {'Field1': 1, # using dict just fill the field you want, those fields you ignored will remain default value
'Field2': True,
'Field3': 0x343434 # if this line is removed , then 'Field3' = 0x0
}
}
# Note: it is not necessary to make your tuple(list or dict) start from root everytime,
# if it has a deep path, the following code is more friendly and it act all the same with the above one.
MyComplexStruct.stStBasic2.Field1 = (1, True, 0x121212)
MyComplexStruct.stStBasic2.Field2 = {'Field1': 1, 'Field2': True, 'Field3': 0x343434}
# the Original way (before V0.1.4)
# innerSt_Basic2 = Make_St_Basic2()
# innerSt_Basic2.Field1 = Make_St_Basic()(1, True, 0x121212)
# innerSt_Basic2.Field2 = Make_St_Basic()(2, False, 0x343434)
## Optional: more directly instead of above, use this:
## innerSt_Basic2(Make_St_Basic()(1, True, 0x121212), Make_St_Basic()(2, False, 0x343434))
# MyComplexStruct.stStBasic2 = innerSt_Basic2
# Note : write the struct or array must use **IDataAccessService.Write** ,
# **IDataAccessService.WriteSingle** not support for complex data
err = IDataAccessService(device).Write((WriteItem("Arp.Plc.Eclr/myComplexStruct", RscVariant.of(MyComplexStruct)),))[0]
if err == DataAccessError.NONE:
print("Success")
Note
In the created instance , all values have been set to its’ default value, means that if user don’t change the element (field) in it, it can also send to PLCnext successfully , but all elements (or fields) are 0 , 0.0 , False or “”(empty str)
As for read data:
with Device('192.168.1.10', secureInfoSupplier=GUISupplierExample) as device:
complexStruct_ForSend = TypeStore.NewSchemaInstance("MyComplexStruct")
Make_St_Basic = lambda: TypeStore.NewSchemaInstance("St_Basic")
Make_St_Basic2 = lambda: TypeStore.NewSchemaInstance("St_Basic2")
service = IDataAccessService(device)
# Prepare value to send
complexStruct_ForSend.stBasic = Make_St_Basic()(1, True, Field3=0x00112233)
complexStruct_ForSend.arrINT[:] = [100, 200, 300]
err = service.Write((WriteItem("Arp.Plc.Eclr/myComplexStruct", RscVariant.of(complexStruct_ForSend)),))[0]
if err == DataAccessError.NONE:
print("Write Success")
read_item = service.Read(("Arp.Plc.Eclr/myComplexStruct",))[0]
if read_item.Error == DataAccessError.NONE:
# Use **TypeStore.ReceiveAsSchemaInstance** to convert RscVariant object
# to python value which is equivalence to'MyComplexStruct'
complexStruct_Received = TypeStore.ReceiveAsSchemaInstance("MyComplexStruct", read_item.Value)
# you can directly compare these two objects :
assert complexStruct_Received == complexStruct_ForSend
complexStruct_ForSend.xINT = 6666
assert complexStruct_Received != complexStruct_ForSend
complexStruct_ForSend.xINT = 0
complexStruct_ForSend.arrINT[2] = 5
assert complexStruct_Received != complexStruct_ForSend
print(f"complexStruct_Received.stBasic = {complexStruct_Received.stBasic}") # direct print the struct
print(f"complexStruct_Received.stBasic.Field1={complexStruct_Received.stBasic.Field1}")
print(f"complexStruct_Received.stBasic.Field2={complexStruct_Received.stBasic.Field2}")
print(f"complexStruct_Received.stBasic.Field3={complexStruct_Received.stBasic.Field3}")
# more common way is to get the reference of inner object
inner_struct = complexStruct_Received.stBasic
print(f"inner_struct.Field2={inner_struct.Field2}")
print(f"complexStruct_Received.arrINT={complexStruct_Received.arrINT}")
# Remember, the array of there is subclass instance of list
# just treat it as list
for idx, element in enumerate(complexStruct_Received.arrINT):
print(f"arrINT<{idx}>{element}")
>>>
Write Success
complexStruct_Received.stBasic = stBasic(Field1=1, Field2=True, Field3=1122867)
complexStruct_Received.stBasic.Field1=1
complexStruct_Received.stBasic.Field2=True
complexStruct_Received.stBasic.Field3=1122867
inner_struct.Field2=True
complexStruct_Received.arrINT=RscList<Int16>[100, 200, 300]
arrINT<0>100
arrINT<1>200
arrINT<2>300
There are some advanced usages:
from PyPlcnextRsc import Device, GUISupplierExample, RscVariant
from PyPlcnextRsc.Arp.Plc.Gds.Services import IDataAccessService, WriteItem, DataAccessError
from PyPlcnextRsc.tools import DataTypeStore
TypeStore = DataTypeStore.fromString(
"""
// Anonymous 4-dimension
CrazyArray : ARRAY[0..2] OF ARRAY[0..2] OF ARRAY[0..2] OF ARRAY[0..2] OF STRING;
// Disassemble 4-dimension
Inner0 : ARRAY[0..2] OF STRING;
Inner1 : ARRAY[0..2] OF Inner0;
Inner2 : ARRAY[0..2] OF Inner1;
CrazyArray2 : ARRAY[0..2] OF Inner2;
// ArrayOfStruct
InnerStruct : STRUCT
Field1 : INT;
Field2 : BOOL;
END_STRUCT
ArrayOfStruct : ARRAY[0..2] OF InnerStruct;
// Chain
StCell : STRUCT
Field1 : String;
END_STRUCT
ArrCell:ARRAY[0..1] of StCell;
StCell2:STRUCT
Field1 : ArrCell;
END_STRUCT
Chain:ARRAY[0..1] of StCell2;
"""
)
with Device('192.168.1.10', secureInfoSupplier=GUISupplierExample) as device:
service = IDataAccessService(device)
MyCrazyArray = TypeStore.NewSchemaInstance("CrazyArray")
# Anonymous 4-dimension
for a in range(3):
for b in range(3):
for c in range(3):
for d in range(3):
MyCrazyArray[a][b][c][d] = f"I am {a} {b} {c} {d}"
err = service.Write((WriteItem("Arp.Plc.Eclr/MyCrazyArray", RscVariant.of(MyCrazyArray)),))[0]
assert err == DataAccessError.NONE
# Disassemble 4 - dimension
# Of course you can still use the above way, they are the same
# this example just to show you how to generate inner array, it will be useful in some case.
def Inner0(a, b, c):
ret = TypeStore.NewSchemaInstance("Inner0")
ret[:] = [f"I am {a} {b} {c} {i}" for i in range(3)]
return ret
def Inner1(a, b):
ret = TypeStore.NewSchemaInstance("Inner1")
ret[:] = [Inner0(a, b, c) for c in range(3)]
return ret
def Inner2(a):
ret = TypeStore.NewSchemaInstance("Inner2")
ret[:] = [Inner1(a, b) for b in range(3)]
return ret
MyCrazyArray2 = TypeStore.NewSchemaInstance("CrazyArray2")
MyCrazyArray2[:] = [Inner2(a) for a in range(3)]
err = service.Write((WriteItem("Arp.Plc.Eclr/MyCrazyArray2", RscVariant.of(MyCrazyArray2)),))[0]
assert err == DataAccessError.NONE
# ArrayOfStruct
MyArrayOfStruct = TypeStore.NewSchemaInstance("ArrayOfStruct")
MyArrayOfStruct[0].Field1 = 100 # typical way
MyArrayOfStruct[0].Field2 = True
# Since V0.1.4 : if element is struct type in array , use this method is quick !
MyArrayOfStruct[1] = {"Field1": 200, "Field2": False} # or (200,False) or [200,False]
MyArrayOfStruct[2] = (300, True)
# Since V0.1.4 : and this can override all above values (this example showed '[:]' you can
# use other slice which you want such as '[0:8]' '[1:-1]'... )
MyArrayOfStruct[:] = [{"Field1": 200, "Field2": False}, (200, True), (300, True)]
# # before V0.1.4
# MyArrayOfStruct[1] = TypeStore.NewSchemaInstance("InnerStruct")(Field1=200, Field2=False)
# # new struct instance using attribute to fill fields
# innerStruct = TypeStore.NewSchemaInstance("InnerStruct")
# innerStruct.Field1 = 300
# innerStruct.Field2 = True
# MyArrayOfStruct[2] = innerStruct
# MyArrayOfStruct[:] = [
# TypeStore.NewSchemaInstance("InnerStruct")(Field1=200, Field2=False),
# TypeStore.NewSchemaInstance("InnerStruct")(Field1=200, Field2=False),
# TypeStore.NewSchemaInstance("InnerStruct")(Field1=200, Field2=False)
# ]
err = service.Write((WriteItem("Arp.Plc.Eclr/MyArrayOfStruct", RscVariant.of(MyArrayOfStruct)),))[0]
assert err == DataAccessError.NONE
# Chain
MyChain = TypeStore.NewSchemaInstance("Chain")
MyChain[0].Field1[0].Field1 = "Hello !!"
err = service.Write((WriteItem("Arp.Plc.Eclr/MyChain", RscVariant.of(MyChain)),))[0]
assert err == DataAccessError.NONE
DataLogger¶
For Detail Instructions, please refer to
The DataLogger is a service component that transfers real-time data from the GDS to a database for recording and storage purposes. When starting and stopping the PLCnext Technology firmware, a configured DataLogger session is started and stopped automatically. The DataLogger then collects the task-synchronous values of the configured GDS ports with a given sampling rate and stores them with a time stamp into a database. To learn more about the DataLogger in general, read the DataLogger topic, Click.
A DataLogger instance can be configured with a configuration file, or by means of PLCnext Engineer (firmware 2020.6 or higher), or with the RSC IDataLoggerService2 as described in this topic.
Note
DataLogger sessions initiated via this RSC service will not perform the Download Changes command in PLCnext Engineer. This is true even if sessions are not currently running but are created already. The blocking is indicated by a notification (Arp.Services.DataLogger.Error, payload string: “Dynamic session detected! Download rejected!”). Unlike sessions that are started via this RSC service, sessions that are created via an XML configuration or in PLCnext Engineer are continued even after a Download All operation.
import datetime
import time
from PyPlcnextRsc import Device, GUISupplierExample, RscVariant, RscType
from PyPlcnextRsc.Arp.Services.DataLogger.Services import IDataLoggerService2, ErrorCode, SessionProperty, SessionPropertyName, SinkType, TriggerRpnItem, RpnItemType
if __name__ == "__main__":
with Device('192.168.1.10', secureInfoSupplier=GUISupplierExample) as device:
data_logger_service = IDataLoggerService2(device) # Get DataLoggerService2
sessionName = "MyDb"
# Create a datalogger session
assert data_logger_service.CreateSession("MyDb", persistent=False) == ErrorCode.NONE
# Configure the session
propertys = [
SessionProperty(SessionPropertyName.SamplingInterval, RscVariant.of("100ms")),
SessionProperty(SessionPropertyName.PublishInterval, RscVariant.of("200ms")),
SessionProperty(SessionPropertyName.BufferCapacity, RscVariant(10, RscType.Uint16)),
SessionProperty(SessionPropertyName.SinkType, RscVariant.ofEnum(SinkType.Database)),
SessionProperty(SessionPropertyName.SinkProperties,
RscVariant.of("writeInterval=1000;dst=/opt/plcnext/test.db;rollover=true;maxFiles=25")
),
]
assert data_logger_service.ConfigureSession(sessionName, propertys) == ErrorCode.NONE
# Add Variables
errors = data_logger_service.SetVariables(sessionName, ["Arp.Plc.Eclr/A", "Arp.Plc.Eclr/A1.v"])
for err in errors:
assert err == ErrorCode.NONE
c = [
TriggerRpnItem(RpnItemType.Variable, RscVariant.of("Arp.Plc.Eclr/A1.v")),
TriggerRpnItem(RpnItemType.Constant, RscVariant.of(True)),
TriggerRpnItem(RpnItemType.Operation, RscVariant(1, RscType.Uint8)),
]
# Set Trigger
assert data_logger_service.SetTriggerCondition(sessionName, taskName="MyTask", preCycleCount=3, postCycleCount=20, triggerCondition=c) == ErrorCode.NONE
# Start
assert data_logger_service.StartSession(sessionName) == ErrorCode.NONE
time.sleep(10)
# Read
start = datetime.datetime(2021, 8, 27, 14, 48, 30, tzinfo=datetime.timezone.utc)
values, err = data_logger_service.ReadVariablesData(sessionName, start, datetime.datetime(2021, 8, 27, 14, 49, 00, tzinfo=datetime.timezone.utc), ["Arp.Plc.Eclr/A1.v"])
for v in values:
print(v)
Files¶
Read file¶
Link to relative service :
The following snippet showed an example for reading the /opt/plcnext/logs/Output.log file and save to local,
also the Traits of Length and Permissions
be read at the same time.
from PyPlcnextRsc.Arp.System.Commons.Services.Io import IFileService, Traits, FileSystemError
...
file_service = IFileService(device)
stream, traits, error = file_service.Read(Traits.Length | Traits.Permissions, "/opt/plcnext/logs/Output.log")
if error == FileSystemError.NONE:
for trait in traits:
print(f"Trait({trait.Trait.name}) = {trait.Value.GetValue()}")
stream.saveToFile("log.txt")
>>>
Trait(Permissions) = 509
Trait(Length) = 1974442
Write file¶
Link to relative service :
The following snippet send two files to PLCnext /tmp folder,
the first file is the current python file (use variable __file__ to represent),
another file is not a real file on filesystem , just python bytes, and you can also use bytearray or str to wrap as a
stream directly. this is convenient in some case. At mean while , these two files are written with permission
owner_all | group_all | others_all (see Permissions)
import os
from PyPlcnextRsc import RscStream, RscVariant, RscType
from PyPlcnextRsc.Arp.System.Commons.Services.Io import IFileService, TraitItem, Traits
...
file_service = IFileService(device)
# just send this py file for example
file_path = __file__
file_name = os.path.split(file_path)[-1]
# prepare RscStream object
MyFileStream = RscStream.ofFile(file_path)
MyDataStream = RscStream.ofData(b'echo Hello World !')
# 511 = owner_all | group_all | others_all
permissionTrait = TraitItem(Traits.Permissions, RscVariant(511, RscType.Int32))
file_service.Write(f'/tmp/{file_name}', overwrite=True, traitItems=(permissionTrait,), data=MyFileStream)
file_service.Write(f'/tmp/say_hello.sh', overwrite=True, traitItems=(permissionTrait,), data=MyDataStream)
Read directory¶
Link to relative service :
This example shows a basic case of how to download a directory from PLCnext to local filesystem,
all files (searchPattern= ” * “) in /opt/plcnext/projects/PCWE will be download to receivedPCWE folder
Note
This is only a basic demo, users should make more exception handler in their code.
import os
import platform
from PyPlcnextRsc.Arp.System.Commons.Services.Io import IDirectoryService, IFileService, Traits
...
# check if we are running on Windows
isWin = platform.system().lower() == 'windows'
# IDirectoryService for enumerating filesystem
directory_service = IDirectoryService(device)
# IFileService for downloading file
file_service = IFileService(device)
save_folder = "receivedPCWE"
remote_folder = "/opt/plcnext/projects/PCWE"
# Read all entrys of the target path on device
entrys = directory_service.EnumerateFileSystemEntries(remote_folder, searchPattern="*", recursive=True)
for entry in entrys:
relative_path = entry.Path.removeprefix(remote_folder + '/')
if isWin:
relative_path.replace('/', '\\')
save_path = os.path.join(save_folder, relative_path)
if entry.IsFile:
os.makedirs(os.path.split(save_path)[0], exist_ok=True)
file_service.Read(Traits.NONE, entry.Path)[0].saveToFile(save_path)
else:
os.makedirs(save_path, exist_ok=True)
Secure¶
Change password¶
Link to relative service :
Warning
The greater the ability, the greater the responsibility
from PyPlcnextRsc.Arp.System.Security.Services import IPasswordConfigurationService2
password_configuration_service = IPasswordConfigurationService2(device)
password_configuration_service.changePassword("testCount", oldPassword="123456", newPassword="654321")
# This operation is an administration function ,if you are admin , you can set password directly:
password_configuration_service.setPassword("testCount", newPassword="654321")