../_images/bar.jfif

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

https://gitlab.phoenixcontact.com/SongYantao/pyplcnextrsc

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:

../_images/gui_supplier.png ../_images/console_supplier.png

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:

../_images/complexStruct.png

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")