Menggunakan gRPC dalam Python untuk Membaca dan Menulis Data Proses pada AXC F 3152

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 berkas yang diperlukan, di luar PLC, misalnya, pada mesin Windows.

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

Menghasilkan _pb2.py dan _pb2_grpc.py dari Berkas .proto

Selanjutnya, kita harus menghasilkan berkas Python yang diperlukan dari berkas .proto yang disediakan. Berkas .proto berada di folder berikut: gRPC-master/protobuf.

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

1. menghasilkan berkas 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

Membuat Proyek Demo PLCnext

Proyek yang ditunjukkan hanya untuk menunjukkan bagaimana antarmuka gRPC berinteraksi dengan GDS. Silakan gunakan proyek yang ada sebagai gantinya. Untuk proyek individual, Anda harus mengedit nama port dalam skrip Python berikut sesuai kebutuhan, misalnya, Arp.Plc.Eclr/MainInstance.strInput.

Menyiapkan PLC

Instal pip untuk mengelola paket Python Anda:

1. Hubungkan kontroler AXC F 3152 ke Internet.
2. Ketik perintah curl https://bootstrap.pypa.io/get-pip.py -o get-pip.py.
3. Kemudian ketik 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 berkas Python yang telah dibuat 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 live view.

Menjalankan Contoh

Sekarang mulai contoh. Login ke PLC via ssh dan navigasikan 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, tipe data diperlukan, misalnya, wi1.Value.TypeCode = 19. Tipe-tipe dijelaskan dalam berkas yang dihasilkan gRPC-master/pxc_grpc/ArpTypes_pb2.py dimulai di 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, misalnya, r._ReturnValue.Value.StringValue, dapat ditemukan dalam berkas yang sama, dimulai di baris 365, misalnya, BoolValue, Int8Value, StringValue.