gRPC Python – Membaca dan Menulis Data Proses

Artikel ini menjelaskan cara mengakses dan menulis data proses sederhana dengan Python dengan AXC F 3152 menggunakan gRPC. (https://www.plcnext.help/te/Service_Components/gRPC_Introduction.htm)

Prasyarat

Pertama kita harus menyiapkan file yang dibutuhkan, di luar PLC, misalnya pada mesin Windows.

1. Instal Python 3.9 (3.10 dapat menyebabkan kesalahan)
2. Instal paket Python yang diperlukan untuk menghasilkan kode dari file .proto:pip install grpcio-tools==1.36.1
3. Unduh dan buka zip repositori yang berisi file .proto dari https://github.com/PLCnext/gRPC

Buat _pb2.py dan _pb2_grpc.py dari File .proto

Selanjutnya, kita harus membuat file python yang diperlukan dari file .proto yang disediakan. Yang terakhir terletak di folder berikut:gRPC-master/protobuf.

Gunakan kode ini untuk membuat skrip Python di Folder gRPC-master, mis., generate_grpc.py. Skrip

1. membuat file yang diperlukan dan menempatkannya di gRPC-master/pxc_grpc
2. menyesuaikan jalur impor

import glob
import os
from pathlib import Path

# create the output directory
Path('pxc_grpc').mkdir(parents=True, exist_ok=True)

grpc_command_base = 'python -m grpc_tools.protoc -I./protobuf --python_out=pxc_grpc --grpc_python_out=pxc_grpc '

import_paths = set()

# generate the *_pb2.py and *_pb2_grpc.py files
for filename in glob.iglob('./protobuf/**', recursive=True):

    if filename.endswith('.proto'):
        
        # store the import path
        path_parts = filename.split(os.sep)
        import_paths.add('.'.join(path_parts[1:-1]))

        grpc_command = ''.join([grpc_command_base, os.path.join('.', os.path.relpath(filename))])
        stream = os.popen(grpc_command)
        output = stream.read()
        if output != '':
            print(''.join(['error/info for file ', os.path.relpath(filename), ' - ', output]))


# get the python files in the base directory
base_pys = set()

for (dirpath, dirnames, filenames) in os.walk('./pxc_grpc'):
    for f in filenames:
        base_pys.add(f.split('.py')[0])
    break

# reformat the stored paths to adapt the import statements
try:
    import_paths.remove('')
except:
    pass

import_paths = list(import_paths)
import_paths.sort(key=len)
import_paths.reverse()

# adapt the imports
for filename in glob.iglob('./pxc_grpc/**', recursive=True):

    if filename.endswith('.py'):

        new_lines = []

        with open(filename, 'r') as file:
            lines = file.readlines()
            for line in lines:
                if line.startswith('from'):
                    for import_path in import_paths:
                        if import_path in line:
                            line = line.replace(import_path, ''.join(['pxc_grpc.', import_path]), 1)
                            break
                elif line.startswith('import'):
                    parts = line.split()
                    if parts[1] in base_pys:
                        line = line.replace('import', 'from pxc_grpc import')
                
                new_lines.append(line)

        with open(filename, 'w') as file:
            file.write(''.join(new_lines))

Buka shell dan jalankan skrip:pyton generate_grpc.py

Buat Proyek Demo PLCnext

Proyek yang ditampilkan seharusnya hanya menunjukkan bagaimana antarmuka gRPC berinteraksi dengan GDS. Jangan ragu untuk menggunakan proyek yang ada sebagai gantinya. Untuk proyek individual, Anda harus mengedit nama port dalam skrip Python berikut, mis., Arp.Plc.Eclr/MainInstance.strInput .

Siapkan PLC

Instal pip untuk mengelola paket Python Anda:

1. Hubungkan pengontrol AXC F 3152 ke Internet.
2. Masukkan perintah curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py.
3. Kemudian masukkan perintah python3 get-pip.py.

Instal paket yang diperlukan:pip install grpcio protobuf==3.20.0

Buat folder 'grpc_test' di direktori proyek (/opt/plcnext/projects/).

Salin folder 'pxc_grpc', yang berisi file Python yang disimpan ke 'grpc_test', misalnya dengan WinSCP.

Buat skrip Python di folder 'grpc_test' bernama 'grpc_test.py' dan masukkan kode berikut:

import grpc
from pxc_grpc.Plc.Gds.IDataAccessService_pb2 import IDataAccessServiceReadSingleRequest, \
    IDataAccessServiceReadRequest, IDataAccessServiceWriteSingleRequest, IDataAccessServiceWriteRequest
from pxc_grpc.Plc.Gds.IDataAccessService_pb2_grpc import IDataAccessServiceStub
from pxc_grpc.Plc.Gds.WriteItem_pb2 import WriteItem


def write_single_string(stub, port_name, value):
    
    single_write_request = IDataAccessServiceWriteSingleRequest()
    single_write_request.data.PortName = port_name
    single_write_request.data.Value.TypeCode = 19
    single_write_request.data.Value.StringValue = value

    return stub.WriteSingle(single_write_request)


def write_single_int(stub, port_name, value):
    
    single_write_request = IDataAccessServiceWriteSingleRequest()
    single_write_request.data.PortName = port_name
    single_write_request.data.Value.TypeCode = 6
    single_write_request.data.Value.Int16Value = value

    return stub.WriteSingle(single_write_request)


def write_multiple_values(stub):

    write_request = IDataAccessServiceWriteRequest()

    wi1 = WriteItem()
    wi1.PortName = 'Arp.Plc.Eclr/MainInstance.strInput'
    wi1.Value.StringValue = "test1"
    wi1.Value.TypeCode = 19
    
    wi2 = WriteItem()
    wi2.PortName = 'Arp.Plc.Eclr/MainInstance.strInput2'
    wi2.Value.StringValue = "test2"
    wi2.Value.TypeCode = 19

    # add multiple WriteItems at once
    write_request.data.extend([wi1, wi2])

    # add WriteItems separately
    # response1.data.append(wi1)
    # response1.data.append(wi2)

    return stub.Write(write_request)


def read_single_value(stub, port_name):

    single_read_request = IDataAccessServiceReadSingleRequest()
    single_read_request.portName=port_name

    return stub.ReadSingle(single_read_request)


def read_multiple_values(stub, port_names):

    read_request = IDataAccessServiceReadRequest()
    read_request.portNames.extend(port_names)

    return stub.Read(read_request)


if __name__ == "__main__":
   
    # create channel and stub
    channel = grpc.insecure_channel('unix:/run/plcnext/grpc.sock')
    stub = IDataAccessServiceStub(channel)

    print(write_single_string(stub, 'Arp.Plc.Eclr/MainInstance.strInput', 'test123'))
    print(write_single_int(stub, 'Arp.Plc.Eclr/MainInstance.iInput', 18))

    print(write_multiple_values(stub))

    r = read_single_value(stub, 'Arp.Plc.Eclr/MainInstance.strInput')
    print(r)
    print(r._ReturnValue.Value.TypeCode)
    print(r._ReturnValue.Value.StringValue)

    r = read_multiple_values(stub, ['Arp.Plc.Eclr/MainInstance.iInput', 'Arp.Plc.Eclr/MainInstance.strInput'])
    for value in r._ReturnValue:
        print(value, value.Value.TypeCode)

Hubungkan PLC Anda ke PLCnext Engineer, unduh proyek dan mulai tampilan langsung.

Jalankan Contoh

Sekarang mulai contohnya. Login di PLC melalui ssh dan arahkan ke 'grpc_test', lalu jalankan skrip Python:

1. cd projects/grpc_test/
2. python3 grpc_test.py

gRPC memungkinkan interaksi dengan variabel GDS.

Tipe Data

Untuk membaca dan menulis variabel, diperlukan tipe data, misalnya wi1.Value.TypeCode = 19 . Jenisnya dijelaskan dalam file yang dibuat gRPC-master/pxc_grpc/ArpTypes_pb2.py mulai dari baris 242:

CT_None = 0
CT_End = 0
CT_Void = 1
CT_Boolean = 2
CT_Char = 3
CT_Int8 = 4
CT_Uint8 = 5
CT_Int16 = 6
CT_Uint16 = 7
CT_Int32 = 8
CT_Uint32 = 9
CT_Int64 = 10
CT_Uint64 = 11
CT_Real32 = 12
CT_Real64 = 13
CT_Struct = 18
CT_String = 19
CT_Utf8String = 19
CT_Array = 20
CT_DateTime = 23
CT_Version = 24
CT_Guid = 25
CT_AnsiString = 26
CT_Object = 28
CT_Utf16String = 30
CT_Stream = 34
CT_Enumerator = 35
CT_SecureString = 36
CT_Enum = 37
CT_Dictionary = 38
CT_SecurityToken = 39
CT_Exception = 40
CT_IecTime = 41
CT_IecTime64 = 42
CT_IecDate = 43
CT_IecDate64 = 44
CT_IecDateTime = 45
CT_IecDateTime64 = 46
CT_IecTimeOfDay = 47
CT_IecTimeOfDay64 = 48

Variabel nilai yang sesuai, mis., r._ReturnValue.Value.StringValue , dapat ditemukan di file yang sama, dimulai pada baris 365, mis., BoolValue , Int8Value , StringValue .